feat: complete ONI Agent project with full Mod API, Python toolchain, and development guide

- Implement all Mod API endpoints (buildings, research, geysers, alerts, critters, deconstruct, prioritize, research, schedule, wardrobe)
- Enhance Python tools with comprehensive CLI, analysis (O2/food/power/temp/water), and 7 blueprints (SPOM, toilet, ranch, farm, cooling, bedroom)
- Add utility scripts: auto_repair, auto_analyze, watch mode, setup
- Write Agent-Mod integration constraints and development guide
- Create skills directory with ONI agent skill definition
This commit is contained in:
root
2026-05-22 08:44:56 +08:00
commit 0e17e8a6ac
14 changed files with 1579 additions and 0 deletions

221
tools/oni_analyzer.py Normal file
View File

@ -0,0 +1,221 @@
import json
import sys
from oni_api import api_get, api_post
def get_game_state():
return {
'game': api_get('/api/state/game'),
'resources': api_get('/api/state/resources'),
'duplicants': api_get('/api/state/duplicants'),
'buildings': api_get('/api/state/buildings'),
'research': api_get('/api/state/research'),
'geysers': api_get('/api/state/geysers'),
'alerts': api_get('/api/state/alert'),
'critters': api_get('/api/state/critters'),
}
def as_dict(resources):
if not isinstance(resources, list):
return {}
return {r.get('name'): r for r in resources}
def analyze_o2(resources):
r = as_dict(resources)
o2 = r.get('Oxygen', {}).get('amount', 0)
algae = r.get('Algae', {}).get('amount', 0)
pw = r.get('PollutedWater', {}).get('amount', 0)
warnings = []
if o2 < 100:
warnings.append(("CRITICAL", f"Oxygen critically low ({o2:.0f} kg)"))
elif o2 < 500:
warnings.append(("WARN", f"Oxygen declining ({o2:.0f} kg)"))
if algae < 1000:
warnings.append(("WARN", f"Algae running out ({algae:.0f} kg) — build electrolyzer"))
elif algae < 5000:
warnings.append(("INFO", f"Algae moderate ({algae:.0f} kg) — plan SPOM"))
if pw > 50000:
warnings.append(("INFO", f"Polluted water abundant ({pw:.0f} kg) — use for reed fiber / pincha pepper"))
return warnings
def analyze_food(resources):
r = as_dict(resources)
cal = r.get('Calories', {}).get('amount', 0)
warnings = []
if cal < 100000:
warnings.append(("CRITICAL", f"Food shortage ({cal:.0f} kcal)"))
elif cal < 500000:
warnings.append(("WARN", f"Food declining ({cal:.0f} kcal)"))
elif cal > 2000000:
warnings.append(("INFO", f"Food surplus ({cal:.0f} kcal) — consider more dupes"))
return warnings
def analyze_power(resources):
r = as_dict(resources)
coal = r.get('Coal', {}).get('amount', 0)
hydrogen = r.get('Hydrogen', {}).get('amount', 0)
natgas = r.get('NaturalGas', {}).get('amount', 0)
warnings = []
if coal < 5000:
warnings.append(("WARN", f"Coal low ({coal:.0f} kg) — diversify power"))
if hydrogen > 20000:
warnings.append(("INFO", f"Hydrogen stockpiled ({hydrogen:.0f} kg) — add generators"))
if natgas > 10000:
warnings.append(("INFO", f"Natural gas abundant ({natgas:.0f} kg) — tap for power"))
return warnings
def analyze_temp(resources):
r = as_dict(resources)
warnings = []
for key in ['Temperature', 'AvgTemp']:
t = r.get(key, {}).get('amount')
if t:
if t > 50:
warnings.append(("CRITICAL", f"Overheating ({t:.0f}°C)"))
elif t > 35:
warnings.append(("WARN", f"High temperature ({t:.0f}°C)"))
elif t < -5:
warnings.append(("WARN", f"Too cold ({t:.0f}°C)"))
return warnings
def analyze_water(resources):
r = as_dict(resources)
water = r.get('Water', {}).get('amount', 0)
pw = r.get('PollutedWater', {}).get('amount', 0)
warnings = []
if water < 10000:
warnings.append(("WARN", f"Clean water low ({water:.0f} kg) — conserve / filter PW"))
if pw > water * 2 and water > 0:
warnings.append(("INFO", f"More polluted water than clean — build water purifier"))
return warnings
def suggest_actions(warnings, alerts):
suggestions = []
for sev, msg in warnings:
if 'Oxygen' in msg or 'oxygen' in msg:
if 'CRITICAL' in sev:
suggestions.append("URGENT: Build algae deoxidizer or electrolyzer immediately")
else:
suggestions.append("Build or expand SPOM (Self-Powered Oxygen Module)")
elif 'Food' in msg or 'food' in msg:
if 'CRITICAL' in sev:
suggestions.append("URGENT: Harvest wild plants or cook mush fry")
else:
suggestions.append("Expand mealwood farm or start hatch ranching")
elif 'Coal' in msg:
suggestions.append("Diversify power: hydrogen generator, natural gas, or solar")
elif 'Hydrogen' in msg:
suggestions.append("Build more hydrogen generators and battery bank")
elif 'NaturalGas' in msg:
suggestions.append("Build natural gas generator + gas pipe system")
elif 'temperature' in msg.lower() or 'overheat' in msg.lower():
suggestions.append("Check cooling loop; add liquid pipe thermo sensor")
elif 'water' in msg.lower() and 'low' in msg.lower():
suggestions.append("Dig more water sources or filter polluted water")
if isinstance(alerts, list):
for a in alerts:
msg = a.get('message', '') or a.get('title', '')
ml = msg.lower()
if 'oxygen' in ml or 'breathable' in ml:
suggestions.append("Build or expand electrolyzer setup (SPOM)")
elif 'food' in ml or 'starving' in ml:
suggestions.append("Expand mealwood farm or start ranching")
elif 'heat' in ml or 'temperature' in ml:
suggestions.append("Check cooling system, expand steam turbine setup")
elif 'power' in ml or 'wattage' in ml:
suggestions.append("Add power generation (hydrogen/natural gas)")
return list(dict.fromkeys(suggestions))
def print_report(state):
g = state.get('game', {})
if 'error' in g:
print(f"[!] Cannot connect to game: {g['error']}")
return False
resources = state.get('resources', [])
alerts = state.get('alerts', [])
all_warnings = []
all_warnings += analyze_o2(resources)
all_warnings += analyze_food(resources)
all_warnings += analyze_power(resources)
all_warnings += analyze_temp(resources)
all_warnings += analyze_water(resources)
critical = [w for w in all_warnings if w[0] == 'CRITICAL']
warns = [w for w in all_warnings if w[0] == 'WARN']
infos = [w for w in all_warnings if w[0] == 'INFO']
suggestions = suggest_actions(all_warnings, alerts)
print("=" * 52)
print(" ONI Analysis Report")
print("=" * 52)
print(f" Cycle: {g.get('cycle', '?')}")
print(f" Duplicants: {g.get('duplicantCount', '?')}")
print(f" World: {g.get('worldName', '?')}")
print(f" Buildings: {len(state.get('buildings', []) or [])}")
print(f" Critters: {len(state.get('critters', []) or [])}")
print(f" Geysers: {len(state.get('geysers', []) or [])}")
print(f" Research done: {sum(1 for t in (state.get('research') or []) if t.get('isComplete'))}")
print()
if critical:
print(f" [CRITICAL] {len(critical)} issues")
for _, msg in critical:
print(f" ! {msg}")
print()
if warns:
print(f" [WARN] {len(warns)} issues")
for _, msg in warns:
print(f" * {msg}")
print()
if infos:
print(f" [INFO] {len(infos)} notes")
for _, msg in infos:
print(f" i {msg}")
print()
if not all_warnings:
print(" Status: All stable")
print()
if suggestions:
print(f" Suggestions ({len(suggestions)}):")
for s in suggestions:
print(f" -> {s}")
print()
print(f" Alerts in-game: {len(alerts) if isinstance(alerts, list) else 0}")
if isinstance(alerts, list):
for a in alerts:
print(f" [{a.get('severity', '?')}] {a.get('title', '?')}: {a.get('message', '')}")
print("=" * 52)
return True
if __name__ == '__main__':
state = get_game_state()
if not print_report(state):
sys.exit(1)

204
tools/oni_api.py Normal file
View File

@ -0,0 +1,204 @@
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)}
def cmd_health():
print(api_get('/health'))
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("\n=== Resources (top 15) ===")
if isinstance(resources, list):
for r in sorted(resources, key=lambda x: x.get('amount', 0), reverse=True)[:15]:
print(f" {r.get('name', '?'):20s} {r.get('amount', 0):>10.1f} {r.get('unit', '')}")
print("\n=== Duplicants ===")
if isinstance(dups, list):
for d in dups:
print(f" {d.get('name'):12s} stress={d.get('stress', '?'):>5} food={d.get('calories', 0)/1000:>6.0f} kcal"
f" stamina={d.get('stamina', '?'):>5} o2={d.get('oxygen', '?'):>5}")
print("\n=== Alerts ===")
if isinstance(alerts, list):
for a in alerts:
print(f" [{a.get('severity', '?')}] {a.get('title', '?')}: {a.get('message', '')}" if alerts else " (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):
print(f"{r.get('name', '?'):30s} {r.get('amount', 0):>12.1f} {r.get('unit', '')}")
else:
print(json.dumps(data, indent=2, ensure_ascii=False))
def cmd_duplicants():
print(json.dumps(api_get('/api/state/duplicants'), indent=2, ensure_ascii=False))
def cmd_buildings():
data = api_get('/api/state/buildings')
if isinstance(data, list):
for b in data:
print(f" {b.get('name', '?'):25s} at ({b.get('x', '?')}, {b.get('y', '?')}) "
f"{'ON' if b.get('isOperational') else 'OFF'}")
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}] ({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', '?')} rate={g.get('emitRate', '?')}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', '?'):.1f} happy={c.get('happiness', '?')}")
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))
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,
'dig': cmd_dig,
'build': cmd_build,
'deconstruct': cmd_deconstruct,
'prioritize': cmd_prioritize,
'research_select': cmd_research_select,
}
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("")
print("Commands:")
print(" health Check Mod connection")
print(" status Game overview (cycle, resources, dups, alerts)")
print(" resources List all resources with amounts")
print(" duplicants Show duplicant details")
print(" buildings List all buildings")
print(" research Show research tree progress")
print(" geysers Show geyser states")
print(" critters Show critter list")
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")
else:
COMMANDS[cmd](sys.argv[2:])

163
tools/oni_builder.py Normal file
View File

@ -0,0 +1,163 @@
import json
import sys
from oni_api import api_post
BLUEPRINTS = {
'spom': {
'name': 'SPOM (Self-Powered Oxygen Module)',
'description': 'Standard Rodriguez SPOM: electrolyzer + hydrogen generators',
'size': {'width': 8, 'height': 6},
'dig': {'x': -1, 'y': -1, 'width': 10, 'height': 8},
'buildings': [
{'id': 'Electrolyzer', 'x': 3, 'y': 2},
{'id': 'GasPump', 'x': 1, 'y': 2},
{'id': 'GasPump', 'x': 5, 'y': 2},
{'id': 'HydrogenGenerator', 'x': 1, 'y': 0},
{'id': 'HydrogenGenerator', 'x': 4, 'y': 0},
{'id': 'GasFilter', 'x': 3, 'y': 0},
],
},
'spom_mini': {
'name': 'Mini SPOM',
'description': 'Compact electrolyzer + 1 hydrogen generator for early game',
'size': {'width': 5, 'height': 4},
'dig': {'x': -1, 'y': -1, 'width': 7, 'height': 6},
'buildings': [
{'id': 'Electrolyzer', 'x': 2, 'y': 1},
{'id': 'GasPump', 'x': 1, 'y': 1},
{'id': 'HydrogenGenerator', 'x': 2, 'y': 0},
],
},
'toilet_loop': {
'name': 'Bathroom Water Loop',
'description': 'Lavatory -> Water Purifier -> Lavatory closed loop',
'size': {'width': 8, 'height': 4},
'dig': {'x': -1, 'y': -1, 'width': 10, 'height': 6},
'buildings': [
{'id': 'Lavatory', 'x': 1, 'y': 1},
{'id': 'Lavatory', 'x': 3, 'y': 1},
{'id': 'WaterPurifier', 'x': 6, 'y': 1},
{'id': 'LiquidPump', 'x': 6, 'y': 2},
],
},
'ranch_hatch': {
'name': 'Hatch Ranch',
'description': 'Standard hatch ranching module with feeder and incubator',
'size': {'width': 10, 'height': 6},
'dig': {'x': -1, 'y': -1, 'width': 12, 'height': 8},
'buildings': [
{'id': 'RanchStation', 'x': 1, 'y': 1},
{'id': 'Incubator', 'x': 4, 'y': 1},
{'id': 'StorageLocker', 'x': 8, 'y': 1},
],
},
'farm_mealwood': {
'name': 'Mealwood Farm',
'description': 'Basic mealwood farm: 5 planter boxes + storage',
'size': {'width': 6, 'height': 4},
'dig': {'x': -1, 'y': -1, 'width': 8, 'height': 6},
'buildings': [
{'id': 'PlanterBox', 'x': 1, 'y': 1},
{'id': 'PlanterBox', 'x': 3, 'y': 1},
{'id': 'PlanterBox', 'x': 5, 'y': 1},
{'id': 'PlanterBox', 'x': 1, 'y': 3},
{'id': 'PlanterBox', 'x': 3, 'y': 3},
{'id': 'StorageLocker', 'x': 5, 'y': 3},
],
},
'cooling': {
'name': 'Steam Turbine Cooler',
'description': 'Liquid cooling loop with steam turbine + aquatuner',
'size': {'width': 8, 'height': 6},
'dig': {'x': -1, 'y': -1, 'width': 10, 'height': 8},
'buildings': [
{'id': 'SteamTurbine', 'x': 1, 'y': 4},
{'id': 'SteamTurbine', 'x': 5, 'y': 4},
{'id': 'Aquatuner', 'x': 2, 'y': 1},
{'id': 'LiquidPump', 'x': 5, 'y': 1},
],
},
'bedroom': {
'name': 'Barracks / Bedroom',
'description': 'Basic bedroom module with cots and decorations',
'size': {'width': 8, 'height': 4},
'dig': {'x': -1, 'y': -1, 'width': 10, 'height': 6},
'buildings': [
{'id': 'Bed', 'x': 1, 'y': 1},
{'id': 'Bed', 'x': 3, 'y': 1},
{'id': 'Bed', 'x': 5, 'y': 1},
{'id': 'LadderBed', 'x': 1, 'y': 3},
{'id': 'LadderBed', 'x': 3, 'y': 3},
{'id': 'FlowerVase', 'x': 6, 'y': 1},
],
},
}
def list_blueprints():
print(f"Available Blueprints ({len(BLUEPRINTS)}):")
print("=" * 60)
for key, bp in BLUEPRINTS.items():
print(f" {key:16s} {bp['name']:28s} {bp['size']['width']}x{bp['size']['height']}")
print(f" {'':16s} {bp['description']}")
print()
def apply_blueprint(name, origin_x, origin_y):
bp = BLUEPRINTS.get(name)
if not bp:
print(f"[!] Blueprint '{name}' not found")
print(f" Use 'list' to see available blueprints")
return False
print(f"Applying blueprint: {bp['name']}")
print(f" Origin: ({origin_x}, {origin_y})")
print(f" Size: {bp['size']['width']} x {bp['size']['height']}")
print()
results = []
dig = bp.get('dig', {'x': -1, 'y': -1, 'width': bp['size']['width'] + 2, 'height': bp['size']['height'] + 2})
dig_result = api_post('/api/action/dig', {
'x': origin_x + dig['x'],
'y': origin_y + dig['y'],
'width': dig['width'],
'height': dig['height'],
})
status = 'OK' if 'error' not in dig_result else dig_result.get('error', 'fail')
print(f" [Dig] area ({dig['width']}x{dig['height']}): {status}")
for b in bp['buildings']:
x = origin_x + b['x']
y = origin_y + b['y']
result = api_post('/api/action/build', {
'buildingId': b['id'],
'x': x,
'y': y,
})
ok = 'error' not in result
results.append({'building': b['id'], 'x': x, 'y': y, 'ok': ok})
status = 'OK' if ok else result.get('error', 'fail')
print(f" [Build] {b['id']:20s} at ({x:3d}, {y:3d}): {status}")
ok_count = sum(1 for r in results if r['ok'])
print()
print(f" Result: {ok_count}/{len(results)} buildings placed")
return ok_count == len(results)
if __name__ == '__main__':
cmd = sys.argv[1] if len(sys.argv) > 1 else 'list'
if cmd == 'list':
list_blueprints()
elif cmd == 'build':
if len(sys.argv) < 4:
print("Usage: python oni_builder.py build <blueprint_name> <origin_x> <origin_y>")
print()
list_blueprints()
sys.exit(1)
success = apply_blueprint(sys.argv[2], int(sys.argv[3]), int(sys.argv[4]))
sys.exit(0 if success else 1)
else:
print("Usage: python oni_builder.py <list|build>")