205ff7668a
New Mod endpoints: - POST /api/action/build_pipe: build gas/liquid pipe path (x1,y1)->(x2,y2) with bridge option - POST /api/action/build_wire: build wire path with types: regular, heavy, conductive, heavy_conductive New Python tool: tools/oni_commander.py High-level commands that combine multiple low-level API calls: - diagnose: full diagnostic (power, CO2, temp, diseases, pipes) - fix_co2: auto-detect CO2 pockets and dig vent shafts - fix_overload: detect overloaded circuits with fix suggestions - emergency_o2: auto-check O2 and build OxygenDiffuser/Electrolyzer - expand_base: dig + build walls/floors in one command (one-click room expansion) - build_pipe_line: simplified CLI for pipe path building - build_wire_line: simplified CLI for wire path building All high-level commands auto-pause/resume the game. CLI: build_pipe_line, build_wire_line added to oni_api.py
327 lines
11 KiB
Python
327 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ONI Commander — 高级指令封装
|
||
============================
|
||
把多个底层 API 调用组合成一条"指挥官指令"，AI 一句话就能执行复杂操作。
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
import os
|
||
|
||
TOOLS_DIR = os.path.dirname(__file__)
|
||
sys.path.insert(0, TOOLS_DIR)
|
||
|
||
from oni_api import api_get, api_post, _print_feedback
|
||
|
||
|
||
def cmd_diagnose(args):
|
||
"""全面诊断：O2、食物、电力、CO2、温度、管道。"""
|
||
print("=" * 56)
|
||
print(" ONI Full Diagnostic")
|
||
print("=" * 56)
|
||
|
||
pause_before()
|
||
pause_after()
|
||
|
||
# 1. 游戏总览
|
||
game = api_get('/api/state/game')
|
||
if 'error' in game:
|
||
print(f"[!] Cannot connect: {game['error']}")
|
||
return False
|
||
print(f" Cycle {game.get('cycle', '?')} | {game.get('duplicantCount', '?')} dupes | "
|
||
f"{game.get('suffocating', 0)} suffocating | {game.get('starving', 0)} starving | "
|
||
f"{game.get('stressed', 0)} stressed")
|
||
|
||
# 2. 电力
|
||
print("\n--- Power ---")
|
||
power = api_get('/api/state/power')
|
||
if 'circuits' in power:
|
||
for c in power['circuits']:
|
||
mark = " *** OVERLOAD ***" if c.get('isOverloaded') else ""
|
||
print(f" Circuit {c.get('id')}: {c.get('wattsUsed', 0):.0f}W / {c.get('maxWatts', 0):.0f}W{mark}")
|
||
|
||
# 3. CO2
|
||
print("\n--- CO2 ---")
|
||
co2 = api_get('/api/state/co2')
|
||
if co2.get('pocketCount', 0) > 0:
|
||
print(f" {co2.get('pocketCount')} pockets ({co2.get('totalMassKg', 0):.0f} kg CO2)")
|
||
for p in co2.get('pockets', [])[:3]:
|
||
print(f" ({p.get('x')},{p.get('y')}) {p.get('mass', 0):.0f} kg")
|
||
else:
|
||
print(" No CO2 pockets detected")
|
||
|
||
# 4. 温度
|
||
print("\n--- Temperature ---")
|
||
temp = api_get('/api/state/temperature/zones')
|
||
if 'averageC' in temp:
|
||
print(f" Avg: {temp['averageC']:.0f}°C Min: {temp.get('minC', 0):.0f}°C Max: {temp.get('maxC', 0):.0f}°C")
|
||
if temp.get('hotSpots'):
|
||
print(f" {len(temp['hotSpots'])} hot spots (>50°C) — risk!")
|
||
if temp.get('coldSpots'):
|
||
print(f" {len(temp['coldSpots'])} cold spots (<5°C)")
|
||
|
||
# 5. 疾病
|
||
print("\n--- Diseases ---")
|
||
diseases = api_get('/api/state/diseases')
|
||
infected = diseases.get('infectedDuplicants', [])
|
||
if infected:
|
||
for d in infected:
|
||
print(f" {d.get('duplicant')} — {d.get('disease')} ({d.get('severity')})")
|
||
else:
|
||
print(" No infections")
|
||
|
||
# 6. 管道
|
||
print("\n--- Pipes ---")
|
||
for pt in ('gas', 'liquid'):
|
||
pipes = api_get(f'/api/state/pipes?type={pt}')
|
||
segs = pipes.get('segmentCount', 0)
|
||
if segs:
|
||
first = pipes.get('segments', [{}])[0]
|
||
print(f" {pt}: {segs} segments (e.g. {first.get('element', '?')})")
|
||
else:
|
||
print(f" {pt}: empty")
|
||
|
||
print()
|
||
print("=" * 56)
|
||
print(" Diagnostic complete")
|
||
print("=" * 56)
|
||
|
||
|
||
def cmd_fix_co2(args):
|
||
"""找到 CO2 并挖掘排气通道。"""
|
||
print("[CO2 Fix] Scanning for CO2 pockets...")
|
||
co2 = api_get('/api/state/co2')
|
||
pockets = co2.get('pockets', [])
|
||
if not pockets:
|
||
print("[OK] No CO2 pockets found.")
|
||
return
|
||
|
||
pause_before()
|
||
|
||
# Find the lowest y-level pocket and dig below it
|
||
bottom = min(pockets, key=lambda p: p.get('y', 0))
|
||
x, y = bottom.get('x', 0), bottom.get('y', 0)
|
||
print(f"[CO2 Fix] Largest pocket at ({x},{y}), {bottom.get('mass', 0):.0f} kg")
|
||
|
||
# Dig a 1-wide shaft down
|
||
dig_y = max(0, y - 5)
|
||
result = api_post('/api/action/dig', {"x": x, "y": dig_y, "width": 1, "height": y - dig_y + 1})
|
||
if result.get('success'):
|
||
print(f"[CO2 Fix] Dug vent shaft at x={x}, y={dig_y}..{y}")
|
||
else:
|
||
print(f"[CO2 Fix] Dig failed: {result.get('errorMessage', 'unknown')}")
|
||
|
||
pause_after()
|
||
|
||
|
||
def cmd_fix_overload(args):
|
||
"""检测过载电路并给出修复建议。"""
|
||
print("[Power Fix] Analyzing circuits...")
|
||
power = api_get('/api/state/power')
|
||
overloaded = [c for c in power.get('circuits', []) if c.get('isOverloaded')]
|
||
if not overloaded:
|
||
print("[OK] No overloaded circuits.")
|
||
return
|
||
|
||
pause_before()
|
||
|
||
print(f"[Power Fix] {len(overloaded)} overloaded circuits:")
|
||
for c in overloaded:
|
||
print(f" Circuit {c.get('id')}: {c.get('wattsUsed', 0):.0f}W / {c.get('maxWatts', 0):.0f}W")
|
||
|
||
# Suggest fixes
|
||
print()
|
||
print(" Suggested fixes:")
|
||
print(" 1. Move heavy consumers (MetalRefinery, Aquatuner) to separate circuit")
|
||
print(" 2. Upgrade wire to HeaviWatt or split into 2 transformers")
|
||
print(" 3. Add PowerTransformer to isolate high-load branches")
|
||
|
||
pause_after()
|
||
|
||
|
||
def cmd_build_pipe_line(args):
|
||
"""铺设管道路径。"""
|
||
if len(args) < 5:
|
||
print("Usage: build_pipe_line gas|liquid <x1> <y1> <x2> <y2> [bridge]")
|
||
return
|
||
ptype = args[0]
|
||
x1, y1, x2, y2 = int(args[1]), int(args[2]), int(args[3]), int(args[4])
|
||
bridge = len(args) > 5 and args[5] == 'bridge'
|
||
|
||
pause_before()
|
||
|
||
if ptype not in ('gas', 'liquid'):
|
||
print("[!] Type must be 'gas' or 'liquid'")
|
||
return
|
||
|
||
result = api_post('/api/action/build_pipe', {
|
||
"type": ptype, "x1": x1, "y1": y1,
|
||
"x2": x2, "y2": y2, "bridge": bridge
|
||
})
|
||
if result.get('success'):
|
||
segs = result.get('data', {}).get('segmentCount', 0)
|
||
print(f"[OK] {ptype} pipe: {segs} segments from ({x1},{y1}) to ({x2},{y2})")
|
||
else:
|
||
print(f"[!] Failed: {result.get('errorMessage', 'unknown')}")
|
||
|
||
pause_after()
|
||
|
||
|
||
def cmd_build_wire_line(args):
|
||
"""铺设电线路径。"""
|
||
if len(args) < 5:
|
||
print("Usage: build_wire_line regular|heavy|conductive <x1> <y1> <x2> <y2> [bridge]")
|
||
return
|
||
wtype = args[0]
|
||
x1, y1, x2, y2 = int(args[1]), int(args[2]), int(args[3]), int(args[4])
|
||
bridge = len(args) > 5 and args[5] == 'bridge'
|
||
|
||
pause_before()
|
||
|
||
if wtype not in ('regular', 'heavy', 'conductive', 'heavy_conductive'):
|
||
print("[!] Type must be 'regular', 'heavy', 'conductive', or 'heavy_conductive'")
|
||
return
|
||
|
||
result = api_post('/api/action/build_wire', {
|
||
"type": wtype, "x1": x1, "y1": y1,
|
||
"x2": x2, "y2": y2, "bridge": bridge
|
||
})
|
||
if result.get('success'):
|
||
segs = result.get('data', {}).get('segmentCount', 0)
|
||
print(f"[OK] {wtype} wire: {segs} segments from ({x1},{y1}) to ({x2},{y2})")
|
||
else:
|
||
print(f"[!] Failed: {result.get('errorMessage', 'unknown')}")
|
||
|
||
pause_after()
|
||
|
||
|
||
def cmd_expand_base(args):
|
||
"""拓展基地：挖掘 + 建造墙壁。"""
|
||
if len(args) < 4:
|
||
print("Usage: expand_base <x> <y> <width> <height>")
|
||
return
|
||
x, y, w, h = int(args[0]), int(args[1]), int(args[2]), int(args[3])
|
||
|
||
pause_before()
|
||
|
||
# Dig
|
||
r1 = api_post('/api/action/dig', {"x": x - 1, "y": y - 1, "width": w + 2, "height": h + 2})
|
||
if not r1.get('success'):
|
||
print(f"[!] Dig failed: {r1.get('errorMessage', 'unknown')}")
|
||
pause_after()
|
||
return
|
||
print(f"[Expand] Dug ({x},{y}) {w}x{h}")
|
||
|
||
# Build floor tiles
|
||
for fx in range(x, x + w):
|
||
api_post('/api/action/build', {"buildingId": "Tile", "x": fx, "y": y})
|
||
print(f"[Expand] Built floor: {w} tiles")
|
||
|
||
# Build walls
|
||
for wx in range(x, x + w):
|
||
api_post('/api/action/build', {"buildingId": "Tile", "x": wx, "y": y + h})
|
||
for wy in range(y + 1, y + h):
|
||
api_post('/api/action/build', {"buildingId": "Tile", "x": x, "y": wy})
|
||
api_post('/api/action/build', {"buildingId": "Tile", "x": x + w - 1, "y": wy})
|
||
print(f"[Expand] Built walls")
|
||
|
||
pause_after()
|
||
print(f"[OK] Room expanded to ({x},{y}) {w}x{h}")
|
||
|
||
|
||
def cmd_emergency_o2(args):
|
||
"""紧急制氧：检查 O2 并自动建造。"""
|
||
print("[Emergency O2] Checking oxygen status...")
|
||
|
||
pause_before()
|
||
|
||
resources = api_get('/api/state/resources')
|
||
buildings = api_get('/api/state/buildings')
|
||
game = api_get('/api/state/game')
|
||
|
||
if isinstance(resources, list):
|
||
o2 = next((r for r in resources if r.get('name') == 'Oxygen'), {})
|
||
algae = next((r for r in resources if r.get('name') == 'Algae'), {})
|
||
o2_kg = o2.get('amount', 0)
|
||
algae_kg = algae.get('amount', 0)
|
||
print(f" O2: {o2_kg:.0f} kg | Algae: {algae_kg:.0f} kg")
|
||
else:
|
||
o2_kg, algae_kg = 0, 0
|
||
|
||
has_electrolyzer = any(b.get('id') == 'Electrolyzer' for b in (buildings or []))
|
||
has_diffuser = any(b.get('id') == 'OxygenDiffuser' for b in (buildings or []))
|
||
|
||
if o2_kg < 500:
|
||
print("[CRITICAL] Oxygen critical!")
|
||
if has_diffuser and algae_kg > 500:
|
||
print(" OxygenDiffuser already exists, checking Algae supply...")
|
||
elif not has_electrolyzer and not has_diffuser:
|
||
# Find a spot near base and build
|
||
game_info = api_get('/api/state/game')
|
||
print(" No O2 production! Building OxygenDiffuser...")
|
||
result = api_post('/api/action/build', {"buildingId": "OxygenDiffuser", "x": 30, "y": 20})
|
||
if result.get('success'):
|
||
print(" [OK] OxygenDiffuser queued at (30,20)")
|
||
else:
|
||
print(f" [!] {result.get('errorMessage', 'build failed')}")
|
||
elif o2_kg < 2000 and not has_electrolyzer:
|
||
print("[WARN] Low O2, recommend SPOM build")
|
||
else:
|
||
print("[OK] Oxygen stable")
|
||
|
||
pause_after()
|
||
|
||
|
||
# ── Helpers ──────────────────────────────────────────────────────────────
|
||
|
||
def pause_before():
|
||
"""High-level ops always pause first."""
|
||
api_post('/api/action/pause', {"reason": "High-level operation"})
|
||
|
||
|
||
def pause_after():
|
||
"""Resume after operation."""
|
||
api_post('/api/action/unpause', {"speed": 1})
|
||
|
||
|
||
# ── Command Registry ─────────────────────────────────────────────────────
|
||
|
||
COMMANDS = {
|
||
'diagnose': cmd_diagnose,
|
||
'fix_co2': cmd_fix_co2,
|
||
'fix_overload': cmd_fix_overload,
|
||
'expand_base': cmd_expand_base,
|
||
'emergency_o2': cmd_emergency_o2,
|
||
'build_pipe_line': cmd_build_pipe_line,
|
||
'build_wire_line': cmd_build_wire_line,
|
||
}
|
||
|
||
if __name__ == '__main__':
|
||
cmd = sys.argv[1] if len(sys.argv) > 1 else 'help'
|
||
|
||
if cmd == 'help' or cmd not in COMMANDS:
|
||
print("ONI Commander — 高级指令")
|
||
print("=" * 56)
|
||
print()
|
||
print("=== Diagnostics ===")
|
||
print(" diagnose 全面诊断（O2/电力/CO2/温度/疾病/管道）")
|
||
print()
|
||
print("=== Automated Fixes ===")
|
||
print(" fix_co2 找到 CO2 并挖掘排气通道")
|
||
print(" fix_overload 检测过载电路并建议修复")
|
||
print(" emergency_o2 紧急制氧（检查+自动建造）")
|
||
print()
|
||
print("=== Room Expansion ===")
|
||
print(" expand_base <x> <y> <w> <h> 挖掘+建造墙壁（一键拓展房间）")
|
||
print()
|
||
print("=== Pipe / Wire Lines ===")
|
||
print(" build_pipe_line <t> <x1> <y1> <x2> <y2> [bridge]")
|
||
print(" t: gas | liquid")
|
||
print(" build_wire_line <t> <x1> <y1> <x2> <y2> [bridge]")
|
||
print(" t: regular | heavy | conductive | heavy_conductive")
|
||
print()
|
||
print("All high-level commands auto-pause/resume the game.")
|
||
else:
|
||
COMMANDS[cmd](sys.argv[2:])
|