"""Run discovery and manifest handling for the FHOPS web service.
The service reads the artefacts the CLI writes. This module classifies a directory into one of the
three planning lenses (operational, hauling, tactical-operational), collects the known files, and
optionally writes a stable ``run.json`` manifest. It never invents a file or a KPI.
"""
from __future__ import annotations
import json
import re
from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
__all__ = [
"DOMAINS",
"RUN_MANIFEST_SCHEMA_VERSION",
"RunRecord",
"build_run_manifest",
"classify_run",
"discover_runs",
"load_run_manifest",
"write_run_manifest",
]
RUN_MANIFEST_SCHEMA_VERSION = "1.0.0"
DOMAINS = ("operational", "hauling", "tactical_operational")
_MANIFEST_NAME = "run.json"
_SKIP_DIRS = {".git", ".hg", ".svn", "__pycache__", "node_modules", ".mypy_cache", ".ruff_cache"}
# Logical artefact name -> candidate filenames/globs. The first match wins.
_KNOWN_FILES: dict[str, tuple[str, ...]] = {
# Operational
"assignments": ("assignments.csv", "*_solution.csv", "mip_solution.csv", "sa_solution.csv"),
"summary": ("summary.json",),
"telemetry": ("telemetry.jsonl", "telemetry-*.jsonl"),
"kpis": ("kpis.json",),
"day_summary": ("day_summary.csv",),
"shift_summary": ("shift_summary.csv",),
"day_parquet": ("day_summary.parquet",),
"shift_parquet": ("shift_summary.parquet",),
"rolling_iterations": ("iterations.csv", "iterations.jsonl"),
# Hauling
"haul_flows": ("haul_flows.csv",),
"haul_truckloads": ("haul_truckloads.csv",),
"haul_yard_inventory": ("haul_yard_inventory.csv",),
"haul_block_inventory": ("haul_block_inventory.csv",),
"haul_contractor_workload": ("haul_contractor_workload.csv",),
"haul_schedule": ("haul_schedule.csv",),
"haul_truck_summary": ("haul_truck_summary.csv",),
"haul_loader_summary": ("haul_loader_summary.csv",),
"haul_kpis": ("haul_kpis.json",),
"haul_summary": ("haul_summary.json",),
# Tactical-operational
"harvest_decisions": ("harvest_decisions.csv",),
"production": ("production.csv",),
"flows": ("flows.csv",),
"purchases": ("purchases.csv",),
"inventory": ("inventory.csv",),
"roads": ("roads.csv",),
"silviculture": ("silviculture.csv",),
"fleet": ("fleet.csv",),
"tactical_summary": ("tactical_summary.json",),
}
[docs]
@dataclass(slots=True)
class RunRecord:
"""One discovered run directory.
Attributes
----------
run_id:
Stable identifier used in API paths.
path:
Absolute run directory.
domain:
``operational``, ``hauling``, or ``tactical_operational`` (``None`` when unclassified).
run_type:
Best-effort run kind, such as ``solve_heur``, ``haul_solve``, or ``tactical_operational``.
status:
Manifest status when present, otherwise ``unknown``.
manifest:
Parsed ``run.json`` when present.
files:
Logical artefact name -> path relative to the run directory.
warnings:
Discovery/classification warnings.
"""
run_id: str
path: Path
domain: str | None
run_type: str | None
status: str
manifest: dict[str, Any] | None = None
files: dict[str, str] = field(default_factory=dict)
warnings: list[str] = field(default_factory=list)
[docs]
def to_dict(self) -> dict[str, Any]:
"""Return a JSON-ready payload."""
return {
"run_id": self.run_id,
"path": str(self.path),
"domain": self.domain,
"run_type": self.run_type,
"status": self.status,
"manifest": self.manifest,
"files": self.files,
"warnings": self.warnings,
}
[docs]
def discover_runs(roots: Sequence[Path], *, max_depth: int = 2) -> list[RunRecord]:
"""Discover run directories under one or more roots.
Parameters
----------
roots:
Directories to scan. A root may itself be a run directory.
max_depth:
Maximum child depth below each root. ``1`` scans immediate children, ``2`` grandchildren.
Returns
-------
list[RunRecord]
Sorted by ``run_id``. Directories with no known artefact are skipped.
"""
discovered: dict[Path, RunRecord] = {}
for root in roots:
root = root.expanduser().resolve()
if not root.exists():
continue
if _is_run_dir(root):
record = _record_for(root, root)
discovered[record.path] = record
continue
for candidate in _walk_candidates(root, max_depth=max_depth):
if _is_run_dir(candidate):
record = _record_for(candidate, root)
discovered[record.path] = record
return sorted(discovered.values(), key=lambda record: record.run_id)
def _walk_candidates(root: Path, *, max_depth: int) -> Iterable[Path]:
yield from _walk(root, depth=1, max_depth=max_depth)
def _walk(path: Path, *, depth: int, max_depth: int) -> Iterable[Path]:
if depth > max_depth:
return
try:
children = sorted(path.iterdir(), key=lambda item: item.name)
except OSError:
return
for child in children:
if not child.is_dir() or child.name in _SKIP_DIRS or child.name.startswith("."):
continue
yield child
yield from _walk(child, depth=depth + 1, max_depth=max_depth)
def _is_run_dir(path: Path) -> bool:
return (path / _MANIFEST_NAME).exists() or bool(_scan_files(path))
def _record_for(run_dir: Path, root: Path) -> RunRecord:
manifest, warnings = _read_manifest(run_dir)
files = _scan_files(run_dir)
warnings = list(warnings)
domain = manifest.get("domain") if manifest else None
run_type = manifest.get("run_type") if manifest else None
if domain is None or run_type is None:
classified_domain, classified_type, classification_warnings = classify_run(files)
domain = domain or classified_domain
run_type = run_type or classified_type
warnings.extend(classification_warnings)
if domain not in DOMAINS and domain is not None:
warnings.append(f"Unknown domain in run.json: {domain!r}")
run_id = _run_id(run_dir, root, manifest)
return RunRecord(
run_id=run_id,
path=run_dir,
domain=domain,
run_type=run_type,
status=(
str(manifest.get("status", "unknown")) if manifest else _infer_status(run_dir, files)
),
manifest=manifest,
files=files,
warnings=warnings,
)
def _run_id(run_dir: Path, root: Path, manifest: dict[str, Any] | None) -> str:
if manifest and manifest.get("run_id"):
return str(manifest["run_id"])
try:
relative = run_dir.relative_to(root)
text = str(relative).replace("\\", "/").strip("/")
except ValueError:
text = run_dir.name
if not text or text == ".":
text = run_dir.name
slug = re.sub(r"[^A-Za-z0-9._-]+", "-", text).strip("-")
return slug or "run"
def _scan_files(run_dir: Path) -> dict[str, str]:
files: dict[str, str] = {}
for logical, patterns in _KNOWN_FILES.items():
for pattern in patterns:
matches = sorted(run_dir.glob(pattern))
if matches:
files[logical] = str(matches[0].relative_to(run_dir))
break
return files
[docs]
def classify_run(files: dict[str, str]) -> tuple[str | None, str | None, list[str]]:
"""Classify a run from its logical file names.
Parameters
----------
files:
Logical artefact names from :func:`_scan_files` (or a manifest's ``files`` map).
Returns
-------
tuple[str | None, str | None, list[str]]
``(domain, run_type, warnings)``. Ambiguous combinations are reported rather than guessed.
"""
names = set(files)
warnings: list[str] = []
hauling = names & {
"haul_flows",
"haul_truckloads",
"haul_schedule",
"haul_summary",
"haul_kpis",
}
tactical = names & {
"harvest_decisions",
"production",
"flows",
"inventory",
"roads",
"silviculture",
"fleet",
}
operational = names & {
"assignments",
"telemetry",
"day_summary",
"shift_summary",
"rolling_iterations",
}
domains = [
name
for name, present in (
("hauling", hauling),
("tactical_operational", tactical),
("operational", operational),
)
if present
]
if len(domains) > 1:
warnings.append(f"Run contains artefacts from multiple domains: {', '.join(domains)}")
domain = domains[0] if domains else None
if domain == "hauling":
if "haul_schedule" in names:
run_type = "haul_solve" if "haul_flows" in names else "haul_schedule"
elif "haul_flows" in names:
run_type = "haul_plan"
else:
run_type = "hauling"
elif domain == "tactical_operational":
run_type = "tactical_operational"
elif domain == "operational":
if "rolling_iterations" in names:
run_type = "plan_rolling"
elif "assignments" in names:
run_type = "solve"
elif "day_summary" in names or "shift_summary" in names:
run_type = "eval_playback"
else:
run_type = "operational"
else:
run_type = None
if names:
warnings.append("Known files did not classify into a planning domain")
return domain, run_type, warnings
[docs]
def load_run_manifest(run_dir: Path) -> dict[str, Any] | None:
"""Load ``run.json`` from a run directory, or ``None`` when absent.
Raises
------
ValueError
When the manifest is not a JSON object.
"""
path = run_dir / _MANIFEST_NAME
if not path.exists():
return None
payload = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise ValueError(f"{path} must contain a JSON object")
return payload
def _read_manifest(run_dir: Path) -> tuple[dict[str, Any] | None, list[str]]:
try:
return load_run_manifest(run_dir), []
except (OSError, ValueError) as exc:
return None, [f"Ignored invalid run.json: {exc}"]
[docs]
def build_run_manifest(
run_dir: Path,
*,
domain: str | None = None,
run_type: str | None = None,
command: str | None = None,
scenario: str | None = None,
solver: str | None = None,
seed: int | None = None,
status: str | None = None,
run_id: str | None = None,
write: bool = True,
) -> dict[str, Any]:
"""Build (and optionally write) a ``run.json`` manifest for a directory.
Parameters
----------
run_dir:
Existing run directory.
domain / run_type:
Explicit overrides. When omitted, the manifest is classified from the files.
command / scenario / solver / seed / status:
Provenance fields. Existing values are preserved unless overridden.
run_id:
Explicit run id. Defaults to the existing manifest value or the directory name/path slug.
write:
When ``True``, write ``run.json`` and return the written payload.
Returns
-------
dict[str, Any]
Manifest payload.
"""
run_dir = run_dir.expanduser().resolve()
if not run_dir.exists():
raise FileNotFoundError(run_dir)
existing, manifest_warnings = _read_manifest(run_dir)
files = _scan_files(run_dir)
classified_domain, classified_type, warnings = classify_run(files)
warnings.extend(manifest_warnings)
payload: dict[str, Any] = dict(existing or {})
payload.update(
{
"schema_version": RUN_MANIFEST_SCHEMA_VERSION,
"run_id": run_id or payload.get("run_id") or _run_id(run_dir, run_dir, None),
"domain": domain or payload.get("domain") or classified_domain,
"run_type": run_type or payload.get("run_type") or classified_type,
"status": status or payload.get("status") or _infer_status(run_dir, files),
"files": files,
"warnings": warnings,
"generated_at": _now(),
}
)
for key, value in (
("command", command),
("scenario", scenario),
("solver", solver),
("seed", seed),
):
if value is not None:
payload[key] = value
if write:
(run_dir / _MANIFEST_NAME).write_text(
json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8"
)
return payload
[docs]
def write_run_manifest(run_dir: Path, **kwargs: Any) -> dict[str, Any]:
"""Write a ``run.json`` manifest and return it. See :func:`build_run_manifest`."""
kwargs["write"] = True
return build_run_manifest(run_dir, **kwargs)
def _infer_status(run_dir: Path, files: dict[str, str]) -> str:
"""Read a status field from a summary artefact when no manifest exists."""
for logical in ("haul_summary", "tactical_summary", "summary"):
filename = files.get(logical)
if not filename:
continue
try:
payload = json.loads((run_dir / filename).read_text(encoding="utf-8"))
except (OSError, ValueError):
continue
if isinstance(payload, dict) and payload.get("status"):
return str(payload["status"])
return "unknown"
def _now() -> str:
"""Return a UTC timestamp."""
return datetime.now(UTC).isoformat()