80485d2a5a
- All api_get/api_post calls now auto-pause game before executing - AUTO_PAUSE_ENABLED flag to disable for event daemon polling - Event daemon explicitly disables auto-pause (frequent light polling) - Iron rules updated: Rule 1: AI must pause BEFORE any reasoning/decision Rule 2: Tools auto-pause on state data fetch Rule 3: AI must pause before asking user questions Rule 4: AI must ensure paused when exiting response state Rule 5: Save before major operations
1041 lines
40 KiB
Python
1041 lines
40 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}")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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,
|
|
}
|
|
|
|
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("=== Event / Queue ===")
|
|
print(" events [since] [limit] Poll new game events")
|
|
print(" queue [batch_id] View pending task queue")
|
|
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("=== Actions ===")
|
|
print(" dig <x> <y> <w> <h> Dig area")
|
|
print(" build <id> <x> <y> Place building")
|
|
print(" deconstruct <id> <x> <y> Remove building")
|
|
print(" prioritize <x> <y> <p> Set priority")
|
|
print(" research_select <id> Select tech to research")
|
|
print(" mop <x> <y> Mop liquid")
|
|
print(" harvest <x> <y> Harvest plant")
|
|
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:])
|