eng-lab/hwe/__main__.py

"""The engine's command line. Run from the eng-lab folder with the venv's Python.

    python -m hwe report                 run every unlocked analysis, write out/report/index.html
    python -m hwe watch                  rerun the report whenever a design, data, or code file changes
    python -m hwe tinker KEY=VALUE ... --predict "what you expect"
                                         change knobs WITHOUT editing the design file, compare
                                         against the baseline, and log your prediction beside the result
    python -m hwe explain N "why"        add your explanation to tinker entry N
    python -m hwe status                 which stages run, which wait for your code, which are locked

The tinker loop (EL01, section 1): predict, run, explain. `tinker` refuses to
run without --predict, because a prediction written after seeing the answer
teaches nothing.
"""
from __future__ import annotations

import argparse
import csv
import datetime as dt
import sys
import time
from pathlib import Path

from hwe.design import ROOT, parse_override

LOG = ROOT / "logs" / "tinker.csv"
LOG_FIELDS = ["id", "time", "change", "prediction", "result", "explanation"]
WATCHED = [ROOT / "design", ROOT / "data", ROOT / "hwe"]
TRACKED = ["mass.predicted", "mass.allowable", "mass.margin", "hover.thrust_to_weight", "hover.hover_fraction",
           "hover.max_climb_accel", "energy.battery_energy", "energy.hover_power", "energy.flight_time", "arm.mass"]


def _utf8():
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(encoding="utf-8", errors="replace")
        except (AttributeError, ValueError):
            pass


def _read_log() -> list[dict]:
    if not LOG.exists():
        return []
    with open(LOG, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))


def _write_log(rows: list[dict]) -> None:
    LOG.parent.mkdir(parents=True, exist_ok=True)
    with open(LOG, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=LOG_FIELDS)
        writer.writeheader()
        writer.writerows(rows)


def _print_summary(run) -> None:
    from hwe.units import fmt
    print(f"\n{run.design.raw.get('meta', {}).get('name', 'design')}")
    for key in TRACKED:
        if key in run.results:
            print(f"  {key:28s} {fmt(run.results[key], 4)}")
    bad = [c for c in run.cross_checks if not c["ok"]]
    print(f"  cross-checks: {len(run.cross_checks) - len(bad)} of {len(run.cross_checks)} agree")
    for s in run.requirement_status:
        print(f"  {s.requirement.id:10s} {s.state}")
    for stage, status in run.stages.items():
        if status not in ("ok", "locked"):
            print(f"  [{stage}] {status}")


def cmd_report(args) -> int:
    from hwe import pipeline
    from hwe.report import html
    run = pipeline.run()
    page = html.write(run, _read_log())
    _print_summary(run)
    print(f"\nreport: {page}")
    if args.open:
        import webbrowser
        webbrowser.open(page.as_uri())
    return 0


def _snapshot() -> dict[Path, float]:
    stamps = {}
    for folder in WATCHED:
        for path in folder.rglob("*"):
            if path.is_file() and "__pycache__" not in path.parts:
                stamps[path] = path.stat().st_mtime
    return stamps


def cmd_watch(args) -> int:
    import importlib
    print("watching design/, data/, and hwe/ (Ctrl+C to stop)")
    last = None
    while True:
        now = _snapshot()
        if now != last:
            if last is not None:
                changed = sorted(p.relative_to(ROOT) for p in now if now.get(p) != last.get(p))
                print(f"\nchanged: {', '.join(map(str, changed[:5]))}")
            for name in [m for m in sys.modules if m.startswith("hwe.") and m != "hwe.__main__"]:
                importlib.reload(sys.modules[name])       # pick up edited code without restarting
            try:
                cmd_report(argparse.Namespace(open=False))
            except Exception as e:                           # a half-typed file should not kill the watch
                print(f"report failed: {type(e).__name__}: {e}")
            last = now
        time.sleep(1.0)


def cmd_tinker(args) -> int:
    from hwe import pipeline
    from hwe.units import fmt
    if not args.predict or not args.predict.strip():
        print('Write your prediction first, for example:\n  python -m hwe tinker "battery.capacity=8000 mA*h" '
              '--predict "flight time up, but less than the 60 percent the capacity grew: the pack gets heavier. T/W down"')
        return 2
    overrides = dict(parse_override(o) for o in args.changes)
    base = pipeline.run()
    new = pipeline.run(overrides)
    lines, parts = [], []
    print(f"\n{'result':28s} {'before':>14s} {'after':>14s} {'change':>9s}")
    for key in TRACKED:
        a, b = base.results.get(key), new.results.get(key)
        if a is None or b is None:
            continue
        rel = ((b - a) / a).to("dimensionless").m if a.m else 0.0
        print(f"{key:28s} {fmt(a, 4):>14s} {fmt(b, 4):>14s} {rel * 100:+8.1f}%")
        if abs(rel) > 1e-12:
            parts.append(f"{key} {fmt(a, 3)} -> {fmt(b, 3)} ({rel * 100:+.1f}%)")
    for s_base, s_new in zip(base.requirement_status, new.requirement_status):
        if s_base.state != s_new.state:
            parts.append(f"{s_new.requirement.id} {s_base.state} -> {s_new.state}")
            print(f"{s_new.requirement.id}: {s_base.state} -> {s_new.state}")
    for name, state in new.stages.items():                  # a stage that stopped running is a result too
        if base.stages.get(name) != state:
            parts.append(f"stage {name}: {state}")
            print(f"stage {name}: {base.stages.get(name)} -> {state}")
    agreed = {c["name"]: c["ok"] for c in base.cross_checks}
    for c in new.cross_checks:
        if not c["ok"] and agreed.get(c["name"], True):
            parts.append(f"cross-check now disagrees: {c['name']}")
            print(f"cross-check now disagrees: {c['name']}")
    rows = _read_log()
    entry_id = str(len(rows) + 1)
    rows.append(dict(id=entry_id, time=dt.datetime.now().isoformat(timespec="seconds"),
                     change="; ".join(f"{k}={v}" for k, v in overrides.items()), prediction=args.predict.strip(),
                     result="; ".join(parts) or "no tracked result changed", explanation=""))
    _write_log(rows)
    print(f"\nlogged as entry {entry_id} in {LOG.relative_to(ROOT)}. Now explain the result in your own words:")
    print(f'  python -m hwe explain {entry_id} "..."')
    return 0


def cmd_explain(args) -> int:
    rows = _read_log()
    for row in rows:
        if row["id"] == str(args.id):
            row["explanation"] = args.text
            _write_log(rows)
            print(f"entry {args.id} explained.")
            return 0
    print(f"no tinker entry {args.id}")
    return 1


def cmd_status(args) -> int:
    from hwe import pipeline
    run = pipeline.run()
    print(f"unlocked modules: {', '.join(run.unlocked)}")
    for stage, module, what in pipeline.STAGES:
        print(f"  {module}  {stage:13s} {run.stages.get(stage, '')}")
    return 0


def main(argv=None) -> int:
    _utf8()
    parser = argparse.ArgumentParser(prog="python -m hwe", description=__doc__.split("\n\n")[0])
    sub = parser.add_subparsers(dest="command", required=True)
    p = sub.add_parser("report"); p.add_argument("--open", action="store_true"); p.set_defaults(func=cmd_report)
    p = sub.add_parser("watch"); p.set_defaults(func=cmd_watch)
    p = sub.add_parser("tinker"); p.add_argument("changes", nargs="+"); p.add_argument("--predict", default="")
    p.set_defaults(func=cmd_tinker)
    p = sub.add_parser("explain"); p.add_argument("id"); p.add_argument("text"); p.set_defaults(func=cmd_explain)
    p = sub.add_parser("status"); p.set_defaults(func=cmd_status)
    args = parser.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())