Spaces:
Running
Running
File size: 4,397 Bytes
b7c46f4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """
Command line for the intel store (run from backend/ with INTEL_DB_PATH set):
py -m intel.cli init create tables, seed sources/streams
py -m intel.cli locations build the state/LGA gazetteer
py -m intel.cli calendar holidays + seasons, 2000 → +10 years
py -m intel.cli places load the Overture extract (run overture_extract first)
py -m intel.cli weather-history [--budget N] [--start YYYY-MM-DD] resumable backfill
py -m intel.cli weather-forecast 16-day forecast at state centres
py -m intel.cli status Layer 1 report (exit 1 if unhealthy)
py -m intel.cli export [--out DIR] CSVs + manifest for the Colab notebooks
py -m intel.cli resolve "TEXT" which location does this text mention?
py -m intel.cli run-daily forecast + history top-up + status, one run id
`run-daily` is what a scheduler will call once one exists (none yet — see the
design doc's "Conflicts").
"""
import argparse
import json
import sys
import uuid
from datetime import date
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
from intel import db # noqa: E402 (env must be loaded first)
EXPORT_DIR = Path(__file__).resolve().parent / "data" / "export"
def _print(obj) -> None:
print(json.dumps(obj, indent=2, default=str))
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(prog="intel")
sub = p.add_subparsers(dest="cmd", required=True)
sub.add_parser("init")
sub.add_parser("locations")
sub.add_parser("calendar")
sub.add_parser("places")
wh = sub.add_parser("weather-history")
wh.add_argument("--budget", type=float, default=None)
wh.add_argument("--start", type=date.fromisoformat, default=None)
wh.add_argument("--estimate", action="store_true", help="only print the weighted calls still needed")
sub.add_parser("weather-forecast")
sub.add_parser("status")
ex = sub.add_parser("export")
ex.add_argument("--out", type=Path, default=EXPORT_DIR)
rs = sub.add_parser("resolve")
rs.add_argument("text")
sub.add_parser("run-daily")
args = p.parse_args(argv)
con = db.connect()
run_id = f"{args.cmd}-{uuid.uuid4().hex[:8]}"
try:
if args.cmd == "init":
db.init(con)
_print({"initialised": db.db_path()})
elif args.cmd == "locations":
from intel.locations import build
_print(build(con))
elif args.cmd == "calendar":
from intel.ingest.calendar import ingest
_print(ingest(con, run_id=run_id))
elif args.cmd == "places":
from intel.ingest.places import ingest
_print(ingest(con, run_id=run_id))
elif args.cmd == "weather-history":
from intel.ingest import weather
start = args.start or weather.DEFAULT_HISTORY_START
if args.estimate:
_print({"weighted_calls_needed": round(weather.estimate_backfill_calls(con, start=start), 1)})
else:
_print(weather.backfill_history(con, start=start, budget=args.budget or weather.DEFAULT_RUN_BUDGET, run_id=run_id))
elif args.cmd == "weather-forecast":
from intel.ingest.weather import fetch_forecast
_print(fetch_forecast(con, run_id=run_id))
elif args.cmd == "status":
from intel.freshness import report
r = report(con)
_print(r)
return 0 if r["healthy"] else 1
elif args.cmd == "export":
from intel.export import export
_print(export(con, args.out)["rows"])
elif args.cmd == "resolve":
from intel.locations import resolve
_print({"text": args.text, "location_id": resolve(con, args.text)})
elif args.cmd == "run-daily":
from intel.freshness import report
from intel.ingest import weather
out = {"run_id": run_id, "forecast": weather.fetch_forecast(con, run_id=run_id),
"history": weather.backfill_history(con, run_id=run_id)}
out["status"] = report(con)
_print(out)
return 0 if out["status"]["healthy"] else 1
finally:
con.close()
return 0
if __name__ == "__main__":
sys.exit(main())
|