1"""Shared helpers: settings, paths, logging, deterministic seeds."""2from __future__ import annotations3import hashlib, json, logging, os, random, sys, time4from pathlib import Path5import numpy as np6import yaml78ROOT = Path(__file__).resolve().parents[1]9SETTINGS = yaml.safe_load((ROOT / "config" / "settings.yaml").read_text())10SEED = int(SETTINGS["seed"])11TZ = SETTINGS.get("timezone_display", "America/Toronto")1213def path(key: str) -> Path:14 p = ROOT / SETTINGS["paths"][key]15 p.mkdir(parents=True, exist_ok=True)16 return p1718def snapshot_db() -> Path:19 return ROOT / SETTINGS["snapshot"]["path"]2021def verify_snapshot() -> str:22 h = hashlib.sha256()23 with open(snapshot_db(), "rb") as f:24 for chunk in iter(lambda: f.read(1 << 20), b""):25 h.update(chunk)26 digest = h.hexdigest()27 if digest != SETTINGS["snapshot"]["sha256"]:28 raise SystemExit(f"snapshot sha256 mismatch: {digest}")29 return digest3031def seed_everything(seed: int = SEED) -> None:32 random.seed(seed); np.random.seed(seed)33 os.environ["PYTHONHASHSEED"] = str(seed)3435def get_logger(name: str) -> logging.Logger:36 log = logging.getLogger(name)37 if not log.handlers:38 fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s", "%Y-%m-%dT%H:%M:%S")39 sh = logging.StreamHandler(sys.stdout); sh.setFormatter(fmt); log.addHandler(sh)40 fh = logging.FileHandler(path("logs") / "pipeline.log"); fh.setFormatter(fmt); log.addHandler(fh)41 log.setLevel(logging.INFO)42 return log4344def ledger(phase: str, action: str, result: str) -> None:45 """Append a machine-readable line to the structured run ledger."""46 rec = {"utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "phase": phase, "action": action, "result": result}47 with open(path("logs") / "run_ledger.jsonl", "a") as f:48 f.write(json.dumps(rec, ensure_ascii=False) + "\n")4950def write_manifest(name: str, files: list[Path], extra: dict | None = None) -> None:51 man = {"name": name, "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "seed": SEED,52 "snapshot": SETTINGS["snapshot"]["id"], "files": {}}53 for f in files:54 f = Path(f)55 if f.exists():56 man["files"][str(f.relative_to(ROOT))] = {"bytes": f.stat().st_size,57 "sha256": hashlib.sha256(f.read_bytes()).hexdigest() if f.stat().st_size < 400_000_000 else None}58 if extra: man.update(extra)59 (path("logs") / f"manifest_{name}.json").write_text(json.dumps(man, indent=2, ensure_ascii=False))60