dodoxandClaude Sonnet 5
4c186ac677
- pyproject.toml + puga/cli.py: `pip install -e .` installs a `puga <tool>` command, so no venv path is needed (replaces requirements.txt) - tools/plan_push.py: `pull <uuid>` reads a plan (incl. UI edits) back into a YAML spec; recipe entries may carry an amount so switched-off recipes round trip; fixes a variable that overwrote the plan name in build_payload, with a regression test - README: usage with the puga command, and a section on working on the same plans in the PRUNplanner UI and from Claude (write, edit in UI, read back) - CLAUDE.md updated accordingly Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
151 lines
8.9 KiB
Python
Executable File
151 lines
8.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Turn a YAML plan spec (plans/*.yaml) into a PRUNplanner plan, via the Api-Key API.
|
|
DEFAULT IS DRY RUN. Guardrails (docs/decisions.md): only plans named '[PuGa] ...' are created/updated; deletes only [PuGa] plans and only when asked;
|
|
--update refuses if the existing plan's name does not start with '[PuGa]'. Show the dry-run to the user and get a yes before --apply.
|
|
|
|
tools/plan_push.py list
|
|
tools/plan_push.py pull <uuid> [-o file.yaml] # read a plan (UI edits included) back into a YAML spec
|
|
tools/plan_push.py plans/examples/base_plus_hwp.yaml # dry run: validated payload summary
|
|
tools/plan_push.py plans/examples/base_plus_hwp.yaml --json # full JSON payload
|
|
tools/plan_push.py plans/examples/base_plus_hwp.yaml --apply # create (after user says yes)
|
|
tools/plan_push.py plans/examples/base_plus_hwp.yaml --apply --update <uuid>
|
|
tools/plan_push.py delete <uuid> # only [PuGa] plans, only when the user asks
|
|
|
|
Spec: name, planet (natural id), permits, cogc (e.g. METALLURGY or null), hq, experts {METALLURGY: 2}, lux {pioneer: [true,true]},
|
|
infrastructure {HB1: 4, HB2: 1}, buildings: [{building: SME, amount: 5, recipes: ["ALO,FLX,C,O=>4AL"|"AL,STL,HE=>BHP"|"EXT#ALO"]}]
|
|
A recipe is an exact PRUNplanner recipe_id, an extraction id like EXT#ALO, or 'IN,IN=>OUT' (ticker sets, matched within the building).
|
|
"""
|
|
import argparse, json, sys
|
|
from pathlib import Path
|
|
import yaml
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
from puga import prunplanner as pp
|
|
|
|
EXPERTS = ["Agriculture", "Chemistry", "Construction", "Electronics", "Food_Industries", "Fuel_Refining",
|
|
"Manufacturing", "Metallurgy", "Resource_Extraction"]
|
|
COGC = {"AGRICULTURE", "CHEMISTRY", "CONSTRUCTION", "ELECTRONICS", "FOOD_INDUSTRIES", "FUEL_REFINING", "MANUFACTURING",
|
|
"METALLURGY", "RESOURCE_EXTRACTION", "PIONEERS", "SETTLERS", "TECHNICIANS", "ENGINEERS", "SCIENTISTS"}
|
|
TIERS = ["pioneer", "settler", "technician", "engineer", "scientist"]
|
|
PREFIX = "[PuGa]"
|
|
|
|
|
|
def resolve_recipe(spec: str, building: str, recipes: list[dict]) -> str:
|
|
rs = [r for r in recipes if r["building_ticker"] == building]
|
|
if any(r["recipe_id"] == spec for r in rs) or spec.startswith(building + "#") and "=>" not in spec:
|
|
return spec # exact id, or extraction id (EXT#ALO)
|
|
if "=>" not in spec:
|
|
raise ValueError(f"cannot parse recipe '{spec}'")
|
|
left, right = spec.split("=>")
|
|
want_in = {x.strip().split("x")[-1] if x.strip()[0].isdigit() and "x" in x else x.strip() for x in left.split(",") if x.strip()}
|
|
want_out = {x.strip().split("x")[-1] if x.strip()[0].isdigit() and "x" in x else x.strip() for x in right.split(",") if x.strip()}
|
|
# tolerate amount prefixes like '4AL' / '4xAL'
|
|
strip = lambda s: {"".join(ch for ch in x.lstrip("0123456789x")) for x in s}
|
|
want_in, want_out = strip(want_in), strip(want_out)
|
|
hits = [r for r in rs if {i["material_ticker"] for i in r["inputs"]} == want_in and {o["material_ticker"] for o in r["outputs"]} == want_out]
|
|
if len(hits) != 1:
|
|
raise ValueError(f"recipe '{spec}' at {building}: {len(hits)} matches" + (": " + ", ".join(h["recipe_id"] for h in hits) if hits else ""))
|
|
return hits[0]["recipe_id"]
|
|
|
|
|
|
def build_payload(spec: dict, recipes: list[dict], building_tickers: set[str]) -> dict:
|
|
name = spec["name"]
|
|
if not name.startswith(PREFIX):
|
|
raise ValueError(f"plan name must start with '{PREFIX}' (guardrail); got '{name}'")
|
|
planet = spec["planet"]
|
|
if len(planet) != 7:
|
|
raise ValueError("planet natural id must be 7 chars, e.g. ZV-759c")
|
|
permits = int(spec.get("permits", 1))
|
|
if not 0 <= permits <= 3:
|
|
raise ValueError("permits must be 0..3")
|
|
cogc = spec.get("cogc")
|
|
if cogc is not None and cogc.upper() not in COGC:
|
|
raise ValueError(f"cogc '{cogc}' not one of {sorted(COGC)}")
|
|
experts = {k.upper(): v for k, v in (spec.get("experts") or {}).items()}
|
|
lux = spec.get("lux") or {}
|
|
buildings = []
|
|
for b in spec["buildings"]:
|
|
tk = b["building"]
|
|
if tk not in building_tickers:
|
|
raise ValueError(f"unknown building {tk}")
|
|
recs = []
|
|
for r in b.get("recipes", []):
|
|
rname, amt = (r["recipe"], r.get("amount", 1)) if isinstance(r, dict) else (r, 1)
|
|
recs.append({"recipeid": resolve_recipe(rname, tk, recipes), "amount": int(amt)})
|
|
buildings.append({"name": tk, "amount": int(b["amount"]), "active_recipes": recs})
|
|
infra = [{"building": k, "amount": int(v)} for k, v in (spec.get("infrastructure") or {}).items()]
|
|
return {
|
|
"plan_name": name, "planet_natural_id": planet, "plan_permits_used": permits,
|
|
"plan_cogc": cogc.upper() if cogc else None, "plan_corphq": bool(spec.get("hq", False)),
|
|
"plan_data": {
|
|
"experts": [{"type": e, "amount": int(experts.get(e.upper(), 0))} for e in EXPERTS],
|
|
"workforce": [{"type": t, "lux1": bool(lux.get(t, [True, True])[0]), "lux2": bool(lux.get(t, [True, True])[1])} for t in TIERS],
|
|
"infrastructure": infra, "buildings": buildings}}
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("spec", help="plans/*.yaml, 'list', 'pull' or 'delete'")
|
|
ap.add_argument("target", nargs="?", help="uuid for 'delete' / 'pull'")
|
|
ap.add_argument("-o", "--out", help="with 'pull': write the YAML spec to this file")
|
|
ap.add_argument("--apply", action="store_true", help="actually write to PRUNplanner (creates a new plan)")
|
|
ap.add_argument("--update", metavar="UUID", help="with --apply: update this existing [PuGa] plan instead of creating")
|
|
ap.add_argument("--json", action="store_true")
|
|
a = ap.parse_args()
|
|
|
|
if a.spec == "list":
|
|
for p in pp.request("GET", "/planning/plan/"):
|
|
print(p["uuid"], p.get("plan_name"), p.get("planet_natural_id"))
|
|
return
|
|
if a.spec == "pull": # read a plan from the account (e.g. after edits in the PRUNplanner UI) into a YAML spec
|
|
cur = pp.request("GET", f"/planning/plan/{a.target}/")
|
|
d = cur["plan_data"]
|
|
cg = cur.get("plan_cogc")
|
|
spec = {"name": cur["plan_name"], "planet": cur["planet_natural_id"], "permits": cur["plan_permits_used"],
|
|
"cogc": None if cg in (None, "---") else cg, "hq": cur["plan_corphq"],
|
|
"experts": {e["type"].upper(): e["amount"] for e in d["experts"] if e["amount"]},
|
|
"lux": {w["type"]: [w["lux1"], w["lux2"]] for w in d["workforce"] if not (w["lux1"] and w["lux2"])},
|
|
"infrastructure": {i["building"]: i["amount"] for i in d["infrastructure"]},
|
|
"buildings": [{"building": b["name"], "amount": b["amount"],
|
|
"recipes": [r["recipeid"] if r["amount"] == 1 else {"recipe": r["recipeid"], "amount": r["amount"]} for r in b["active_recipes"]]}
|
|
for b in d["buildings"]]}
|
|
text = yaml.safe_dump(spec, sort_keys=False, default_flow_style=None, width=140)
|
|
if a.out:
|
|
Path(a.out).write_text(text)
|
|
print("wrote", a.out)
|
|
else:
|
|
print(text)
|
|
return
|
|
if a.spec == "delete":
|
|
# delete is allowed only on request, and only for plans this tool made ([PuGa] prefix).
|
|
cur = pp.request("GET", f"/planning/plan/{a.target}/")
|
|
if not str(cur.get("plan_name", "")).startswith(PREFIX):
|
|
sys.exit(f"REFUSED: '{cur.get('plan_name')}' is not a {PREFIX} plan")
|
|
pp.request("DELETE", f"/planning/plan/{a.target}/")
|
|
print("DELETED:", cur["plan_name"], a.target)
|
|
return
|
|
spec = yaml.safe_load(Path(a.spec).read_text())
|
|
payload = build_payload(spec, pp.recipes(), {b["building_ticker"] for b in pp.buildings()})
|
|
d = payload["plan_data"]
|
|
print(f"PLAN {payload['plan_name']} planet {payload['planet_natural_id']} permits {payload['plan_permits_used']} cogc {payload['plan_cogc']} hq {payload['plan_corphq']}")
|
|
print("experts:", {e["type"]: e["amount"] for e in d["experts"] if e["amount"]} or "-")
|
|
print("infrastructure:", {i["building"]: i["amount"] for i in d["infrastructure"]} or "-")
|
|
for b in d["buildings"]:
|
|
print(f" {b['amount']:>3} x {b['name']:4}", [r["recipeid"] for r in b["active_recipes"]])
|
|
if a.json:
|
|
print(json.dumps(payload, indent=1))
|
|
if not a.apply:
|
|
print("\nDRY RUN: nothing sent. Re-run with --apply after the user confirms.")
|
|
return
|
|
if a.update:
|
|
cur = pp.request("GET", f"/planning/plan/{a.update}/")
|
|
if not str(cur.get("plan_name", "")).startswith(PREFIX):
|
|
sys.exit(f"REFUSED: existing plan '{cur.get('plan_name')}' is not a {PREFIX} plan")
|
|
out = pp.request("PUT", f"/planning/plan/{a.update}/", payload)
|
|
else:
|
|
out = pp.request("POST", "/planning/plan/", payload)
|
|
print("WRITTEN:", out.get("uuid") if isinstance(out, dict) else out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|