Jobsystem 

axqua.jobs.paths.index_path()[Quellcode]

The SQLite job index. Derived - deleting it is safe, list --rebuild restores.

Rückgabetyp:: Path

axqua.jobs.paths.profiles_path()[Quellcode]

The solver-profile file. Machine-local by design: it names install paths.

Rückgabetyp:: Path

class axqua.jobs.paths.JobDir(root)[Quellcode]

Bases: object

Every path inside one job directory, in one place.

The per-job directory is authoritative (plan §5); the SQLite index is derived from it. So this class is the schema of a job on disk, and anything that needs a path asks here rather than joining strings.

Parameter:: root (str | os.PathLike)

property job_id: str

property spec: Path: job.json - the immutable submission record.

property status: Path: status.json - mutable, replaced atomically, runner-only.

property lock: Path

property cancel_request: Path: Its own file, so the one-writer-per-file rule survives cancellation.

property runner_log: Path: axqua’s narration. The solver’s own listing is a separate artifact.

property launch_env: Path: The captured solver environment, as systemd’s EnvironmentFile.

property answers: Path: Answers to questions a study asked while detached.

property input: Path

property frozen_config: Path

property preprocessing: Path

property simulation: Path

property solver_logs: Path

property calibration: Path

property results: Path

property results_json: Path

property qgis: Path: Where SolverBackend.export_qgis_results() writes. Never PyQGIS output.

exists()[Quellcode]

Rückgabetyp:: bool

create()[Quellcode]

Claim the directory. Raises FileExistsError if the id is taken.

The claim is the collision check - mkdir(exist_ok=False) is atomic, so two concurrent submits cannot both win the same id no matter how the tails were drawn.

Rückgabetyp:: JobDir

Reading and writing the two job files, safely.

Two properties matter more than anything else here, because QGIS may be polling status.json at 1 Hz while the runner rewrites it:

Writes are atomic. Temporary file, fsync, os.replace. Not os.rename - that fails on Windows when the destination exists, and a „write then rename“ that silently falls back to „unlink then rename“ has a window where the file does not exist at all. os.replace is atomic on both platforms.

Reads tolerate a torn moment. Even with an atomic replace, a reader on a network filesystem can observe an incomplete file, and the correct response is to look again rather than to report the job as broken. So the reader retries a few times over a few tens of milliseconds before giving up.

Schema versioning follows plan §5 exactly: older migrates in memory and is never rewritten on read (a read must not mutate a record the user may still want to inspect); equal proceeds; newer is refused with a message naming both versions, because running a job description you do not fully understand is how a job that looks submitted quietly runs something else.

axqua.jobs.store.read_json_tolerant(path, *, retries=4, delay=0.05)[Quellcode]

Read a JSON file, retrying through a concurrent replace.

Returns None when the file is absent - an absent status.json is a normal state (the job directory exists but nothing has run yet), not an error.

Parameter:

path (str | PathLike)
retries (int)
delay (float)

Rückgabetyp:

dict[str, Any] | None

axqua.jobs.store.read_spec(job_dir)[Quellcode]

Parameter:: job_dir (JobDir | str | PathLike)
Rückgabetyp:: JobSpec

axqua.jobs.store.read_status(job_dir)[Quellcode]

The last complete status, or None if nothing has been written yet.

Parameter:: job_dir (JobDir | str | PathLike)
Rückgabetyp:: JobStatus | None

axqua.jobs.store.write_json_atomic(path, payload, *, fsync=True)[Quellcode]

Replace path with payload atomically. Returns the path.

Parameter:

path (str | PathLike)
payload (dict[str, Any])
fsync (bool)

Rückgabetyp:

axqua.jobs.store.write_spec(job_dir, spec, *, freeze=True)[Quellcode]

Write job.json once, then make it read-only.

The read-only bit is not security - it is a statement. job.json is the record of what was asked for; anything that wants to change it wants a new job.

Parameter:

job_dir (JobDir | str | PathLike)
spec (JobSpec)
freeze (bool)

Rückgabetyp:

axqua.jobs.store.write_status(job_dir, status)[Quellcode]

Replace status.json.

Callers must hold job.lock; that is enforced by convention and by the executor’s structure rather than by a check here, because reaper.reconcile legitimately writes after proving the previous owner is gone.

Parameter:

job_dir (JobDir | str | PathLike)
status (JobStatus)

Rückgabetyp:

job.lock: one writer at a time, and a way back from a crash.

Two QGIS windows, or a CLI and QGIS, can act on the same job at once (plan §5). The lock gives status.json exactly one writer. What makes it more than a file is what happens when the holder dies: a lock nobody can break is worse than no lock, because a job that crashed at 3 a.m. could then never be resubmitted.

So the record carries enough to judge staleness without guessing, and the checks run in order of confidence:

A different host is never broken automatically. On a shared job root the PID means nothing here, so the honest answer is „cannot tell“, and --force is a human decision.
A different boot id is stale: the machine restarted, so every PID it recorded is meaningless. This is also the wsl --shutdown case.
A dead PID is stale.
A live PID with a different start time is stale - PID reuse. Without this check a long-running job’s recycled PID makes an unrelated process look like the lock holder, and the lock is never broken.

Readers never lock (plan §5); they read the last complete status.json.

class axqua.jobs.lock.JobLock(job_dir)[Quellcode]

Bases: object

An advisory lock on one job directory.

Parameter:: job_dir (JobDir | str | os.PathLike)

owner()[Quellcode]

Rückgabetyp:: LockRecord | None

is_stale()[Quellcode]

(stale, reason). A missing lock is not stale - it is absent.

Rückgabetyp:: tuple[bool, str]

acquire(*, purpose='execute', break_stale=True, force=False)[Quellcode]

Parameter:

purpose (str)
break_stale (bool)
force (bool)

Rückgabetyp:

JobLock

release()[Quellcode]

Rückgabetyp:: None

property held: bool

holding(*, purpose='execute', break_stale=True, force=False)[Quellcode]

Parameter:

purpose (str)
break_stale (bool)
force (bool)

Rückgabetyp:

Iterator[JobLock]

class axqua.jobs.lock.LockRecord(pid: 'int', started: 'str', purpose: 'str' = 'execute', host: 'str' = '', boot_id: 'str | None' = None, proc_started: 'float | None' = None, extra: 'dict[str, Any]'=<factory>)[Quellcode]

Bases: object

Parameter:

pid (int)
started (str)
purpose (str)
host (str)
boot_id (str | None)
proc_started (float | None)
extra (dict[str, Any])

pid: int

started: str

purpose: str = 'execute'

host: str = ''

boot_id: str | None = None

proc_started: float | None = None

extra: dict[str, Any]

as_dict()[Quellcode]

Rückgabetyp:: dict[str, Any]

classmethod from_dict(data)[Quellcode]

Parameter:: data (dict[str, Any])
Rückgabetyp:: LockRecord

classmethod for_self(purpose)[Quellcode]

Parameter:: purpose (str)
Rückgabetyp:: LockRecord

exception axqua.jobs.lock.LockedError(message, *, subject=None, remedy=None, **details)[Quellcode]

Bases: ConfigError

Someone else holds the lock, and it is not stale.

Parameter:

message (str)
subject (str | None)
remedy (str | None)
details (Any)

Rückgabetyp:

None

code = 'axqua.job.locked'

A SQLite index over the job root - derived, never authoritative.

Listing hundreds of jobs by opening hundreds of status.json files is slow enough to be felt in a dashboard, so there is an index. But the index is a cache of the directories, and the directories win every disagreement (plan §5). Two consequences, both deliberate:

rebuild() reconstructs the whole thing by scanning, and it is a tested path rather than a repair tool nobody exercises. That covers the first crash mid-write, a restored backup, and a job directory the user moved by hand.
Every index failure is swallowed. A job must never fail because its bookkeeping could not be written; that is what makes plan §24’s „a solver failure must not corrupt the job registry“ true by construction rather than by care. Callers that cannot open the database fall back to a live directory scan.

WAL journalling, so a QGIS reader polling the list never blocks the runner writing it.

class axqua.jobs.index.JobRow(job_id, root, kind=None, solver=None, case_name=None, source_config=None, state=None, created=None, updated=None, finished=None, schema_version=None, indexed=None)[Quellcode]

Bases: object

One row, flattened for a table view. Never carries solver field data (plan §26).

Parameter:

job_id (str)
root (str)
kind (str | None)
solver (str | None)
case_name (str | None)
source_config (str | None)
state (str | None)
created (str | None)
updated (str | None)
finished (str | None)
schema_version (int | None)
indexed (str | None)

job_id: str

root: str

kind: str | None = None

solver: str | None = None

case_name: str | None = None

source_config: str | None = None

state: str | None = None

created: str | None = None

updated: str | None = None

finished: str | None = None

schema_version: int | None = None

indexed: str | None = None

property job_dir: JobDir

as_dict()[Quellcode]

Rückgabetyp:: dict[str, Any]

class axqua.jobs.index.JobIndex(db_path=None)[Quellcode]

Bases: object

The index. Constructing one never raises; using one never raises.

Parameter:: db_path (str | os.PathLike | None)

property usable: bool: False when the database could not be opened - callers then scan instead.

upsert(row)[Quellcode]

Parameter:: row (JobRow)
Rückgabetyp:: None

remove(job_id)[Quellcode]

Parameter:: job_id (str)
Rückgabetyp:: None

get(job_id)[Quellcode]

Parameter:: job_id (str)
Rückgabetyp:: JobRow | None

list(*, state=None, kind=None, solver=None, case=None, since=None, limit=None)[Quellcode]

Parameter:

state (str | None)
kind (str | None)
solver (str | None)
case (str | None)
since (str | None)
limit (int | None)

Rückgabetyp:

list[JobRow]

rebuild(job_root=None, *, reconcile=True)[Quellcode]

Rebuild from a directory scan. Returns how many jobs were found.

This is the authority restoring the cache, so it clears first: an index row for a directory the user deleted is exactly the kind of disagreement the scan exists to settle.

Parameter:

job_root (str | PathLike | None)
reconcile (bool)

Rückgabetyp:

int

axqua.jobs.index.row_for(job_dir, *, reconcile=False)[Quellcode]

Read one job directory into a row, or None if it is not a job.

Never raises: a half-written or hand-mangled directory should drop out of a listing, not take the listing down with it.

Parameter:

job_dir (JobDir | str | PathLike)
reconcile (bool)

Rückgabetyp:

JobRow | None

axqua.jobs.index.scan(job_root=None, *, reconcile=False)[Quellcode]

Yield every job directory under job_root, newest first.

The directory name must look like a job id, so unrelated folders under a shared scratch root are skipped rather than misread.

Parameter:

job_root (str | PathLike | None)
reconcile (bool)

Rückgabetyp:

Iterator[JobRow]

axqua.jobs.index.list_jobs(*, job_root=None, index=None, rebuild=False, **filters)[Quellcode]

List jobs, preferring the index and falling back to a scan.

The fallback is not a corner case: a fresh checkout, a read-only home, or a corrupt database all land here, and in every one of them a slower correct answer beats an error.

Parameter:

job_root (str | PathLike | None)
index (JobIndex | None)
rebuild (bool)
filters (Any)

Rückgabetyp:

list[JobRow]

results/results.json: what a job produced, as small structured metadata.

The rule this file exists to enforce is plan §26: the runner and the plugin communicate through job metadata, status metadata, file paths and small structured results - never through solver field arrays. A TELEMAC result is hundreds of megabytes and a reconstructed OpenFOAM case is gigabytes; serializing any of that into JSON, or copying it so the plugin can find it, would be the single easiest way to make the system unusable.

So a result entry is a path plus a hint. The plugin reads the manifest, decides what to add to the layer tree, and lets QGIS/MDAL open the file itself.

The manifest is written by SolverBackend.export_qgis_results(), which is also why that method takes no PyQGIS: it produces files and describes them, and the plugin does the loading (plan §14).

class axqua.jobs.results.ResultEntry(name, path, kind='file', style='', variable='', description='', extra=<factory>)[Quellcode]

Bases: object

One loadable artifact.

Parameter:

name (str)
path (str)
kind (str)
style (str)
variable (str)
description (str)
extra (dict[str, Any])

name: str

path: str

kind: str = 'file'

style: str = ''

variable: str = ''

description: str = ''

extra: dict[str, Any]

as_dict()[Quellcode]

Rückgabetyp:: dict[str, Any]

classmethod from_dict(data)[Quellcode]

Parameter:: data (dict[str, Any])
Rückgabetyp:: ResultEntry

class axqua.jobs.results.ResultManifest(job_id='', solver='', kind='', generated=<factory>, schema_version=1, objective=None, entries=<factory>, summary=<factory>, crs_epsg=None)[Quellcode]

Bases: object

Everything one job produced that is worth opening.

Parameter:

job_id (str)
solver (str)
kind (str)
generated (str)
schema_version (int)
objective (float | None)
entries (list[ResultEntry])
summary (dict[str, Any])
crs_epsg (int | None)

job_id: str = ''

solver: str = ''

kind: str = ''

generated: str

schema_version: int = 1

objective: float | None = None

entries: list[ResultEntry]

summary: dict[str, Any]

crs_epsg: int | None = None: The coordinate reference system of the case. A SELAFIN file carries none, so a result opened in QGIS had no CRS at all and did not line up with a base map unless the project happened to use the same one.

add(name, path, *, kind='file', style='', variable='', description='', base=None, **extra)[Quellcode]

Record one artifact, skipping it if the file is not there.

Silently skipping is right here: backends list everything they might have produced, and a manifest entry pointing at a missing file would make the plugin show a broken layer rather than simply not showing one.

Parameter:

name (str)
path (str | PathLike)
kind (str)
style (str)
variable (str)
description (str)
base (Path | None)
extra (Any)

Rückgabetyp:

ResultManifest

as_dict()[Quellcode]

Rückgabetyp:: dict[str, Any]

classmethod from_dict(data)[Quellcode]

Parameter:: data (dict[str, Any])
Rückgabetyp:: ResultManifest

axqua.jobs.results.read_manifest(job_dir)[Quellcode]

Parameter:: job_dir (JobDir | str | PathLike)
Rückgabetyp:: ResultManifest | None

axqua.jobs.results.write_manifest(job_dir, manifest)[Quellcode]

Parameter:

job_dir (JobDir | str | PathLike)
manifest (ResultManifest)

Rückgabetyp:

Prozesse und Staatenabstimmung

Is that process still there? - the primitives, without psutil.

A detached job records a PID. Some time later, possibly after a reboot, something has to decide whether that PID still means what it meant. Three facts are needed and none of them is a plain os.kill(pid, 0):

PID reuse. On Linux, PIDs wrap; on a busy box a multi-day job’s PID can be reassigned to something unrelated. os.kill(pid, 0) then says „alive“ about a process that is not the job. So the start time is recorded alongside the PID and compared - the pair is what identifies a process, not the number alone.
Reboot. After a restart every recorded PID is meaningless. The boot id makes that a single comparison rather than a heuristic, and it is also the wsl --shutdown case: the distro’s VM restarted and every job in it is gone.
Another machine. A job root on a shared filesystem may hold jobs launched elsewhere. Their PIDs say nothing here, so the host is recorded and a foreign job is reported rather than judged.

psutil would supply all three, but the dependency policy (plan §28) is to stay light, and this is ~100 lines of standard library.

axqua.jobs.procs.boot_id()[Quellcode]

An identifier that changes on every boot, or None if unobtainable.

Linux exposes one directly. Elsewhere it is derived from the boot instant, rounded hard: the point is only to differ across restarts, and monotonic-clock jitter must not make two readings on the same boot disagree.

Rückgabetyp:: str | None

axqua.jobs.procs.host_id()[Quellcode]

A stable-enough machine name. Used only to refuse to judge a foreign job.

Rückgabetyp:: str

axqua.jobs.procs.pid_alive(pid)[Quellcode]

True when pid names a live process. Never raises.

A zombie counts as dead, and getting this wrong is not academic. os.kill(pid, 0) succeeds against a zombie, because the process table entry still exists until the parent reaps it. So a job whose runner has exited but whose parent has not yet called wait reads as RUNNING - which makes cancel report that it failed to stop something that already stopped, and leaves the dashboard waiting forever. In production the launcher’s parent exits and init reaps the child, so the window is short; in any long-lived parent (a test runner, a plugin process that submitted) it is not.

Parameter:: pid (int | None)
Rückgabetyp:: bool

axqua.jobs.procs.process_start_time(pid)[Quellcode]

When pid started, in whatever unit the platform offers.

The value is only ever compared with another reading from the same platform, so the unit does not need to be portable - only stable. None means „cannot tell“, which callers must treat as „do not conclude anything“, not as „changed“.

Parameter:: pid (int | None)
Rückgabetyp:: float | None

axqua.jobs.procs.self_start_time()[Quellcode]

Rückgabetyp:: float | None

Reconciling a job’s recorded state with reality.

A detached job can die in ways it cannot report: the machine reboots, wsl --shutdown takes the distro with every job in it, an OOM killer removes the runner, someone runs kill -9. In all of those the last thing written to status.json says RUNNING, and it will keep saying so forever.

A job stuck in RUNNING is worse than a job marked FAILED: the dashboard keeps polling it, the user keeps waiting for it, and nothing ever prompts a resubmission. So every read path that shows state to a human goes through here first, and one implementation covers the POSIX fallback, systemd, WSL and Windows because they all reduce to the same question - is the recorded process still the process we launched?

The one thing this must not do is guess about another machine. On a shared job root a PID from another host says nothing here, so such a job is reported unchanged.

axqua.jobs.reaper.reconcile(job_dir, *, write=True)[Quellcode]

Return the job’s true state, correcting status.json when it is wrong.

write exists for read-only callers (a listing over a mounted archive) that still want the corrected answer in memory.

Parameter:

job_dir (JobDir | str | PathLike)
write (bool)

Rückgabetyp:

JobStatus | None

SolverProfile

Solver profiles: how to reach TELEMAC or OpenFOAM on this machine.

A profile is deliberately machine-local and lives in the user’s config directory, never in the project file. It names install paths, core counts and scratch volumes - exactly the things that differ between the workstation, the cluster and a colleague’s laptop. Putting them in a shared .axqua-prj would destroy its portability the first time it was committed (plan §3, §6).

The schema is the plan’s, verbatim, so the example in §6 parses unchanged:

profiles:
  telemac_linux:
    solver: telemac
    environment: posix
    setup_script: /opt/telemac/v9.1/configs/pysource.sh
    mpi_launcher: mpirun
    mpi_processes: 16
    working_root: /scratch/axqua

  openfoam_windows:            # Foundation OpenFOAM: WSL is the route
    solver: openfoam
    environment: wsl
    distro: Ubuntu-22.04
    setup_script: /opt/openfoam9/etc/bashrc     # a path INSIDE the distro
    working_root: /scratch/axqua            # inside the distro

validate() is the pre-submission check plan §6 asks for, and it is SolverEnvironment.validate() unchanged - including its provenance warning about sentinels that came from the ambient shell rather than the setup script. That warning is the „works in my terminal, fails in the runner“ trap, and a detached job is precisely where it bites.

class axqua.jobs.profiles.Profile(name, solver='telemac', environment='posix', setup_script=None, shell=None, distro=None, config_name=None, mpi_launcher=None, mpi_processes=1, working_root=None, launcher='auto', postprocess=<factory>, overrides=<factory>)[Quellcode]

Bases: object

One way of reaching one solver.

Parameter:

name (str)
solver (str)
environment (str)
setup_script (Path | None)
shell (str | None)
distro (str | None)
config_name (str | None)
mpi_launcher (str | None)
mpi_processes (int)
working_root (Path | None)
launcher (str)
postprocess (list[str])
overrides (dict[str, str])

name: str

solver: str = 'telemac'

environment: str = 'posix'

setup_script: Path | None = None

shell: str | None = None

distro: str | None = None

config_name: str | None = None

mpi_launcher: str | None = None

mpi_processes: int = 1

working_root: Path | None = None

launcher: str = 'auto'

postprocess: list[str]

overrides: dict[str, str]

classmethod from_dict(name, data, *, base=None)[Quellcode]

Parameter:

name (str)
data (dict[str, Any])
base (Path | None)

Rückgabetyp:

Profile

as_dict()[Quellcode]

Rückgabetyp:: dict[str, Any]

solver_environment()[Quellcode]

Rückgabetyp:: SolverEnvironment

validate(*, timeout=120.0)[Quellcode]

Can this profile actually reach its solver? Never raises.

Parameter:: timeout (float)
Rückgabetyp:: EnvStatus

describe()[Quellcode]

Rückgabetyp:: str

axqua.jobs.profiles.from_plugin_profile(plugin, solver='telemac', cfg=None)[Quellcode]

The job-system view of one binding of an *.axq-profile.

None when the profile does not bind solver, so the caller can fall back. The case’s own process count fills in where the binding names none: the profile says what the computer has, and a binding without a number has no opinion.

Parameter:

plugin (Any)
solver (str)
cfg (Any)

Rückgabetyp:

Profile | None

axqua.jobs.profiles.implicit(cfg, solver='telemac')[Quellcode]

The profile a job uses when none is named.

The active *.axq-profile where it binds solver, otherwise whatever the case config resolved to (from_config()). The second half is what keeps a computer without a profile working exactly as before.

The environment variable of the code and a solvers.local.yml beside the case outrank the profile here as they do when the case is loaded (axqua.core.machine.resolve()); a job would otherwise compute with another installation than the one its case was built with. The profile then still says how this computer enters an environment and how many processes it runs. The script is the one that outranks it, and what the profile holds for its own installation (the systel configuration, the variables) is not applied to another one.

Parameter:

cfg (Any)
solver (str)

Rückgabetyp:

Profile

axqua.jobs.profiles.load_profiles(path=None)[Quellcode]

Read profiles.yml. A missing file is an empty mapping, not an error.

A fresh install has no profiles and must still be able to run axqua profiles to be told so.

Parameter:: path (str | PathLike | None)
Rückgabetyp:: dict[str, Profile]

axqua.jobs.profiles.resolve(name, *, solver=None, cfg=None, path=None)[Quellcode]

Find the profile to use, or synthesize one from the case config.

The fallback matters: every existing case already carries telemac.pysource (or openfoam.bashrc), so a user who has never written a profiles.yml can still submit a job. The profile system is an addition, not a new prerequisite.

Parameter:

name (str | None)
solver (str | None)
cfg (Any)
path (str | PathLike | None)

Rückgabetyp:

Profile

axqua.jobs.profiles.save_profiles(profiles, path=None)[Quellcode]

Write profiles.yml, keeping the profiles: wrapper.

Parameter:

profiles (dict[str, Profile])
path (str | PathLike | None)

Rückgabetyp:

Fortschrittsereignisse und Protokolle

Turning a stream of solver lines into job state.

axqua’s existing progress reporting is console-shaped: on_line callbacks that draw a bar on a TTY. A detached job has no TTY and needs the same information as structured state in status.json, so that QGIS never has to parse a human-readable log (plan §11, explicitly).

The wrong fix would be a second parser. SolverProgress already extracts the iteration and the simulated time from TELEMAC’s listing header, and OpenFoamProgress already extracts the Courant number and the time step from interFoam’s - both hard-won against real output. So this module adds a sink those classes also emit to, and everything composes:

ConsoleSink is today’s behavior, unchanged, for terminal runs;
StatusFileSink writes status.json;
TeeSink does both, which is what axqua execute wants - a debugging run should look exactly like the old scripts and leave a machine-readable trail.

StatusFileSink is throttled. Without it a TELEMAC listing at a few headers a second produces several fsync-ed rewrites a second on a scratch volume, which is both wasteful and a reliable way to make a network filesystem unhappy. State and step changes are never throttled - they are what the dashboard is waiting for.

class axqua.jobs.events.ConsoleSink(*, echo=True)[Quellcode]

Bases: NullSink

Print. This is what a terminal run has always done.

Parameter:: echo (bool)

line(text)[Quellcode]

Parameter:: text (str)
Rückgabetyp:: None

step(name, *, done=False, seconds=None, failed=False)[Quellcode]

Parameter:

name (str)
done (bool)
seconds (float | None)
failed (bool)

Rückgabetyp:

None

class axqua.jobs.events.EventSink(*args, **kwargs)[Quellcode]

Bases: Protocol

What a long-running operation reports.

Five verbs, because five different things happen and collapsing them into one log(str) is exactly what forces a GUI to parse prose.

line(text)[Quellcode]

One raw line of solver output.

Parameter:: text (str)
Rückgabetyp:: None

progress(**fields)[Quellcode]

Typed progress fields for the current job kind.

Parameter:: fields (Any)
Rückgabetyp:: None

step(name, *, done=False, seconds=None, failed=False)[Quellcode]

A named phase started, finished or failed.

Parameter:

name (str)
done (bool)
seconds (float | None)
failed (bool)

Rückgabetyp:

None

artifact(kind, path)[Quellcode]

A file worth recording in the job’s artifact map.

Parameter:

kind (str)
path (Path | str)

Rückgabetyp:

None

decision(question, answer, source)[Quellcode]

A question the job answered on the user’s behalf, and why.

Parameter:

question (str)
answer (bool)
source (str)

Rückgabetyp:

None

class axqua.jobs.events.NullSink[Quellcode]

Bases: object

Discards everything. The default, so nothing has to check for None.

line(text)[Quellcode]

Parameter:: text (str)
Rückgabetyp:: None

progress(**fields)[Quellcode]

Parameter:: fields (Any)
Rückgabetyp:: None

step(name, *, done=False, seconds=None, failed=False)[Quellcode]

Parameter:

name (str)
done (bool)
seconds (float | None)
failed (bool)

Rückgabetyp:

None

artifact(kind, path)[Quellcode]

Parameter:

kind (str)
path (Path | str)

Rückgabetyp:

None

decision(question, answer, source)[Quellcode]

Parameter:

question (str)
answer (bool)
source (str)

Rückgabetyp:

None

class axqua.jobs.events.StatusFileSink(job_dir, status, *, min_interval=2.0)[Quellcode]

Bases: NullSink

Fold events into status.json.

Holds the JobStatus in memory and replaces the file; the executor owns the lock for the duration of the run, so there is exactly one writer.

Parameter:

job_dir (Any)
status (JobStatus)
min_interval (float)

progress(**fields)[Quellcode]

Parameter:: fields (Any)
Rückgabetyp:: None

step(name, *, done=False, seconds=None, failed=False)[Quellcode]

Parameter:

name (str)
done (bool)
seconds (float | None)
failed (bool)

Rückgabetyp:

None

artifact(kind, path)[Quellcode]

Parameter:

kind (str)
path (Path | str)

Rückgabetyp:

None

decision(question, answer, source)[Quellcode]

Parameter:

question (str)
answer (bool)
source (str)

Rückgabetyp:

None

pending(question)[Quellcode]

Record a question the job is about to answer, so the dashboard can show it.

Parameter:: question (str | None)
Rückgabetyp:: None

state_changed()[Quellcode]

Write the status now, because its state was just changed.

The executor assigns a transition on the status object itself, so this sink still considers itself clean and flush() skips the write. The new state then reached the file only when a progress event happened to follow. A steady run reports progress within seconds; a calibration reports none, and stood in the dashboard as STARTING for as long as it ran.

Rückgabetyp:: None

flush()[Quellcode]

Rückgabetyp:: None

close()[Quellcode]

Rückgabetyp:: None

class axqua.jobs.events.TeeSink(*sinks)[Quellcode]

Bases: NullSink

Fan out to several sinks. A failing sink never stops the others.

Parameter:: sinks (Any)

line(text)[Quellcode]

Parameter:: text (str)
Rückgabetyp:: None

progress(**fields)[Quellcode]

Parameter:: fields (Any)
Rückgabetyp:: None

step(name, *, done=False, seconds=None, failed=False)[Quellcode]

Parameter:

name (str)
done (bool)
seconds (float | None)
failed (bool)

Rückgabetyp:

None

artifact(kind, path)[Quellcode]

Parameter:

kind (str)
path (Path | str)

Rückgabetyp:

None

decision(question, answer, source)[Quellcode]

Parameter:

question (str)
answer (bool)
source (str)

Rückgabetyp:

None

flush()[Quellcode]

Rückgabetyp:: None

close()[Quellcode]

Rückgabetyp:: None

class axqua.jobs.events.TextLogSink(writer)[Quellcode]

Bases: NullSink

Append raw lines to a rotating text file - the solver’s own narration.

Deliberately separate from runner.log (plan §23): axqua’s narration is state transitions and exit codes, while this is the solver talking. Mixing them makes both harder to read and makes the rotation policy wrong for at least one of them.

Parameter:: writer (Any)

line(text)[Quellcode]

Parameter:: text (str)
Rückgabetyp:: None

step(name, *, done=False, seconds=None, failed=False)[Quellcode]

Parameter:

name (str)
done (bool)
seconds (float | None)
failed (bool)

Rückgabetyp:

None

axqua.jobs.events.current_sink: ContextVar = <ContextVar name='axqua_event_sink' default=<axqua.jobs.events.NullSink object>>: The ambient sink, so log_step can report phases without every one of its ~40 call sites growing a parameter. A ContextVar rather than a module global because a convergence study runs nested steps and must not leak its sink to a sibling.

axqua.jobs.events.using_sink(sink)[Quellcode]

Parameter:: sink (Any)
Rückgabetyp:: Iterator[Any]

Job logging: bounded, and split by who is talking.

Two separate streams, because they answer different questions and want different policies (plan §23):

runner.log: axqua’s own narration - state transitions, the commands it issued, exit codes, log_step timings. Small, and the thing to read first when a job fails.
solver/<name>.log: The solver talking. TELEMAC prints a block per time step and interFoam at ~1e-3 s steps over hours produces hundreds of thousands of lines; a multi-day run will otherwise fill the scratch volume and take itself down.

Both rotate, and rotation keeps the last parts. That is worth stating because the naive alternative - stop writing at N bytes - keeps the first, and when a run fails after two days the beginning is precisely the part nobody needs. logging.handlers.RotatingFileHandler already behaves correctly here: the live file is always the newest and .1 the most recent rollover.

The disk guard is here too. „Insufficient disk space where detectable“ (plan §24) is six lines and turns a mid-run death with a corrupt half-written result into a refusal at submit time carrying a remedy.

class axqua.jobs.logs.RotatingTextSink(path, *, max_bytes=134217728, backups=3)[Quellcode]

Bases: object

Append raw solver output to a size-bounded file.

Built on RotatingFileHandler rather than hand-rolled, so the rollover semantics are the standard library’s and the last parts survive. Line-buffered writes: a job that is killed must still have its most recent output on disk.

Parameter:

path (str | os.PathLike)
max_bytes (int)
backups (int)

write(line)[Quellcode]

Parameter:: line (str)
Rückgabetyp:: None

close()[Quellcode]

Rückgabetyp:: None

axqua.jobs.logs.free_bytes(path)[Quellcode]

Free space on the volume holding path, or None if it cannot be told.

Parameter:: path (str | PathLike)
Rückgabetyp:: int | None

axqua.jobs.logs.require_free_space(path, minimum=5368709120)[Quellcode]

Refuse to start when the volume is already too full to finish.

Parameter:

path (str | PathLike)
minimum (int)

Rückgabetyp:

None

axqua.jobs.logs.runner_log(path, *, level=10, console=False, max_bytes=33554432, backups=5)[Quellcode]

Route the axqua logger into a rotating runner.log for the duration.

Scoped rather than global because a job runs inside a process that may also be doing something else (axqua execute in a terminal), and the handler must come off cleanly afterwards.

Parameter:

path (str | PathLike)
level (int)
console (bool)
max_bytes (int)
backups (int)

Rückgabetyp:

Iterator[Path]

Answering, in a detached job, the questions an interactive study asks.

convergence.run_mesh_convergence prompts twice: whether to reuse completed levels on a re-run, and whether to refine one level further when the study has not converged. Its default handler reads stdin with a timeout and, off a TTY, returns the default - so a detached job does not hang, which is the important part.

But „does not hang“ is not the same as „does the right thing“: the default for extend is no, so a study submitted specifically to march finer would quietly stop at its first ladder. The fix is not to make the prompt smarter; it is to make the answer part of the submission. PolicyAsk answers from job.json, never blocks, and records every question and answer in runner.log so the decision is auditable rather than invisible.

An answer file is also honored, which is what lets a future axqua answer JOB_ID --yes reach a running study without restarting it. The file format is fixed here even though the CLI verb is not built yet, so adding it later needs no migration.

class axqua.jobs.interaction.PolicyAsk(answers=None, *, sink=None, answer_file=None, default=False)[Quellcode]

Bases: object

A drop-in replacement for the study’s ask callable.

Signature-compatible with convergence._default_ask (prompt, keyword default, and a tolerated timeout), so it substitutes without the study knowing.

Parameter:

answers (dict[str, bool] | None)
sink (Any)
answer_file (Path | str | None)
default (bool)

axqua.jobs.interaction.classify(prompt)[Quellcode]

Map a prompt to a policy key, or "other".

Parameter:: prompt (str)
Rückgabetyp:: str

Einreichung und Ausführung

Turning a case configuration into a job, and starting it detached.

Submission does three things, in an order chosen so that a failure never leaves a half-started job behind:

Freeze. The resolved configuration is written into the job directory, both as the record in job.json and as a loadable input/case-config.yml. From that moment the job is reproducible from its own directory and immune to later edits of the case - which is the whole reason job.json stores the configuration rather than a path to it (plan §3).
Validate. The solver environment is probed before anything is launched, because „the setup script is wrong“ discovered at submit time is a message, while the same fact discovered inside a detached process is an archaeology exercise.
Launch, and only then record the handle.

The environment is captured here, once, and handed to the launcher as a dict - the first real use of SolverEnvironment.capture(). That is what keeps a shell out of the job’s process tree.

axqua.jobs.submit.build_spec(cfg, kind, *, profile=None, job_id=None, options=None, np=None, workspace=None, source_job=None, launcher='auto', source_config=None)[Quellcode]

Assemble the immutable submission record. Writes nothing.

Parameter:

cfg (Any)
kind (str | JobKind)
profile (Profile | None)
job_id (str | None)
options (dict[str, Any] | None)
np (int | None)
workspace (str | None)
source_job (str | None)
launcher (str)
source_config (str | PathLike | None)

Rückgabetyp:

JobSpec

axqua.jobs.submit.create_job(cfg, spec, *, job_root=None)[Quellcode]

Claim the directory and freeze the inputs. Nothing is launched.

Parameter:

cfg (Any)
spec (JobSpec)
job_root (str | PathLike | None)

Rückgabetyp:

JobDir

axqua.jobs.submit.submit_job(cfg, kind, *, profile=None, options=None, job_root=None, np=None, workspace=None, source_job=None, launcher='auto', source_config=None, validate_env=True, dry_run=False)[Quellcode]

Freeze, validate, launch. Returns the JobDir, or the spec on a dry run.

Parameter:

cfg (Any)
kind (str | JobKind)
profile (Profile | None)
options (dict[str, Any] | None)
job_root (str | PathLike | None)
np (int | None)
workspace (str | None)
source_job (str | None)
launcher (str)
source_config (str | PathLike | None)
validate_env (bool)
dry_run (bool)

Rückgabetyp:

JobSpec | JobDir

Running one job: dispatch, state, and turning any failure into a record.

This is the only module in axqua.jobs that touches a solver, and it touches it through exactly one seam - registry.get(solver).load(). There is no if solver == "telemac" here and there must never be (plan §12): the kind table says which verb to call, the backend decides what that verb means, and adding a third solver is a pip install, not an edit to this file.

The shape of a run:

lock -> runner.log -> load the frozen config -> rebase the workspace
     -> disk check -> STARTING -> RUNNING -> backend.<verb>(...)
     -> POSTPROCESSING -> backend.postprocess / export_qgis_results
     -> results.json -> COMPLETED

Every failure becomes a ErrorRecord in status.json and an ERROR line in runner.log. Plan §24 requires both, and the reason is practical: the record is what a GUI can act on, the log is what a human can read.

Cancellation is cooperative first. cancel writes a request file; the token below is checked between steps and inside the streaming callback, so the job tears itself down and writes CANCELLED on its own. The launcher’s tree-kill is the fallback for a job that is too busy or too stuck to notice.

class axqua.jobs.executor.CancelToken(job_dir, *, interval=1.0)[Quellcode]

Bases: object

Cooperative cancellation, driven by a file.

A file rather than a signal because the request comes from a different process (possibly a different QGIS session, possibly days later), and because it preserves the one-writer-per-file rule: cancel never touches status.json.

Parameter:

job_dir (JobDir)
interval (float)

property requested: bool

check()[Quellcode]

Rückgabetyp:: None

class axqua.jobs.executor.ExecutionContext(job_dir, spec, cfg, sink=<factory>, cancel=None, manifest=None, log=<Logger axqua.jobs.executor (WARNING)>, _ask=None, _open_sinks=<factory>)[Quellcode]

Bases: object

Everything a backend needs that is not the case configuration.

Passed as ctx= rather than as loose keyword arguments so a backend can grow a new need without every caller changing, and so the standalone case scripts - which pass no context at all - keep working against the same backend methods.

Parameter:

job_dir (JobDir)
spec (JobSpec)
cfg (Any)
sink (Any)
cancel (CancelToken | None)
manifest (ResultManifest | None)
log (Logger)
_ask (Any)
_open_sinks (list[Any])

job_dir: JobDir

spec: JobSpec

cfg: Any

sink: Any

cancel: CancelToken | None = None

manifest: ResultManifest | None = None

log: Logger = <Logger axqua.jobs.executor (WARNING)>

property capability: Capability | None

property options: Any

should_stop()[Quellcode]

Rückgabetyp:: bool

check_cancelled()[Quellcode]

Rückgabetyp:: None

ask(prompt, *, default=False, timeout=None)[Quellcode]

Answer a study’s question from the job record. Never reads stdin.

Parameter:

prompt (str)
default (bool)
timeout (float | None)

Rückgabetyp:

progress(**fields)[Quellcode]

Parameter:: fields (Any)
Rückgabetyp:: None

solver_sink(name)[Quellcode]

A rotating text sink for one solver invocation’s raw output.

Parameter:: name (str)
Rückgabetyp:: Any

record(name, path, **kwargs)[Quellcode]

Add an artifact to the result manifest.

Parameter:

name (str)
path (str | PathLike)
kwargs (Any)

Rückgabetyp:

None

close()[Quellcode]

Rückgabetyp:: None

exception axqua.jobs.executor.JobCancelled[Quellcode]

Bases: Exception

Raised inside the job when a cancellation request is noticed.

axqua.jobs.executor.execute(job_dir, *, sink=None, console=False, break_stale_lock=True)[Quellcode]

Run the job in job_dir to completion. Returns a process exit code.

Synchronous by design - this is what axqua execute calls, and what every detached launcher ultimately starts. Keeping the detached path and the debugging path identical means a failure reproduces under a terminal.

Parameter:

job_dir (JobDir | str | PathLike)
sink (Any)
console (bool)
break_stale_lock (bool)

Rückgabetyp:

int

Abgestellte Trägerraketen

Starting a job that outlives whatever started it.

This is the interface plan §8 asks for, and the reason it is an interface rather than one implementation is that „detached“ means four genuinely different things:

systemd user services on Linux, where the kernel already provides the thing this module wants - a named unit whose whole cgroup can be queried, killed and logged as a unit. MPI ranks belong to it automatically.
A new POSIX session everywhere else, where the process group is the unit.
A Windows Job Object, because Windows has no process groups in the POSIX sense and without a Job Object there is no reliable tree-kill at all.
WSL, where the job must be detached inside the distro and the Windows side acts on it only by running commands through wsl.exe.

Two rules cut across all four:

The environment goes on the process, not into a wrapper shell. Every launcher takes an already-captured environment dict (SolverEnvironment.capture()) and sets it directly. A bash -lc 'source ...; axqua execute' in the middle would add a layer between the unit and the ranks, and each such layer is somewhere a signal can fail to propagate.

A handle must survive the launching process. Everything needed to query or cancel the job later - unit name, pid, pgid, job-object name, the recording host and boot id - is serialized into status.json. A fresh axqua cancel days later reconstructs the launcher from that alone.

class axqua.jobs.launcher.JobLauncher(*args, **kwargs)[Quellcode]

Bases: Protocol

name: str

classmethod available(env=None)[Quellcode]

Can this launcher be used on this machine, right now?

Parameter:: env (SolverEnvironment | None)
Rückgabetyp:: bool

submit(job_dir, argv, *, env, cwd)[Quellcode]

Parameter:

job_dir (Any)
argv (list[str])
env (dict[str, str])
cwd (Path | str)

Rückgabetyp:

status(handle)[Quellcode]

Parameter:: handle (LaunchHandle)
Rückgabetyp:: LaunchState

cancel(handle, *, grace=20.0)[Quellcode]

Parameter:

handle (LaunchHandle)
grace (float)

Rückgabetyp:

logs(handle)[Quellcode]

Parameter:: handle (LaunchHandle)
Rückgabetyp:: Path | None

class axqua.jobs.launcher.LaunchHandle(backend, pid=None, pgid=None, proc_started=None, unit=None, job_object=None, distro=None, host=<factory>, boot_id=<factory>, started=<factory>, log_path=None)[Quellcode]

Bases: object

Everything needed to find the job again from a different process.

Parameter:

backend (str)
pid (int | None)
pgid (int | None)
proc_started (float | None)
unit (str | None)
job_object (str | None)
distro (str | None)
host (str)
boot_id (str | None)
started (str)
log_path (str | None)

backend: str

pid: int | None = None

pgid: int | None = None

proc_started: float | None = None

unit: str | None = None

job_object: str | None = None

distro: str | None = None

host: str

boot_id: str | None

started: str

log_path: str | None = None

as_dict()[Quellcode]

Rückgabetyp:: dict[str, Any]

classmethod from_dict(data)[Quellcode]

Parameter:: data (dict[str, Any])
Rückgabetyp:: LaunchHandle

class axqua.jobs.launcher.LaunchState(*values)[Quellcode]

Bases: str, Enum

What the launcher can say about a handle.

UNKNOWN is a real answer and is treated as „still running“ by the reaper. Reporting a job dead because its supervisor was momentarily unreachable would be worse than saying nothing.

RUNNING = 'running'

GONE = 'gone'

UNKNOWN = 'unknown'

axqua.jobs.launcher.available_launchers(env=None)[Quellcode]

Which launchers this machine can actually use. For axqua profiles.

Parameter:: env (SolverEnvironment | None)
Rückgabetyp:: list[str]

axqua.jobs.launcher.launcher_for(handle)[Quellcode]

Reconstruct the launcher that started a recorded job.

By name, from the handle - not by re-running the selection. A job launched under systemd must be cancelled through systemd even if the user manager has since become unavailable and selection would now choose the POSIX fallback.

Parameter:: handle (LaunchHandle)
Rückgabetyp:: JobLauncher

axqua.jobs.launcher.select_launcher(env=None, *, prefer=None)[Quellcode]

Pick a launcher, first hit wins.

The WSL rule is a forced match rather than a preference: a captured Linux environment cannot start a Windows process, so a WSL solver has exactly one way to be launched and letting a preference override it would produce a job that cannot run.

Parameter:

env (SolverEnvironment | None)
prefer (str | None)

Rückgabetyp:

JobLauncher

systemd user services: the preferred Linux launcher.

A transient user unit gives, for free, all four things plan §8 asks for: the solver survives the shell that started it, its state can be queried, its entire process tree can be terminated as a unit, and its logs are retrievable. MPI ranks belong to the unit’s cgroup automatically, which is what makes cancellation reliable rather than best-effort - a cgroup kill cannot miss a descendant the way signalling a process group can if a child deliberately changes its own group.

Three details are load-bearing:

--collect: The unit name is deterministic (axqua-<job_id>.service), which is what lets a later process find it. Without --collect a failed unit lingers in that name and the same job could never be resubmitted.
EnvironmentFile: Rather than N x --setenv. A captured OpenFOAM environment is large enough to exceed ARG_MAX, and the failure is an opaque „Argument list too long“ at submit time.
KillMode=control-group: The default already kills the cgroup, but stating it makes the guarantee explicit rather than inherited.

available() deliberately checks three things. A bare SSH session without lingering has systemd-run on PATH and no user manager at all, so testing for the binary alone is a false positive that produces a job which fails the moment it is submitted.

axqua.jobs.launchers.systemd.unit_name(job_id)[Quellcode]

Parameter:: job_id (str)
Rückgabetyp:: str

class axqua.jobs.launchers.systemd.SystemdUserLauncher[Quellcode]

Bases: object

systemd-run --user transient units.

name = 'systemd'

classmethod available(env=None)[Quellcode]

Parameter:: env (Any)
Rückgabetyp:: bool

submit(job_dir, argv, *, env, cwd)[Quellcode]

Parameter:

job_dir (Any)
argv (list[str])
env (dict[str, str])
cwd (Path | str)

Rückgabetyp:

status(handle)[Quellcode]

Parameter:: handle (LaunchHandle)
Rückgabetyp:: LaunchState

cancel(handle, *, grace=20.0)[Quellcode]

Stop the unit - which stops the whole cgroup, MPI ranks included.

Parameter:

handle (LaunchHandle)
grace (float)

Rückgabetyp:

logs(handle)[Quellcode]

The unit’s captured stdout.

A file rather than journalctl: the plugin needs a path it can tail, and the unit is configured to append to one, so the journal is the fallback for a human rather than the interface.

Parameter:: handle (LaunchHandle)
Rückgabetyp:: Path | None

The POSIX fallback: a new session, and a process group to kill.

Used wherever systemd’s user manager is not available - a container, a machine without lingering enabled, macOS. Less capable than a systemd unit (no supervision, no restart, no journal) but sufficient for the one thing that matters: the job survives the shell that started it, and it can still be terminated as a whole.

start_new_session=True is the entire trick. It calls setsid, which makes the child a session and process-group leader, so its own children - TELEMAC’s launcher, mpirun, the ranks - inherit that process group. killpg then reaches all of them. Without it the child joins the submitter’s process group, and cancelling would either miss the ranks or kill the terminal that launched them.

The parent never waits. It returns immediately, exits, and the child is reparented to init; that is what „detached“ means here (plan §8).

class axqua.jobs.launchers.posix.PosixDetachedLauncher[Quellcode]

Bases: object

Popen(..., start_new_session=True) plus a recorded process group.

name = 'posix'

classmethod available(env=None)[Quellcode]

Parameter:: env (Any)
Rückgabetyp:: bool

submit(job_dir, argv, *, env, cwd)[Quellcode]

Parameter:

job_dir (Any)
argv (list[str])
env (dict[str, str])
cwd (Path | str)

Rückgabetyp:

status(handle)[Quellcode]

Parameter:: handle (LaunchHandle)
Rückgabetyp:: LaunchState

cancel(handle, *, grace=20.0)[Quellcode]

SIGTERM the group, then SIGKILL what is left.

The group, never the pid: the ranks are grandchildren, and signalling the leader alone leaves orphan mpirun processes running at full tilt (plan §9).

Parameter:

handle (LaunchHandle)
grace (float)

Rückgabetyp:

logs(handle)[Quellcode]

Parameter:: handle (LaunchHandle)
Rückgabetyp:: Path | None

Windows: detached processes inside a named Job Object.

Windows has no process groups in the POSIX sense, so killpg has no equivalent and the naive proc.terminate() kills only the process you named - leaving every MPI rank it started running at full tilt. A Job Object is the actual equivalent: a kernel object that owns a set of processes and can terminate all of them at once.

Two flags decide whether this works, and the plan’s sketch gets one of them subtly wrong:

JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE must be cleared: Otherwise the job dies when the last handle to the object closes - i.e. when axqua submit exits, which is the opposite of detached.
The object must be named: An unnamed Job Object is reachable only through the handle that created it, and that handle dies with the submitting process. A later axqua cancel, in a different process, then has nothing to open - so cancellation would be impossible, which is exactly the failure the Job Object was introduced to prevent. Hence Local\axqua-<job_id> plus OpenJobObjectW at cancel time.

Never executed against a real Windows install. The argv and flag construction are unit-tested because they are host-independent; the kernel32 calls are not, and this module says so rather than implying a coverage it does not have.

axqua.jobs.launchers.windows.job_object_name(job_id)[Quellcode]

Session-local, so two users on one machine cannot collide.

Parameter:: job_id (str)
Rückgabetyp:: str

class axqua.jobs.launchers.windows.WindowsJobObjectLauncher[Quellcode]

Bases: object

Detached CreateProcess assigned to a named Job Object.

name = 'windows'

classmethod available(env=None)[Quellcode]

Parameter:: env (Any)
Rückgabetyp:: bool

submit(job_dir, argv, *, env, cwd)[Quellcode]

Parameter:

job_dir (Any)
argv (list[str])
env (dict[str, str])
cwd (Path | str)

Rückgabetyp:

status(handle)[Quellcode]

Parameter:: handle (LaunchHandle)
Rückgabetyp:: LaunchState

cancel(handle, *, grace=20.0)[Quellcode]

TerminateJobObject on the named object, with taskkill /T as fallback.

Parameter:

handle (LaunchHandle)
grace (float)

Rückgabetyp: