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') 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 api_get(endpoint): 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): 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) 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}") 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, } 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("=== 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 Dig area") print(" build Place building") print(" deconstruct Remove building") print(" prioritize

Set priority") print(" research_select Select tech to research") print(" mop Mop liquid") print(" harvest 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("=== 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:])