ONI Agent Bridge - initial mod implementation
This commit is contained in:
JianFeeeee
281
tools/oni_analyzer.py
Normal file
281
tools/oni_analyzer.py
Normal file
@ -0,0 +1,281 @@
|
||||
import json
|
||||
import sys
|
||||
from oni_api import api_get
|
||||
|
||||
|
||||
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'),
|
||||
'plants': api_get('/api/state/plants'),
|
||||
'rooms': api_get('/api/state/rooms'),
|
||||
}
|
||||
|
||||
|
||||
def as_dict(resources):
|
||||
if not isinstance(resources, list):
|
||||
return {}
|
||||
return {r.get('name'): r for r in resources}
|
||||
|
||||
|
||||
def buildings_by_cat(buildings):
|
||||
cats = {}
|
||||
for b in (buildings or []):
|
||||
cat = b.get('category', 'Other')
|
||||
cats.setdefault(cat, []).append(b)
|
||||
return cats
|
||||
|
||||
|
||||
def analyze_o2(resources, buildings):
|
||||
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)
|
||||
|
||||
has_electrolyzer = any(b.get('id') == 'Electrolyzer' for b in (buildings or []))
|
||||
has_diffuser = any(b.get('id') == 'OxygenDiffuser' for b in (buildings or []))
|
||||
|
||||
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 o2 < 1000 and not has_electrolyzer and not has_diffuser:
|
||||
warnings.append(("CRITICAL", "No oxygen production buildings found! Build OxygenDiffuser or Electrolyzer"))
|
||||
|
||||
if algae < 1000:
|
||||
warnings.append(("WARN", f"Algae running out ({algae:.0f} kg) — build electrolyzer"))
|
||||
elif algae < 5000 and not has_electrolyzer:
|
||||
warnings.append(("INFO", f"Algae moderate ({algae:.0f} kg) — plan SPOM transition"))
|
||||
|
||||
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, buildings):
|
||||
r = as_dict(resources)
|
||||
cal = r.get('Calories', {}).get('amount', 0)
|
||||
|
||||
has_farm = any(b.get('id') in ('PlanterBox', 'FarmTile') for b in (buildings or []))
|
||||
has_grill = any(b.get('id') == 'ElectricGrill' for b in (buildings or []))
|
||||
|
||||
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)"))
|
||||
|
||||
if cal < 500000 and not has_farm:
|
||||
warnings.append(("WARN", "No farm plots found. Build PlanterBox and plant Mealwood"))
|
||||
if cal < 500000 and not has_grill:
|
||||
warnings.append(("INFO", "Build ElectricGrill to improve food quality"))
|
||||
return warnings
|
||||
|
||||
|
||||
def analyze_power(resources, buildings):
|
||||
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)
|
||||
|
||||
generators = [b for b in (buildings or []) if b.get('id') in (
|
||||
'CoalGenerator', 'HydrogenGenerator', 'NaturalGasGenerator',
|
||||
'ManualGenerator', 'PetroleumGenerator', 'WoodBurner'
|
||||
)]
|
||||
batteries = [b for b in (buildings or []) if b.get('id') in ('Battery', 'JumboBattery', 'SmartBattery')]
|
||||
|
||||
warnings = []
|
||||
if not generators:
|
||||
warnings.append(("CRITICAL", "No power generators found! Build ManualGenerator or CoalGenerator"))
|
||||
else:
|
||||
powered_on = sum(1 for g in generators if g.get('isOperational'))
|
||||
warnings.append(("INFO", f"Power: {len(generators)} generators ({powered_on} operational), {len(batteries)} batteries"))
|
||||
|
||||
if coal < 5000 and any(g.get('id') == 'CoalGenerator' for g in generators):
|
||||
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)"))
|
||||
|
||||
return warnings
|
||||
|
||||
|
||||
def analyze_temp():
|
||||
from oni_api import api_get
|
||||
data = api_get('/api/state/temperature/zones', auto_pause=False)
|
||||
warnings = []
|
||||
if 'error' in data:
|
||||
return warnings
|
||||
avg = data.get('averageC', 0)
|
||||
if avg > 50:
|
||||
warnings.append(("CRITICAL", f"Overheating ({avg:.0f}°C)"))
|
||||
elif avg > 35:
|
||||
warnings.append(("WARN", f"High temperature ({avg:.0f}°C)"))
|
||||
elif avg < -5:
|
||||
warnings.append(("WARN", f"Too cold ({avg:.0f}°C)"))
|
||||
if data.get('hotSpots'):
|
||||
warnings.append(("WARN", f"{len(data['hotSpots'])} hot spots (>50°C)"))
|
||||
if data.get('coldSpots'):
|
||||
warnings.append(("WARN", f"{len(data['coldSpots'])} cold spots (<5°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)
|
||||
sw = r.get('SaltWater', {}).get('amount', 0)
|
||||
|
||||
total_water = water + pw + sw
|
||||
warnings = []
|
||||
if total_water < 10000:
|
||||
warnings.append(("WARN", f"Total water low ({total_water:.0f} kg across all sources)"))
|
||||
elif total_water < 50000:
|
||||
warnings.append(("INFO", f"Water reserves moderate ({total_water:.0f} kg)"))
|
||||
if water < 10000 and pw > 10000:
|
||||
warnings.append(("INFO", f"Filter polluted water ({pw:.0f} kg available)"))
|
||||
|
||||
return warnings
|
||||
|
||||
|
||||
def analyze_research(research):
|
||||
if not isinstance(research, list):
|
||||
return []
|
||||
warnings = []
|
||||
done = sum(1 for t in research if t.get('isComplete'))
|
||||
total = len(research)
|
||||
if done == 0 and total > 0:
|
||||
warnings.append(("WARN", "No research completed! Start with Research Station"))
|
||||
elif done < total * 0.3:
|
||||
warnings.append(("INFO", f"Research progress: {done}/{total} ({done*100//total}%)"))
|
||||
return warnings
|
||||
|
||||
|
||||
def suggest_actions(warnings, alerts, buildings):
|
||||
suggestions = []
|
||||
|
||||
has_electrolyzer = any(b.get('id') == 'Electrolyzer' for b in (buildings or []))
|
||||
has_lavatory = any(b.get('id') == 'Lavatory' for b in (buildings or []))
|
||||
has_sieve = any(b.get('id') == 'WaterSiever' for b in (buildings or []))
|
||||
|
||||
for sev, msg in warnings:
|
||||
ml = msg.lower()
|
||||
if 'oxygen' in ml:
|
||||
suggestions.append("Build SPOM: Electrolyzer + Hydrogen Generator at a water source")
|
||||
elif 'food' in ml and 'shortage' in ml:
|
||||
suggestions.append("Build PlanterBoxes x5, plant Mealwood (no irrigation needed)")
|
||||
elif 'food' in ml and 'declining' in ml:
|
||||
suggestions.append("Expand farm or start hatch ranching (Hatch eats Sedimentary Rock)")
|
||||
elif 'power' in ml and 'generator' in ml:
|
||||
suggestions.append("Build ManualGenerator (early) or CoalGenerator (durable)")
|
||||
elif 'coal' in ml:
|
||||
suggestions.append("Diversify power: build HydrogenGenerator + SmartBattery")
|
||||
elif 'hydrogen' in ml:
|
||||
suggestions.append("Connect HydrogenGenerator to your hydrogen vent/SPOM")
|
||||
elif 'water' in ml and 'low' in ml:
|
||||
suggestions.append("Dig to find water geyser or filter polluted water")
|
||||
elif 'heat' in ml or 'overheat' in ml:
|
||||
suggestions.append("Build insulated tiles around heat sources; add cooling loop")
|
||||
|
||||
if not has_lavatory:
|
||||
suggestions.append("Build Lavatory + Water Sieve for renewable water loop")
|
||||
if not has_electrolyzer:
|
||||
suggestions.append("Plan SPOM once Algae < 5t or you have renewable water")
|
||||
if not has_sieve and has_lavatory:
|
||||
suggestions.append("Build Water Sieve to close the bathroom loop")
|
||||
|
||||
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', [])
|
||||
buildings = state.get('buildings', [])
|
||||
alerts = state.get('alerts', [])
|
||||
|
||||
all_warnings = []
|
||||
all_warnings += analyze_o2(resources, buildings)
|
||||
all_warnings += analyze_food(resources, buildings)
|
||||
all_warnings += analyze_power(resources, buildings)
|
||||
all_warnings += analyze_temp()
|
||||
all_warnings += analyze_water(resources)
|
||||
all_warnings += analyze_research(state.get('research', []))
|
||||
|
||||
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, buildings)
|
||||
|
||||
print("=" * 56)
|
||||
print(" ONI Analysis Report")
|
||||
print("=" * 56)
|
||||
print(f" Cycle: {g.get('cycle', '?')}")
|
||||
print(f" Duplicants: {g.get('duplicantCount', '?')}")
|
||||
print(f" World: {g.get('worldName', '?')}")
|
||||
print(f" Grid: {g.get('gridWidth', '?')} x {g.get('gridHeight', '?')}")
|
||||
print(f" Buildings: {len(buildings or [])}")
|
||||
print(f" Critters: {len(state.get('critters', []) or [])}")
|
||||
print(f" Geysers: {len(state.get('geysers', []) or [])}")
|
||||
print(f" Plants: {len(state.get('plants', []) or [])}")
|
||||
print(f" Rooms: {len(state.get('rooms', []) or [])}")
|
||||
print(f" Research done: {sum(1 for t in (state.get('research') or []) if t.get('isComplete'))}/{len(state.get('research', []) or [])}")
|
||||
print()
|
||||
|
||||
if critical:
|
||||
print(f" [CRITICAL] {len(critical)} issues — act immediately!")
|
||||
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" In-game alerts: {len(alerts) if isinstance(alerts, list) else 0}")
|
||||
if isinstance(alerts, list):
|
||||
for a in alerts:
|
||||
print(f" [{a.get('severity', '?')}] {a.get('title', '?')}")
|
||||
print("=" * 56)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
state = get_game_state()
|
||||
if not print_report(state):
|
||||
sys.exit(1)
|
||||
1540
tools/oni_api.py
Normal file
1540
tools/oni_api.py
Normal file
File diff suppressed because it is too large
Load Diff
163
tools/oni_builder.py
Normal file
163
tools/oni_builder.py
Normal 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>")
|
||||
355
tools/oni_commander.py
Normal file
355
tools/oni_commander.py
Normal file
@ -0,0 +1,355 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
ONI Commander — 高级指令封装
|
||||
============================
|
||||
把多个底层 API 调用组合成一条"指挥官指令"，AI 一句话就能执行复杂操作。
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
import inspect
|
||||
|
||||
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> [mode]")
|
||||
print(" mode: 'line'(默认,与已有管线合并) | 'cross'(跨接器跳过) | 'single'(单段)")
|
||||
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'
|
||||
|
||||
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, "mode": mode
|
||||
})
|
||||
if result.get('success'):
|
||||
d = result.get('data', {})
|
||||
segs = d.get('segmentCount', 0)
|
||||
bridges = d.get('bridgesPlaced', 0)
|
||||
print(f"[OK] {ptype} pipe: {segs} segments from ({x1},{y1}) to ({x2},{y2})")
|
||||
if bridges > 0:
|
||||
print(f" {bridges} bridges placed at crossings (mode={mode})")
|
||||
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> [mode]")
|
||||
print(" mode: 'line' (default), 'cross' (use bridges), 'single'")
|
||||
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'
|
||||
|
||||
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, "mode": mode
|
||||
})
|
||||
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,
|
||||
'snapshot': cmd_snapshot,
|
||||
'camera': cmd_camera,
|
||||
}
|
||||
|
||||
def cmd_snapshot(args):
|
||||
"""Take screenshot + view. Usage: snapshot [name]"""
|
||||
from oni_api import cmd_snapshot as api_snapshot
|
||||
api_snapshot(args)
|
||||
|
||||
def cmd_camera(args):
|
||||
"""Move camera. Usage: camera <x> <y> [zoom]"""
|
||||
from oni_api import cmd_camera as api_camera
|
||||
api_camera(args)
|
||||
|
||||
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> [mode]")
|
||||
print(" t: gas | liquid | mode: line(merge) | cross(bridge)")
|
||||
print(" build_wire_line <t> <x1> <y1> <x2> <y2> [mode]")
|
||||
print(" t: regular | heavy | conductive | heavy_conductive")
|
||||
print()
|
||||
print("=== Screenshot / Camera ===")
|
||||
print(" snapshot [file.png] Take screenshot")
|
||||
print(" camera <x> <y> [zoom] Move camera view")
|
||||
print()
|
||||
print("All high-level commands auto-pause/resume the game.")
|
||||
else:
|
||||
args = sys.argv[2:]
|
||||
fn = COMMANDS[cmd]
|
||||
sig = inspect.signature(fn)
|
||||
if len(sig.parameters) > 0:
|
||||
fn(args)
|
||||
else:
|
||||
fn()
|
||||
Reference in New Issue
Block a user