neu_morpho.run_records

A work dir’s run records: one directory per run, and a current view built from them.

<work_dir>/runs/<run id>/run_plan.json written when the run starts <work_dir>/runs/<run id>/run_summary.json written when it ends, completed or aborted <work_dir>/run_plan.json, run_summary.json the current view

A run’s own files are written once and never merged, so every run sharing a work dir (neu-morpho run --stages index,mesh, neu-skel run, a retry) stays on record in full. The view holds, for each stage, the latest run that executed it, naming that run, and lists every run. A run with a plan and no summary is still going, or was killed.

Imports only the standard library: neu-morpho progress reads through it.

Attributes

PLAN
RUNS
SUMMARY
UNFINISHED

Functions

new_run_id(work_dir, started, stages) → str	<YYYYmmddTHHMMSS.mmm>-<stages>, suffixed -2, -3 … if taken.
planned_blocks(plan_view) → dict[str, Any]
rebuild_view(work_dir, dst) → tuple[dict, dict]	Rewrite the work dir's run_plan.json and run_summary.json from its runs to
runs(work_dir) → list[tuple[str, dict, dict | None]]	(run id, plan, summary or None) for every recorded run, oldest first.
seg_scales(plan_view) → list
unfinished_stages(summary_view) → list[str]	Stages whose latest run has written no summary.
write_run_plan(work_dir, run_id, record) → None
write_run_summary(work_dir, run_id, record) → None

Module Contents

neu_morpho.run_records.new_run_id(work_dir: str, started: datetime.datetime, stages: Iterable[str]) → str

<YYYYmmddTHHMMSS.mmm>-<stages>, suffixed -2, -3 … if taken.

Ids sort in start order, which is the order the view applies runs in.

neu_morpho.run_records.planned_blocks(plan_view: dict) → dict[str, Any]

neu_morpho.run_records.rebuild_view(work_dir: str, dst: str) → tuple[dict, dict]

Rewrite the work dir’s run_plan.json and run_summary.json from its runs to dst; returns (plan view, summary view).

neu_morpho.run_records.runs(work_dir: str) → list[tuple[str, dict, dict | None]]

(run id, plan, summary or None) for every recorded run, oldest first.

neu_morpho.run_records.seg_scales(plan_view: dict) → list

neu_morpho.run_records.unfinished_stages(summary_view: dict | None) → list[str]

Stages whose latest run has written no summary.

neu_morpho.run_records.write_run_plan(work_dir: str, run_id: str, record: dict) → None

neu_morpho.run_records.write_run_summary(work_dir: str, run_id: str, record: dict) → None

neu_morpho.run_records.PLAN = 'run_plan.json'

neu_morpho.run_records.RUNS = 'runs'

neu_morpho.run_records.SUMMARY = 'run_summary.json'

neu_morpho.run_records.UNFINISHED = 'unfinished'