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 ") 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 ") 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 [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 [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 [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 ") 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 [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 ") 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 ") 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 ") 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 ") 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 ") 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 """ if not args: print("Usage: batch ") 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 """ if len(args) < 2: print("Usage: priority_global ") 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 """ if len(args) < 2: print("Usage: priority_type ") 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 """ if not args: print("Usage: save_as ") return result = api_post('/api/action/save_as', {"name": ' '.join(args)}) _print_feedback(result) def cmd_load(args): """Load a save. Usage: load """ if not args: print("Usage: load ") 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 """ if len(args) < 2: print("Usage: assign_job ") 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 [mode]""" if len(args) < 5: print("Usage: build_pipe_line gas|liquid [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 [mode]""" if len(args) < 5: print("Usage: build_wire_line regular|heavy|conductive [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 """ if len(args) < 2: print("Usage: toggle ") 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 """ if len(args) < 3: print("Usage: set_recipe ") 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 """ if len(args) < 2: print("Usage: empty ") 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 """ if len(args) < 2: print("Usage: cancel_errand ") 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 """ if len(args) < 2: print("Usage: building_detail ") 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 """ if len(args) < 3: print("Usage: set_building_priority ") 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 """ if len(args) < 3: print("Usage: set_automation ") 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 """ if len(args) < 2: print("Usage: critter_attack ") 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 """ if len(args) < 3: print("Usage: door_lock ") 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 """ if len(args) < 3: print("Usage: door_one_way ") return result = api_post('/api/action/door_one_way', { "x": int(args[0]), "y": int(args[1]), "direction": args[2] }) _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 [zoom]""" if len(args) < 2: print("Usage: camera [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, } 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 Get single cell details") print(" cells Get region as grid") print(" slice Row/column scan") print(" gas [r] Gas analysis in radius r") print(" explore AI-friendly region summary") print("") print("=== Door / Critter / Suits ===") print(" door_lock on|off Lock/unlock a door") print(" door_one_way Set door to one-way (left/right/up/down/none)") print(" critter_attack Toggle critter attack mode") print(" atmo_suits Check atmo suit dock status") print("") 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 Select tech to research") print(" research_cancel [id] Cancel research (all or specific tech)") print("") print("=== Building Interaction ===") print(" building_detail Full detail for a building (health/contents/automation)") print(" toggle Toggle building on/off") print(" set_recipe Set building recipe") print(" empty Empty building storage") print(" cancel_errand Cancel errands at building") print(" set_building_priority

Set building priority 1-9") print(" set_automation 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 Save with custom name") print(" saves List save files") print(" load 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 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 [zoom] Move camera view to coordinates") print(" camera_status Get current camera position/zoom") print("") print("=== Batch / Priority (Advanced) ===") print(" batch Execute batch plan") print(" priority_global

Set global default priority") print(" priority_type

Set per-building-type priority") else: COMMANDS[cmd](sys.argv[2:])