runner_utils¶

runner_utils ¶

PyAERMOD runner UX helpers.

Additions on top of runner.py / BatchRunner:

extract_errmsg / tail_output / summarize_failure: pull useful diagnostics from AERMOD's ERRMSG.TMP and .OUT files when a run fails.
ProgressReporter protocol with TqdmProgress (if tqdm is installed) and LoggingProgress / NoOpProgress fallbacks.
resume_batch: given a list of input files and an output dir, return which already have valid .out files and which still need to run.
RunManifest: JSON-backed batch state for resume / inspection.
generate_slurm_script: produce a SLURM job-array template for a directory of .inp files.

ERRMSGInfo `dataclass` ¶

Parsed contents of AERMOD's ERRMSG.TMP (or equivalent).

NoOpProgress ¶

Silent progress reporter.

LoggingProgress ¶

Progress reporter that emits INFO-level log lines.

TqdmProgress ¶

Progress reporter using tqdm. Only works if tqdm is installed.

RunManifest `dataclass` ¶

Tracks a batch's per-run state in a JSON file.

Use-cases: - Persist partial batch progress across restarts - Post-hoc inspection of which inputs succeeded / failed

extract_errmsg ¶

extract_errmsg(path: Union[str, Path]) -> Optional[ERRMSGInfo]

Parse an AERMOD ERRMSG.TMP-style file.

Returns None if the file doesn't exist.

tail_output ¶

tail_output(path: Union[str, Path], n_lines: int = 40) -> List[str]

Return the last n_lines of a file as a list of strings.

summarize_failure ¶

summarize_failure(input_file: Union[str, Path], working_dir: Union[str, Path]) -> str

Return a human-readable failure summary.

Gathers ERRMSG.TMP content plus the tail of the .OUT file.

resume_batch ¶

resume_batch(input_files: Sequence[Union[str, Path]], output_dir: Union[str, Path]) -> Dict[str, List[Path]]

Partition input_files into 'done' and 'todo' lists.

An input is 'done' if a sibling .out in output_dir has the AERMOD success marker.

generate_slurm_script ¶

generate_slurm_script(input_files: Sequence[Union[str, Path]], output_dir: Union[str, Path], script_path: Union[str, Path], input_list_path: Union[str, Path], *, aermod_exe: str = 'aermod', job_name: str = 'pyaermod', log_dir: str = 'logs', partition: str = 'general', cpus: int = 1, mem: str = '4G', wallclock: str = '02:00:00', max_concurrent: Optional[int] = None) -> Path

Write a SLURM job-array script + input-list file for a batch.

Returns the path to the generated script.