Add runway tool with auto-detected inbound ships; fix multi-base state tagging
- tools/runway.py: days of runway per material before a base stalls, from simulate()'s real net consumption, base storage, plus (by default) any ship currently flying to that planet - tools/state.py: tag each STORE entry with its planet (FIO AddressableId -> site SiteId) and each ship with its live flight origin/destination (keyed by ShipId, not StlFuelStoreId). Needed now that a 2nd base exists - storage/ship data was previously unscoped, which would have silently mixed bases' inventories together - tests, docs updated (docs/library.md notes the multi-base tagging as a reusable pattern) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5
Executable
+85
@@ -0,0 +1,85 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Days of runway per material before the base stalls, from the plan's real NET consumption
|
||||
(simulate()'s flows, so self-produced inputs like AL correctly don't count against you), the
|
||||
base's own storage, plus (by default) any ship currently flying TO this planet, auto-detected
|
||||
from empire state (tools/state.py tags each ship with its live flight destination).
|
||||
|
||||
puga runway plans/examples/base_plus_hwp.yaml # base storage + inbound ships (default)
|
||||
puga runway plans/examples/base_plus_hwp.yaml --no-inbound # base storage only, ignore ships in flight
|
||||
"""
|
||||
import argparse, sys
|
||||
from pathlib import Path
|
||||
import yaml
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from puga import config, prunplanner as pp, market
|
||||
from puga.simulate import simulate
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import plan_push
|
||||
from buy import stock_items # net-consumption-per-day logic, shared with `buy`
|
||||
|
||||
|
||||
def inbound_ships(state: dict, planet: str) -> list[str]:
|
||||
"""Registrations of ships whose live flight destination is this planet (natural id substring match
|
||||
against the flight's destination string, e.g. '... - Deimos (ZV-759c)')."""
|
||||
out = []
|
||||
for s in state.get("ships", []):
|
||||
f = s.get("flight")
|
||||
if f and f.get("destination") and planet in f["destination"]:
|
||||
out.append(s["reg"])
|
||||
return out
|
||||
|
||||
|
||||
def on_hand(state: dict, planet: str, ships: list[str]) -> dict:
|
||||
"""This planet's own STORE items plus the named ships' SHIP_STORE cargo, summed."""
|
||||
totals: dict[str, float] = {}
|
||||
for s in state.get("storage", []):
|
||||
if (s.get("type") == "STORE" and s.get("planet") == planet) or (s.get("type") == "SHIP_STORE" and s.get("name") in ships):
|
||||
for tk, qty in s.get("items", {}).items():
|
||||
totals[tk] = totals.get(tk, 0) + qty
|
||||
return totals
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("spec", nargs="?", help="plans/*.yaml describing the WHOLE base (see docs/library.md)")
|
||||
ap.add_argument("--uuid", help="read the plan from the PRUNplanner account instead")
|
||||
ap.add_argument("--no-inbound", action="store_true", help="ignore ships currently flying to this base; base storage only")
|
||||
ap.add_argument("--cx", default=config.DEFAULT_CX)
|
||||
a = ap.parse_args()
|
||||
|
||||
recipes, blds = pp.recipes(), pp.buildings()
|
||||
if a.uuid:
|
||||
plan = pp.request("GET", f"/planning/plan/{a.uuid}/")
|
||||
elif a.spec:
|
||||
plan = plan_push.build_payload(yaml.safe_load(Path(a.spec).read_text()), recipes, {b["building_ticker"] for b in blds})
|
||||
else:
|
||||
sys.exit("give a spec file or --uuid")
|
||||
|
||||
st = yaml.safe_load(config.state_path().read_text())
|
||||
if not st.get("hq"):
|
||||
plan["plan_corphq"] = False
|
||||
faction = st.get("company", {}).get("faction")
|
||||
perm = (st.get("permits", {}).get("used", 1), st.get("permits", {}).get("total", 2))
|
||||
planet = pp._g(f"/data/planet/{plan['planet_natural_id']}/", 3600)
|
||||
snap = market.snapshot()
|
||||
price = lambda t, side="buy": (lambda q: q.ask if q else None)(snap.get((t, a.cx)))
|
||||
r = simulate(plan, recipes, blds, planet["resources"], planet["fertility"], price, faction, perm)
|
||||
|
||||
ships = [] if a.no_inbound else inbound_ships(st, plan["planet_natural_id"])
|
||||
have = on_hand(st, plan["planet_natural_id"], ships)
|
||||
daily = stock_items(r["flows"], days=1.0) # net units consumed per day, per material (self-production already netted)
|
||||
if not daily:
|
||||
print("nothing net-consumed by this plan; no runway to compute")
|
||||
return
|
||||
|
||||
rows = sorted(((tk, have.get(tk, 0), rate, have.get(tk, 0) / rate) for tk, rate in daily.items()), key=lambda x: x[3])
|
||||
ships_note = f" + inbound {', '.join(ships)}" if ships else ""
|
||||
print(f"RUNWAY (base storage{ships_note})\n{'material':8} {'have':>8} {'burn/day':>9} {'days':>7}")
|
||||
for tk, have, rate, days in rows:
|
||||
flag = " <- CRITICAL" if days < 1 else (" <- low" if days < 2 else "")
|
||||
print(f"{tk:8} {have:8.1f} {rate:9.2f} {days:7.2f}{flag}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+11
-2
@@ -29,6 +29,8 @@ def sync():
|
||||
prod = fio.own(f"/production/{u}", ttl=0)
|
||||
stores = fio.own(f"/storage/{u}", ttl=0)
|
||||
ships = fio.own(f"/ship/ships/{u}", ttl=0)
|
||||
flights = fio.own(f"/ship/flights/{u}", ttl=0)
|
||||
by_ship = {f["ShipId"]: f for f in flights}
|
||||
st["permits"] = {"used": sites[0]["InvestedPermits"] if sites else 0, "total": sites[0]["MaximumPermits"] if sites else 0}
|
||||
if st.get("permits_total_override"): # e.g. the HQ screen (base permits x / y) disagrees with FIO's MaximumPermits
|
||||
st["permits"]["total"] = st["permits_total_override"]
|
||||
@@ -42,10 +44,17 @@ def sync():
|
||||
for l in prod if l["SiteId"] == s["SiteId"]]
|
||||
bases.append(b)
|
||||
st["bases"] = bases
|
||||
st["storage"] = [dict(type=x["Type"], name=x["Name"], weight=f"{x['WeightLoad']:.0f}/{x['WeightCapacity']:.0f}t",
|
||||
site_planet = {s["SiteId"]: s["PlanetIdentifier"] for s in sites} # STORE.AddressableId == a site's SiteId
|
||||
|
||||
st["storage"] = [dict(type=x["Type"], name=x["Name"], planet=site_planet.get(x.get("AddressableId")),
|
||||
weight=f"{x['WeightLoad']:.0f}/{x['WeightCapacity']:.0f}t",
|
||||
items={i["MaterialTicker"]: i["MaterialAmount"] for i in x["StorageItems"] if i.get("MaterialTicker")})
|
||||
for x in stores if x["StorageItems"] or x["Type"] == "SHIP_STORE"]
|
||||
st["ships"] = [dict(name=x["Name"], reg=x["Registration"], mass=x["Mass"], in_flight=bool(x.get("FlightId"))) for x in ships]
|
||||
def ship_dest(x):
|
||||
f = by_ship.get(x.get("ShipId"))
|
||||
return None if not f else {"origin": f.get("Origin"), "destination": f.get("Destination"), "eta_ms": f.get("ArrivalTimeEpochMs")}
|
||||
|
||||
st["ships"] = [dict(name=x["Name"], reg=x["Registration"], mass=x["Mass"], in_flight=bool(x.get("FlightId")), flight=ship_dest(x)) for x in ships]
|
||||
P.write_text(yaml.safe_dump(st, sort_keys=False, default_flow_style=None, width=140))
|
||||
print("synced", st["as_of"])
|
||||
|
||||
|
||||
Reference in New Issue
Block a user