6a7ac87c1b
- Add cell/tile map APIs: /api/state/cell, /api/state/cells, /api/state/cells/slice, /api/state/gas - Add entity registry APIs: /api/registry/buildings, /api/registry/elements, /api/registry/techs - Add plants, rooms, mop, harvest endpoints - Rich semantic metadata: element state/category, building category/power, duplicant chore/cell - AI-friendly coordinate system with (x,y) + cell index in all responses - Build AI_KNOWLEDGE_BASE.md with building IDs, element IDs, tech trees, game mechanics - Rewrite SKILL.md with data model explanation, coordinate guide, operation patterns - Update Python tools: explore, cell, cells, slice, gas, registry subcommands - Update MOD_DEV_GUIDE.md with AI data design principles
491 lines
19 KiB
Python
491 lines
19 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')
|
|
|
|
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 <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))
|
|
|
|
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,
|
|
'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,
|
|
}
|
|
|
|
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("=== 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("")
|
|
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")
|
|
else:
|
|
COMMANDS[cmd](sys.argv[2:])
|