ABQflow package ¶

run_preflight(mode)[source]¶

Run an Abaqus syntax/datacheck on the INP before the real solve.

Uses a temporary job name <job>_chk so preflight output files (.dat, .odb) never overwrite the real job’s files.

Parameters¶

modestr: 'syntaxcheck' or 'datacheck'.

Returns¶

tuple[bool, list[str]]: (passed, errors) — errors are harvested from the temporary .dat file via harvest_errors() (IMP-01/04 synergy).

Parameters:: mode (str)
Return type:: tuple[bool, list[str]]

run_solver()[source]¶

Submit the INP file to the Abaqus solver and wait for completion.

Uses Popen with process-group isolation so that the terminate escalation ladder can reach solver child processes (standard.exe / explicit.exe) — something subprocess.run cannot do.

Escalation ladder (IMP-03):

Normal wait up to self.timeout.
Graceful: abaqus terminate job=<name>.
Grace period G = clamp(0.05 × T, 30, 300) s.
Force-kill the process tree (taskkill /T or os.killpg).
Remove <job>.lck so the job can be re-run.

After the solver process exits (by any means), diagnose() is called and the truth table applied.

Returns¶

SolverResult: Success/failure judgment with diagnostics.

Return type:: SolverResult

subroutine_needs_recompile(subroutine)[source]¶

Return True if subroutine has changed since the last successful compile.

Compares the sha256 of subroutine.source_path against a sidecar hash file written by _record_compile_hash() after a successful compile (same hash-compare-and-skip pattern as _stage_hookkit()). Always True if no prior compile record exists.

Parameters:: subroutine (SubroutineSpec)
Return type:: bool

Parameters:

ctx (JobContext)
logger (logging.Logger)
timeout (float | None)
record_only (bool)

class ABQflow.BatchAbaqusProcessor(batch_data, base_output_dir, cpus_per_job, abaqus_exe='abaqus', duplicate_mode='fail', prompt_fn=<built-in function input>, timeout=None, preflight_only=False)[source]¶

Bases: object

Orchestrate a batch of Abaqus jobs through a three-phase lifecycle.

plan() — inspect for directory conflicts, compute decisions. Pure computation; no side effects.
prepare() — apply decisions (delete, rename, skip) and build the AbaqusCalculation list.
run_batch() — execute via ProcessPoolExecutor; one failure never affects sibling jobs.

Attributes¶

specslist[JobSpec]: Normalised list of job specifications.
calculationslist[AbaqusCalculation] or None: Built calculations (populated by prepare()).
loggerlogging.Logger: Logger writing to batch_processor.log in the output directory.

__init__(batch_data, base_output_dir, cpus_per_job, abaqus_exe='abaqus', duplicate_mode='fail', prompt_fn=<built-in function input>, timeout=None, preflight_only=False)[source]¶

Parameters¶

batch_datalist[dict] or list[JobSpec]

Job configs as dicts or JobSpec objects. Dicts are converted via JobSpec.from_dict().

base_output_dirstr

Absolute directory where all job subdirectories will be created.

cpus_per_jobint

Number of CPUs to request for each Abaqus job.

abaqus_exestr

Absolute path to the Abaqus executable (default 'abaqus').

duplicate_modestr

How to handle existing job directories (default 'fail'):

'fail' — raise FileExistsError on any conflict.
'skip' — skip jobs whose directory already exists.
'overwrite' — delete the existing directory and re-run.
'interactive' — prompt the user for each conflict.

prompt_fncallable

Function for interactive prompts (default input()).

timeoutfloat or None

Per-subprocess timeout in seconds; None means no limit.

preflight_onlybool

If True, only run preparation + preflight, skip solver & extraction (IMP-04 batch inspection mode).

Parameters:

batch_data (list[dict] | list[JobSpec])
base_output_dir (str)
cpus_per_job (int)
abaqus_exe (str)
duplicate_mode (str)
timeout (float | None)
preflight_only (bool)

dry_run(level='plan')[source]¶

Inspect what the batch would do without executing it.

Two levels (see IMP-05):

'plan' (default): Zero side effects. Inspects each spec and builds a command plan without touching the filesystem.
'stage': Runs the real preparation phase, but substitutes a record_only runner so solver and hook commands are logged, not executed. Has filesystem side effects (output_dir is created, INPs are staged).

Parameters¶

levelstr: 'plan' (L1, default) or 'stage' (L2).

Returns¶

list[JobPlan]: One plan per job.

Parameters:: level (str)
Return type:: list[JobPlan]

plan()[source]¶

Inspect output directory for existing job subdirectories.

Pure read-only check — no directories are created, deleted, or renamed. The decision for each job is one of: 'run', 'skip', 'overwrite', or a new name string (rename).

Returns¶

dict[str, str]: {job_name: decision} mapping.

Raises¶

FileExistsError: If duplicate_mode='fail' and any job directory already exists.

Return type:: dict[str, str]

prepare(decisions=None)[source]¶

Apply plan decisions and build the AbaqusCalculation list.

Side effects: directories may be deleted ('overwrite') or specs may be renamed ('rename'). Results are stored in self.calculations.

Parameters¶

decisionsdict[str, str] or None: Decision map from plan(). If None, plan() is called first.

Parameters:: decisions (dict[str, str] | None)

run_batch(num_parallel_jobs, license_tokens=None)[source]¶

Execute all prepared calculations via ProcessPoolExecutor.

If prepare() has not been called yet it is invoked with a fresh call to plan().

Parameters¶

num_parallel_jobsint: Desired maximum concurrent jobs.
license_tokensint or None: Total license tokens available; None means no license limit.

Returns¶

list[JobOutcome]: One outcome per executed job. Failed jobs are included with their error state — they do not halt the batch.

Parameters:

num_parallel_jobs (int)
license_tokens (int | None)

Return type:

run_extraction(num_parallel_jobs=1, license_tokens=None)[source]¶

Run only the post-extraction phase for every spec.

Assumes ctx.odb_path already exists for every job (e.g. from a prior run_simulation() call). Only workflow='modular' specs are supported.

Parameters¶

num_parallel_jobsint: Desired maximum concurrent jobs (default 1).
license_tokensint or None: Total license tokens available; None means no license limit.

Returns¶

list[JobOutcome]: One outcome per job.

Parameters:

num_parallel_jobs (int)
license_tokens (int | None)

Return type:

run_preparation(num_parallel_jobs=1, license_tokens=None)[source]¶

Run only the preparation phase (produce INPs, and compile a subroutine if configured) for every spec in self.specs.

Builds AbaqusCalculations directly from self.specs (see _build_calc()) rather than through plan()/ prepare()’s duplicate-directory handling — call plan()/ prepare() first if you need overwrite/rename/skip semantics. Only workflow='modular' specs are supported; a monolithic spec raises NotImplementedError (it has no separable preparation phase).

Parameters¶

num_parallel_jobsint: Desired maximum concurrent jobs (default 1).
license_tokensint or None: Total license tokens available; None means no license limit.

Returns¶

list[JobOutcome]: One outcome per job.

Parameters:

num_parallel_jobs (int)
license_tokens (int | None)

Return type:

run_simulation(num_parallel_jobs=1, license_tokens=None)[source]¶

Run only the solver phase (pre-extraction + solve) for every spec.

Assumes ctx.inp_path already exists for every job (e.g. from a prior run_preparation() call — possibly in an earlier process/session; reconstruct the BatchAbaqusProcessor with the same batch_data/base_output_dir to resume). Only workflow='modular' specs are supported.

Parameters¶

num_parallel_jobsint: Desired maximum concurrent jobs (default 1).
license_tokensint or None: Total license tokens available; None means no license limit.

Returns¶

list[JobOutcome]: One outcome per job.

Parameters:

num_parallel_jobs (int)
license_tokens (int | None)

Return type:

Parameters:

batch_data (list[dict] | list[JobSpec])
base_output_dir (str)
cpus_per_job (int)
abaqus_exe (str)
duplicate_mode (str)
timeout (float | None)
preflight_only (bool)

class ABQflow.CommandRecord(stage, cmd, cwd)[source]¶

Bases: object

One command that was (or would be) executed.

Parameters:

stage (str)
cmd (list[str])
cwd (str)

stage: str¶

cmd: list[str]¶

cwd: str¶

class ABQflow.ExistingInpStrategy(source_inp_path, staging_mode='copy', resolve_includes=True)[source]¶

Bases: PreparationStrategy

Use a pre-existing INP file directly — no generation or modification.

This strategy satisfies the preparation contract (“ensure an INP at ctx.inp_path”) by copying an already-complete INP file. It is the entry point for the UC-03 “pre-existing INP batch” use case.

Key features beyond a plain file copy:

INCLUDE resolution: scans for *INCLUDE, INPUT=... lines and rewrites relative paths to absolute paths so Abaqus can find referenced files regardless of the working directory.
Template detection: rejects INPs that still contain {{...}} placeholders, steering the user toward kind='inp_based' instead.
STEP presence check: confirms the file contains at least one *STEP keyword.

Attributes¶

source_inp_pathstr: Absolute or relative path to the existing INP file.
staging_modestr: 'copy' (default) — copy the INP (with resolved paths) to output_dir.
resolve_includesbool: If True (default), rewrite *INCLUDE, INPUT=rel_path to use absolute paths resolved against the source INP’s directory.

prepare(ctx, runner, logger)[source]¶

Produce the INP file.

Parameters¶

ctxJobContext: Job context providing inp_path and output_dir.
runnerAbaqusRunner: Subprocess runner (may not be used by every strategy).
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

bool: True if the INP was produced, False otherwise.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

Parameters:

source_inp_path (str)
staging_mode (str)
resolve_includes (bool)

class ABQflow.ExtractionStrategy[source]¶

Bases: ABC

Interface for extraction: read results from model or ODB files.

Subclasses¶

OdbExtractionStrategy: Post-simulation extraction from ODB (requires odbAccess).
ModelPropertiesExtractionStrategy: Pre-simulation extraction from INP (requires mdb / CAE kernel).

abstractmethod extract(ctx, runner, logger)[source]¶

Extract results.

Parameters¶

ctxJobContext: Job context providing file paths.
runnerAbaqusRunner: Subprocess runner for launching hook scripts.
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

dict: {result_name: value, ...}. Failed tasks map to None.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

class ABQflow.HookSpec(script_path, tasks=<factory>)[source]¶

Bases: object

Description of one extraction/pre-extraction hook script and its tasks.

Attributes¶

script_pathstr: Path to the Python script that processes the hook.
taskslist[dict]: List of task descriptors; each dict typically contains result_name, script_path, and task-specific parameters.

script_path: str¶

tasks: list[dict]¶

Parameters:

script_path (str)
tasks (list[dict])

class ABQflow.InpModifyStrategy(base_inp_path, data_params)[source]¶

Bases: PreparationStrategy

Replace {{placeholder}} tokens in a base INP template file.

Performs coverage validation: if the INP references a placeholder that is missing from data_params, preparation fails. If data_params contains keys that are not used in the INP, a warning is emitted.

Attributes¶

base_inp_pathstr: Path to the template INP file containing {{key}} placeholders.
data_paramsdict: Mapping of placeholder names to substitution values.

prepare(ctx, runner, logger)[source]¶

Produce the INP file.

Parameters¶

ctxJobContext: Job context providing inp_path and output_dir.
runnerAbaqusRunner: Subprocess runner (may not be used by every strategy).
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

bool: True if the INP was produced, False otherwise.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

Parameters:

base_inp_path (str)
data_params (dict)

class ABQflow.JobContext(job_name, output_dir, cpus, abaqus_exe='abaqus', user_subroutine=None)[source]¶

Bases: object

Immutable data contract holding all information a strategy can observe.

Strategies depend on this object but cannot mutate it, preventing accidental cross-strategy side effects. Every field is read-only after construction.

Attributes¶

job_namestr: Unique identifier for this job (also used as directory name).
output_dirstr: Absolute path to the job’s working directory.
cpusint: Number of CPUs requested for the Abaqus solver run.
abaqus_exestr: Path or command name for the Abaqus executable (default "abaqus").

abaqus_exe: str = 'abaqus'¶

property dat_path: str¶: Absolute path to the data file (<output_dir>/<job_name>.dat).

property exec_log_path: str¶

Absolute path to ABQflow’s own execution log (<output_dir>/<job_name>_abqflow.log).

Deliberately distinct from log_path (Abaqus’s native job log) so the two writers never collide on the same file.

property inp_path: str¶: Absolute path to the input file (<output_dir>/<job_name>.inp).

property log_path: str¶

Absolute path to Abaqus’s own native job log (<output_dir>/<job_name>.log).

This file is owned and written by the Abaqus solver process itself (abaqus job=... interactive writes it as a side effect). ABQflow never opens this path for writing — see exec_log_path for ABQflow’s own execution log, which uses a distinct filename to avoid colliding with this one.

property msg_path: str¶: Absolute path to the message file (<output_dir>/<job_name>.msg).

property odb_path: str¶: Absolute path to the output database (<output_dir>/<job_name>.odb).

property sta_path: str¶: Absolute path to the status file (<output_dir>/<job_name>.sta).

user_subroutine: str | None = None¶

job_name: str¶

output_dir: str¶

cpus: int¶

Parameters:

job_name (str)
output_dir (str)
cpus (int)
abaqus_exe (str)
user_subroutine (str | None)

class ABQflow.JobOutcome(job_name, status, results=None, error=None, diagnostics=None, output_dir=None, phases=None, duration_s=None)[source]¶

Bases: object

Unified result envelope returned from every job, pass or fail.

Status is normalised to a plain string (JobStatus.value) so it serialises cleanly across process boundaries.

Attributes¶

job_namestr: Name of the job.
statusstr: String status, e.g. "COMPLETED" or "SIMULATION_FAILED".
resultsdict or None: Extracted result values, or None if the job did not reach extraction.
errorstr or None: Error message if the job failed, None otherwise.
diagnosticsdict or None: Solver diagnostics snapshot (IMP-02). Populated on failure and on the rc≠0 + COMPLETED edge case. None for clean success or jobs that never reached the solver phase.
phaseslist[dict] or None: Phase-by-phase history (name/status/duration/error) collected from JobStatusManager. None for strategies that don’t populate it (e.g. monolithic workflows).
duration_sfloat or None: Wall-clock seconds spent in AbaqusCalculation.execute() for this job.

diagnostics: dict | None = None¶

duration_s: float | None = None¶

error: str | None = None¶

output_dir: str | None = None¶

phases: list[dict] | None = None¶

results: dict | None = None¶

job_name: str¶

status: str¶

Parameters:

job_name (str)
status (str)
results (dict | None)
error (str | None)
diagnostics (dict | None)
output_dir (str | None)
phases (list[dict] | None)
duration_s (float | None)

class ABQflow.JobPlan(job_name, commands=<factory>, paths=<factory>, resource_summary=<factory>)[source]¶

Bases: object

Dry-run output for a single job — commands, paths, and resource summary.

Attributes¶

job_namestr: Job identifier.
commandslist: List of CommandRecord instances.
pathsdict[str, str]: Expected file paths (inp, odb, output_dir).
resource_summarydict: Planned CPUs, token estimate, and parallelism info.

job_name: str¶

commands: list¶

paths: dict¶

resource_summary: dict¶

Parameters:

job_name (str)
commands (list)
paths (dict)
resource_summary (dict)

class ABQflow.JobSpec(job_name, workflow='modular', preparation=None, preflight=None, monolithic_script=None, monolithic_params=<factory>, pre_extraction=<factory>, post_extraction=<factory>, subroutine=None, meta=<factory>)[source]¶

Bases: object

Single-job configuration validated at construction time.

Fails fast — validation runs in __post_init__ so invalid configs are rejected before any Abaqus process is launched.

Attributes¶

job_namestr: Unique name for this job (also used as the working directory name).
workflowstr: 'modular' (default, 4-phase pipeline) or 'monolithic' (single-script).
preparationPreparationSpec or None: Preparation spec; required when workflow='modular', ignored for monolithic.
preflightstr, default=None: Preflight mode for modular workflows. - None: No preflight checks (default) - ‘syntaxcheck’: Run abaqus syntax check - ‘datacheck’: Run abaqus datacheck
monolithic_scriptstr or None: Path to the monolithic script; required when workflow='monolithic'.
monolithic_paramsdict: Parameters forwarded to the monolithic script as --key value args.
pre_extractionlist[HookSpec]: Hooks run before the solver (e.g. model property extraction).
post_extractionlist[HookSpec]: Hooks run after the solver (e.g. ODB result extraction).
subroutineSubroutineSpec or None: User subroutine to compile and pass via user= to the solver (modular workflow only; ignored for workflow='monolithic').
metadict: Arbitrary user metadata

classmethod from_dict(d)[source]¶

Migration bridge: construct a JobSpec from a legacy dict.

Deep-copies the input dict so the returned spec owns all of its mutable data (no shared references with the caller).

Parameters¶

ddict: Legacy configuration dict. Recognised keys: job_name, workflow, type, base_inp_path, model_script_path, script_path, params, pre_extraction, post_extraction.

Returns¶

JobSpec: Fully validated spec.

Parameters:: d (dict)
Return type:: JobSpec

monolithic_script: str | None = None¶

preflight: str | None = None¶

preparation: PreparationSpec | None = None¶

subroutine: SubroutineSpec | None = None¶

workflow: str = 'modular'¶

job_name: str¶

monolithic_params: dict¶

pre_extraction: list[HookSpec]¶

post_extraction: list[HookSpec]¶

meta: dict¶

Parameters:

job_name (str)
workflow (str)
preparation (PreparationSpec | None)
preflight (str | None)
monolithic_script (str | None)
monolithic_params (dict)
pre_extraction (list[HookSpec])
post_extraction (list[HookSpec])
subroutine (SubroutineSpec | None)
meta (dict)

class ABQflow.JobStatus(value)[source]¶

Bases: Enum

Lifecycle state for a single batch job.

Key values¶

CREATED: Initial state — job has been constructed but not yet started.
COMPLETED: Terminal success — the full workflow finished without error.
PREPARATION_FAILED: Terminal failure — the preparation phase could not produce an INP.
SIMULATION_FAILED: Terminal failure — the Abaqus solver exited with an error.
EXTRACTION_FAILED: Terminal failure — one or more post-extraction tasks returned None.
MONOLITHIC_SCRIPT_FAILED: Terminal failure — the monolithic script exited with a non-zero code.
JSON_DECODE_ERROR: Terminal failure — monolithic or hook script output could not be parsed as JSON.
SCRIPT_ERROR: Terminal failure — an unhandled exception occurred in a hook or monolithic script.
SUBROUTINE_COMPILE_FAILED: Terminal failure — abaqus make (user subroutine compilation) exited with an error.
UNKNOWN_ERROR: Terminal failure — an exception escaped the worker process.
UNKNOWN: Fallback value used when no explicit status is available.

CREATED = 'CREATED'¶

COMPLETED = 'COMPLETED'¶

PREPARING = 'PREPARING'¶

PREPARATION_FAILED = 'PREPARATION_FAILED'¶

PREPARATION_SUCCESS = 'PREPARATION_SUCCESS'¶

PREFLIGHT_FAILED = 'PREFLIGHT_FAILED'¶

SIMULATING = 'SIMULATING'¶

SIMULATION_FAILED = 'SIMULATION_FAILED'¶

SIMULATION_SUCCESS = 'SIMULATION_SUCCESS'¶

EXTRACTING = 'EXTRACTING'¶

EXTRACTION_FAILED = 'EXTRACTION_FAILED'¶

EXTRACTION_SUCCESS = 'EXTRACTION_SUCCESS'¶

MONOLITHIC_SCRIPT_FAILED = 'MONOLITHIC_SCRIPT_FAILED'¶

JSON_DECODE_ERROR = 'JSON_DECODE_ERROR'¶

SCRIPT_ERROR = 'SCRIPT_ERROR'¶

SUBROUTINE_COMPILE_FAILED = 'SUBROUTINE_COMPILE_FAILED'¶

UNKNOWN_ERROR = 'UNKNOWN_ERROR'¶

UNKNOWN = 'UNKNOWN'¶

class ABQflow.JobStatusManager[source]¶

Bases: object

State machine for a single job with terminal-state protection.

Calling record_preparation(), record_simulation(), or record_extraction() advances the state. Once a terminal failure state is reached, all subsequent transitions are silently ignored — the first failure is the one that is kept.

The mark_* methods set the live in-progress status (PREPARING/ SIMULATING/EXTRACTING) and open a PhaseRecord; the paired record_* method closes it. mark_* calls are optional — callers that only care about the final outcome (as before) can skip them and just call record_* directly.

Attributes¶

error_messagestr or None: Error message from the first terminal failure, or None.

property current_status: JobStatus¶: Read-only access to the live status (updated at phase start, not just at the end).

property error_message: str | None¶: Read-only access to the first-failure error message.

get_final_status()[source]¶

Return the current state or COMPLETED if no failure was recorded.

Returns¶

JobStatus: The terminal failure state if one was reached, otherwise JobStatus.COMPLETED.

Return type:: JobStatus

mark_compiling()[source]¶: Mark the start of subroutine compilation.

mark_extracting(label='extraction')[source]¶

Mark the start of an extraction phase.

Parameters¶

labelstr: 'pre_extraction' or 'post_extraction' — distinguishes the two extraction phases in phase_history.

Parameters:: label (str)

mark_preflight()[source]¶: Mark the start of the preflight check.

mark_preparing()[source]¶: Mark the start of the preparation phase.

mark_simulating()[source]¶: Mark the start of the solver run.

property phase_history: list[dict]¶: Closed phases so far, as plain dicts (picklable across process boundaries).

record_compile(success, error=None)[source]¶

Record the outcome of user-subroutine compilation.

Parameters¶

successbool: True if abaqus make (or an equivalent compile step) succeeded.
errorstr or None: Error message on failure; a default is used if omitted.

Parameters:

success (bool)
error (str | None)

record_extraction(results)[source]¶

Record extraction results; fails if any task returned None.

On success this now sets current_status to JobStatus.EXTRACTION_SUCCESS (previously a no-op — the final status reported by get_final_status() is unaffected, since EXTRACTION_SUCCESS is not a terminal state).

Parameters¶

resultsdict: {result_name: value} mapping. Any None value triggers EXTRACTION_FAILED.

Parameters:: results (dict)

record_preflight(success, error=None)[source]¶

Record the outcome of the preflight phase (IMP-04).

Parameters¶

successbool: True if syntax/datacheck passed.
errorstr or None: Error message on failure; a default is used if omitted.

Parameters:

success (bool)
error (str | None)

record_preparation(success, error=None)[source]¶

Record the outcome of the preparation phase.

Parameters¶

successbool: True if the INP was produced successfully.
errorstr or None: Error message on failure; a default is used if omitted.

Parameters:

success (bool)
error (str | None)

record_simulation(success, error=None)[source]¶

Record the outcome of the Abaqus solver run.

Parameters¶

successbool: True if the solver exited with code 0.
errorstr or None: Error message on failure; a default is used if omitted.

Parameters:

success (bool)
error (str | None)

class ABQflow.JobWorkflowStrategy[source]¶

Bases: ABC

Interface for a complete job workflow.

Subclasses¶

MonolithicWorkflowStrategy: Single-script workflow that handles everything itself.
ModularWorkflowStrategy: Multi-phase pipeline: optional subroutine compile, preparation, optional preflight, pre-extraction, simulation, post-extraction.

Optional phase-separated protocol¶

Subclasses may additionally implement prepare_only(ctx, runner, logger, status_manager=None) -> tuple[dict, JobStatusManager], simulate_only(...) -> tuple[dict, JobStatusManager, bool] (the bool signals whether the pipeline should stop), and extract_only(...) -> tuple[dict, JobStatusManager] so that AbaqusCalculation can invoke a single phase (see its execute(phase=...) parameter). This is not required by the ABC — MonolithicWorkflowStrategy and user-defined strategies that don’t implement it simply raise NotImplementedError when a phase-only call is attempted.

abstractmethod execute(ctx, runner, logger)[source]¶

Run the full workflow and return a result dict.

Parameters¶

ctxJobContext: Job context.
runnerAbaqusRunner: Subprocess runner for all subprocess calls.
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

dict: Must contain at least a 'status' key (a JobStatus or its string value). May include extracted results.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

class ABQflow.ModelGenerationStrategy(model_script_path, script_params)[source]¶

Bases: PreparationStrategy

Run a model-generation script (requires CAE kernel / mdb access).

The script is launched via abaqus cae noGUI=<script> and is expected to produce an INP file at ctx.inp_path. Common arguments (--job_name, user params) are forwarded as CLI flags.

Attributes¶

model_script_pathstr: Path to the model-generation script.
script_paramsdict: Key-value pairs forwarded as --key value arguments.

prepare(ctx, runner, logger)[source]¶

Produce the INP file.

Parameters¶

ctxJobContext: Job context providing inp_path and output_dir.
runnerAbaqusRunner: Subprocess runner (may not be used by every strategy).
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

bool: True if the INP was produced, False otherwise.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

Parameters:

model_script_path (str)
script_params (dict)

class ABQflow.ModelPropertiesExtractionStrategy(hooks)[source]¶

Bases: ExtractionStrategy

Extract material/property data from the INP before simulation.

Runs in the CAE kernel environment (abaqus cae noGUI) because it needs mdb access. Each hook script receives --inp_path as a common argument and a JSON task list via --tasks_json.

Attributes¶

hookslist[HookSpec]: List of hook descriptors, each with script_path and tasks.

extract(ctx, runner, logger)[source]¶

Extract results.

Parameters¶

ctxJobContext: Job context providing file paths.
runnerAbaqusRunner: Subprocess runner for launching hook scripts.
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

dict: {result_name: value, ...}. Failed tasks map to None.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

Parameters:: hooks (list[HookSpec])

class ABQflow.ModularWorkflowStrategy(preparation_strategy, pre_extraction_strategies, post_extraction_strategies, preflight_mode=None, preflight_only=False, compile_strategy=None)[source]¶

Bases: JobWorkflowStrategy

Multi-phase pipeline: [compile], preparation, [preflight], pre-extraction, simulation, post-extraction.

Uses a JobStatusManager internally to track the job through each phase. If any phase fails the pipeline stops and returns the terminal status immediately.

execute() composes three independently callable phase methods — prepare_only(), simulate_only(), extract_only() — so that AbaqusCalculation (and BatchAbaqusProcessor’s run_preparation/run_simulation/run_extraction) can invoke a single phase without running the rest of the pipeline. The external contract of execute() — return-dict shape and terminal-status semantics — is unchanged by this split.

Attributes¶

preparation_strategyPreparationStrategy: Strategy that produces the INP file.
preflight_modestr or None: 'syntaxcheck', 'datacheck', or None (IMP-04).
pre_extraction_strategieslist[ExtractionStrategy]: Strategies run before the solver (e.g. property extraction from INP).
post_extraction_strategieslist[ExtractionStrategy]: Strategies run after the solver (e.g. result extraction from ODB).
compile_strategySubroutineCompileStrategy or None: Optional user-subroutine compile step, run before preparation.

execute(ctx, runner, logger)[source]¶

Run the full modular workflow by composing the three phase methods.

Returns a dict with at least a 'status' key plus any results from pre- and post-extraction hooks. Failing early means later phases are skipped — return-dict shape and terminal-status semantics are unchanged from before the phase-separation refactor.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

extract_only(ctx, runner, logger, status_manager=None)[source]¶

Phase 3: post-extraction hooks only.

Assumes ctx.odb_path already exists. No existence guard is needed — OdbExtractionStrategy already reports every task as None when the ODB is missing, and JobStatusManager.record_extraction() already turns that into EXTRACTION_FAILED.

Returns¶

tuple[dict, JobStatusManager]: (results, status_manager).

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)
status_manager (JobStatusManager | None)

Return type:

tuple[dict, JobStatusManager]

prepare_only(ctx, runner, logger, status_manager=None)[source]¶

Phase 1: optional subroutine compile, preparation, optional preflight.

Standalone entry point for “produce an INP (and compiled subroutine) only”. Does not run pre-extraction, the solver, or post-extraction — those live in simulate_only() / extract_only().

Returns¶

tuple[dict, JobStatusManager]: (results, status_manager) — results has at least 'status' and '_phase_history'; the manager is returned so execute() can thread it into the next phase.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)
status_manager (JobStatusManager | None)

Return type:

tuple[dict, JobStatusManager]

simulate_only(ctx, runner, logger, status_manager=None)[source]¶

Phase 2: pre-extraction hooks, then the solver run.

Assumes ctx.inp_path already exists (produced by a prior prepare_only() call — possibly in an earlier process/session, e.g. via BatchAbaqusProcessor.run_simulation()). Mirrors the original monolithic behavior: a pre-extraction failure does not stop the solver from running, but a solver failure does stop the pipeline.

Returns¶

tuple[dict, JobStatusManager, bool]: (results, status_manager, stop) — stop is True when the caller (e.g. execute()) should not proceed to extract_only() (INP missing or solver failed).

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)
status_manager (JobStatusManager | None)

Return type:

tuple[dict, JobStatusManager, bool]

Parameters:

preparation_strategy (PreparationStrategy)
pre_extraction_strategies (List[ExtractionStrategy])
post_extraction_strategies (List[ExtractionStrategy])
preflight_mode (str | None)
preflight_only (bool)
compile_strategy (SubroutineCompileStrategy | None)

class ABQflow.MonolithicWorkflowStrategy(script_path, params)[source]¶

Bases: JobWorkflowStrategy

Single-script workflow: one script does everything.

The script is launched via the CAE kernel (abaqus cae noGUI) and must print its JSON results wrapped in the sentinel markers ===ABQ_RESULT_BEGIN=== / ===ABQ_RESULT_END===. The result dict is expected to contain at least a 'status' key.

Attributes¶

script_pathstr: Path to the monolithic script.
paramsdict: Key-value parameters forwarded as --key value CLI arguments.

execute(ctx, runner, logger)[source]¶

Run the full workflow and return a result dict.

Parameters¶

ctxJobContext: Job context.
runnerAbaqusRunner: Subprocess runner for all subprocess calls.
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

dict: Must contain at least a 'status' key (a JobStatus or its string value). May include extracted results.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

Parameters:

script_path (str)
params (dict)

class ABQflow.OdbExtractionStrategy(hooks)[source]¶

Bases: ExtractionStrategy

Extract results from the ODB file via hook scripts.

Runs in the odbAccess environment (abaqus python), NOT the CAE kernel. Each hook script receives --odb_path as a common argument and a JSON task list via --tasks_json.

Attributes¶

hookslist[HookSpec]: List of hook descriptors, each with script_path and tasks.

extract(ctx, runner, logger)[source]¶

Extract results.

Parameters¶

ctxJobContext: Job context providing file paths.
runnerAbaqusRunner: Subprocess runner for launching hook scripts.
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

dict: {result_name: value, ...}. Failed tasks map to None.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type:

Parameters:: hooks (list[HookSpec])

class ABQflow.PreparationSpec(kind, source_path, params=<factory>, options=<factory>)[source]¶

Bases: object

Specification for the preparation phase of a modular workflow.

Attributes¶

kindstr: Preparation strategy identifier. Currently 'inp_based', 'existing_inp', or 'model_generation'.
source_pathstr: Path to the base INP file (for inp_based) or model-generation script (for model_generation).
paramsdict: Key-value parameters forwarded to the preparation strategy (e.g. placeholder replacements for inp_based).
optionsdict: Additional options for the preparation strategy (Currently only used by existing_inp): - ‘staging_mode’ (str): 'copy' (default) - ‘resolve_includes’ (bool): Whether to resolve *INCLUDE directives in the INP file (default: True).

kind: str¶

source_path: str¶

params: dict¶

options: dict¶

Parameters:

kind (str)
source_path (str)
params (dict)
options (dict)

class ABQflow.PreparationStrategy[source]¶

Bases: ABC

Interface for preparation: produce an INP file at ctx.inp_path.

Subclasses¶

InpModifyStrategy: Template-based INP generation ({{placeholder}} substitution).
ModelGenerationStrategy: Run an external script that produces the INP (requires CAE kernel).

abstractmethod prepare(ctx, runner, logger)[source]¶

Produce the INP file.

Parameters¶

ctxJobContext: Job context providing inp_path and output_dir.
runnerAbaqusRunner: Subprocess runner (may not be used by every strategy).
loggerlogging.Logger: Logger for progress and error messages.

Returns¶

bool: True if the INP was produced, False otherwise.

Parameters:

ctx (JobContext)
runner (AbaqusRunner)
logger (Logger)

Return type: