"""Shared configuration, paths and logging for the WP39 pipeline.""" from __future__ import annotations import logging import os from dataclasses import dataclass from pathlib import Path import yaml ROOT = Path(__file__).resolve().parents[2] @dataclass(frozen=True) class Paths: root: Path configs: Path artifacts: Path manifests: Path logs: Path evaluation: Path models: Path processed: Path results: Path figures: Path paper_tables: Path paper_figures: Path cache: Path @classmethod def default(cls) -> "Paths": cache = Path(os.path.expanduser(os.environ.get("WP39_CACHE", "~/wp39-cache"))) p = cls(root=ROOT, configs=ROOT / "configs", artifacts=ROOT / "artifacts", manifests=ROOT / "artifacts/manifests", logs=ROOT / "artifacts/logs", evaluation=ROOT / "artifacts/evaluation", models=ROOT / "artifacts/models", processed=ROOT / "data/processed", results=ROOT / "results", figures=ROOT / "figures", paper_tables=ROOT / "paper/tables", paper_figures=ROOT / "paper/figures", cache=cache) for d in (p.manifests, p.logs, p.evaluation, p.models, p.processed, p.results, p.figures, p.paper_tables, p.paper_figures, p.cache / "bars", p.cache / "fundamentals", p.cache / "features", p.cache / "external"): d.mkdir(parents=True, exist_ok=True) return p def load_config(name: str) -> dict: with open(ROOT / "configs" / f"{name}.yaml") as fh: return yaml.safe_load(fh) def get_logger(name: str, logfile: Path | None = None) -> logging.Logger: log = logging.getLogger(name) if log.handlers: return log log.setLevel(logging.INFO) fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") h = logging.StreamHandler() h.setFormatter(fmt) log.addHandler(h) if logfile: logfile.parent.mkdir(parents=True, exist_ok=True) fh = logging.FileHandler(logfile) fh.setFormatter(fmt) log.addHandler(fh) return log