48dd1075ed
New endpoints (total now 65+): - clear: clear debris/POI (bottom toolbar button) - rotate: rotate building during placement - copy_settings: copy building config to another building - overlay: switch overlay view (power/temp/gas/liquid/automation/rooms/decor/light/germs/materials/suits/crops/radiation/priority) - dupe_personal_priority: set per-duplicate task priority sliders - battery_charge: set smart battery high/low charge limits - valve_flow: set liquid/gas valve flow rate - vent_pressure: set vent overpressure limit - incubator_setting: set incubator egg priority - fridge_temp: set refrigerator temperature - GET /api/state/overlay: current overlay + available overlays list CLI: clear, rotate, copy_settings, overlay, dupe_personal_priority, battery_charge, valve_flow, vent_pressure, incubator_setting, fridge_temp
1533 lines
60 KiB
Python
1533 lines
60 KiB
Python
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
import sys
|
|
import os
|
|
|
|
CONFIG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'config.json')
|
|
|
|
# Auto-pause control
|
|
AUTO_PAUSE_ENABLED = True
|
|
"""When True, api_get and api_post auto-pause game before executing.
|
|
Set to False for event daemon polling (lightweight, frequent checks)."""
|
|
|
|
def load_config():
|
|
with open(CONFIG_PATH) as f:
|
|
return json.load(f)
|
|
|
|
def api_url(endpoint):
|
|
cfg = load_config()
|
|
return f"http://{cfg['modHost']}:{cfg['modPort']}{endpoint}"
|
|
|
|
def _auto_pause():
|
|
"""Auto-pause the game before any AI operation, ensuring data freshness."""
|
|
if not AUTO_PAUSE_ENABLED:
|
|
return
|
|
try:
|
|
cfg = load_config()
|
|
url = f"http://{cfg['modHost']}:{cfg['modPort']}/api/action/pause"
|
|
req = urllib.request.Request(url, data=b'{}', headers={'Content-Type': 'application/json'}, method='POST')
|
|
with urllib.request.urlopen(req, timeout=3):
|
|
pass
|
|
except:
|
|
pass
|
|
|
|
def api_get(endpoint, auto_pause=True):
|
|
if auto_pause and AUTO_PAUSE_ENABLED:
|
|
_auto_pause()
|
|
url = api_url(endpoint)
|
|
cfg = load_config()
|
|
try:
|
|
req = urllib.request.Request(url, method='GET')
|
|
with urllib.request.urlopen(req, timeout=cfg['timeout']) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.URLError as e:
|
|
return {"error": str(e)}
|
|
|
|
def api_post(endpoint, data, auto_pause=True):
|
|
if auto_pause and AUTO_PAUSE_ENABLED:
|
|
_auto_pause()
|
|
url = api_url(endpoint)
|
|
cfg = load_config()
|
|
try:
|
|
req = urllib.request.Request(
|
|
url, data=json.dumps(data).encode(),
|
|
headers={'Content-Type': 'application/json'},
|
|
method='POST'
|
|
)
|
|
with urllib.request.urlopen(req, timeout=cfg['timeout']) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.URLError as e:
|
|
return {"error": str(e)}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command implementations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_health():
|
|
print(json.dumps(api_get('/health'), indent=2, ensure_ascii=False))
|
|
|
|
def cmd_status():
|
|
game = api_get('/api/state/game')
|
|
resources = api_get('/api/state/resources')
|
|
dups = api_get('/api/state/duplicants')
|
|
alerts = api_get('/api/state/alert')
|
|
|
|
if 'error' in game:
|
|
print(f"Error: {game['error']}")
|
|
return
|
|
|
|
print("=== Game ===")
|
|
print(f" Cycle: {game.get('cycle', '?')}")
|
|
print(f" Duplicants: {game.get('duplicantCount', '?')}")
|
|
print(f" World: {game.get('worldName', '?')}")
|
|
print(f" Grid: {game.get('gridWidth', '?')} x {game.get('gridHeight', '?')}")
|
|
|
|
print("\n=== Resources (top 15) ===")
|
|
if isinstance(resources, list):
|
|
for r in sorted(resources, key=lambda x: x.get('amount', 0), reverse=True)[:15]:
|
|
state_mark = {'solid': '■', 'liquid': '≈', 'gas': '◌', 'vacuum': ' '}.get(r.get('state', ''), '?')
|
|
cat = r.get('category', '')
|
|
print(f" {state_mark} {r.get('name', '?'):20s} {r.get('amount', 0):>10.1f} kg [{cat}]")
|
|
|
|
print("\n=== Duplicants ===")
|
|
if isinstance(dups, list):
|
|
for d in dups:
|
|
chore = d.get('currentChore', '?').replace('Chore', '')
|
|
print(f" {d.get('name'):12s} at ({d.get('x', '?'):3d},{d.get('y', '?'):3d}) "
|
|
f"stress={d.get('stress', 0):.0f}% food={d.get('calories', 0)/1000:.0f} kcal "
|
|
f"doing={chore}")
|
|
|
|
print("\n=== Alerts ===")
|
|
if isinstance(alerts, list):
|
|
if alerts:
|
|
for a in alerts:
|
|
print(f" [{a.get('severity', '?')}] {a.get('title', '?')}: {a.get('message', '')}")
|
|
else:
|
|
print(" (none)")
|
|
|
|
def cmd_resources():
|
|
data = api_get('/api/state/resources')
|
|
if isinstance(data, list):
|
|
for r in sorted(data, key=lambda x: x.get('amount', 0), reverse=True):
|
|
state_mark = {'solid': '■', 'liquid': '≈', 'gas': '◌'}.get(r.get('state', ''), ' ')
|
|
print(f"{state_mark} {r.get('name', '?'):25s} {r.get('amount', 0):>12.1f} kg {r.get('state', '?'):8s} [{r.get('category', '?')}]")
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_duplicants():
|
|
data = api_get('/api/state/duplicants')
|
|
if isinstance(data, list):
|
|
for d in data:
|
|
print(f" Name: {d.get('name', '?')}")
|
|
print(f" Cell: ({d.get('x', '?')}, {d.get('y', '?')})")
|
|
print(f" Stress: {d.get('stress', 0):.1f}%")
|
|
print(f" Food: {d.get('calories', 0)/1000:.0f} kcal")
|
|
print(f" Stamina: {d.get('stamina', 0):.0f}%")
|
|
print(f" Oxygen: {d.get('oxygen', 0):.0f}%")
|
|
print(f" Chore: {d.get('currentChore', '?')}")
|
|
print()
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_buildings():
|
|
data = api_get('/api/state/buildings')
|
|
if isinstance(data, list):
|
|
cats = {}
|
|
for b in data:
|
|
cat = b.get('category', 'Other')
|
|
if cat not in cats:
|
|
cats[cat] = []
|
|
cats[cat].append(b)
|
|
for cat, blist in sorted(cats.items()):
|
|
print(f"[{cat}] ({len(blist)})")
|
|
for b in blist:
|
|
op = 'ON' if b.get('isOperational') else 'OFF'
|
|
pw = f" {b.get('powerWatt', 0)}W" if b.get('powerWatt', 0) > 0 else ''
|
|
print(f" {b.get('name', '?'):25s} at ({b.get('x', '?')},{b.get('y', '?')}) [{op}]{pw}")
|
|
print()
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_research():
|
|
data = api_get('/api/state/research')
|
|
if isinstance(data, list):
|
|
for t in data:
|
|
status = "DONE" if t.get('isComplete') else f"{t.get('progress', 0)*100:.0f}%"
|
|
print(f" {t.get('name', '?'):25s} [{status:5s}] ({t.get('category', '?')})")
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_geysers():
|
|
data = api_get('/api/state/geysers')
|
|
if isinstance(data, list):
|
|
for g in data:
|
|
print(f" {g.get('name', '?'):25s} at ({g.get('x', '?')}, {g.get('y', '?')}) "
|
|
f"state={g.get('state', '?'):8s} rate={g.get('emitRate', 0):.1f} g/s")
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_critters():
|
|
data = api_get('/api/state/critters')
|
|
if isinstance(data, list):
|
|
for c in data:
|
|
print(f" {c.get('name', '?'):20s} ({c.get('species', '?')}) at ({c.get('x', '?')},{c.get('y', '?')}) "
|
|
f"age={c.get('age', 0):.1f} happy={c.get('happiness', 0)}")
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_plants():
|
|
data = api_get('/api/state/plants')
|
|
if isinstance(data, list):
|
|
for p in data:
|
|
grown = "GROWN" if p.get('isGrown') else f"{p.get('progress', 0)*100:.0f}%"
|
|
wilt = " WILT" if p.get('isWilting') else ""
|
|
print(f" {p.get('name', '?'):25s} at ({p.get('x', '?')},{p.get('y', '?')}) [{grown}]{wilt}")
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_rooms():
|
|
data = api_get('/api/state/rooms')
|
|
if isinstance(data, list):
|
|
for r in data:
|
|
print(f" {r.get('name', '?'):25s} cells={r.get('cellCount', 0):4d} "
|
|
f"buildings={r.get('buildings', 0)} creatures={r.get('creatures', 0)} plants={r.get('plants', 0)}")
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_cell(args):
|
|
if len(args) < 2:
|
|
print("Usage: cell <x> <y>")
|
|
return
|
|
data = api_get(f"/api/state/cell?x={args[0]}&y={args[1]}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Cell ({data.get('x', '?')}, {data.get('y', '?')}) [{data.get('cell', '?')}]")
|
|
print(f" Element: {data.get('element', '?')} ({data.get('elementState', '?')})")
|
|
print(f" Mass: {data.get('massKg', 0):.1f} kg")
|
|
print(f" Temp: {data.get('temperatureC', 0):.1f} °C")
|
|
print(f" Building: {data.get('buildingName', 'none')}")
|
|
print(f" Duplicant: {data.get('duplicantName', 'none')}")
|
|
print(f" Pressure: {data.get('pressure', 0):.3f} kg")
|
|
print(f" Visible: {data.get('isVisible', False)}")
|
|
|
|
def cmd_cells(args):
|
|
if len(args) < 4:
|
|
print("Usage: cells <x> <y> <width> <height>")
|
|
return
|
|
data = api_get(f"/api/state/cells?x={args[0]}&y={args[1]}&width={args[2]}&height={args[3]}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
region = data.get('region', {})
|
|
cells = data.get('cells', [])
|
|
print(f"Region ({region.get('x', '?')},{region.get('y', '?')}) {region.get('width', '?')}x{region.get('height', '?')} ({len(cells)} cells)")
|
|
print()
|
|
# Print as a grid
|
|
grid = {}
|
|
for c in cells:
|
|
key = (c.get('x'), c.get('y'))
|
|
grid[key] = c
|
|
rx, ry = region.get('x', 0), region.get('y', 0)
|
|
rw, rh = region.get('width', 0), region.get('height', 0)
|
|
# Header row
|
|
header = " "
|
|
for cx in range(rx, rx + rw):
|
|
header += f"{cx % 10} "
|
|
print(header)
|
|
for cy in range(ry + rh - 1, ry - 1, -1):
|
|
row = f"{cy:3d} "
|
|
for cx in range(rx, rx + rw):
|
|
c = grid.get((cx, cy))
|
|
if c is None:
|
|
row += " "
|
|
elif c.get('isVacuum'):
|
|
row += " "
|
|
elif c.get('hasDuplicant'):
|
|
row += "D "
|
|
elif c.get('hasBuilding'):
|
|
row += "B "
|
|
elif c.get('isLiquid'):
|
|
row += "~ "
|
|
elif c.get('isGas'):
|
|
row += ". "
|
|
elif c.get('isSolid'):
|
|
row += "# "
|
|
else:
|
|
row += " "
|
|
print(row)
|
|
|
|
def cmd_cell_slice(args):
|
|
if len(args) < 2:
|
|
print("Usage: slice <x|y> <index> [start] [end]")
|
|
return
|
|
axis = args[0]
|
|
index = args[1]
|
|
start = args[2] if len(args) > 2 else "0"
|
|
end = args[3] if len(args) > 3 else "100"
|
|
data = api_get(f"/api/state/cells/slice?axis={axis}&index={index}&start={start}&end={end}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
for c in data.get('cells', []):
|
|
state = "VAC" if c.get('isVacuum') else c.get('element', '?')
|
|
building = f" [{c.get('buildingName', '')}]" if c.get('hasBuilding') else ""
|
|
print(f" ({c.get('x', '?')},{c.get('y', '?')}) {state:15s} {c.get('temperatureC', 0):6.1f}°C {c.get('massKg', 0):8.1f}kg{building}")
|
|
|
|
def cmd_gas(args):
|
|
if len(args) < 2:
|
|
print("Usage: gas <x> <y> [radius]")
|
|
return
|
|
x, y = args[0], args[1]
|
|
radius = args[2] if len(args) > 2 else "20"
|
|
data = api_get(f"/api/state/gas?x={x}&y={y}&radius={radius}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Gas analysis at ({x},{y}) radius={radius}")
|
|
for g in data.get('gases', []):
|
|
print(f" {g.get('gas', '?'):25s} {g.get('mass', 0):10.1f} kg ({g.get('count', 0)} cells)")
|
|
|
|
def cmd_registry(args):
|
|
if len(args) < 1:
|
|
print("Usage: registry <buildings|elements|techs> [filter]")
|
|
return
|
|
kind = args[0]
|
|
filt = args[1].lower() if len(args) > 1 else ""
|
|
data = api_get(f"/api/registry/{kind}")
|
|
if isinstance(data, list):
|
|
count = 0
|
|
for item in data:
|
|
name = item.get('name', item.get('id', '?'))
|
|
item_id = item.get('id', '')
|
|
if filt and filt not in name.lower() and filt not in item_id.lower():
|
|
continue
|
|
count += 1
|
|
if kind == 'buildings':
|
|
print(f" {item_id:35s} {item.get('name', '?'):30s} {item.get('width', 1)}x{item.get('height', 1)} {item.get('powerCost', 0)}W")
|
|
elif kind == 'elements':
|
|
print(f" {item_id:25s} {item.get('name', '?'):20s} {item.get('state', '?'):7s} [{item.get('category', '?')}]")
|
|
elif kind == 'techs':
|
|
status = "DONE" if item.get('isComplete') else "PENDING"
|
|
unlocks = ', '.join(item.get('unlockedBuildings', [])[:5])
|
|
print(f" {item_id:30s} {item.get('name', '?'):25s} [{status}] -> {unlocks}")
|
|
print(f"\n Total: {count} matches")
|
|
else:
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_dig(args):
|
|
if len(args) < 4:
|
|
print("Usage: dig <x> <y> <width> <height>")
|
|
return
|
|
result = api_post('/api/action/dig', {
|
|
"x": int(args[0]), "y": int(args[1]),
|
|
"width": int(args[2]), "height": int(args[3])
|
|
})
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_build(args):
|
|
if len(args) < 3:
|
|
print("Usage: build <buildingId> <x> <y> [rotation]")
|
|
return
|
|
data = {"buildingId": args[0], "x": int(args[1]), "y": int(args[2])}
|
|
if len(args) >= 4:
|
|
data["rotation"] = args[3]
|
|
result = api_post('/api/action/build', data)
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_deconstruct(args):
|
|
if len(args) < 3:
|
|
print("Usage: deconstruct <buildingId> <x> <y>")
|
|
return
|
|
result = api_post('/api/action/deconstruct', {
|
|
"buildingId": args[0], "x": int(args[1]), "y": int(args[2])
|
|
})
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_prioritize(args):
|
|
if len(args) < 3:
|
|
print("Usage: prioritize <x> <y> <priority>")
|
|
return
|
|
result = api_post('/api/action/prioritize', {
|
|
"x": int(args[0]), "y": int(args[1]), "priority": int(args[2])
|
|
})
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_research_select(args):
|
|
if len(args) < 1:
|
|
print("Usage: research_select <techId>")
|
|
return
|
|
result = api_post('/api/action/research', {"techId": args[0]})
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_mop(args):
|
|
if len(args) < 2:
|
|
print("Usage: mop <x> <y>")
|
|
return
|
|
result = api_post('/api/action/mop', {"x": int(args[0]), "y": int(args[1])})
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
def cmd_harvest(args):
|
|
if len(args) < 2:
|
|
print("Usage: harvest <x> <y>")
|
|
return
|
|
result = api_post('/api/action/harvest', {"x": int(args[0]), "y": int(args[1])})
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Event / Queue / Batch / Priority
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_events(args):
|
|
"""Poll game events. Usage: events [since] [limit]"""
|
|
since = args[0] if len(args) > 0 else "0"
|
|
limit = args[1] if len(args) > 1 else "50"
|
|
data = api_get(f"/api/state/events?since={since}&limit={limit}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
events = data.get('events', [])
|
|
next_seq = data.get('next_seq', 0)
|
|
has_more = data.get('has_more', False)
|
|
|
|
if not events:
|
|
print("No new events.")
|
|
print(f"Next sequence: {next_seq}")
|
|
return
|
|
|
|
print(f"Events ({len(events)} new, next_seq={next_seq}, has_more={has_more}):")
|
|
print()
|
|
for e in events:
|
|
severity = e.get('severity', '?')
|
|
sev_mark = {'critical': '!!!', 'warning': '!!', 'info': 'i'}.get(severity.lower(), '?')
|
|
cat = e.get('category', '?')
|
|
title = e.get('title', '?')
|
|
msg = e.get('message', '')
|
|
cycle = e.get('cycle', '?')
|
|
entity = e.get('entity', '')
|
|
cell = e.get('cell', '')
|
|
print(f" [{sev_mark}] ({cycle}) {title}")
|
|
if msg:
|
|
print(f" {msg}")
|
|
if entity:
|
|
print(f" entity: {entity}")
|
|
if isinstance(cell, int) and cell >= 0:
|
|
print(f" cell index: {cell}")
|
|
print()
|
|
|
|
def cmd_queue(args):
|
|
"""View pending task queue. Usage: queue [batch_id]"""
|
|
params = ""
|
|
if args:
|
|
params = f"?batch_id={args[0]}"
|
|
data = api_get(f"/api/state/queue{params}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Task Queue:")
|
|
print(f" Length: {data.get('queue_length', '?')}")
|
|
print(f" Batch ID: {data.get('batch_id', 'none')}")
|
|
for t in data.get('tasks', []):
|
|
print(f" - {t.get('type', '?')}: {t.get('value', '?')}")
|
|
|
|
def cmd_batch(args):
|
|
"""Execute a batch of actions. Usage: batch <json_file>"""
|
|
if not args:
|
|
print("Usage: batch <json_file>")
|
|
print(" JSON format: { \"actions\": [ { \"type\": \"build|dig|...\", ... } ] }")
|
|
return
|
|
try:
|
|
with open(args[0]) as f:
|
|
plan = json.load(f)
|
|
except Exception as e:
|
|
print(f"Error reading file: {e}")
|
|
return
|
|
|
|
result = api_post('/api/action/batch', plan)
|
|
if 'error' in result:
|
|
print(f"Error: {result['error']}")
|
|
return
|
|
|
|
print(f"Batch: {result.get('batchId', '?')}")
|
|
print(f" Total: {result.get('total', 0)}")
|
|
print(f" OK: {result.get('successCount', 0)}")
|
|
print(f" Failed: {result.get('failCount', 0)}")
|
|
print(f" Summary: {result.get('summary', '?')}")
|
|
print()
|
|
|
|
for action in result.get('actions', []):
|
|
status = 'OK' if action.get('success') else 'FAIL'
|
|
result_type = action.get('result', '?')
|
|
error = action.get('error', '')
|
|
err_msg = action.get('errorMessage', '')
|
|
suggestion = action.get('suggestion', '')
|
|
|
|
print(f" [{status}] {result_type}")
|
|
if error:
|
|
print(f" error: {error}")
|
|
if err_msg:
|
|
print(f" msg: {err_msg}")
|
|
if suggestion:
|
|
print(f" -> {suggestion}")
|
|
print()
|
|
|
|
def cmd_priority_global(args):
|
|
"""Set global priority. Usage: priority_global <target> <priority>"""
|
|
if len(args) < 2:
|
|
print("Usage: priority_global <target> <priority>")
|
|
print(" target: 'dig', 'build', 'clear', or 'all'")
|
|
print(" priority: 1 (lowest) to 9 (emergency)")
|
|
return
|
|
result = api_post('/api/action/priority_global', {
|
|
"target": args[0],
|
|
"priority": int(args[1])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_priority_type(args):
|
|
"""Set priority for a building type. Usage: priority_type <buildingType> <priority>"""
|
|
if len(args) < 2:
|
|
print("Usage: priority_type <buildingType> <priority>")
|
|
return
|
|
result = api_post('/api/action/priority_type', {
|
|
"buildingType": args[0],
|
|
"priority": int(args[1])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_pause(args):
|
|
"""Pause the game. Usage: pause [reason]"""
|
|
reason = ' '.join(args) if args else 'AI operation in progress'
|
|
result = api_post('/api/action/pause', {} if not args else {"reason": reason})
|
|
_print_feedback(result)
|
|
|
|
def cmd_unpause(args):
|
|
"""Unpause the game. Usage: unpause [speed]"""
|
|
speed = int(args[0]) if args else 1
|
|
result = api_post('/api/action/unpause', {"speed": speed})
|
|
_print_feedback(result)
|
|
|
|
def cmd_speed(args):
|
|
"""Set game speed. Usage: speed <1|2|3>"""
|
|
if not args:
|
|
print("Usage: speed <1|2|3>")
|
|
return
|
|
speed = int(args[0])
|
|
if speed < 1 or speed > 3:
|
|
print("Speed must be 1, 2, or 3")
|
|
return
|
|
result = api_post('/api/action/speed', {"speed": speed})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Save / Load
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_saves(args):
|
|
"""List saves. Usage: saves"""
|
|
data = api_get('/api/state/saves')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
for s in data.get('saves', []):
|
|
print(f" {s.get('name', '?'):40s} {s.get('size', 0)/1024:.0f} KB")
|
|
|
|
def cmd_save(args):
|
|
"""Save game. Usage: save [name]"""
|
|
name = ' '.join(args) if args else None
|
|
result = api_post('/api/action/save', {"name": name} if name else {})
|
|
_print_feedback(result)
|
|
|
|
def cmd_save_as(args):
|
|
"""Save with specific name. Usage: save_as <name>"""
|
|
if not args:
|
|
print("Usage: save_as <name>")
|
|
return
|
|
result = api_post('/api/action/save_as', {"name": ' '.join(args)})
|
|
_print_feedback(result)
|
|
|
|
def cmd_load(args):
|
|
"""Load a save. Usage: load <name>"""
|
|
if not args:
|
|
print("Usage: load <name>")
|
|
return
|
|
name = ' '.join(args)
|
|
print(f"[!] Loading save '{name}' — game will restart!")
|
|
result = api_post('/api/action/load', {"name": name})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Power Grid
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_power(args):
|
|
"""Analyze power grid. Usage: power"""
|
|
data = api_get('/api/state/power')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Power Grid — {data.get('circuitCount', 0)} circuits")
|
|
print()
|
|
for c in data.get('circuits', []):
|
|
overload = " *** OVERLOAD ***" if c.get('isOverloaded') else ""
|
|
print(f" Circuit {c.get('id', '?')}: {c.get('wattsUsed', 0):.0f}W / {c.get('maxWatts', 0):.0f}W used{overload}")
|
|
print()
|
|
print(f"Generators ({len(data.get('generators', []))}):")
|
|
for g in data.get('generators', []):
|
|
status = 'ON' if g.get('isActive') else 'OFF'
|
|
print(f" {g.get('name', '?'):25s} {g.get('watts', 0):.0f}W [{status}]")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pipes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_pipes(args):
|
|
"""Inspect pipe contents. Usage: pipes [gas|liquid]"""
|
|
ptype = args[0] if args else 'all'
|
|
if ptype not in ('gas', 'liquid', 'all'):
|
|
print("Usage: pipes [gas|liquid|all]")
|
|
return
|
|
data = api_get(f"/api/state/pipes?type={ptype}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Pipe Contents ({data.get('pipeType', '?')}): {data.get('segmentCount', 0)} segments")
|
|
for s in data.get('segments', [])[:20]:
|
|
t = s.get('type', '?')
|
|
el = s.get('element', '?')
|
|
m = s.get('mass', 0)
|
|
temp = s.get('temperature', 0)
|
|
print(f" [{t}] {el:20s} {m:8.1f} kg {temp:6.1f} K")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CO2
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_co2(args):
|
|
"""Find CO2 pockets. Usage: co2"""
|
|
data = api_get('/api/state/co2')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"CO2 Pockets: {data.get('pocketCount', 0)} Total: {data.get('totalMassKg', 0):.0f} kg")
|
|
for p in data.get('pockets', [])[:10]:
|
|
print(f" ({p.get('x', '?')},{p.get('y', '?')}) {p.get('mass', 0):.1f} kg {p.get('temp', 0):.0f}°C")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Temperature
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_temp_zones(args):
|
|
"""Analyze temperature zones. Usage: temp_zones"""
|
|
data = api_get('/api/state/temperature/zones')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Temperature Analysis:")
|
|
print(f" Average: {data.get('averageC', 0):.0f}°C")
|
|
print(f" Min: {data.get('minC', 0):.0f}°C")
|
|
print(f" Max: {data.get('maxC', 0):.0f}°C")
|
|
print(f" Samples: {data.get('sampleCount', 0)}")
|
|
print()
|
|
hotspots = data.get('hotSpots', [])
|
|
if hotspots:
|
|
print(f"Hot spots (>50°C):")
|
|
for h in hotspots:
|
|
print(f" ({h.get('x', '?')},{h.get('y', '?')}) {h.get('tempC', 0):.0f}°C")
|
|
coldspots = data.get('coldSpots', [])
|
|
if coldspots:
|
|
print(f"Cold spots (<5°C):")
|
|
for c in coldspots:
|
|
print(f" ({c.get('x', '?')},{c.get('y', '?')}) {c.get('tempC', 0):.0f}°C")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Morale
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_morale(args):
|
|
"""Check morale. Usage: morale"""
|
|
data = api_get('/api/state/morale')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
for m in data if isinstance(data, list) else []:
|
|
print(f" {m.get('name', '?'):12s} morale={m.get('morale', 0):.0f} qol={m.get('qualityOfLife', 0):.0f}")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Diseases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_diseases(args):
|
|
"""Check disease status. Usage: diseases"""
|
|
data = api_get('/api/state/diseases')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
infected = data.get('infectedDuplicants', [])
|
|
if infected:
|
|
print(f"Infected dupes: {len(infected)}")
|
|
for d in infected:
|
|
print(f" {d.get('duplicant', '?')} - {d.get('disease', '?')} ({d.get('severity', '?')})")
|
|
else:
|
|
print("No infected duplicants.")
|
|
germs = data.get('environmentGerms', {})
|
|
if germs:
|
|
print(f"Environmental germs:")
|
|
for name, count in sorted(germs.items(), key=lambda x: -x[1])[:5]:
|
|
print(f" {name}: {count:.0f}")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Storage
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_storage(args):
|
|
"""Check storage. Usage: storage"""
|
|
data = api_get('/api/state/storage')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Storage buildings: {data.get('storageCount', 0)}")
|
|
for s in data.get('storages', [])[:10]:
|
|
print(f" {s.get('building', '?'):25s} ({s.get('x', '?')},{s.get('y', '?')}) "
|
|
f"{s.get('totalMass', 0):.0f}/{s.get('capacity', 0):.0f} kg")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skills / Job
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_skills(args):
|
|
"""Show duplicant skills. Usage: skills"""
|
|
data = api_get('/api/state/duplicants/skills')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
for d in data if isinstance(data, list) else []:
|
|
print(f" {d.get('name', '?'):12s} skill_pts={d.get('skillPoints', 0)} total_pts={d.get('totalSkillPointsGained', 0)}")
|
|
for a in d.get('attributes', [])[:5]:
|
|
print(f" {a.get('name', '?'):15s} = {a.get('value', 0)}")
|
|
|
|
def cmd_assign_job(args):
|
|
"""Assign dupe to a job. Usage: assign_job <dupe_name> <chore_group>"""
|
|
if len(args) < 2:
|
|
print("Usage: assign_job <dupe_name> <chore_group>")
|
|
print(" chore groups: Build, Dig, Cook, Farm, Ranch, Operate, Research, Store, Tidy, LifeSupport")
|
|
return
|
|
result = api_post('/api/action/assign_job', {
|
|
"duplicantId": args[0],
|
|
"choreGroup": args[1]
|
|
})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pipe / Wire Lines
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_build_pipe_line(args):
|
|
"""Build a pipe line with crossing mode. Usage: build_pipe_line <gas|liquid> <x1> <y1> <x2> <y2> [mode]"""
|
|
if len(args) < 5:
|
|
print("Usage: build_pipe_line gas|liquid <x1> <y1> <x2> <y2> [mode]")
|
|
print(" mode: 'line' (default, auto-merge), 'cross' (use bridges at intersections), 'single' (one segment)")
|
|
return
|
|
ptype = args[0]
|
|
x1, y1, x2, y2 = int(args[1]), int(args[2]), int(args[3]), int(args[4])
|
|
mode = args[5] if len(args) > 5 else 'line'
|
|
|
|
if mode not in ('line', 'cross', 'single'):
|
|
print("[!] mode must be 'line', 'cross', or 'single'")
|
|
return
|
|
|
|
result = api_post('/api/action/build_pipe', {
|
|
"type": ptype, "x1": x1, "y1": y1,
|
|
"x2": x2, "y2": y2, "mode": mode
|
|
})
|
|
if result.get('success'):
|
|
d = result.get('data', {})
|
|
segs = d.get('segmentCount', 0)
|
|
bridges = d.get('bridgesPlaced', 0)
|
|
xing = sum(1 for s in d.get('segments', []) if s.get('crossing'))
|
|
print(f"[OK] {ptype} pipe: {segs} segments, {bridges} bridges, {xing} crossings handled")
|
|
if xing > 0:
|
|
print(f" Mode '{mode}': {'bridges used at crossings' if mode == 'cross' else 'merged at crossings'}")
|
|
else:
|
|
_print_feedback(result)
|
|
|
|
def cmd_build_wire_line(args):
|
|
"""Build a wire line with crossing mode. Usage: build_wire_line <type> <x1> <y1> <x2> <y2> [mode]"""
|
|
if len(args) < 5:
|
|
print("Usage: build_wire_line regular|heavy|conductive <x1> <y1> <x2> <y2> [mode]")
|
|
print(" mode: 'line' (default, auto-merge), 'cross' (use bridges at intersections), 'single' (one segment)")
|
|
return
|
|
wtype = args[0]
|
|
x1, y1, x2, y2 = int(args[1]), int(args[2]), int(args[3]), int(args[4])
|
|
mode = args[5] if len(args) > 5 else 'line'
|
|
|
|
if mode not in ('line', 'cross', 'single'):
|
|
print("[!] mode must be 'line', 'cross', or 'single'")
|
|
return
|
|
|
|
result = api_post('/api/action/build_wire', {
|
|
"type": wtype, "x1": x1, "y1": y1,
|
|
"x2": x2, "y2": y2, "mode": mode
|
|
})
|
|
if result.get('success'):
|
|
d = result.get('data', {})
|
|
segs = d.get('segmentCount', 0)
|
|
bridges = d.get('bridgesPlaced', 0)
|
|
xing = sum(1 for s in d.get('segments', []) if s.get('crossing'))
|
|
print(f"[OK] {wtype} wire: {segs} segments, {bridges} bridges, {xing} crossings handled")
|
|
else:
|
|
_print_feedback(result)
|
|
|
|
def _print_feedback(result):
|
|
"""Pretty-print action feedback."""
|
|
if result.get('success'):
|
|
print(f"OK: {result.get('result', 'done')}")
|
|
for k, v in result.get('data', {}).items():
|
|
print(f" {k}: {v}")
|
|
else:
|
|
print(f"FAIL: {result.get('error', 'unknown_error')}")
|
|
print(f" {result.get('errorMessage', '')}")
|
|
sug = result.get('suggestion')
|
|
if sug:
|
|
print(f" -> {sug}")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Buildable / Research / Building Interaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_buildable(args):
|
|
"""List all buildable buildings. Usage: buildable [filter]"""
|
|
filt = args[0].lower() if args else ""
|
|
data = api_get('/api/state/buildable')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
total = data.get('total', 0)
|
|
unlocked = data.get('unlocked', 0)
|
|
print(f"Buildings: {unlocked}/{total} unlocked")
|
|
print()
|
|
for b in data.get('buildings', []):
|
|
name = b.get('name', '?')
|
|
bid = b.get('id', '')
|
|
if filt and filt not in name.lower() and filt not in bid.lower():
|
|
continue
|
|
mark = "✓" if b.get('unlocked') else "✗"
|
|
cat = b.get('category', '?')
|
|
pw = f" {b.get('powerCost', 0)}W" if b.get('powerCost', 0) > 0 else ""
|
|
print(f" [{mark}] {name:30s} {bid:30s} {cat:15s}{pw}")
|
|
|
|
def cmd_toggle(args):
|
|
"""Toggle building on/off. Usage: toggle <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: toggle <x> <y>")
|
|
return
|
|
result = api_post('/api/action/toggle', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
def cmd_set_recipe(args):
|
|
"""Set a building's recipe. Usage: set_recipe <x> <y> <recipeId>"""
|
|
if len(args) < 3:
|
|
print("Usage: set_recipe <x> <y> <recipeId>")
|
|
return
|
|
result = api_post('/api/action/set_recipe', {
|
|
"x": int(args[0]), "y": int(args[1]), "recipeId": args[2]
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_empty(args):
|
|
"""Empty a building's storage. Usage: empty <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: empty <x> <y>")
|
|
return
|
|
result = api_post('/api/action/empty', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
def cmd_cancel_errand(args):
|
|
"""Cancel errands at a building. Usage: cancel_errand <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: cancel_errand <x> <y>")
|
|
return
|
|
result = api_post('/api/action/cancel_errand', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Research Enhancements
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_research_detail(args):
|
|
"""Detailed research status. Usage: research_detail"""
|
|
data = api_get('/api/state/research/detail')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print("=== Research Status ===")
|
|
print(f" Research Station: {'✓' if data.get('researchStationOperational') else '✗'} "
|
|
f"(built={data.get('hasResearchStation', False)})")
|
|
print(f" Super Computer: {'✓' if data.get('hasSuperComputer') else '✗'}")
|
|
active = data.get('activeResearch', [])
|
|
if active:
|
|
for a in active:
|
|
print(f" Active: {a.get('name', '?')} ({a.get('progress', 0)*100:.0f}%)")
|
|
else:
|
|
print(f" Active: none {'[all complete]' if data.get('researchComplete') else '[idle]'}")
|
|
stations = data.get('stations', [])
|
|
if stations:
|
|
print(f" Stations: {len(stations)}")
|
|
for s in stations:
|
|
op = 'ON' if s.get('isOperational') else 'OFF'
|
|
print(f" {s.get('name', '?')} at ({s.get('x', '?')},{s.get('y', '?')}) [{op}]")
|
|
|
|
def cmd_research_cancel(args):
|
|
"""Cancel research. Usage: research_cancel [techId]"""
|
|
payload = {"techId": args[0]} if args else {}
|
|
result = api_post('/api/action/research_cancel', payload)
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Building Detail
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_building_detail(args):
|
|
"""Detailed info about a building. Usage: building_detail <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: building_detail <x> <y>")
|
|
return
|
|
data = api_get(f"/api/state/building_detail?x={args[0]}&y={args[1]}")
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Building: {data.get('name', '?')} ({data.get('id', '?')})")
|
|
print(f" Position: ({data.get('x', '?')},{data.get('y', '?')}) size={data.get('width')}x{data.get('height')}")
|
|
print(f" Operational: {'ON' if data.get('isOperational') else 'OFF'}")
|
|
print(f" Powered: {'YES' if data.get('isPowered') else 'NO'} {data.get('powerWatt', 0)}W")
|
|
print(f" Health: {data.get('health', '?')}/{data.get('maxHealth', '?')}")
|
|
storage = data.get('storageItems', [])
|
|
if storage:
|
|
print(f" Storage: {data.get('storageMass', 0):.0f}/{data.get('storageCapacity', 0):.0f} kg")
|
|
for s in storage[:5]:
|
|
print(f" {s.get('name', '?')}: {s.get('mass', 0):.1f} kg")
|
|
print(f" Automation: {'YES' if data.get('hasAutomation') else 'NO'}")
|
|
print(f" Materials: {', '.join(data.get('material', []))}")
|
|
|
|
def cmd_set_building_priority(args):
|
|
"""Set a building's priority. Usage: set_building_priority <x> <y> <priority>"""
|
|
if len(args) < 3:
|
|
print("Usage: set_building_priority <x> <y> <priority 1-9>")
|
|
return
|
|
result = api_post('/api/action/set_building_priority', {
|
|
"x": int(args[0]), "y": int(args[1]), "priority": int(args[2])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_set_automation(args):
|
|
"""Toggle automation on a building. Usage: set_automation <x> <y> <on|off>"""
|
|
if len(args) < 3:
|
|
print("Usage: set_automation <x> <y> <on|off>")
|
|
return
|
|
enabled = args[2].lower() in ('on', 'true', '1', 'yes')
|
|
result = api_post('/api/action/set_automation', {
|
|
"x": int(args[0]), "y": int(args[1]), "enabled": enabled
|
|
})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Printing Pod
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_printing_pod(args):
|
|
"""Check Printing Pod status. Usage: printing_pod"""
|
|
data = api_get('/api/state/printing_pod')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
if data.get('isReady'):
|
|
print("=== Printing Pod: READY ===")
|
|
print(f" Options available:")
|
|
for opt in data.get('options', []):
|
|
print(f" [{opt.get('index')}] {opt.get('description', '?')} ({opt.get('type', '?')})")
|
|
print()
|
|
print(" Select with: python3 tools/oni_api.py printing_pod_select <0|1|2>")
|
|
else:
|
|
cycles = data.get('cyclesUntilNext', 0)
|
|
if cycles > 0:
|
|
print(f"Printing Pod: not ready ({cycles:.1f} cycles remaining)")
|
|
else:
|
|
print("Printing Pod: checking...")
|
|
|
|
def cmd_printing_pod_select(args):
|
|
"""Select a Printing Pod option. Usage: printing_pod_select <0|1|2>"""
|
|
if not args:
|
|
print("Usage: printing_pod_select <0|1|2>")
|
|
return
|
|
index = int(args[0])
|
|
if index < 0 or index > 2:
|
|
print("Index must be 0, 1, or 2")
|
|
return
|
|
result = api_post('/api/action/printing_pod_select', {"index": index})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Atmo Suits / Critter Attack / Door Control
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_atmo_suits(args):
|
|
"""Check atmo suit docks. Usage: atmo_suits"""
|
|
data = api_get('/api/state/atmo_suits')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Atmo Suit Docks: {data.get('dockCount', 0)}")
|
|
has = data.get('hasAtmoSuits', False)
|
|
print(f"Has suits available: {'YES' if has else 'NO'}")
|
|
for d in data.get('docks', []):
|
|
suit = "HAS SUIT" if d.get('hasSuit') else "empty"
|
|
o2 = d.get('o2Level', 0)
|
|
print(f" {d.get('name', '?'):20s} at ({d.get('x', '?')},{d.get('y', '?')}) "
|
|
f"{suit:10s} O2={o2:.0f} {'ON' if d.get('isOperational') else 'OFF'}")
|
|
|
|
def cmd_critter_attack(args):
|
|
"""Toggle critter attack. Usage: critter_attack <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: critter_attack <x> <y>")
|
|
return
|
|
result = api_post('/api/action/critter_attack', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
def cmd_door_lock(args):
|
|
"""Lock/unlock a door. Usage: door_lock <x> <y> <on|off>"""
|
|
if len(args) < 3:
|
|
print("Usage: door_lock <x> <y> <on|off>")
|
|
return
|
|
locked = args[2].lower() in ('on', 'true', '1', 'lock', 'locked', 'yes')
|
|
result = api_post('/api/action/door_lock', {
|
|
"x": int(args[0]), "y": int(args[1]), "locked": locked
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_door_one_way(args):
|
|
"""Set door one-way. Usage: door_one_way <x> <y> <left|right|up|down|none>"""
|
|
if len(args) < 3:
|
|
print("Usage: door_one_way <x> <y> <left|right|up|down|none>")
|
|
return
|
|
result = api_post('/api/action/door_one_way', {
|
|
"x": int(args[0]), "y": int(args[1]), "direction": args[2]
|
|
})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dupe / Critter / Plant / Sensor / Storage / Sweep / Disinfect
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_dupe_move(args):
|
|
"""Move a duplicant. Usage: dupe_move <name> <x> <y>"""
|
|
if len(args) < 3:
|
|
print("Usage: dupe_move <dupe_name> <x> <y>")
|
|
return
|
|
result = api_post('/api/action/dupe_move', {
|
|
"duplicantId": args[0], "x": int(args[1]), "y": int(args[2])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_dupe_cancel_task(args):
|
|
"""Cancel a dupe's current task. Usage: dupe_cancel_task <name>"""
|
|
if not args:
|
|
print("Usage: dupe_cancel_task <dupe_name>")
|
|
return
|
|
result = api_post('/api/action/dupe_cancel_task', {"duplicantId": args[0]})
|
|
_print_feedback(result)
|
|
|
|
def cmd_critter_wrangle(args):
|
|
"""Wrangle a critter. Usage: critter_wrangle <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: critter_wrangle <x> <y>")
|
|
return
|
|
result = api_post('/api/action/critter_wrangle', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
def cmd_plant_uproot(args):
|
|
"""Uproot a plant. Usage: plant_uproot <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: plant_uproot <x> <y>")
|
|
return
|
|
result = api_post('/api/action/plant_uproot', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
def cmd_storage_filter(args):
|
|
"""Set storage filter. Usage: storage_filter <x> <y> <filter_tag>"""
|
|
if len(args) < 3:
|
|
print("Usage: storage_filter <x> <y> <filter_tag>")
|
|
return
|
|
result = api_post('/api/action/storage_filter', {
|
|
"x": int(args[0]), "y": int(args[1]), "filter": args[2]
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_door_open(args):
|
|
"""Open/close a door. Usage: door_open <x> <y> <on|off>"""
|
|
if len(args) < 3:
|
|
print("Usage: door_open <x> <y> <on|off>")
|
|
return
|
|
open_door = args[2].lower() in ('on', 'true', '1', 'open', 'yes')
|
|
result = api_post('/api/action/door_open', {
|
|
"x": int(args[0]), "y": int(args[1]), "open": open_door
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_sensors(args):
|
|
"""List all sensors. Usage: sensors [filter]"""
|
|
filt = args[0].lower() if args else ""
|
|
data = api_get('/api/state/sensors')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Sensors: {data.get('sensorCount', 0)}")
|
|
for s in data.get('sensors', []):
|
|
sname = f"{s.get('type', '?')}Sensor"
|
|
if filt and filt not in sname.lower() and filt not in str(s.get('threshold', '')):
|
|
continue
|
|
thr = f" threshold={s.get('threshold')}" if s.get('threshold') is not None else ""
|
|
print(f" {sname:15s} at ({s.get('x', '?')},{s.get('y', '?')}){thr} {'ON' if s.get('isOperational') else 'OFF'}")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Extended interactions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_clear(args):
|
|
"""Clear debris. Usage: clear <x> <y> [radius]"""
|
|
if len(args) < 2:
|
|
print("Usage: clear <x> <y> [radius]")
|
|
return
|
|
payload = {"x": int(args[0]), "y": int(args[1]), "radius": int(args[2]) if len(args) > 2 else 1}
|
|
result = api_post('/api/action/clear', payload)
|
|
_print_feedback(result)
|
|
|
|
def cmd_rotate(args):
|
|
"""Rotate a building. Usage: rotate <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: rotate <x> <y>")
|
|
return
|
|
result = api_post('/api/action/rotate', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
def cmd_copy_settings(args):
|
|
"""Copy building settings. Usage: copy_settings <from_x> <from_y> <to_x> <to_y>"""
|
|
if len(args) < 4:
|
|
print("Usage: copy_settings <from_x> <from_y> <to_x> <to_y>")
|
|
return
|
|
result = api_post('/api/action/copy_settings', {
|
|
"x1": int(args[0]), "y1": int(args[1]), "x2": int(args[2]), "y2": int(args[3])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_overlay(args):
|
|
"""Set overlay view. Usage: overlay <type>"""
|
|
if not args:
|
|
print("Usage: overlay <power|temp|gas|liquid|automation|rooms|decor|light|germs|materials|suits|crops|radiation|priority|none>")
|
|
return
|
|
result = api_post('/api/action/overlay', {"type": args[0]})
|
|
_print_feedback(result)
|
|
|
|
def cmd_dupe_personal_priority(args):
|
|
"""Set dupe personal priority. Usage: dupe_personal_priority <name> <taskType> <priority 1-9>"""
|
|
if len(args) < 3:
|
|
print("Usage: dupe_personal_priority <name> <taskType> <priority 1-9>")
|
|
print(" taskType: Dig, Build, Cook, Farm, Ranch, Operate, Research, Store, Tidy, LifeSupport, Supply")
|
|
return
|
|
result = api_post('/api/action/dupe_personal_priority', {
|
|
"duplicantId": args[0], "taskType": args[1], "priority": int(args[2])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_battery_charge(args):
|
|
"""Set battery charge limits. Usage: battery_charge <x> <y> <high%> <low%>"""
|
|
if len(args) < 4:
|
|
print("Usage: battery_charge <x> <y> <high_percent> <low_percent>")
|
|
return
|
|
result = api_post('/api/action/battery_charge', {
|
|
"x": int(args[0]), "y": int(args[1]),
|
|
"high": int(args[2]), "low": int(args[3])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_valve_flow(args):
|
|
"""Set valve flow limit. Usage: valve_flow <x> <y> <flow_kg>"""
|
|
if len(args) < 3:
|
|
print("Usage: valve_flow <x> <y> <flow_kg>")
|
|
return
|
|
result = api_post('/api/action/valve_flow', {
|
|
"x": int(args[0]), "y": int(args[1]), "flow": float(args[2])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_vent_pressure(args):
|
|
"""Set vent overpressure. Usage: vent_pressure <x> <y> <pressure_kg>"""
|
|
if len(args) < 3:
|
|
print("Usage: vent_pressure <x> <y> <pressure_kg>")
|
|
return
|
|
result = api_post('/api/action/vent_pressure', {
|
|
"x": int(args[0]), "y": int(args[1]), "pressure": float(args[2])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_incubator_setting(args):
|
|
"""Set incubator egg priority. Usage: incubator_setting <x> <y> <egg_type>"""
|
|
if len(args) < 3:
|
|
print("Usage: incubator_setting <x> <y> <egg_type>")
|
|
return
|
|
result = api_post('/api/action/incubator_setting', {
|
|
"x": int(args[0]), "y": int(args[1]), "egg": args[2]
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_fridge_temp(args):
|
|
"""Set fridge temperature. Usage: fridge_temp <x> <y> <temp_c>"""
|
|
if len(args) < 3:
|
|
print("Usage: fridge_temp <x> <y> <temp_celsius>")
|
|
return
|
|
result = api_post('/api/action/fridge_temp', {
|
|
"x": int(args[0]), "y": int(args[1]), "temperature": float(args[2])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_sensor_threshold(args):
|
|
"""Set sensor threshold. Usage: sensor_threshold <x> <y> <value>"""
|
|
if len(args) < 3:
|
|
print("Usage: sensor_threshold <x> <y> <value>")
|
|
return
|
|
result = api_post('/api/action/sensor_threshold', {
|
|
"x": int(args[0]), "y": int(args[1]), "threshold": float(args[2])
|
|
})
|
|
_print_feedback(result)
|
|
|
|
def cmd_sweep(args):
|
|
"""Mark for sweeping. Usage: sweep <x> <y> [radius]"""
|
|
if len(args) < 2:
|
|
print("Usage: sweep <x> <y> [radius]")
|
|
return
|
|
payload = {"x": int(args[0]), "y": int(args[1])}
|
|
if len(args) > 2:
|
|
payload["radius"] = int(args[2])
|
|
result = api_post('/api/action/sweep', payload)
|
|
_print_feedback(result)
|
|
|
|
def cmd_disinfect(args):
|
|
"""Disinfect a cell. Usage: disinfect <x> <y>"""
|
|
if len(args) < 2:
|
|
print("Usage: disinfect <x> <y>")
|
|
return
|
|
result = api_post('/api/action/disinfect', {"x": int(args[0]), "y": int(args[1])})
|
|
_print_feedback(result)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Screenshot / Camera
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def cmd_snapshot(args):
|
|
"""Take a screenshot and save locally. Usage: snapshot [output.png]"""
|
|
output = args[0] if args else f"oni_snapshot_{int(__import__('time').time())}.png"
|
|
result = api_post('/api/action/screenshot', {})
|
|
if not result.get('success'):
|
|
print(f"[!] Screenshot failed: {result.get('errorMessage', '')}")
|
|
return
|
|
filename = result.get('data', {}).get('filename', '?')
|
|
print(f"[OK] Screenshot taken: {filename}")
|
|
|
|
# Download the image
|
|
try:
|
|
import urllib.request
|
|
from oni_api import api_url
|
|
url = api_url('/api/screenshot/latest')
|
|
urllib.request.urlretrieve(url, output)
|
|
print(f"[OK] Saved to {output} ({os.path.getsize(output)} bytes)")
|
|
except Exception as e:
|
|
print(f"[!] Download failed: {e}")
|
|
|
|
def cmd_camera(args):
|
|
"""Control camera. Usage: camera <x> <y> [zoom]"""
|
|
if len(args) < 2:
|
|
print("Usage: camera <x> <y> [zoom]")
|
|
print(" zoom: 5 (close) to 80 (far), default 30")
|
|
return
|
|
x, y = int(args[0]), int(args[1])
|
|
zoom = float(args[2]) if len(args) > 2 else None
|
|
payload = {"x": x, "y": y}
|
|
if zoom is not None:
|
|
payload["zoom"] = zoom
|
|
result = api_post('/api/action/camera', payload)
|
|
_print_feedback(result)
|
|
|
|
def cmd_camera_status(args):
|
|
"""Get current camera position. Usage: camera_status"""
|
|
data = api_get('/api/state/camera')
|
|
if 'error' in data:
|
|
print(f"Error: {data['error']}")
|
|
return
|
|
print(f"Camera at ({data.get('x', 0):.0f}, {data.get('y', 0):.0f}), zoom={data.get('zoom', 30):.0f}")
|
|
|
|
def cmd_explore(args):
|
|
"""AI-friendly exploration: reads a region and returns structured text summary."""
|
|
x = int(args[0]) if len(args) > 0 else 0
|
|
y = int(args[1]) if len(args) > 1 else 0
|
|
w = int(args[2]) if len(args) > 2 else 20
|
|
h = int(args[3]) if len(args) > 3 else 20
|
|
|
|
cells_data = api_get(f"/api/state/cells?x={x}&y={y}&width={w}&height={h}")
|
|
buildings_data = api_get("/api/state/buildings")
|
|
dups_data = api_get("/api/state/duplicants")
|
|
|
|
summary = {
|
|
"region": f"({x},{y}) to ({x+w-1},{y+h-1})",
|
|
"cell_count": len(cells_data.get('cells', [])),
|
|
"buildings_in_region": [],
|
|
"duplicants_in_region": [],
|
|
"elements_summary": {},
|
|
"interesting_cells": []
|
|
}
|
|
|
|
region_cells = cells_data.get('cells', [])
|
|
|
|
# Buildings overlapping region
|
|
if isinstance(buildings_data, list):
|
|
for b in buildings_data:
|
|
bx, by = b.get('x', -1), b.get('y', -1)
|
|
if x <= bx < x + w and y <= by < y + h:
|
|
summary["buildings_in_region"].append({
|
|
"name": b.get('name', '?'),
|
|
"id": b.get('id', '?'),
|
|
"position": (bx, by),
|
|
"operational": b.get('isOperational', False)
|
|
})
|
|
|
|
# Dupes in region
|
|
if isinstance(dups_data, list):
|
|
for d in dups_data:
|
|
dx, dy = d.get('x', -1), d.get('y', -1)
|
|
if x <= dx < x + w and y <= dy < y + h:
|
|
summary["duplicants_in_region"].append({
|
|
"name": d.get('name', '?'),
|
|
"position": (dx, dy),
|
|
"stress": d.get('stress', 0),
|
|
"chore": d.get('currentChore', '?')
|
|
})
|
|
|
|
# Element summary
|
|
elem_counts = {}
|
|
for c in region_cells:
|
|
ename = c.get('element', 'Vacuum')
|
|
elem_counts[ename] = elem_counts.get(ename, 0) + 1
|
|
summary["elements_summary"] = elem_counts
|
|
|
|
# Interesting cells (buildings, dupes, liquids, hot)
|
|
for c in region_cells:
|
|
if c.get('hasDuplicant') or c.get('hasBuilding') or (c.get('isLiquid') and c.get('massKg', 0) > 100):
|
|
summary["interesting_cells"].append({
|
|
"pos": (c.get('x'), c.get('y')),
|
|
"element": c.get('element'),
|
|
"mass_kg": c.get('massKg', 0),
|
|
"temp_c": c.get('temperatureC', 0),
|
|
"building": c.get('buildingName'),
|
|
"dupe": c.get('duplicantName')
|
|
})
|
|
|
|
print(json.dumps(summary, indent=2, ensure_ascii=False))
|
|
|
|
|
|
COMMANDS = {
|
|
'health': cmd_health,
|
|
'status': cmd_status,
|
|
'resources': cmd_resources,
|
|
'duplicants': cmd_duplicants,
|
|
'buildings': cmd_buildings,
|
|
'research': cmd_research,
|
|
'geysers': cmd_geysers,
|
|
'critters': cmd_critters,
|
|
'plants': cmd_plants,
|
|
'rooms': cmd_rooms,
|
|
'queue': cmd_queue,
|
|
'events': cmd_events,
|
|
'cell': cmd_cell,
|
|
'cells': cmd_cells,
|
|
'slice': cmd_cell_slice,
|
|
'gas': cmd_gas,
|
|
'registry': cmd_registry,
|
|
'dig': cmd_dig,
|
|
'build': cmd_build,
|
|
'deconstruct': cmd_deconstruct,
|
|
'prioritize': cmd_prioritize,
|
|
'research_select': cmd_research_select,
|
|
'mop': cmd_mop,
|
|
'harvest': cmd_harvest,
|
|
'explore': cmd_explore,
|
|
'batch': cmd_batch,
|
|
'priority_global': cmd_priority_global,
|
|
'priority_type': cmd_priority_type,
|
|
'pause': cmd_pause,
|
|
'unpause': cmd_unpause,
|
|
'speed': cmd_speed,
|
|
'save': cmd_save,
|
|
'save_as': cmd_save_as,
|
|
'load': cmd_load,
|
|
'saves': cmd_saves,
|
|
'power': cmd_power,
|
|
'pipes': cmd_pipes,
|
|
'co2': cmd_co2,
|
|
'temp_zones': cmd_temp_zones,
|
|
'morale': cmd_morale,
|
|
'diseases': cmd_diseases,
|
|
'storage': cmd_storage,
|
|
'skills': cmd_skills,
|
|
'assign_job': cmd_assign_job,
|
|
'build_pipe_line': cmd_build_pipe_line,
|
|
'build_wire_line': cmd_build_wire_line,
|
|
'snapshot': cmd_snapshot,
|
|
'camera': cmd_camera,
|
|
'camera_status': cmd_camera_status,
|
|
'buildable': cmd_buildable,
|
|
'toggle': cmd_toggle,
|
|
'set_recipe': cmd_set_recipe,
|
|
'empty': cmd_empty,
|
|
'cancel_errand': cmd_cancel_errand,
|
|
'research_detail': cmd_research_detail,
|
|
'research_cancel': cmd_research_cancel,
|
|
'building_detail': cmd_building_detail,
|
|
'set_building_priority': cmd_set_building_priority,
|
|
'set_automation': cmd_set_automation,
|
|
'printing_pod': cmd_printing_pod,
|
|
'printing_pod_select': cmd_printing_pod_select,
|
|
'atmo_suits': cmd_atmo_suits,
|
|
'critter_attack': cmd_critter_attack,
|
|
'door_lock': cmd_door_lock,
|
|
'door_one_way': cmd_door_one_way,
|
|
'dupe_move': cmd_dupe_move,
|
|
'dupe_cancel_task': cmd_dupe_cancel_task,
|
|
'critter_wrangle': cmd_critter_wrangle,
|
|
'plant_uproot': cmd_plant_uproot,
|
|
'storage_filter': cmd_storage_filter,
|
|
'door_open': cmd_door_open,
|
|
'sensors': cmd_sensors,
|
|
'sensor_threshold': cmd_sensor_threshold,
|
|
'sweep': cmd_sweep,
|
|
'disinfect': cmd_disinfect,
|
|
'clear': cmd_clear,
|
|
'rotate': cmd_rotate,
|
|
'copy_settings': cmd_copy_settings,
|
|
'overlay': cmd_overlay,
|
|
'dupe_personal_priority': cmd_dupe_personal_priority,
|
|
'battery_charge': cmd_battery_charge,
|
|
'valve_flow': cmd_valve_flow,
|
|
'vent_pressure': cmd_vent_pressure,
|
|
'incubator_setting': cmd_incubator_setting,
|
|
'fridge_temp': cmd_fridge_temp,
|
|
}
|
|
|
|
if __name__ == '__main__':
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else 'help'
|
|
|
|
if cmd == 'help' or cmd not in COMMANDS:
|
|
print("ONI Agent API Client")
|
|
print("=" * 60)
|
|
print("")
|
|
print("=== State Queries ===")
|
|
print(" health Check Mod connection")
|
|
print(" status Game overview (cycle, resources, dups, alerts)")
|
|
print(" resources [filter] List all resources with amounts")
|
|
print(" duplicants Show duplicant details")
|
|
print(" buildings List all buildings (grouped by category)")
|
|
print(" research Show research tree progress")
|
|
print(" geysers Show geyser states")
|
|
print(" critters Show critter list")
|
|
print(" plants Show plant list")
|
|
print(" rooms Show rooms")
|
|
print("")
|
|
print("=== Map / Cell Data ===")
|
|
print(" cell <x> <y> Get single cell details")
|
|
print(" cells <x> <y> <w> <h> Get region as grid")
|
|
print(" slice <x|y> <idx> <s> <e> Row/column scan")
|
|
print(" gas <x> <y> [r] Gas analysis in radius r")
|
|
print(" explore <x> <y> <w> <h> AI-friendly region summary")
|
|
print("")
|
|
print("=== Door / Critter / Suits ===")
|
|
print(" door_lock <x> <y> on|off Lock/unlock a door")
|
|
print(" door_one_way <x> <y> <dir> Set door to one-way (left/right/up/down/none)")
|
|
print(" door_open <x> <y> on|off Open/close a door manually")
|
|
print(" critter_attack <x> <y> Toggle critter attack mode")
|
|
print(" critter_wrangle <x> <y> Wrangle a critter")
|
|
print(" atmo_suits Check atmo suit dock status")
|
|
print("")
|
|
print("=== Duplicant / Plant ===")
|
|
print(" dupe_move <name> <x> <y> Move a duplicant to coordinates")
|
|
print(" dupe_cancel_task <name> Cancel a dupe's current task")
|
|
print(" plant_uproot <x> <y> Uproot a plant")
|
|
print("")
|
|
print("=== Sensors / Automation ===")
|
|
print(" sensors [filter] List all sensors and thresholds")
|
|
print(" sensor_threshold <x> <y> <v> Set sensor threshold")
|
|
print("")
|
|
print("=== Storage / Cleaning ===")
|
|
print(" storage_filter <x> <y> <tag> Set storage building filter")
|
|
print(" sweep <x> <y> [r] Mark area for sweeping")
|
|
print(" disinfect <x> <y> Disinfect a cell/building")
|
|
print(" empty <x> <y> Empty building storage")
|
|
print(" events [since] [limit] Poll new game events")
|
|
print(" printing_pod Check Printing Pod status (ready/options)")
|
|
print(" printing_pod_select <0|1|2> Select Printing Pod option")
|
|
print("")
|
|
print("=== Registries (AI Reference) ===")
|
|
print(" registry buildings [f] List all building IDs with metadata")
|
|
print(" registry elements [f] List all element IDs with properties")
|
|
print(" registry techs [f] List all tech IDs with unlocks")
|
|
print(" registry priorities Show priority level meanings")
|
|
print("")
|
|
print("=== Research / Buildable ===")
|
|
print(" buildable [filter] List buildings unlocked by current research")
|
|
print(" research Show research tree progress")
|
|
print(" research_detail Detailed research status (active tech / stations)")
|
|
print(" research_select <id> Select tech to research")
|
|
print(" research_cancel [id] Cancel research (all or specific tech)")
|
|
print("")
|
|
print("=== Building Interaction ===")
|
|
print(" building_detail <x> <y> Full detail for a building (health/contents/automation)")
|
|
print(" toggle <x> <y> Toggle building on/off")
|
|
print(" set_recipe <x> <y> <id> Set building recipe")
|
|
print(" empty <x> <y> Empty building storage")
|
|
print(" cancel_errand <x> <y> Cancel errands at building")
|
|
print(" set_building_priority <x> <y> <p> Set building priority 1-9")
|
|
print(" set_automation <x> <y> on|off Toggle automation input")
|
|
print("")
|
|
print("=== Game Speed Control ===")
|
|
print(" pause [reason] Pause the game (AI should always pause before ops)")
|
|
print(" unpause [speed] Unpause the game at 1x/2x/3x speed")
|
|
print(" speed <1|2|3> Set game speed while running")
|
|
print("")
|
|
print("=== Save / Load ===")
|
|
print(" save [name] Save game (AI undo button)")
|
|
print(" save_as <name> Save with custom name")
|
|
print(" saves List save files")
|
|
print(" load <name> Load a save (rollback)")
|
|
print("")
|
|
print("=== Grid Analysis ===")
|
|
print(" power Power grid (circuits, load, overloads)")
|
|
print(" pipes [gas|liquid|all] Pipe contents (debug plumbing)")
|
|
print(" co2 Find CO2 pockets (colony killer!)")
|
|
print(" temp_zones Temperature hot/cold spots")
|
|
print("")
|
|
print("=== Colony Management ===")
|
|
print(" storages Storage building contents")
|
|
print(" diseases Disease/infection overview")
|
|
print(" skills Duplicant skills and attributes")
|
|
print(" assign_job <name> <job> Assign dupe to a chore group")
|
|
print(" morale Duplicant morale levels")
|
|
print("")
|
|
print("=== Screenshot / Camera ===")
|
|
print(" snapshot [file.png] Take screenshot + save locally")
|
|
print(" camera <x> <y> [zoom] Move camera view to coordinates")
|
|
print(" camera_status Get current camera position/zoom")
|
|
print("")
|
|
print("=== Batch / Priority (Advanced) ===")
|
|
print(" batch <json_file> Execute batch plan")
|
|
print(" priority_global <t> <p> Set global default priority")
|
|
print(" priority_type <type> <p> Set per-building-type priority")
|
|
else:
|
|
COMMANDS[cmd](sys.argv[2:])
|