ABQflow.helpers package

Submodules

ABQflow.helpers.constant module

ABQflow.helpers.convert module

Data conversion utilities — array generation, result flattening, outcome serialisation.

These are the most commonly used helper functions extracted from the main orchestrator so they can be imported lightweight without pulling in the entire batch-processing machinery.

ABQflow.helpers.convert.is_sidecar(value)[source]

Return True if value is a sidecar envelope (dict with __file__).

Parameters

valueany

Value to test.

Returns

bool

Return type:

bool

ABQflow.helpers.convert.resolve_sidecar(value, output_dir, load=False)[source]

Resolve a sidecar envelope to an absolute path and optional data.

Parameters

valuedict

Sidecar envelope: {'__file__': path, 'format': 'csv', ...}.

output_dirstr

Directory that __file__ is relative to.

loadbool

If True, load the file and return a numpy.ndarray. Default False (lazy — returns the absolute path).

Returns

tuple[str, dict] or tuple[numpy.ndarray, dict]

(absolute_path, metadata) when load=False; (ndarray, metadata) when load=True.

Raises

ValueError

If the envelope is missing __file__ or the file doesn’t exist.

Parameters:
ABQflow.helpers.convert.load_field(outcome, result_name, numeric_only=True)[source]

Load a single named field from a JobOutcome.

Normalises inline values and sidecar CSV envelopes into a uniform numpy.ndarray. This is the single-job entry point for consuming sidecar results; for batch consumption use iter_fields().

Parameters

outcomeJobOutcome

Outcome from BatchAbaqusProcessor.run_batch().

result_namestr

Key in outcome.results to load.

numeric_onlybool

If True (default), non-numeric CSV columns are dropped with a warning. If False, return an object array preserving string columns (for callers that need label columns).

Returns

numpy.ndarray or None

None when the result is missing, the extraction failed, the sidecar file is gone, or the path is unsafe.

ABQflow.helpers.convert.iter_fields(outcomes, result_name, on_missing='skip')[source]

Yield (job_name, ndarray) pairs for a named field across a batch.

Outcomes are sorted by _natural_key() on job_name so the iteration order is deterministic and matches the row order of degenerate_from_array() (row-order contract).

Parameters

outcomeslist[JobOutcome]

Outcomes from BatchAbaqusProcessor.run_batch().

result_namestr

Key in outcome.results to load.

on_missingstr

How to handle jobs where load_field() returns None:

  • 'skip' (default) — omit the job; a single summary warning lists all skipped job names at generator exit.

  • 'none' — yield (job_name, None) so the caller can align rows with degenerate_from_array().

  • 'raise' — raise ValueError on the first missing field.

Yields

tuple[str, numpy.ndarray or None]

(job_name, ndarray) pairs. ndarray is None only when on_missing='none'.

ABQflow.helpers.convert.sanitize_job_name(name, max_len=80)[source]

Clean name so it is a valid Abaqus job name.

Replaces any character outside [A-Za-z0-9_-] with '_', collapses consecutive underscores, strips leading/trailing underscores, ensures the result starts with a letter, and truncates to max_len.

Returns name unchanged if it is already valid.

Parameters:
Return type:

str

ABQflow.helpers.convert.generate_from_inp_files(inp_files, base_spec, naming='stem', sort=True)[source]

Create N JobSpec objects from a list (or glob) of existing INP files.

This is the batch-spec generator for the UC-03 “pre-existing INP batch” use case. Each INP file becomes a spec with kind='existing_inp'.

Parameters

inp_fileslist[str] or str

List of INP paths, or a glob pattern (e.g. './legacy/*.inp').

base_specJobSpec or dict

Template spec whose workflow, extraction hooks, and other non-preparation fields are copied. The preparation field is overwritten for each generated spec.

namingstr

Job-name generation rule:

  • 'stem' (default) - use the INP filename without extension,

sanitised via sanitize_job_name(). * 'indexed' - {base_spec.job_name}_{i:04d}.

sortbool

If True (default), sort files by natural key order.

Returns

list[JobSpec]

One spec per INP file, ready for BatchAbaqusProcessor.

Raises

ValueError

If glob expands to zero files, or if sanitised stem names collide.

Parameters:
Return type:

list[JobSpec]

ABQflow.helpers.convert.generate_from_array(samples_array, param_names, base_spec)[source]

Create N JobSpec objects from an (N, D) parameter array.

Each row of samples_array becomes a new spec via copy.deepcopy() of base_spec, so every spec owns independent mutable state.

Parameters

samples_arrayndarray or Tensor

Shape (N, D) parameter matrix. Torch tensors are converted to NumPy internally.

param_nameslist[str]

Length-D list of parameter names.

base_specJobSpec or dict

Template spec. Dicts are upgraded via JobSpec.from_dict().

Returns

list[JobSpec]

N specs with zero-padded names (e.g. job_0001, job_0002).

Raises

ValueError

If the array column count does not match len(param_names).

Return type:

list[JobSpec]

ABQflow.helpers.convert.degenerate_from_array(outcomes, output_names, default_value=nan, require_completed=True)[source]

Extract a 2D NumPy array of output values from a list of outcomes.

Outcomes are sorted by natural key on job_name so rows appear in the order the jobs were generated. Jobs that are not COMPLETED are filled with default_value and trigger a warning.

Parameters

outcomeslist[JobOutcome]

Outcomes from BatchAbaqusProcessor.run_batch().

output_nameslist[str]

Keys to extract from each outcome’s results dict.

default_valuefloat

Value to use for missing or non-completed results (default NaN).

require_completedbool

If True (default), warn when non-COMPLETED jobs are encountered.

Returns

np.ndarray

Shape (len(outcomes), len(output_names)) float array.

Parameters:
Return type:

ndarray

ABQflow.helpers.convert.outcomes_to_list(outcomes)[source]

Convert a list of JobOutcome objects to a list of plain dicts.

Convenience for callers that prefer the legacy list-of-dicts shape.

Parameters

outcomeslist[JobOutcome]

Outcomes from BatchAbaqusProcessor.run_batch().

Returns

list[dict]

Each dict contains 'job_name', 'status', flattened results, and optionally 'error'.

Parameters:

outcomes (list)

Return type:

list[dict]

ABQflow.helpers.convert.outcomes_to_dict(outcomes)[source]

Convert a list of JobOutcome objects to a {job_name: {...}} dict.

Parameters

outcomeslist[JobOutcome]

Outcomes from BatchAbaqusProcessor.run_batch().

Returns

dict[str, dict]

Each value dict contains 'status', flattened results, and optionally 'error'.

Raises

ValueError

If two outcomes share the same job_name.

Parameters:

outcomes (list)

Return type:

dict[str, dict]

Module contents