220 lines
7.8 KiB
Python
220 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
ONI Commander — High-Level Game Operations
|
|
===========================================
|
|
Combines multiple low-level API calls into one "commander directive".
|
|
|
|
Usage:
|
|
python oni_commander.py diagnose Full game diagnostic
|
|
python oni_commander.py fix_co2 Auto-vent CO2 pockets
|
|
python oni_commander.py fix_overload Diagnose power overloads
|
|
python oni_commander.py emergency_o2 Emergency oxygen setup
|
|
python oni_commander.py expand_base <cx> <cy> <w> <h> Dig expansion area
|
|
"""
|
|
|
|
import json, sys, io
|
|
# Fix GBK encoding
|
|
if sys.stdout.encoding and sys.stdout.encoding.upper() in ('GBK', 'GB2312', 'CP936'):
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
|
|
|
from oni_api import api_get, api_post, get_data
|
|
|
|
def pause_before():
|
|
api_post('/api/action/pause', {"reason": "commander operation"})
|
|
|
|
def unpause_after():
|
|
api_post('/api/action/unpause', {"speed": 1})
|
|
|
|
def cmd_diagnose(args):
|
|
pause_before()
|
|
game = get_data('/api/state/game')
|
|
resources = get_data('/api/state/resources') or []
|
|
buildings = get_data('/api/state/buildings') or []
|
|
dups = get_data('/api/state/duplicants') or []
|
|
alerts = get_data('/api/state/alert') or []
|
|
|
|
rdict = {}
|
|
for r in resources if isinstance(resources, list) else []:
|
|
rdict[r.get('name','')] = r.get('amountKg', 0)
|
|
|
|
btypes = {}
|
|
for b in buildings if isinstance(buildings, list) else []:
|
|
bid = b.get('id', '')
|
|
btypes[bid] = btypes.get(bid, 0) + 1
|
|
|
|
print("=" * 60)
|
|
print(" ONI Full Diagnostic")
|
|
print("=" * 60)
|
|
if game:
|
|
speed_str = f"{game.get('gameSpeed','?')}x" if not game.get('isPaused') else "PAUSED"
|
|
print(f" Cycle {game.get('cycle','?')} | {game.get('duplicantCount','?')} dupes | {speed_str}")
|
|
|
|
print(f"\n── Resources ──")
|
|
for name in ['Oxygen', 'Water', 'PollutedWater', 'Dirt', 'Carbon', 'Hydrogen']:
|
|
v = rdict.get(name, 0)
|
|
print(f" {name:20s} {v:>10.1f} kg")
|
|
|
|
print(f"\n── Buildings ({len(btypes)} types) ──")
|
|
for bid, cnt in sorted(btypes.items()):
|
|
print(f" {bid:30s} x{cnt}")
|
|
|
|
print(f"\n── Dupes ({len(dups)}) ──")
|
|
for d in dups if isinstance(dups, list) else []:
|
|
print(f" {d.get('name','?'):12s} ({d.get('x',0)},{d.get('y',0)}) hp={d.get('health',0)}")
|
|
|
|
print(f"\n── Alerts ({len(alerts)}) ──")
|
|
for a in alerts:
|
|
print(f" [{a.get('severity','?')}] {a.get('title','?')}")
|
|
|
|
# Suggestions
|
|
print(f"\n── Suggestions ──")
|
|
o2 = rdict.get('Oxygen', 0)
|
|
cal = rdict.get('Calories', 0)
|
|
water = rdict.get('Water', 0)
|
|
coal = rdict.get('Carbon', 0)
|
|
|
|
if o2 < 500: print(f" 🔴 O2 CRISIS: {o2:.0f} kg — build SPOM immediately")
|
|
elif o2 < 2000: print(f" 🟡 O2 low: {o2:.0f} kg — plan oxygen production")
|
|
|
|
if cal < 200000: print(f" 🔴 FOOD CRISIS: {cal:.0f} kcal — build farm")
|
|
elif cal < 500000: print(f" 🟡 Food low: {cal:.0f} kcal")
|
|
|
|
if water < 5000: print(f" 🔴 WATER CRISIS: {water:.0f} kg")
|
|
|
|
if coal < 1000 and 'CoalGenerator' in btypes:
|
|
print(f" 🟡 Coal low ({coal:.0f} kg) — diversify power")
|
|
|
|
unpause_after()
|
|
|
|
def cmd_fix_co2(args):
|
|
"""Find CO2 pockets below base and dig to let it settle."""
|
|
pause_before()
|
|
game = get_data('/api/state/game')
|
|
if not game:
|
|
print("Cannot connect")
|
|
return
|
|
|
|
gw, gh = game.get('gridWidth', 256), game.get('gridHeight', 384)
|
|
print("Scanning for CO2 pockets...")
|
|
|
|
# Scan bottom portion of the map for CO2
|
|
scan_y = max(0, gh - 40)
|
|
r = api_get(f'/api/state/cells/slice?axis=y&index={scan_y}&start=0&end={gw-1}')
|
|
if not r.get("success"):
|
|
print("Cannot scan area")
|
|
unpause_after()
|
|
return
|
|
|
|
cells = r.get("data", {}).get("cells", [])
|
|
co2_cells = [c for c in cells if c.get('element') == 'CarbonDioxide' and c.get('isSolid') == False]
|
|
if not co2_cells:
|
|
print("No accessible CO2 pockets found in scan area")
|
|
unpause_after()
|
|
return
|
|
|
|
print(f"Found {len(co2_cells)} CO2 cells. Digging to the right for ventilation...")
|
|
for c in co2_cells[:5]:
|
|
dig_x = c['x'] + 1
|
|
r2 = api_post('/api/action/dig', {"x": dig_x, "y": scan_y, "width": 3, "height": 3})
|
|
if r2.get("success"):
|
|
print(f" Dig at ({dig_x},{scan_y}) 3x3 — queued")
|
|
|
|
unpause_after()
|
|
|
|
def cmd_fix_overload(args):
|
|
pause_before()
|
|
print("Diagnosing power...")
|
|
buildings = get_data('/api/state/buildings') or []
|
|
resources = get_data('/api/state/resources') or []
|
|
rdict = {}
|
|
for r in resources if isinstance(resources, list) else []:
|
|
rdict[r.get('name','')] = r.get('amountKg', 0)
|
|
|
|
btypes = {}
|
|
for b in buildings if isinstance(buildings, list) else []:
|
|
bid = b.get('id', '')
|
|
btypes[bid] = btypes.get(bid, 0) + 1
|
|
|
|
has_coal = 'CoalGenerator' in btypes
|
|
has_hydro = 'HydrogenGenerator' in btypes
|
|
has_manual = 'ManualGenerator' in btypes
|
|
coal = rdict.get('Carbon', 0)
|
|
|
|
print(f" Coal: {coal:.0f} kg")
|
|
print(f" Generators: Manual={has_manual} Coal={has_coal} Hydrogen={has_hydro}")
|
|
print(f" Coal plants: {btypes.get('CoalGenerator', 0)}")
|
|
print(f" Total buildings: {len(buildings)}")
|
|
|
|
if has_coal and coal < 2000:
|
|
print(f" ⚠ Low coal — supplement with manual generators")
|
|
if not has_hydro and 'Electrolyzer' in btypes:
|
|
print(f" ⚠ Wasteful: Electrolyzer running without HydrogenGenerator")
|
|
unpause_after()
|
|
|
|
def cmd_emergency_o2(args):
|
|
pause_before()
|
|
print("Emergency O2 response...")
|
|
resources = get_data('/api/state/resources') or []
|
|
buildings = get_data('/api/state/buildings') or []
|
|
rdict = {}
|
|
for r in resources if isinstance(resources, list) else []:
|
|
rdict[r.get('name','')] = r.get('amountKg', 0)
|
|
|
|
has_diffuser = any(b.get('id') == 'OxygenDiffuser' for b in buildings if isinstance(buildings, list))
|
|
algae = rdict.get('Algae', 0)
|
|
|
|
if has_diffuser:
|
|
print(" ✓ OxygenDiffuser exists — ensure it has power and algae")
|
|
elif algae > 200:
|
|
print(" Building OxygenDiffuser (uses algae)...")
|
|
r = api_post('/api/action/build', {"buildingId": "OxygenDiffuser", "x": 30, "y": 30})
|
|
if r.get("success"):
|
|
print(" → OxygenDiffuser queued")
|
|
else:
|
|
print(f" → Build failed: {r.get('errorMessage', r.get('error', 'unknown'))}")
|
|
else:
|
|
print(" No algae for diffuser. Need SPOM (Electrolyzer)")
|
|
water = rdict.get('Water', 0)
|
|
if water > 5000:
|
|
print(f" Water: {water:.0f} kg — enough for SPOM")
|
|
else:
|
|
print(f" Water: {water:.0f} kg — insufficient for electrolysis")
|
|
unpause_after()
|
|
|
|
def cmd_expand_base(args):
|
|
if len(args) < 4:
|
|
print("Usage: expand_base <center_x> <center_y> <width> <height>", file=sys.stderr)
|
|
return
|
|
cx, cy, w, h = int(args[0]), int(args[1]), int(args[2]), int(args[3])
|
|
dig_x, dig_y = cx - w // 2, cy - h // 2
|
|
pause_before()
|
|
print(f"Expanding: dig ({dig_x},{dig_y}) {w}x{h}")
|
|
r = api_post('/api/action/dig', {"x": dig_x, "y": dig_y, "width": w, "height": h})
|
|
if r.get("success"):
|
|
info = r.get("data", {})
|
|
print(f" Queued {info.get('count', '?')} dig orders")
|
|
else:
|
|
print(f" Error: {r.get('errorMessage', r.get('error', 'unknown'))}")
|
|
unpause_after()
|
|
|
|
def main():
|
|
if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help"):
|
|
print(__doc__)
|
|
return
|
|
cmd = sys.argv[1]
|
|
args = sys.argv[2:]
|
|
|
|
cmds = {
|
|
"diagnose": cmd_diagnose, "fix_co2": cmd_fix_co2,
|
|
"fix_overload": cmd_fix_overload, "emergency_o2": cmd_emergency_o2,
|
|
"expand_base": cmd_expand_base,
|
|
}
|
|
if cmd in cmds:
|
|
cmds[cmd](args)
|
|
else:
|
|
print(f"Unknown: {cmd}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|