ABQflow.core package ¶

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:

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:

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)

ABQflow.core.context module¶

JobContext — frozen data contract that strategies read but cannot mutate.

class ABQflow.core.context.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").

job_name: str¶

output_dir: str¶

cpus: int¶

abaqus_exe: str = 'abaqus'¶

user_subroutine: str | None = None¶

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

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

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 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 sta_path: str¶: Absolute path to the status file (<output_dir>/<job_name>.sta).

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

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

Parameters:

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

ABQflow.core.registry module¶

Strategy registry — open/closed mapping from preparation kind to factory.

Replaces hardcoded if/else dispatch. Users can add custom preparation strategies at runtime via register_preparation() without modifying framework code.

ABQflow.core.registry.register_preparation(kind, factory)[source]¶

After registration, users can set PreparationSpec.kind to kind and build_workflow() will dispatch to factory automatically — no framework source changes required.

Parameters¶

kindstr: Unique key for the preparation strategy (referenced in PreparationSpec.kind).
factorycallable: Callable that receives a PreparationSpec and returns a PreparationStrategy.

Parameters:

kind (str)
factory (callable)

ABQflow.core.registry.build_workflow(spec, preflight_only=False)[source]¶

Assemble a concrete JobWorkflowStrategy from a spec.

Monolithic specs produce a MonolithicWorkflowStrategy.
Modular specs look up the preparation kind in PREPARATION_REGISTRY, wrap pre/post-extraction hooks, and return a ModularWorkflowStrategy.

Parameters¶

specJobSpec: Validated job configuration.
preflight_onlybool: If True, the workflow stops after preflight (IMP-04).

Returns¶

JobWorkflowStrategy: Ready-to-execute strategy chain.

Raises¶

ValueError: If spec.preparation.kind is not registered.

Parameters:

spec (JobSpec)
preflight_only (bool)

Return type:

JobWorkflowStrategy

ABQflow.core.runner module¶

AbaqusRunner — subprocess gateway that encapsulates every shell call a strategy needs.

Provides environment detection (abqpy / CAE kernel / odbAccess), sentinel-based JSON extraction, timeout-safe command execution, solver diagnostics, and a record_only dry-run mode (IMP-05).

class ABQflow.core.runner.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¶

ABQflow.core.runner.extract_json(text)[source]¶

Extract a JSON object from subprocess stdout.

Protocol: the script wraps its JSON payload between sentinel markers ===ABQ_RESULT_BEGIN=== and ===ABQ_RESULT_END===. If both are present the payload between them is parsed directly. Otherwise falls back to a legacy brace-scan that searches from the end of the output (useful when Abaqus prints a banner before user code runs).

Parameters¶

textstr: Raw stdout captured from a subprocess call.

Returns¶

dict: Parsed JSON payload.

Raises¶

ValueError: If no JSON object can be found or parsed.

Parameters:: text (str)
Return type:: dict

class ABQflow.core.runner.AbaqusRunner(ctx, logger, timeout=None, record_only=False)[source]¶

Bases: object

Encapsulates every subprocess call a strategy may need.

Detects the execution environment and routes commands accordingly:

abqpy installed — uses plain python (abqpy wraps the Abaqus API).
Needs CAE kernel (mdb) — uses abaqus cae noGUI=<script>.
Only needs odbAccess — uses abaqus python <script>.

Attributes¶

ctxJobContext: Frozen context providing job name, paths, CPU count, and Abaqus exe.
loggerlogging.Logger: Logger instance for this runner.
timeoutfloat or None: Per-command timeout in seconds; None means no limit.

static build_script_command(script, needs_cae_kernel, abaqus_exe, has_abqpy)[source]¶

Select the correct interpreter and Abaqus entry-point for script.

Pure function — no instance state required — so both the real execution path (_base_command()) and dry-run planning (_dry_run_plan()) can share one definition instead of maintaining separate copies.

Decision logic (first match wins):

abqpy available — ['python', script].
needs_cae_kernel is True — [exe, 'cae', 'noGUI=<script>', '--']. The '--' separator prevents custom args from being consumed by the Abaqus CLI.
Otherwise — [exe, 'python', script] (odbAccess-only scripts).

Parameters¶

scriptstr: Path to the Python script to execute.
needs_cae_kernelbool: Whether the script requires the CAE kernel (mdb access).
abaqus_exestr: Path or command name for the Abaqus executable.
has_abqpybool: Whether the abqpy package is importable in this environment.

Returns¶

list[str]: Command line as a list of tokens ready for subprocess.run.

Parameters:

script (str)
needs_cae_kernel (bool)
abaqus_exe (str)
has_abqpy (bool)

Return type:

list[str]

static build_solver_command(ctx)[source]¶

Build the abaqus job=... input=... cpus=... [user=...] interactive command line.

Pure function of ctx — shared by run_solver() and dry-run planning so the two never drift apart. user=<ctx.user_subroutine> is inserted (before interactive) when a subroutine is configured.

Parameters:: ctx (JobContext)
Return type:: list[str]

static build_preflight_command(ctx, mode)[source]¶

Build the abaqus <mode> job=<job>_chk input=... [user=...] command line.

Returns¶

tuple[list[str], str]: (cmd, chk_name) — chk_name is the temporary job name used so preflight output never overwrites the real job’s files.

Parameters:

ctx (JobContext)
mode (str)

Return type:

tuple[list[str], str]

static build_make_command(ctx, subroutine)[source]¶

Build the abaqus make library=<source> [explicit|cfd] command line.

Pure function shared by run_compile() and dry-run planning. solver='standard' needs no extra flag; 'explicit'/'cfd' are appended as bare flags — mirrors the convention documented in reference/abaqus-cli (verify against the installed Abaqus version before relying on this in production).

Parameters:

ctx (JobContext)
subroutine (SubroutineSpec)

Return type:

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

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]]

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

run_compile(subroutine)[source]¶

Run abaqus make to compile subroutine.

No regex parsing of compiler errors is performed — stdout/stderr are captured and returned as-is for the caller to log (matches the reference tool’s approach: compiler-error classification is left to a human/LLM reading the raw output, not this library).

Parameters¶

subroutineSubroutineSpec: Subroutine to compile.

Returns¶

tuple[bool, str, str]: (success, stdout, stderr).

Parameters:: subroutine (SubroutineSpec)
Return type:: tuple[bool, str, str]

run_hook(script_path, tasks, common_args, needs_cae_kernel)[source]¶

Execute a hook script with a JSON task list, return per-task results.

Writes tasks to a temporary JSON file, launches the script via _base_command() (so the correct environment is used), appends common_args, --job_name, and --tasks_json, then extracts the JSON result payload from stdout.

Before execution, _stage_hookkit() copies hookkit.py into the job output directory so hooks can import hookkit.

After execution, every sidecar envelope in the results dict passes through _validate_envelope() for path-safety, existence, and metadata-augmentation checks.

Parameters¶

script_pathstr: Path to the hook script.
taskslist[dict]: List of task descriptors, each expected to contain a result_name key.
common_argsdict[str, str]: Extra CLI arguments forwarded to every task (e.g. --odb_path).
needs_cae_kernelbool: Passed through to _base_command() for environment selection.

Returns¶

dict: Mapping {result_name: value, ...}. Tasks that could not run map to None. Returns an empty dict when tasks is empty.

Parameters:

script_path (str)
tasks (list[dict])
common_args (dict[str, str])
needs_cae_kernel (bool)

Return type:

Parameters:

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

ABQflow.core.spec module¶

JobSpec and related configuration dataclasses — typed, validated at construction.

Replaces the legacy dict-based config format. JobSpec validates itself in __post_init__ so errors are caught before batch execution begins.

class ABQflow.core.spec.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.core.spec.SubroutineSpec(source_path, language='fortran', solver='standard', precompiled=False)[source]¶

Bases: object

Specification for an Abaqus user subroutine (UMAT/VUMAT/UEL/…).

Attributes¶

source_pathstr: Path to the subroutine source file (or, when precompiled=True, to the already-compiled object/library).
languagestr: 'fortran' (default), 'c', or 'cpp'.
solverstr: Target solver: 'standard' (default), 'explicit', or 'cfd'. Controls the flag passed to abaqus make (see build_make_command()).
precompiledbool: If True, skip the compile phase entirely and pass source_path straight through to user= on the solver/preflight commands. Default False.

source_path: str¶

language: str = 'fortran'¶

solver: str = 'standard'¶

precompiled: bool = False¶

Parameters:

source_path (str)
language (str)
solver (str)
precompiled (bool)

class ABQflow.core.spec.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.core.spec.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

job_name: str¶

workflow: str = 'modular'¶

preparation: PreparationSpec | None = None¶

preflight: str | None = None¶

monolithic_script: str | None = None¶

monolithic_params: dict¶

pre_extraction: list[HookSpec]¶

post_extraction: list[HookSpec]¶

subroutine: SubroutineSpec | None = None¶

meta: dict¶

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

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)

ABQflow.core.status module¶

Job status enumeration and state machine — terminal-state protection.

Tracks every job through its lifecycle, from CREATED to COMPLETED or a terminal failure state. Once a job enters a failure state no further state transitions are allowed.

class ABQflow.core.status.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.core.status.PhaseRecord(phase, status='RUNNING', started_at=None, ended_at=None, duration_s=None, error=None)[source]¶

Bases: object

One phase’s start/end/outcome — the unit of a job’s phase history.

Attributes¶

phasestr: 'compile' | 'preparation' | 'preflight' | 'pre_extraction' | 'simulation' | 'post_extraction'.
statusstr: Phase outcome string (e.g. 'RUNNING' while open, then a JobStatus value or 'PASSED'/'COMPILED' once closed).
started_atfloat or None: time.time() when the phase was opened.
ended_atfloat or None: time.time() when the phase was closed.
duration_sfloat or None: ended_at - started_at, populated on close.
errorstr or None: Error message if the phase failed.

phase: str¶

status: str = 'RUNNING'¶

started_at: float | None = None¶

ended_at: float | None = None¶

duration_s: float | None = None¶

error: str | None = None¶

Parameters:

phase (str)
status (str)
started_at (float | None)
ended_at (float | None)
duration_s (float | None)
error (str | None)

class ABQflow.core.status.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 error_message: str | None¶: Read-only access to the first-failure error message.

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

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

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

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

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

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

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)

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_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_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_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)

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)

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

ABQflow.core.strategies module¶

Job workflow strategies — the ABC hierarchy and all concrete implementations.

Strategies are stateless (configuration only in __init__) and depend on three injected arguments at call time: JobContext, AbaqusRunner, and logging.Logger.

class ABQflow.core.strategies.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:

class ABQflow.core.strategies.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.core.strategies.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.core.strategies.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.core.strategies.SubroutineCompileStrategy(subroutine, cache=True)[source]¶

Bases: object

Compiles a user subroutine via abaqus make before preparation.

Not part of the PreparationStrategy/ExtractionStrategy ABC hierarchies — compilation is its own concern with a single implementation today (YAGNI; add a registry like PREPARATION_REGISTRY if multiple compile backends are ever needed).

Attributes¶

subroutineSubroutineSpec: Subroutine to compile.
cachebool: If True (default), skip recompilation when the source file’s content hash matches the last successful compile (see subroutine_needs_recompile()).

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

Compile the subroutine, or skip if precompiled/cached.

Returns¶

tuple[bool, str]: (success, message) — message is empty on success (or a skip note), or the compiler’s raw stdout+stderr on failure. No regex parsing of compiler errors is performed (see run_compile()).

Parameters:

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

Return type:

tuple[bool, str]

Parameters:

subroutine (SubroutineSpec)
cache (bool)

class ABQflow.core.strategies.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.core.strategies.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.core.strategies.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.core.strategies.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.core.strategies.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.core.strategies.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.

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]

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]

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: