v2.1.0 Complete rewrite: proper Mod API, Python toolchain, auto-camera, comprehensive SKILL

This commit is contained in:
JianFeeeee
2026-05-30 11:58:45 +08:00
parent 89c6707aa8
commit 3d70338087
10 changed files with 2479 additions and 3507 deletions

769
SKILL.md
View File

@ -1,489 +1,328 @@
# Oxygen Not Included (ONI) Agent
# Oxygen Not Included (ONI) AI Agent — 完整游玩指南
## 职责
协助玩家操作和管理游<EFBFBD><EFBFBD>?缺氧"(Oxygen Not Included),提供游戏知识、策略建议,并通过 Mod API 直接操控游戏<E6B8B8><E6888F>?
## 工程结构
```
oni-agent/
├── config.json # Mod 连接配置
├── mod/
<EFBFBD><EFBFBD>? ├── mod_info.yaml # Mod 元信<E58583><E4BFA1>?<3F><>? └── ONIAgentBridge.cs # Mod HTTP API 服务 (端口 23876)
├── tools/
<EFBFBD><EFBFBD>? ├── oni_api.py # Mod API 客户<E5AEA2><E688B7>?<3F><>? ├── oni_analyzer.py # 游戏状态分<E68081><E58886>?<3F><>? └── oni_builder.py # 蓝图建造规<E980A0><E8A784>?├── scripts/
<EFBFBD><EFBFBD>? ├── auto_repair.sh # 连接诊断
<EFBFBD><EFBFBD>? ├── auto_analyze.sh # 一键分<E994AE><E58886>?<3F><>? ├── watch.sh # 持续监控
<EFBFBD><EFBFBD>? └── setup.sh # 环境初始<E5889D><E5A78B>?├── docs/
<EFBFBD><EFBFBD>? ├── MOD_DEV_GUIDE.md # Mod 开发指<E58F91><E68C87>?<3F><>? └── AI_KNOWLEDGE_BASE.md # AI 知识<E79FA5><E8AF86>?(ID注册<E6B3A8><E5868C>?语义标签)
├── skills/
<EFBFBD><EFBFBD>? └── oni_agent.md # Agent skill 定义
└── SKILL.md # 本文<E69CAC><E69687>?```
你通过 REST API + Python CLI 工具链完全控制《缺氧》。你的目标是**让复制人存活并建立自持基地**。
---
## 重要概念<EFBFBD><EFBFBD>?ONI 的数据模<E68DAE><E6A8A1>?
### 1. 坐标<E59D90><E6A087>?
ONI 使用二维方格tile系统。AI 必须理解坐标系才能正确操作:
## 一、核心概念
### 1.1 坐标系
- (0,0) = 地图**左下角**x→右y→上
- 每个格子 (cell) 1×1通过 (x,y) 唯一定位
- 建筑坐标 = 其**左下角锚点**
- 地图大小:`/api/state/game` → gridWidth × gridHeight通常 256×384
- **自动拉视角**dig/build/deconstruct/prioritize 会自动将镜头移到坐标
### 1.2 格子数据
通过 `cell <x> <y>` 查看:
```
y <20><>? <20><>? ┌────┬────┬────<E29480><E29480>? <20><>? <20><>?5,5)<29><>?6,5)<29><>?7,5)<29><>? <20><>? ├────┼────┼────<E29480><E29480>? <20><>? <20><>?5,4)<29><>?6,4)<29><>?7,4)<29><>? <20><>?这个格子 (6,4) 包含一个电解器
<20><>? ├────┼────┼────<E29480><E29480>? <20><>? <20><>?5,3)<29><>?6,3)<29><>?7,3)<29><>? <20><>? └────┴────┴────<E29480><E29480>? └──────────────────────────<E29480><E29480>?x
(0,0)
element: "Oxygen" | "Water" | "SandStone" | "Vacuum"
massKg: 质量 | temperatureC: 温度
isSolid / isLiquid / isGas / isVacuum: 物态
isDiggable: 可挖掘固体且非Unobtanium
hasBuilding: 有建筑 | hasDuplicant: 有复制人
```
- **原点 (0,0)** 在地<E59CA8><E59CB0>?*左下<E5B7A6><E4B88B>?*
- **x <20><>?*向右增加<E5A29E><E58AA0>?*y <20><>?*向上增加
- 每个格子 (cell) 有唯一<E594AF><E4B880>?(x, y) 坐标
- 建筑占用 w×h 个格子,其坐标是**左下角锚<E8A792><E9949A>?*
- 世界大小通过 `/api/state/game` 查询(`gridWidth` x `gridHeight`<60><>?- 典型地图: ~256 x 384 <20><>?
### 2. 理解格子状<E5AD90><E78AB6>?
每格的数据结构如下(通过 `/api/state/cell?x=&y=` 查询):
### 1.3 可达性 (AI 必须理解)
复制人只能通过**开放空间**移动。障碍物和规则:
- **固体方块**isSolid=true阻挡移动——需要挖掘
- **液体/气体** 不阻挡移动
- **梯子 (Ladder)** 提供垂直移动
- **火棒 (FirePole)** 提供快速下落
- **门 (Door/PneumaticDoor)** 控制通行
- **砖块 (Tile)** 提供稳固地面
- **区域必须互相连通**:被固体完全包围的空间不可达
- **水位**:超过复制人高度的液体(约800kg/tile)会减慢甚至阻止移动
```json
{
"x": 10, "y": 5,
"element": "Oxygen", // 该格包含的元素名<E7B4A0><E5908D>? "elementState": "gas", // solid/liquid/gas/vacuum
"massKg": 1.8, // 该格中元素的质量
"temperatureC": 23.5, // 温度(摄氏<E69184><E6B08F>?
"hasBuilding": true, // 是否有建<E69C89><E5BBBA>? "buildingName": "Electrolyzer",// 建筑名称(如有)
"hasDuplicant": false, // 是否有复制人
"isVacuum": false, // 是否为真<E4B8BA><E79C9F>? "isSolid": false, // 是否为固<E4B8BA><E59BBA>? "isLiquid": false,
"isGas": true,
"isVisible": true // 是否已探<E5B7B2><E68EA2>?}
```
### 3. 理解地图区域
通过 `/api/state/cells?x=&y=&width=&height=` 获取矩形区域的格子数组<E695B0><E7BB84>?
通过 `explore <x> <y> <w> <h>` 命令获取 AI 友好的结构化摘要<E69198><E8A681>?- 该区域的建筑列表(带是否可运行)
- 该区域的复制人列表带压<E5B8A6><E58E8B>?当前任务<E4BBBB><E58AA1>?- 元素分布统计
- 感兴趣的关键格子
---
## 通信方式
- Mod 在游戏内启动 HTTP 服务<EFBC8C><E69AB4>?RESTful API
- 通过 `http://127.0.0.1:PORT` 与游戏通信
- 端口<E7ABAF><E58FA3>?`config.json` 中配置(默认 23876<37><36>?
---
## 可用 API 端点
### 状态查<E68081><E69FA5>?(GET)
| 端点 | 说明 | 用<><E794A8>?|
### 1.4 方块类型
| 方块 | 功能 | 材料 |
|------|------|------|
| `/api/state/buildable` | 当前科技解锁的建<E79A84><E5BBBA>?| 查看 AI 现在能造什<E980A0><E4BB80>?|
| `/api/state/research` | 科技树(进度/解锁的建筑) | 科研规划 |
| `/api/state/geysers` | 喷泉<EFBC88><E4BD8D>?状<><E78AB6>?排放率) | 资源规划 |
| `/api/state/alert` | 警报列表 | 紧急处<E680A5><E5A484>?|
| `/api/state/critters` | 小动物(位置/种类/幸福<E5B9B8><E7A68F>?年龄<E5B9B4><E9BE84>?| 养殖管理 |
| `/api/state/plants` | 植物<EFBC88><E4BD8D>?生长进度/是否枯萎<E69EAF><E8908E>?| 农业管理 |
| `/api/state/rooms` | 房间<EFBC88><E7B1BB>?格数/建筑数) | 房间判定 |
| Tile (砖块) | 地板/天花板,提供站立面 | RawMineral |
| Ladder (梯子) | 垂直攀爬 | RawMineral |
| FirePole (火棒) | 快速下落(比梯子快) | Metal |
| PneumaticDoor (气动门) | 控制进出 | Metal |
| ManualDoor (手动门) | 初期手动门 | RawMineral |
| InsulationTile (隔热砖) | 阻止热传递 | RawMineral + Ceramic |
| WindowTile (透光砖) | 透光地板 | Glass |
| MeshTile (网格砖) | 透水透气的砖块 | Metal |
| PlasticTile (塑料砖) | 高装饰地板 | Plastic |
| CarpetTile (地毯砖) | 高装饰 | Reed Fiber + Plastic |
### 地图/格子数据 (GET)
### 1.5 材料系统
建筑需要**特定类别的材料**。游戏把材料分为以下类别:
| 端点 | 说明 | 示例 |
| 材料类别 | 包含元素 | 如何获得 |
|----------|---------|---------|
| **BuildableRaw** | SandStone, Granite, IgneousRock, SedimentaryRock, Obsidian, MaficRock | 挖掘 |
| **RawMineral** | 同上BuildableRaw的子集 | 挖掘 |
| **Metal** | Cuprite(铜矿), IronOre(铁矿), GoldAmalgam(金汞齐), Wolframite(钨矿), Cobaltite(钴矿) | 挖掘 → 精炼 |
| **RefinedMetal** | Copper(铜), Iron(铁), Gold(金), Steel(钢), Tungsten(钨) | 在精炼炉/金属精炼器中生产 |
| **Plastic** | Polypropylene(塑料) | 聚合压机制造 |
| **Glass** | Glass(玻璃) | 窑炉烧制砂→熔融玻璃→冷却 |
| **BuildingFiber** | Reed Fiber(芦苇纤维) | 种植芦苇→收获 |
| **Transparent** | Diamond(钻石), Glass(玻璃) | 挖掘或制造 |
| **Farmable** | Dirt(泥土) | 挖掘或堆肥 |
**AI 必须理解**:造建筑时,需要查看世界库存中**有哪些元素满足该类别**。
例如 `ManualGenerator` 需要 Material=`['Metal']`,世界库存中有 Cuprite(铜矿) 和 IronOre(铁矿)AI 应选择存量最多的。
### 1.6 挖掘系统
- 挖掘命令:`dig <x> <y> <width> <height>` → 在矩形区域内标记所有可挖掘格子
- 可挖掘条件:`isSolid === true && element !== Unobtanium && element !== Vacuum`
- 挖掘需要**任务排队**:复制人把挖掘任务加入队列后才会执行
- 挖掘完成后:该格变为真空或气体(取决于背后的元素)
- 挖掘产物:掉落物 → 复制人会捡起→放入附近储存箱
- **必须在要建造的区域先挖掘**,因为建筑不能放置在固体方块上
### 1.7 任务队列与优先级
- 游戏内所有操作(挖掘/建造/运输/研究)都由复制人执行
- 优先级 1-99最高影响复制人选择顺序
- 全局优先级默认5。紧急任务设为9非紧急设为1
---
## 二、分阶段游戏目标
### 第一阶段:初期基地 (Cycle 1-20) — 生存基础
**目标:建立基本生存设施**
**检查清单:**
```
□ 手动发电机(ManualGenerator) ×1 — 提供电力
□ 电池(Battery) ×1 — 储电
□ 氧气扩散器(OxygenDiffuser) ×1 — 用藻类产氧(如有)
□ 或电解器(Electrolyzer) + 水泵 — 用水产氧
□ 厕所(Outhouse/Lavatory) ×1-2 — 控制污染
□ 洗手盆(WashBasin) ×1 — 防止食物中毒
□ 床(Cot) ×3 — 每复制人1张
□ 配给盒(RationBox) ×1 — 储食放在CO₂中防腐
□ 种植箱(PlanterBox) ×3-5 — 种Mealwood产食物
□ 研究站(ResearchStation) ×1 — 开始科研
□ 储存箱(StorageLocker) ×N — 分类储料
```
**关键决策点:**
- **氧气路径选择**:藻类 >200kg 可以用OxygenDiffuser否则必须尽快建造 Electrolyzer
- **电力规划**:先手工(ManualGenerator),尽快过渡到燃煤(CoalGenerator)
- **食物规划**Mealwood用泥土是最简单的初期食物来源
**我该怎么做:**
```bash
# 1. 暂停游戏
python tools/oni_api.py pause "Phase 1 - Initial base"
# 2. 查看状态
python tools/oni_api.py status
# 3. 探索起始区域通常在120-140, 180-200附近
python tools/oni_api.py explore 120 180 40 30
# 4. 检查资源
python tools/oni_api.py resources
# 5. 挖掘基地空间
python tools/oni_api.py dig 120 180 20 15
# 6. 建造初期设施(检查材料是否足够)
python tools/oni_api.py build ManualGenerator 125 192
python tools/oni_api.py build Battery 123 192
python tools/oni_api.py build ResearchStation 125 189
python tools/oni_api.py build Cot 120 192
# 7. 开始研究(农牧技术 → 电力调节)
python tools/oni_api.py research_select FarmingTech
# 8. 恢复游戏
python tools/oni_api.py unpause 1
```
### 第二阶段:中期扩张 (Cycle 20-100) — 自持系统
**目标:建立闭环系统,摆脱对初始资源的依赖**
**检查清单:**
```
□ SPOM电解器+氢气发电机)— 无限氧气+部分电力
□ 卫生间水循环(马桶→净水器→马桶)— 无限水循环
□ 燃煤发电机 ×1-2 — 主电力来源
□ 变压器(Transformer) — 电路分离,防止过载
□ 食物农场扩张 — ≥6个种植箱或4个农场砖
□ 医疗床(TriageCot) — 治疗受伤复制人
□ 装饰物 — 缓解压力(参考房间系统)
□ 隔热砖包围基地 — 阻止外部高温进入
□ 气压服存放柜(OxygenMaskLocker) — 探索外部
```
**关键系统详解:**
- **SPOM**水→电解器→O₂+H₂→H₂发电机(发电)+O₂排入基地
- 需要:水源(初始水池或喷泉)+ 电解器 + 气泵 + 氢气发电机 + 气体过滤器
- 使用蓝图:`python tools/oni_builder.py build spom <x> <y>`
- **卫生间水循环**:抽水马桶→水泵→净水器→抽水马桶
- 净水器消耗过滤介质(沙/砂石),排出清水+污染物(污染土)
- 使用蓝图:`python tools/oni_builder.py build toilet_loop <x> <y>`
- **电力管理**单路电线最大1000W超过=过载损坏
- 使用变压器分成小电路:发电线路(重导线) → 变压器 → 用电线路(普通导线)
- 检查电力:`/api/state/power` 查看各电路负载需要Mod实现
**科研优先级:**
1. FarmingTech → 食物生产
2. PowerRegulation → 变压器+重导线
3. Plumbing → 厕所水循环
4. Ventilation → 气泵+管道
5. Refinement → 精炼金属+玻璃
6. TemperatureModulation → 温度控制
### 第三阶段:后期自动化 (Cycle 100-300) — 全面自持
**目标:全自动化,应对长期挑战**
**检查清单:**
```
□ 石油发电 / 天然气发电 — 稳定大容量电力
□ 液冷模块(蒸汽机+导热管)— 主动降温
□ 自动化控制(自动化线+传感器)— 智能管理
□ 喷泉开发(冷水泉/蒸汽泉/天然气泉)— 无限资源
□ 畜牧模块(哈奇/滑鳞/飞鱼)— 稳定食物+材料
□ 太空探索(望远镜+火箭)— 终极资源
□ 精炼系统(金属精炼器/聚合压机/窑炉)
```
**温度控制:**
- 生物生存温度:-10°C ~ 40°C
- 高温来源:发电机/精炼器/电池 → 需隔热+冷却
- 冷却方案Aquatuner(导热管)+ SteamTurbine(蒸汽机) → 主动热删除
- 隔热方案InsulationTile + Vacuum气密层
**房间系统:**
房间提供重要加成。房间判定条件:
- **卧室(Bedroom)**4×床 + 封闭空间 + 门 → +1士气
- **卫生间(Washroom)**:洗手盆+马桶+封闭 → +1士气
- **餐厅(MessHall)**:餐桌+食物盒 → +2士气
- **温室(Greenhouse)**:种植箱+光照+封闭 → 植物生长+50%
---
## 三、建筑数据库AI 常用)
| buildingId | 尺寸 | 材料 | 耗电 | 用途 |
|-----------|------|------|------|------|
| ManualGenerator | 2×2 | Metal | -400W | 人力发电 |
| CoalGenerator | 2×2 | Metal | -600W | 燃煤发电 |
| HydrogenGenerator | 2×2 | Metal | -800W | 燃氢发电 |
| Electrolyzer | 2×2 | Metal | +120W | 水→O₂+H₂ |
| OxygenDiffuser | 2×2 | Metal | +120W | 藻类→O₂ |
| GasPump | 1×2 | Metal | +240W | 抽气 |
| GasFilter | 1×2 | Metal | +120W | 过滤气体 |
| WaterPump | 1×2 | Metal | +240W | 抽水 |
| LiquidPump | 1×2 | Metal | +240W | 抽任意液体 |
| LiquidFilter | 1×2 | Metal | +120W | 过滤液体 |
| WaterPurifier | 2×2 | Metal | +120W | 污水→清水 |
| PlanterBox | 1×1 | BuildableRaw | 0 | 种植食物 |
| FarmTile | 1×1 | BuildableRaw | 0 | 灌溉种植 |
| StorageLocker | 1×1 | BuildableRaw | 0 | 储存 |
| RationBox | 2×2 | BuildableRaw | 0 | 储食 |
| Cot | 1×1 | BuildableRaw | 0 | 床 |
| Lavatory | 1×2 | Metal | 0 | 抽水马桶 |
| WashBasin | 1×1 | BuildableRaw | 0 | 洗手 |
| Outhouse | 1×2 | BuildableRaw | 0 | 初期厕所 |
| Tile | 1×1 | BuildableRaw | 0 | 地板 |
| InsulationTile | 1×1 | BuildableRaw | 0 | 隔热墙 |
| Ladder | 1×1 | BuildableRaw | 0 | 梯子 |
| FirePole | 1×1 | Metal | 0 | 火棒 |
| PneumaticDoor | 1×2 | Metal | 0 | 气压门 |
| Battery | 1×1 | Metal | 0 | 电池(1000J) |
| Transformer | 2×1 | Metal | 0 | 变压器 |
| ResearchStation | 2×2 | BuildableRaw | +60W | 初级研究 |
| SuperComputer | 2×2 | Plastic | +120W | 高级研究 |
| SteamTurbine | 4×2 | Metal | -850W | 发电+热删除 |
| Aquatuner | 1×2 | RefinedMetal | +1200W | 液体冷却 |
| Kiln | 1×2 | BuildableRaw | 0 | 烧陶瓷/玻璃 |
| MetalRefinery | 3×2 | Metal | +1200W | 精炼金属 |
| PolymerPress | 2×3 | Metal | +240W | 石油→塑料 |
| Incubator | 2×2 | BuildableRaw | 0 | 孵化蛋 |
---
## 四、AI 操作铁律
### 4.1 标准操作流程(每一步自动拉视角)
```
STEP 1: 暂停
pause "我要做什么" ← 必须!任何决策前暂停
STEP 2: 感知
status ← 游戏总览
resources ← 资源清单
events ← 最近事件
explore x y w h ← 探索目标区域
STEP 3: 验证坐标
cell x y ← 检查目标格子是否可建造/可挖掘
registry buildings ID ← 确认buildingId正确
STEP 4: 执行
dig x y w h ← 先挖(自动拉视角)
build ID x y ← 再建(自动拉视角)
deconstruct x y ← 拆除(自动拉视角)
STEP 5: 验证与恢复
events ← 检查执行结果
unpause 1 ← 恢复运行
```
### 4.2 错误处理
| 错误 | 原因 | 对策 |
|------|------|------|
| `/api/state/cell?x=10&y=5` | 单格详情 | 查看某个格子是气<E698AF><E6B094>?液体/建筑 |
| `/api/state/cells?x=0&y=0&width=10&height=10` | 矩形区域 | 查看 10x10 区域 |
| `/api/state/cells/slice?axis=y&index=20&start=0&end=50` | <20><>?列扫<E58897><E689AB>?| 查看<E69FA5><E79C8B>?20 <20><>?|
| `/api/state/gas?x=10&y=10&radius=20` | 区域气体分析 | 查看周围气体成分 |
| `nothing_to_dig` | 区域已经挖空 | 跳过挖掘,直接建造 |
| `cell_occupied` | 该格已有建筑 | `deconstruct` 或选相邻位置 |
| `unknown_building` | buildingId 不存在 | 查 `registry buildings <关键词>` |
| `material_shortage` | 材料不够 | `resources` 查看缺什么,先挖材料 |
| `bounds` | 坐标越界 | 确认坐标在 gridWidth×gridHeight 内 |
| `missing_prerequisites` | 科技未解锁 | 先 `research_select` 研发前置科技 |
### 实体注册<EFBFBD><EFBFBD>?(GETAI 参考用)
### 4.3 紧急处理
| 端点 | 说明 |
|------|------|
| `/api/registry/buildings` | 所有建<E69C89><E5BBBA>?ID 及尺<E58F8A><E5B0BA>?功<><E58A9F>?发热 |
| `/api/registry/elements` | 所有元<E69C89><E58583>?ID 及比热容/导热/熔沸<E78694><E6B2B8>?|
| `/api/registry/techs` | 所有科技 ID 及前<E58F8A><E5898D>?解锁内容 |
### 操作 (POST)
| 端点 | 请求<E8AFB7><E6B182>?| 用<><E794A8>?|
|------|--------|------|
| `/api/action/toggle` | `{x, y}` | 开关建筑(省电/控制流程<E6B581><E7A88B>?|
| `/api/action/set_recipe` | `{x, y, recipeId}` | 设置建筑配方 |
| `/api/action/empty` | `{x, y}` | 清空建筑储物 |
| `/api/action/cancel_errand` | `{x, y}` | 取消建筑处的任务 |
| `/api/action/dig` | `{x, y, width, height}` | 挖掘区域 |
| `/api/action/build` | `{buildingId, x, y}` | 建造建<E980A0><E5BBBA>?|
| `/api/action/deconstruct` | `{buildingId, x, y}` | 拆除建筑 |
| `/api/action/prioritize` | `{x, y, priority}` | 设优先级 |
| `/api/action/research` | `{techId}` | 选研究项<E7A9B6><E9A1B9>?|
| `/api/action/mop` | `{x, y}` | 清理液体 |
| `/api/action/harvest` | `{x, y}` | 收获植物 |
---
## AI 如何进行推理和操<E5928C><E6938D>?
### 第一步获取全局上下<E4B88A><E4B88B>?
```bash
python3 tools/oni_api.py status
python3 tools/oni_api.py buildings
python3 tools/oni_analyzer.py
```
### 第二步:理解地图
```bash
# 探索基地中心区域(假设基地在 50,50<35><30>?python3 tools/oni_api.py explore 40 40 40 30
# 检查某个格子的详细信息
python3 tools/oni_api.py cell 45 48
# 查看气体分布
python3 tools/oni_api.py gas 50 50 30
```
### 第三步:参考知识库
```bash
# 查找某个建筑<E5BBBA><E7AD91>?ID
python3 tools/oni_api.py registry buildings Electrolyzer
# 查看元素属<E7B4A0><E5B19E>?python3 tools/oni_api.py registry elements Water
# 查看科技<E7A791><E68A80>?python3 tools/oni_api.py registry techs
```
### 第四步:执行操作
```bash
# 建造电解器
python3 tools/oni_api.py build Electrolyzer 45 48
# 挖掘空间
python3 tools/oni_api.py dig 40 45 8 6
# 选择科研方向
python3 tools/oni_api.py research_select ImprovedOxygen
# 使用蓝图
python3 tools/oni_builder.py build spom 42 42
```
---
## AI 如何理解常见游戏场景
### 场景 1氧气不<E6B094><E4B88D>?
**AI 推理过程<E8BF87><E7A88B>?*
1.<><E6A380>?`/api/state/resources` 中的 O2 <20><>?Algae 存量
2.<><E6A380>?`/api/state/buildings` 是否有电解器或氧气扩散器
3.<><E6A380>?`/api/state/cell?x=&y=` 查询基地气体分布
4. 如果 Algae < 1t 且无电解<EFBFBD><EFBFBD>?<3F><>?建议建<EFBFBD><EFBFBD>?SPOM
5. SPOM 需要水源 + 电解<EFBFBD><EFBFBD>?+ 气体<EFBFBD><EFBFBD>?+ 氢气发电<EFBFBD><EFBFBD>?+ 气体过滤<EFBFBD><EFBFBD>?6. 通过 `explore` 找到一<EFBFBD><EFBFBD>?8x6 的空<EFBFBD><EFBFBD>?7. 执行 `build Electrolyzer x y` + `build GasPump ...` + `build HydrogenGenerator ...`
### 场景 2食物短<E789A9><E79FAD>?
**AI 推理过程<EFBFBD><EFBFBD>?*
1. <EFBFBD><EFBFBD>?Calories < 500,000 kcal <EFBFBD><EFBFBD>?食物预警
2. 检查是否有 PlanterBox/FarmTile <EFBFBD><EFBFBD>?ElectricGrill
3. 如果没有农场 <EFBFBD><EFBFBD>?建议建<EFBFBD><EFBFBD>?5 <EFBFBD><EFBFBD>?PlanterBox <EFBFBD><EFBFBD>?Mealwood
4. Mealwood 不需要灌溉或施肥只需 Dirt
5. <EFBFBD><EFBFBD>?Dirt 存量如果足<EFBFBD><EFBFBD>?<3F><>?执行建<EFBFBD><EFBFBD>?6. 如果有污<EFBFBD><EFBFBD>?<3F><>?建议建<EFBFBD><EFBFBD>?Water Sieve + 厕所水循<EFBFBD><EFBFBD>?
### 场景 3温度过<E5BAA6><E8BF87>?
**AI 推理过程<EFBFBD><EFBFBD>?*
1. 检查温度数据通过资源中的 Temperature 或格子数据
2. 查看热源煤发电机精炼厂等靠近基地的位置<EFBFBD><EFBFBD>?3. 建议用隔热门包围热<EFBFBD><EFBFBD>?+ 建造液冷模<EFBFBD><EFBFBD>?4. 液冷模块需要Aquatuner + SteamTurbine + 导热液体管道
---
## 暂停与速度参<E5BAA6><E58F82>?
游戏状态中<EFBFBD><EFBFBD>?`isPaused` <EFBFBD><EFBFBD>?`gameSpeed` 字段<EFBFBD><EFBFBD>?
```
isPaused: true <20><>?是否暂停
gameSpeed: 0 <20><>?0=暂停, 1=1x, 2=2x, 3=3x
```
暂停规则已集成在"统一操作协议"的标<EFBFBD><EFBFBD>?SOP 详见下节核心原则
- **所有写操作前必须暂<EFBFBD><EFBFBD>?*dig/build/deconstruct/batch/pipe/wire<EFBFBD><EFBFBD>?- **只读查询不需要暂<EFBFBD><EFBFBD>?*status/resources/events<EFBFBD><EFBFBD>?- **操作完成后必须恢<EFBFBD><EFBFBD>?*
---
---
## AI 统一操作协议
这是 AI 操作缺氧的标准协议所有决策和操作必须遵循此协议<EFBFBD><EFBFBD>?
### 铁律(必须遵守)
当事件流中出现以下情况:
```
铁律 1: 任何时<E4BD95><E697B6>?AI 开始推<E5A78B><E68EA8>?决策<E586B3><E7AD96>?<3F><>?必须先暂停游戏<E6B8B8><E6888F>? 工具在获取游戏状态数据时会自动触发暂停,
保证 AI 获取的信息是当前时刻的准确快照<E5BFAB><E785A7>? <20><>?工具自动执行 pause无需手动调用<E8B083><E794A8>? <20><>?AI 完成所有操作后主动 unpause
铁律 2: 任何时<E4BD95><E697B6>?AI 需要向用户提问必须先暂停游戏<E6B8B8><E6888F>? <20><>?工具自动暂停,用户回答后 AI 恢复<E681A2><E5A48D>?unpause
铁律 3: 任何时<E4BD95><E697B6>?AI 结束回答/退出操作状态,必须确保游戏处于暂停态,
除非用户明确要求不暂停<E69A82><E5819C>? <20><>?防止游戏<E6B8B8><E6888F>?AI 不监控时状态恶化(窒息/过载/高温/CO₂
铁律 4: 任何时<E4BD95><E697B6>?AI 执行写操作dig/build/deconstruct/batch/pipe/wire
工具自动确保暂停态<E5819C><E68081>?
铁律 5: 重大操作前必须先 save 存档,失败后允许 load 回滚<E59B9E><E6BB9A>?```
### 核心操作循环
每次 AI 与游戏交互都必须遵循这个五步循环<E5BEAA><E78EAF>?
```
┌─────────────────────────────────────────────────────────<EFBFBD><EFBFBD>?<3F><>? 1. 上下文感<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?<3F><>? snapshot + diagnose + events <EFBFBD><EFBFBD>?<3F><>? "我现在看到什么当前状态是什么发生了什么" <EFBFBD><EFBFBD>?└──────────────────────┬──────────────────────────────────<E29480><E29480>? <EFBFBD><EFBFBD>?┌─────────────────────────────────────────────────────────<E29480><E29480>?<3F><>? 2. 决策与规<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?<3F><>? pause <EFBFBD><EFBFBD>?分析数据 <EFBFBD><EFBFBD>?确定目标 <EFBFBD><EFBFBD>?选择工具 <EFBFBD><EFBFBD>?<3F><>? "基于现状我需要做什么用什么工具在哪个坐标" <EFBFBD><EFBFBD>?└──────────────────────┬──────────────────────────────────<E29480><E29480>? <EFBFBD><EFBFBD>?┌─────────────────────────────────────────────────────────<E29480><E29480>?<3F><>? 3. 执行前保<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?<3F><>? save <EFBFBD><EFBFBD>?camera <EFBFBD><EFBFBD>?cell <EFBFBD><EFBFBD>?snapshot <EFBFBD><EFBFBD>?<3F><>? "先存档然后把视野移过去确认坐标正<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?└──────────────────────┬──────────────────────────────────<E29480><E29480>? <EFBFBD><EFBFBD>?┌─────────────────────────────────────────────────────────<E29480><E29480>?<3F><>? 4. 执行操作 <EFBFBD><EFBFBD>?<3F><>? dig/build/build_pipe_line/batch <EFBFBD><EFBFBD>?<3F><>? 每一步检查反馈success/fail + suggestion<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?└──────────────────────┬──────────────────────────────────<E29480><E29480>? <EFBFBD><EFBFBD>?┌─────────────────────────────────────────────────────────<E29480><E29480>?<3F><>? 5. 验证与恢<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?<3F><>? unpause <EFBFBD><EFBFBD>?snapshot <EFBFBD><EFBFBD>?检查状<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?<3F><>? 恶化 <EFBFBD><EFBFBD>?load 回滚 <EFBFBD><EFBFBD>?换方<EFBFBD><EFBFBD>? <EFBFBD><EFBFBD>?└─────────────────────────────────────────────────────────<E29480><E29480>?```
### 标准操作 SOP
任何时<EFBFBD><EFBFBD>?AI 执行操作必须按以下流程<E6B581><E7A88B>?
```
步骤 0: 暂停
python3 tools/oni_api.py pause "操作说明"
<20><>?确认 success=true否则重<E58899><E9878D>?
步骤 1: 视觉确认
python3 tools/oni_api.py camera <x> <y> <zoom>
python3 tools/oni_api.py snapshot
<20><>?下载截图确认目标位置正<E7BDAE><E6ADA3>?
步骤 2: 数据确认
python3 tools/oni_api.py cell <x> <y>
<20><>?hasBuilding=true <20><>?<3F><>?deconstruct 或换位置
<20><>?isSolid=true <20><>?<3F><>?dig
<20><>?isVacuum=true <20><>?确认原因
步骤 3: 安全存档
python3 tools/oni_api.py save "before_任务<E4BBBB><E58AA1>?
<20><>?确认 success=true
步骤 4: 执行
dig / build / build_pipe_line / build_wire_line / batch
<20><>?每次检查反馈:
success=true <20><>?继续
success=false <20><>?<3F><>?suggestion <20><>?调整重试 <20><>?3次失败则 load 回滚
步骤 5: 验证
snapshot <20><>?截图对比
resources <20><>?资源变化
未改<E69CAA><E694B9>?<3F><>?load 回滚
步骤 6: 恢复
unpause 1
⚠️ 氧气<500kg → pause → emergency_o2 → build Electrolyzer
⚠️ 食物<200kkcal → pause → build PlanterBox ×5
⚠️ CO₂积聚 → pause → fix_co2 → 挖排气管路
⚠️ 温度>40°C → pause → 隔热+冷却模块
⚠️ 污水满溢 → pause → build WaterPurifier
```
### 决策框架
AI 面对任何问题按此框架思考:
### 4.4 工具调用速查
```
【数据收集<EFBFBD><EFBFBD>? status <20><>?diagnose <20><>?power <20><>?co2 <20><>?temp_zones <20><>?resources <20><>?buildings
# 基本监控
python tools/oni_api.py status
python tools/oni_api.py resources
python tools/oni_api.py events
python tools/oni_api.py registry buildings <关键词>
【定位分析<EFBFBD><EFBFBD>? cells <x> <y> <w> <h> <20><>?cell <x> <y> <20><>?camera + snapshot
# 坐标操作
python tools/oni_api.py cell <x> <y> # 看格子
python tools/oni_api.py cells <x> <y> <w> <h> # 看区域
python tools/oni_api.py explore <x> <y> <w> <h> # AI摘要
python tools/oni_api.py gas <x> <y> <r> # 气体
【方案选择<EFBFBD><EFBFBD>? 紧<><E7B4A7>?<3F><>?emergency_o2 / fix_co2 / fix_overload
<><E5BBBA>?<3F><>?定坐<E5AE9A><E59D90>?<3F><>?build / build_pipe_line / build_wire_line
批量 <20><>?batch JSON
扩展 <20><>?expand_base
<><E6A380>?<3F><>?diagnose
回滚 <20><>?load
# 执行
python tools/oni_api.py pause "原因"
python tools/oni_api.py dig <x> <y> <w> <h>
python tools/oni_api.py build <id> <x> <y>
python tools/oni_api.py deconstruct <x> <y>
python tools/oni_api.py prioritize <x> <y> <1-9>
python tools/oni_api.py research_select <id>
python tools/oni_api.py batch <file.json>
python tools/oni_api.py save <name>
python tools/oni_api.py camera <x> <y> <zoom>
python tools/oni_api.py unpause <1-3>
# 高级
python tools/oni_analyzer.py
python tools/oni_commander.py diagnose
python tools/oni_commander.py emergency_o2
python tools/oni_commander.py fix_co2
python tools/oni_commander.py fix_overload
python tools/oni_builder.py build spom <x> <y>
python tools/oni_builder.py build toilet_loop <x> <y>
python tools/oni_builder.py build bedroom <x> <y>
python tools/oni_builder.py build cooling <x> <y>
```
### 标准方案手册
#### 方案 A初期基地Cycle 1-20<32><30>?
```
1. pause "Initial base"
2. diagnose
3. expand_base <中心> <<3C><>? <<3C><>?
4. build ResearchStation <坐标>
5. build ManualGenerator <坐标>
6. build Battery <坐标>
7. build OxygenDiffuser <坐标>
8. build PlanterBox x5
9. build Cot x3 / LadderBed x3
10. build Outhouse + WashBasin
11. research_select FarmingTech
12. research_select PowerRegulation
13. unpause 1
```
#### 方案 BCO2 危机
```
症状duplicants 窒息co2 显示大量 CO2
1. pause "CO2"
2. co2 <20><>?找到最<E588B0><E69C80>?CO2 聚集<E8819A><E99B86>?3. camera <x> <最<><E69C80>?y> 20
4. snapshot <20><>?确认地形
5. save "before_co2"
6. fix_co2 <20><>?自动挖排气管
7. snapshot <20><>?确认挖<E8AEA4><E68C96>?8. unpause 1
9. 30s <20><>?co2 确认下降
```
#### 方案 C电力过<E58A9B><E8BF87>?
```
症状power 显示 *** OVERLOAD ***
1. pause "Power overload"
2. power <20><>?哪个电路过载
3. diagnose <20><>?资源检<E6BA90><E6A380>?4. save "before_power"
5. fix_overload <20><>?修复建议
6. 如电力不<E58A9B><E4B88D>?<3F><>?build <发电<E58F91><E794B5>? <20><>?build_wire_line <连接>
7. power <20><>?确认过载消失
8. unpause 1
```
#### 方案 D氧气危<E6B094><E58DB1>?
```
症状duplicants 显示 oxygen<20%
1. pause "O2"
2. snapshot <20><>?视觉确认
3. resources <20><>?O2/Algae 存量
4. buildings <20><>?电解<E794B5><E8A7A3>?扩散<E689A9><E695A3>?5. save "before_o2"
6. 分支<E58886><E694AF>? 无设<E697A0><E8AEBE>?<3F><>?build OxygenDiffuser
无藻<E697A0><E897BB>?<3F><>?build Electrolyzer + 接水管电<E7AEA1><E794B5>? 设备不工<E4B88D><E5B7A5>?<3F><>?cell 检查供水和供电
7. unpause 1
8. 30s <20><>?resources <20><>?O2 回升<E59B9E><E58D87>?```
#### 方案 E<EFBC9A><E5BBBA>?SPOM
```
1. pause "SPOM"
2. explore <20><>?8x6 空地
3. camera <20><>?snapshot 确认
4. save "before_spom"
5. oni_builder.py build spom <x> <y>
6. build_pipe_line liquid <水源> <电解<E794B5><E8A7A3>?
7. build_wire_line heavy <发电<E58F91><E794B5>? <电池> cross
8. unpause 1
```
#### 方案 F管道铺设带交叉处理
```
需求:液体管道横穿已有气体管道
1. pause "Plumbing"
2. camera <起点> <终点> 25
3. save "before_pipe"
4. 横穿段用 cross 模式<E6A8A1><E5BC8F>? build_pipe_line liquid <起点> <终点> cross
<20><>?Mod 自动在交叉处放跨接器
5. pipes liquid <20><>?确认流动
6. unpause 1
```
#### 方案 G电线布<E7BABF><E5B883>?
```
需求:为新建筑拉电线到电网
1. pause "Wiring"
2. camera <建筑> <电源> 25
3. power <20><>?查空余容<E4BD99><E5AEB9>?4. save "before_wire"
5. 布线(横穿用 cross
build_wire_line regular <建筑> <变压<E58F98><E58E8B>? cross
6. unpause 1
```
### 错误恢复
```
操作失败 <20><>?AI 必须读取 error + errorMessage + suggestion
错误处理表:
cell_occupied <20><>?换坐标或 deconstruct
cell_solid <20><>?<3F><>?dig
cell_occupied_by_dupe<EFBFBD><EFBFBD>?等待
material_shortage <20><>?查资<E69FA5><E8B584>?+ 安排生产
unknown_building <20><>?registry buildings 查询
unknown_tech <20><>?registry techs 查询
missing_prerequisites<EFBFBD><EFBFBD>?先研究前置科技
invalid_priority <20><>?<3F><>?1-9
save_not_found <20><>?saves 列出
重试 3 次失<E6ACA1><E5A4B1>?<3F><>?load 回滚
```
### 坐标定位方法
```
方法 1: 基于已有建筑偏移
buildings <20><>?查已有建筑坐<E7AD91><E59D90>?<3F><>?偏移放置
方法 2: 基于区域探索
explore <x> <y> <w> <h> <20><>?找空<E689BE><E7A9BA>?
方法 3: 基于资源位置
co2 <20><>?CO2 聚集点下方挖排气<E68E92><E6B094>?
方法 4: 基于 cell 验证
cell <x> <y> <20><>?确认 isSolid=false + hasBuilding=false + isVisible=true
AI 习惯头脑规划坐<E58892><E59D90>?<3F><>?cell 验证 <20><>?确认无误再建<E5868D><E5BBBA>?```
### 事件自动响应
event_daemon 运行<E8BF90><E8A18C>?AI 自动响应规则<E8A784><E58899>?
```
[CRITICAL] 窒息
<20><>?diagnose <20><>?执行方案 D 或方<E68896><E696B9>?B
[CRITICAL] 电力中断
<20><>?power <20><>?执行方案 C
[WARNING] 食物短缺
<20><>?Calories<200k <20><>?<3F><>?PlanterBox x5
[WARNING] 温度过高
<20><>?temp_zones <20><>?隔热<E99A94><E783AD>?+ 冷却
[INFO] 新周<E696B0><E591A8>? <20><>?research <20><>?继续科研
<20><>?resources <20><>?安排生产
AI 不应等待指令——事件本身就是指令<E68C87><E4BBA4>?```
---
## AI 如何表达"在哪个格子做什<E5819A><E4BB80>?
### 定位语法
AI 在描述操作时应使用以下格式:
```
在坐<EFBFBD><EFBFBD>?(x, y) 建<><E5BBBA>?<buildingId>
在区<EFBFBD><EFBFBD>?(x, y, width, height) 进行挖掘
<EFBFBD><EFBFBD>?(x1,y1) <20><>?(x2,y2) 铺设管道/电线
在格<EFBFBD><EFBFBD>?(x, y) 设置优先级为 <priority>
```
### 坐标查找策略
<EFBFBD><EFBFBD>?AI 不确定在哪里建造时<E980A0><E697B6>?1. 先用 `explore` 找一个空闲区域(没有建筑和固体阻挡)
2. 检查空闲区域的元素和温度是否适合
3. <20><>?`cell` 命令确认目标格子状<E5AD90><E78AB6>?4. 然后<E784B6><E5908E>?`dig` 清理空间
5. 最后用 `build` 建<><E5BBBA>?
### 建筑放置规则
- 建筑坐标是其**左下<E5B7A6><E4B88B>?*的位<E79A84><E4BD8D>?- 建筑占用<E58DA0><E794A8>?w×h 区域必须全部是空<E698AF><E7A9BA>?- 需要确认目标区域无建筑、无固体自然方块
- 气体/液体不会阻挡建筑
- 如果建筑需要特定环境如电解器需要水AI 需要先检查环<E69FA5><E78EAF>?
---
## 工具列表
| 工具 | 用<><E794A8>?|
|------|------|
| `tools/oni_api.py` | Mod API 客户端<EFBC88><E78AB6>?格子/注册<E6B3A8><E5868C>?操作/批量/优先<E4BC98><E58588>?事件<E4BA8B><E4BBB6>?|
| `tools/oni_analyzer.py` | 自动分析游戏状态、生成预警和建议 |
| `tools/oni_builder.py` | 预置蓝图建造SPOM/农场/养殖等) |
| `scripts/auto_repair.sh` | 诊断 Mod 连接问题 |
| `scripts/auto_analyze.sh` | 一键健康检<E5BAB7><E6A380>?状<><E78AB6>?分析 |
| `scripts/watch.sh []` | 循环监控模式 |
| `scripts/setup.sh` | 环境初始化与检<E4B88E><E6A380>?|
| `scripts/event_daemon.py` | **事件守护进程** <20><>?持续轮询事件 <20><>?AI 输入<E8BE93><E585A5>?|
| `docs/AI_KNOWLEDGE_BASE.md` | 建筑/元素/科技 ID 注册表和游戏机制参<E588B6><E58F82>?|
| `docs/batch_example.json` | 批量任务示例文件 |
---
## 核心游戏知识
### 生存优先<E4BC98><E58588>?1. **氧气** <20><>?电解<E794B5><E8A7A3>?> 藻类制氧(前期过渡)
2. **食物** <EFBFBD><EFBFBD>?浆果 > 烤肉 > 营养<E890A5><E585BB>?3. **温度控制** <20><>?液冷 + 蒸汽<E892B8><E6B1BD>?4. **电力** <20><>?氢气发电 > 煤炭 > 手动
5. **水资源管<E6BA90><E7AEA1>?* <20><>?净水器、污水过<E6B0B4><E8BF87>?
### 常用布局
- SPOM: 电解制氧 + 氢气发电闭环
- 卫生间水循环: 卫生<E58DAB><E7949F>?<3F><>?净水器 <20><>?卫生<E58DAB><E7949F>?- 冷却系统: 液冷 + 蒸汽<E892B8><E6B1BD>?+ 导热<E5AFBC><E783AD>?- Ranch 模块: 养殖哈奇/滑鳞/飞鱼
### 关键事件预警
- 氧气不足 (< 500g/tile) <EFBFBD><EFBFBD>?增加制氧
- 温度超标 (> 40°C <20><>?< -10°C) <EFBFBD><EFBFBD>?增加温控
- 食物短缺 (< 5 周期余量) <EFBFBD><EFBFBD>?扩大种植/养殖
- 电力不足 <EFBFBD><EFBFBD>?增加发电或减少负<EFBFBD><EFBFBD>?- 污水满溢 <EFBFBD><EFBFBD>?增加净<EFBFBD><EFBFBD>?扩大存储

File diff suppressed because it is too large Load Diff

View File

@ -1,3 +1,4 @@
title: "ONI Agent Bridge"
staticID: "oni_agent_bridge"
description: "Expose HTTP API for AI agents to read ONI game state and perform actions"
description: "Expose HTTP RESTful API for AI agents to read ONI game state and perform actions (dig/build/deconstruct/research/camera)"
tags: ["Mod", "API", "AI"]

View File

@ -1,4 +1,4 @@
supportedContent: ALL
minimumSupportedBuild: 722606
version: 1.0.0
version: 2.0.0
APIVersion: 2

View File

@ -2,294 +2,96 @@
"""
ONI Agent Event Daemon
======================
Continuous event poller that feeds game events to the AI's input stream.
Continuous event poller. Polls game events every N seconds and displays
critical/warning/info events to the console (for AI input stream).
Architecture:
Game Mod --> Event Queue (via HTTP) --> Event Daemon --> AI Input Stream
The daemon:
1. Polls GET /api/state/events?since=<seq> every N seconds
2. Classifies events by severity (critical/warning/info)
3. For critical events: immediately triggers full analysis + prints alert
4. For warning events: logs and optionally triggers targeted checks
5. For info events: accumulates and reports periodically
6. Maintains a compact event log for AI context
Usage:
python event_daemon.py [interval_seconds]
"""
import json
import os
import sys
import time
import datetime
import json, os, sys, time, datetime
# Add tools to path
TOOLS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'tools')
sys.path.insert(0, TOOLS_DIR)
# Disable auto-pause for event daemon — we poll frequently and should not pause the game
import oni_api
oni_api.AUTO_PAUSE_ENABLED = False
from oni_api import api_get, api_post, api_url, load_config
from oni_api import api_get, api_post, api_url
# ── Configuration ──────────────────────────────────────────────────────────
POLL_INTERVAL = 5 # seconds between event polls
CRITICAL_POLL_INTERVAL = 2 # poll faster when critical events detected
MAX_EVENT_HISTORY = 200 # events kept in rolling buffer
CRITICAL_SEVERITIES = {'critical', 'duplicantdeath', 'buildingdamage', 'poweroutage'}
WARNING_SEVERITIES = {'warning', 'duplicantstress', 'lowoxygen', 'foodshortage'}
# ── Event History ──────────────────────────────────────────────────────────
POLL_INTERVAL = 5
MAX_EVENTS = 200
class EventHistory:
"""Rolling buffer of events + statistics for AI context."""
def __init__(self, maxlen=MAX_EVENT_HISTORY):
def __init__(self, maxlen=200):
self.events = []
self.maxlen = maxlen
self.stats = {
'total': 0,
'critical': 0,
'warning': 0,
'info': 0,
'by_category': {},
'by_type': {},
'last_poll_cycle': 0,
}
self.last_seq = -1
self.stats = {"critical": 0, "warning": 0, "info": 0}
def push(self, events):
for e in events:
self.events.append(e)
self.stats['total'] += 1
sev = (e.get('severity') or 'info').lower()
cat = e.get('category', 'unknown')
etype = e.get('type', 'unknown')
if sev in ('critical', 'duplicantdeath', 'buildingdamage'):
self.stats['critical'] += 1
elif sev in ('warning',):
self.stats['warning'] += 1
else:
self.stats['info'] += 1
self.stats['by_category'][cat] = self.stats['by_category'].get(cat, 0) + 1
self.stats['by_type'][etype] = self.stats['by_type'].get(etype, 0) + 1
self.stats['last_poll_cycle'] = e.get('cycle', 0)
# Trim
def add(self, event):
self.events.append(event)
if len(self.events) > self.maxlen:
self.events = self.events[-self.maxlen:]
self.events.pop(0)
sev = event.get('severity', 'info')
self.stats[sev] = self.stats.get(sev, 0) + 1
def get_summary(self):
return {
'total_events': self.stats['total'],
'critical_count': self.stats['critical'],
'warning_count': self.stats['warning'],
'info_count': self.stats['info'],
'categories': dict(sorted(self.stats['by_category'].items(),
key=lambda x: -x[1])[:10]),
'last_cycle': self.stats['last_poll_cycle'],
'recent_critical': [e for e in self.events[-20:]
if (e.get('severity') or '').lower() in CRITICAL_SEVERITIES][-5:],
}
def get_recent(self, n=10):
return self.events[-n:]
# ── Event Classifier ──────────────────────────────────────────────────────
def classify_event(e):
"""Return the action type for a given event."""
sev = (e.get('severity') or '').lower()
title = (e.get('title') or '').lower()
msg = (e.get('message') or '').lower()
cat = (e.get('category') or '').lower()
if sev in CRITICAL_SEVERITIES:
return 'critical'
if sev in WARNING_SEVERITIES:
return 'warning'
if cat == 'action':
return 'action_feedback'
# Content-based classification
combined = title + ' ' + msg
if any(w in combined for w in ['suffocat', 'choking', 'no oxygen', 'out of air']):
return 'critical'
if any(w in combined for w in ['starving', 'food', 'hungry']):
return 'warning'
if any(w in combined for w in ['heat', 'overheat', 'temperature', 'melt']):
return 'warning'
if any(w in combined for w in ['power', 'wattage', 'shutoff']):
return 'warning'
if any(w in combined for w in ['duplicant', 'stress', 'break']):
return 'warning'
if any(w in combined for w in ['research complete', 'research completed']):
return 'info_research'
if any(w in combined for w in ['printing pod']):
return 'info_printing_pod'
return 'info'
def format_event_for_ai(e):
"""Format an event as a structured string for AI input."""
ts = datetime.datetime.fromtimestamp(e.get('timestamp', time.time())).strftime('%H:%M:%S')
cycle = e.get('cycle', '?')
severity = e.get('severity', 'info').upper()
title = e.get('title', '?')
message = e.get('message', '')
lines = [f"[EVENT {severity}] Cycle {cycle} @ {ts}"]
lines.append(f" Title: {title}")
if message:
lines.append(f" Message: {message}")
entity = e.get('entity')
if entity:
lines.append(f" Entity: {entity}")
cell = e.get('cell')
if isinstance(cell, int) and cell >= 0:
lines.append(f" Cell: {cell}")
return '\n'.join(lines)
# ── Polling Loop ──────────────────────────────────────────────────────────
def poll_loop(event_history):
seq = 0
def main():
interval = int(sys.argv[1]) if len(sys.argv) > 1 else POLL_INTERVAL
history = EventHistory(MAX_EVENTS)
consecutive_errors = 0
print("[ONI Event Daemon] Starting event poll...")
print(f"[ONI Event Daemon] Poll interval: {POLL_INTERVAL}s")
print()
print(f"[EventDaemon] Starting — poll interval {interval}s")
print(f"[EventDaemon] Config: {json.dumps(load_config())}")
while True:
try:
data = api_get(f"/api/state/events?since={seq}&limit=50")
if 'error' in data:
r = api_get(f'/api/state/events?since={history.last_seq}')
if not r.get("success"):
consecutive_errors += 1
if consecutive_errors == 1:
print(f"[!] Cannot reach game: {data['error']}")
print(" Waiting for game connection...")
time.sleep(POLL_INTERVAL * 2)
if consecutive_errors > 3:
print(f"[EventDaemon] {consecutive_errors} consecutive errors — mod may be offline")
time.sleep(interval)
continue
consecutive_errors = 0
events = data.get('events', [])
next_seq = data.get('next_seq', seq)
data = r.get("data", {})
events = data.get("events", [])
next_seq = data.get("nextSeq", history.last_seq)
for ev in events:
history.add(ev)
sev = ev.get('severity', 'info')
title = ev.get('title', '')
message = ev.get('message', '')
ts = ev.get('timestamp', 0)
dt = datetime.datetime.fromtimestamp(ts).strftime('%H:%M:%S') if ts else ''
if sev in ('critical', 'error'):
print(f"\n⚠️ [{dt}] CRITICAL: {title} {message}")
elif sev in ('warning',):
print(f"\n🟡 [{dt}] WARNING: {title} {message}")
else:
print(f"🔵 [{dt}] INFO: {title} {message}")
if events:
event_history.push(events)
history.last_seq = next_seq
# Classify and report
critical_events = []
for e in events:
cls = classify_event(e)
if cls == 'critical':
critical_events.append(e)
# Print alert with clear marker
print("=" * 56)
print(" *** CRITICAL EVENT ***")
print(format_event_for_ai(e))
print("=" * 56)
print()
# Auto-trigger full analysis on critical events
_trigger_emergency_analysis(e)
elif cls == 'warning':
print(format_event_for_ai(e))
print()
elif cls == 'info_research':
# Research completed — show with unlocks
print("=" * 40)
print(format_event_for_ai(e))
print(" -> Check what's new: python3 tools/oni_api.py buildable")
print("=" * 40)
print()
elif cls == 'info_printing_pod':
# Printing pod ready
print("=" * 40)
print(format_event_for_ai(e))
print(" -> View options: python3 tools/oni_api.py printing_pod")
print("=" * 40)
print()
else:
# Only print non-info events or batch feedback
cat = e.get('category', '')
if cat != 'general' or cls != 'info':
print(format_event_for_ai(e))
print()
# If critical events happened, poll faster for a bit
if critical_events:
seq = next_seq
time.sleep(CRITICAL_POLL_INTERVAL)
continue
seq = next_seq
time.sleep(POLL_INTERVAL)
# Show summary every 30 seconds
if history.events and int(time.time()) % 30 < interval:
c = history.stats.get('critical', 0)
w = history.stats.get('warning', 0)
print(f"[EventDaemon] Stats | critical={c} warning={w} total={len(history.events)} | seq={history.last_seq}")
except KeyboardInterrupt:
print("\n[ONI Event Daemon] Shutting down.")
summary = event_history.get_summary()
print(f" Total events seen: {summary['total_events']}")
print(f" Critical: {summary['critical_count']}, Warning: {summary['warning_count']}")
print("\n[EventDaemon] Stopped")
break
except Exception as e:
consecutive_errors += 1
if consecutive_errors <= 2:
print(f"[!] Poll error: {e}")
time.sleep(POLL_INTERVAL)
except Exception as ex:
print(f"[EventDaemon] Error: {ex}")
time.sleep(interval)
continue
time.sleep(interval)
def _trigger_emergency_analysis(event):
"""On critical events, pull game state snapshot for AI context."""
try:
print(" -> Triggering emergency snapshot...")
game = api_get('/api/state/game')
alerts = api_get('/api/state/alert')
dups = api_get('/api/state/duplicants')
if 'error' not in game:
print(f" [SNAPSHOT] Cycle {game.get('cycle', '?')}, "
f"{game.get('duplicantCount', '?')} dupes, "
f"{game.get('suffocating', 0)} suffocating, "
f"{game.get('starving', 0)} starving, "
f"{game.get('stressed', 0)} stressed")
if isinstance(alerts, list) and alerts:
print(f" [ALERTS] {len(alerts)} active:")
for a in alerts[:3]:
print(f" - [{a.get('severity', '?')}] {a.get('title', '?')}")
print()
except:
pass
# ── Main ──────────────────────────────────────────────────────────────────
def main():
history = EventHistory()
try:
poll_loop(history)
except KeyboardInterrupt:
pass
# Print final summary
summary = history.get_summary()
print()
print("=" * 56)
print(" Event Daemon Session Summary")
print("=" * 56)
print(f" Total events: {summary['total_events']}")
print(f" Critical: {summary['critical_count']}")
print(f" Warning: {summary['warning_count']}")
print(f" Info: {summary['info_count']}")
print(f" Top categories: {', '.join(summary['categories'].keys())}")
print("=" * 56)
if __name__ == '__main__':
if __name__ == "__main__":
main()

View File

@ -1,466 +1,110 @@
# ONI Agent Skill
AI 操作《缺氧》(Oxygen Not Included) 的完整技能定义。
## 触发条件
- 用户提到《缺氧》/ Oxygen Not Included / ONI
- 用户询问游戏策略、建造方案、资源管理
- 用户希望 Mod 工具链执行操作
- 用户需要实时游戏操作协助
## 游戏核心玩法
缺氧是一款太空殖民模拟游戏。核心目标是在恶劣环境中维持复制人的生存并建设可持续基地。
### 生存要素优先级
| 优先级 | 要素 | 危险条件 | 后果 | 解决方案路线 |
|--------|------|---------|------|-------------|
| ★★★ | 氧气 | O₂ < 500 kg 或无制氧设备 | 复制人 110s 内窒息死亡 | 藻类制氧(前期) 电解器+氢气发电(SPOM) |
| ★★★ | 食物 | Calories < 1000 kcal/ | 复制人饿死 | 浆果种植 烤肉 营养膏棒 |
| ★★☆ | 温度 | > 40°C 或 < 10°C | 作物枯萎/复制人受伤 | 隔热/液冷+蒸汽机/加热 |
| ★★☆ | 电力 | 电量不足/电路过载 | 设备停摆/电线起火 | 手动天然气太阳能 |
| ★☆☆ | 士气 | Stress > 80% | 精神崩溃/破坏设备 | 房间奖励/装饰/按摩/高品质食物 |
| ★☆☆ | 病菌 | Slimelung/Food Poisoning | 生病死亡 | 洗手/净化/消毒/氧气服 |
### 游戏世界规则
- 二维网格地图,原点 (0,0) 在左下角x 向右 y 向上
- 气体按密度分层CO₂(沉底) < PollutedO < O < NaturalGas < H₂(上升)
- 建筑占用 w×h 格子坐标指左下角锚点
- 科技通过研究站超级计算机逐步解锁新建筑
- 3 周期传送舱提供一次新复制人或补给选择
---
name: oni_agent
description: "Oxygen Not Included 完全游戏操控 — REST API + CLI 工具链,含分阶段目标/材料系统/可达性/建筑数据库"
version: "2.1.0"
---
## API 总览92 个端点)
# ONI Agent Skill — 缺氧 AI 完全操控
工具链让 AI 能像人类玩家一样理解游戏并执行操作所有 API 通过 `tools/oni_api.py` 命令行调用
通过 `tools/oni_api.py` CLI + Mod HTTP API 完全控制游戏。**每次决策前必须先 pause**,所有操作自动拉视角到坐标。
### 一、状态查询 — 理解"当前发生了什么"
## 一、坐标系与格子
每次 AI 推理前工具自动暂停游戏保证数据时效性AI 操作完成后主动 unpause
- (0,0)=左下角x→右y→上典型地图 256×384
- 建筑坐标=左下角锚点
- 通过 `cell <x> <y>` 查看格子element / massKg / temperatureC / isSolid / isLiquid / isGas / isDiggable / hasBuilding / hasDuplicant
- `explore <x> <y> <w> <h>` AI友好区域摘要
- 可达性:复制人只能在开放空间移动,固体阻挡需挖掘,液体/气体不阻挡
- 梯子(Ladder)提供垂直移动,砖块(Tile)提供地面
## 二、材料系统
建筑需要特定类别材料:
- **BuildableRaw** = SandStone, Granite, IgneousRock, Obsidian (挖掘获得)
- **Metal** = Cuprite, IronOre, GoldAmalgam, Wolframite (挖掘→精炼)
- **RefinedMetal** = Copper, Iron, Gold, Steel (精炼生产)
- **Plastic** = Polypropylene (聚合压机制造)
- **Glass** = Glass (窑炉烧制)
- **Farmable** = Dirt (挖掘/堆肥)
AI 必须检查世界库存中哪些元素满足建筑的材料类别,选择存量最多的。
## 三、分阶段目标
### Phase 1 (Cycle 1-20): 生存基础
- **氧气**: OxygenDiffuser(藻类) 或 Electrolyzer(水→O₂+H₂)
- **电力**: ManualGenerator → [研究PowerRegulation] → CoalGenerator
- **食物**: PlanterBox×3-5 种 Mealwood (消耗Dirt)
- **卫生**: Outhouse + WashBasin → [研究Plumbing] → Lavatory循环
- **科研**: ResearchStation → FarmingTech先 → PowerRegulation → Plumbing
- **执行**: `dig` 挖空间 → `build IDs` 建造 → `research_select ID` 研究
### Phase 2 (Cycle 20-100): 自持系统
- **SPOM**: Electrolyzer+H₂发电机 → 无限O₂+部分电 (`oni_builder.py build spom`)
- **厕所循环**: Lavatory→WaterPurifier→Lavatory (`oni_builder.py build toilet_loop`)
- **电力升级**: 变压器(Transformer)防过载 + CoalGenerator
- **隔热**: InsulationTile 包围基地
- **养殖**: RanchStation + Incubator (哈奇哈奇产煤+食物)
### Phase 3 (Cycle 100+): 自动化
- 液冷模块 Aquatuner+SteamTurbine
- 石油/天然气发电
- 喷泉开发
- 太空探索
## 四、关键建筑ID速查
| ID | 尺寸 | 材料 | 耗电 | 用途 |
|----|------|------|------|------|
| ManualGenerator | 2×2 | Metal | -400W | 人力发电 |
| CoalGenerator | 2×2 | Metal | -600W | 燃煤发电 |
| HydrogenGenerator | 2×2 | Metal | -800W | 燃氢发电 |
| Electrolyzer | 2×2 | Metal | +120W | 水→O₂+H₂ |
| GasPump | 1×2 | Metal | +240W | 抽气 |
| WaterPurifier | 2×2 | Metal | +120W | 污水净化 |
| PlanterBox | 1×1 | BuildableRaw | 0 | 种植 |
| StorageLocker | 1×1 | BuildableRaw | 0 | 储存 |
| RationBox | 2×2 | BuildableRaw | 0 | 储食 |
| Tile | 1×1 | BuildableRaw | 0 | 地板 |
| InsulationTile | 1×1 | BuildableRaw | 0 | 隔热墙 |
| Ladder | 1×1 | BuildableRaw | 0 | 梯子 |
| PneumaticDoor | 1×2 | Metal | 0 | 门 |
| Battery | 1×1 | Metal | 0 | 电池 |
| Transformer | 2×1 | Metal | 0 | 变压器 |
| ResearchStation | 2×2 | BuildableRaw | +60W | 初级科研 |
| SteamTurbine | 4×2 | Metal | -850W | 热删除+发电 |
| Aquatuner | 1×2 | RefinedMetal | +1200W | 液冷 |
| Cot | 1×1 | BuildableRaw | 0 | 床 |
## 五、标准操作流程
```
1. pause "原因" ← 必须暂停
2. status / resources / events ← 感知
3. explore / cell ← 验证坐标
4. dig → build / deconstruct ← 执行(自动拉视角)
5. events ← 检查结果
6. unpause 1 ← 恢复
```
**紧急响应**: O₂<500emergency_o2 | 食物<200k建农场 | COfix_co2 | 电力过载fix_overload
## 六、工具速查
```bash
# 全局总览 — 最常用的命令
python3 tools/oni_api.py status
# 输出:周期、窒息人数、饥饿人数、压力人数、是否暂停、游戏速度
# 监控
oni_api.py status | resources | events | events N
oni_api.py explore x y w h | cell x y | cells x y w h | gas x y r
oni_api.py registry buildings [filter]
# 全面诊断
python3 tools/oni_commander.py diagnose
# 输出:电力+CO₂+温度+疾病+管道的完整报告
```
**各状态查询命令的用途:**
| 命令 | 等价于玩家做什么 | AI 用它做什么 |
|------|-----------------|-------------|
| `status` | 看一眼基地总览面板 | 快速评估危机有人在窒息吗在挨饿吗 |
| `resources` | 打开资源面板 | 检查氧气储量够多久煤还剩多少 |
| `duplicants` | 逐一查看每个复制人 | 谁在窒息谁压力爆了谁没事干 |
| `buildings` | 按分类查看建筑 | 数一数有多少发电机电解器种植箱 |
| `power` | 打开电力覆盖层 | 哪个电路过载了负载多少 |
| `co2` | CO 在哪聚集 | CO 是前中期最大杀手找到它并挖排气道 |
| `temp_zones` | 看温度覆盖层 | 找到过热区域>50°C和过冷区域<5°C |
| `pipes gas/liquid` | 点开管道看看 | 管道堵塞了吗内容物在流动吗 |
| `rooms` | 打开房间覆盖层 | 确认宿舍/卫生间/餐厅生效了吗 |
| `morale` | 查看士气面板 | 士气够不够支撑当前技能数 |
| `diseases` | 打开病菌覆盖层 | 谁被感染了环境病菌多吗 |
| `storage` | 查看储物建筑 | 储备粮在哪存了多少 |
| `storages` | 同上 | storage |
| `resources` | 资源面板 | 所有物资的精确余量 |
| `skills` | 技能面板 | 每个复制人的属性值和已学技能 |
| `research` | 科技树 | 哪些科技已完成哪些可研究 |
| `research_detail` | 点开研究站 | 研究站在工作吗现在在研究什么 |
| `buildable` | 建造菜单 | 当前科技解锁了哪些建筑 |
| `geysers` | 查看喷泉 | 找到水源/天然气/火山位置 |
| `critters` | 查看生物 | 哈奇/飞鱼/滑鳞的数量和位置 |
| `plants` | 查看植物 | 作物生长进度是否枯萎 |
### 二、格子地图 — 理解"每个格子有什么"
所有建造操作前必须先用格子查询确认位置
```bash
# 查看单个格子(最常用)
python3 tools/oni_api.py cell 45 48
# 输出:元素、温度、有无建筑、有无复制人、是否可挖掘、是否安全
# 查看 10x10 区域
python3 tools/oni_api.py cells 40 40 10 10
# 输出:区域内所有格子的地图 + AI 友好摘要
# 扫描某一行/列
python3 tools/oni_api.py slice y 20 0 50
# 输出:第 20 行从 0 到 50 的每个格子
# 气体分析
python3 tools/oni_api.py gas 50 50 30
# 输出:以 50,50 为中心半径 30 内的气体分布
# AI 友好区域探索
python3 tools/oni_api.py explore 40 40 20 20
# 输出:区域内建筑列表+复制人+元素摘要+感兴趣格子
```
**AI 坐标定位方法:**
```
方法 1: buildings → 查已有建筑坐标 → 在附近偏移找空地
方法 2: explore <x> <y> <w> <h> → 找 buildings_in_region 为空的区域
方法 3: cell <x> <y> → 确认 isSolid=false + hasBuilding=false + isVisible=true
方法 4: co2 → 在 CO₂ 聚集点下方挖排气通道
方法 5: gas → 在气体聚集区放气体泵
```
### 三、实体注册表 — "这个建筑是什么"
AI 在运行时查询任何不懂的 ID
```bash
# 查建筑
python3 tools/oni_api.py registry buildings Electrolyzer
# 输出:名称、分类、尺寸(2x2)、功耗(120W)、所需材料
# 查元素
python3 tools/oni_api.py registry elements Water
# 输出:比热容、导热系数、沸点(99°C)、物态
# 查科技
python3 tools/oni_api.py registry techs ImprovedOxygen
# 输出:前置科技、解锁建筑(Electrolyzer)
# 查优先级含义
python3 tools/oni_api.py registry priorities
# 输出1=最低 5=默认 9=紧急
```
### 四、建造操作 — "建造世界"
建造 = 挖掘(dig) 放置(build)。坐标是建筑左下角
```bash
# 挖掘区域
python3 tools/oni_api.py dig 40 40 10 10
# 在 (40,40) 处挖一个 10x10 的空间
# 放置建筑
python3 tools/oni_api.py build Electrolyzer 45 42
# 在 (45,42) 放电解器(左下角定位,占 2x2 格)
# 铺设管路
python3 tools/oni_api.py build_pipe_line liquid 42 40 48 40 line
# 从 (42,40) 到 (48,40) 铺液体管道
# mode=line: 与已有管线合并
# mode=cross: 交叉处自动用跨接器跳过
# mode=single: 单段
# 铺设电线
python3 tools/oni_api.py build_wire_line heavy 30 20 45 20 cross
# 从 (30,20) 到 (45,20) 铺重载电线
# 交叉处用跨接器不连通
# 拆除
python3 tools/oni_api.py deconstruct Tile 42 45
# 拆除 (42,45) 处的 Tile
# 旋转
python3 tools/oni_api.py rotate 45 48
# 旋转 (45,48) 处的建筑
# 清碎片
python3 tools/oni_api.py clear 40 40 5
# 清除 (40,40) 半径 5 内的碎片
# 复制设置
python3 tools/oni_api.py copy_settings 45 48 50 48
# 把 (45,48) 建筑的设置复制到 (50,48)
```
### 五、建筑操作 — "使用已建成的建筑"
相当于玩家点击建筑后选择功能
```bash
# 开关
python3 tools/oni_api.py toggle 45 42
# 关闭/打开电解器(省电或恢复运行)
# 设优先级
python3 tools/oni_api.py set_building_priority 45 42 9
# 电解器优先级设为 9紧急
# 设配方
python3 tools/oni_api.py set_recipe 30 25 Crush
# 碎岩机设为粉碎模式
# 开关自动化
python3 tools/oni_api.py set_automation 45 42 on
# 电解器开启自动化控制
# 清空储物
python3 tools/oni_api.py empty 50 50
# 清空 (50,50) 储物箱全部内容
# 设储物过滤
python3 tools/oni_api.py storage_filter 50 50 Coal
# 储物箱只收煤炭
# 设冰箱温度
python3 tools/oni_api.py fridge_temp 60 30 2
# 冰箱设为 2°C
# 设电池充放电阈值
python3 tools/oni_api.py battery_charge 35 25 90 20
# 智能电池 90% 停充、20% 启动
# 设阀门流量
python3 tools/oni_api.py valve_flow 42 42 1000
# 阀门限流 1000g/s
# 设排气口压力
python3 tools/oni_api.py vent_pressure 45 50 2000
# 高压排气口设为 2000g
# 设孵化器
python3 tools/oni_api.py incubator_setting 50 40 HatchEgg
# 孵化器优先孵哈奇蛋
# 设传感器阈值
python3 tools/oni_api.py sensor_threshold 70 30 500
# 气压传感器阈值 500g
# 消毒
python3 tools/oni_api.py disinfect 42 42
# 对 (42,42) 消毒
# 清扫
python3 tools/oni_api.py sweep 40 40 10
# 标记 (40,40) 半径 10 内的物品为待清扫
# 取消任务
python3 tools/oni_api.py cancel_errand 45 42
# 取消电解器处的排队任务
```
### 六、门控制
```bash
# 锁门
python3 tools/oni_api.py door_lock 42 45 on
# 锁上 (42,45) 的门,复制人不能通过
# 单向通行
python3 tools/oni_api.py door_one_way 42 45 left
# 门只允许向左通过
# 手动开门/关门
python3 tools/oni_api.py door_open 42 45 off
# 手动关闭 (42,45) 的门
```
### 七、复制人管理
```bash
# 移动复制人
python3 tools/oni_api.py dupe_move Dup1 30 25
# 命令 Dup1 移动到 (30,25)
# 取消任务
python3 tools/oni_api.py dupe_cancel_task Dup1
# 命令 Dup1 取消当前正在做的事
# 分配工作
python3 tools/oni_api.py assign_job Dup1 Dig
# 让 Dup1 专注挖掘工作
# 个人优先级
python3 tools/oni_api.py dupe_personal_priority Dup1 Dig 9
# Dup1 的挖掘优先级设为 9最高
```
### 八、科研管理
```bash
# 查看研究状态
python3 tools/oni_api.py research_detail
# 输出:研究站是否工作、当前研究什么、超级计算机是否造好
# 选择研究
python3 tools/oni_api.py research_select ImprovedOxygen
# 开始研究 ImprovedOxygen
# 取消研究
python3 tools/oni_api.py research_cancel
# 取消全部当前研究
```
### 九、生物/植物管理
```bash
# 攻击生物
python3 tools/oni_api.py critter_attack 120 80
# 标记 (120,80) 的生物为攻击目标
# 抓捕生物
python3 tools/oni_api.py critter_wrangle 120 80
# 抓捕 (120,80) 的生物运到存放点
# 收获植物
python3 tools/oni_api.py harvest 50 45
# 收获 (50,45) 的成熟作物
# 拔除植物
python3 tools/oni_api.py plant_uproot 50 45
# 拔掉 (50,45) 的植物
```
### 十、游戏控制
```bash
# 暂停/恢复
python3 tools/oni_api.py pause "Building SPOM"
python3 tools/oni_api.py unpause 1
# 速度
python3 tools/oni_api.py speed 3
# 存档回滚
python3 tools/oni_api.py save "before_experiment"
python3 tools/oni_api.py saves
python3 tools/oni_api.py load "before_experiment"
# 截图视角
python3 tools/oni_api.py snapshot /tmp/now.png
python3 tools/oni_api.py camera 120 80 20
# 覆盖层切换
python3 tools/oni_api.py overlay power
python3 tools/oni_api.py overlay temp
python3 tools/oni_api.py overlay gas
python3 tools/oni_api.py overlay rooms
# 传送舱
python3 tools/oni_api.py printing_pod
python3 tools/oni_api.py printing_pod_select 0
```
### 十一、批量操作
```bash
python3 tools/oni_api.py batch docs/batch_example.json
# 一次执行多个动作,逐个报告结果
```
### 十二、高级指令
```bash
# 全面诊断
python3 tools/oni_commander.py diagnose
# 自动处理 CO₂
python3 tools/oni_commander.py fix_co2
# 处理过载电路
python3 tools/oni_commander.py fix_overload
# 紧急制氧
python3 tools/oni_commander.py emergency_o2
# 一键拓展房间
python3 tools/oni_commander.py expand_base 40 40 10 8
```
---
## 常见场景查询路线
### "复制人要窒息了"
```
status → 看 suffocating 计数
duplicants → 找 oxygen < 20 的复制人
cell <窒息复制人坐标> → 所在格子的元素
co2 → 如果是 CO₂ 窒息fix_co2 挖排气道
resources → O₂ 存量和 Algae 存量
buildings → 过滤 Oxygen 分类看有无制氧设备
无设备 → emergency_o2 或 build OxygenDiffuser
```
### "基地没电了"
```
power → 看各电路负载和过载
resources → Coal/Hydrogen/NaturalGas 存量
buildings → Power 分类有哪些发电机
过载 → fix_overload
缺电 → 根据资源选发电机类型
```
### "研究卡住了"
```
research_detail → 研究站通不通电、有没有复制人在研究
building_detail <研究站坐标> → 供上电了吗
resources → Dirt 够吗(研究站消耗)
需超级计算机 → build SuperComputer <坐标>
```
### "管道不走了"
```
pipes liquid → 看管道内容
building_detail <水泵> → 泵有没有电
building_detail <目的地建筑> → 输入是否已满
build_pipe_line liquid <起点> <终点> cross → 重建
```
### "传送舱来了"
```
event_daemon 显示 [EVENT] Printing Pod ready
printing_pod → 看选项
printing_pod_select 0|1|2 → 选择
```
---
## AI 操作铁律
```
铁律 1: AI 开始推理/决策前 → 工具自动暂停游戏
铁律 2: AI 需要问用户时 → 工具自动暂停
铁律 3: AI 退出回答时 → 确保游戏已暂停(无人监控时不能运行)
铁律 4: 写操作前 → 工具自动确保暂停态
铁律 5: 重大操作前 → save 存档,失败可 load 回滚
铁律 6: 每次操作后检查反馈 → 读 error + errorMessage + suggestion
```
## 工具列表
| 工具 | 用途 |
|------|------|
| `tools/oni_api.py` | 92 API 端点的 CLI 客户端所有查询/操作 |
| `tools/oni_commander.py` | 高级指令diagnose/fix_co2/expand_base |
| `tools/oni_analyzer.py` | 自动分析+6 维度预警建议 |
| `tools/oni_builder.py` | 预置 SPOM/农场/养殖等蓝图 |
| `scripts/event_daemon.py` | 事件守护进程持续轮询推送游戏事件到 AI |
## 常用操作速查
```bash
# 每分钟检查
python3 scripts/event_daemon.py
# 紧急三步
python3 tools/oni_api.py pause "Emergency"
python3 tools/oni_commander.py diagnose
python3 tools/oni_api.py status
# ...操作...
python3 tools/oni_api.py unpause 1
# 存档保护
python3 tools/oni_api.py save "before_操作"
# ...操作...
# 如果搞砸了:
python3 tools/oni_api.py load "before_操作"
# 截图确认
python3 tools/oni_api.py camera 50 50 20
python3 tools/oni_api.py snapshot
# 操作(自动拉视角)
oni_api.py pause "r" | unpause [speed]
oni_api.py dig x y w h | build id x y | deconstruct x y
oni_api.py prioritize x y 1-9 | research_select id | batch file
oni_api.py camera x y [zoom] | save [name]
# 分析
oni_analyzer.py | oni_commander.py diagnose
oni_commander.py emergency_o2 | fix_co2 | fix_overload
oni_builder.py build spom/toilet_loop/bedroom x y
```

View File

@ -1,281 +1,149 @@
import json
import sys
#!/usr/bin/env python3
"""
ONI Agent — Game State Analyzer
================================
Fetches comprehensive game state and produces actionable analysis across
6 dimensions: oxygen, food, power, temperature, water, research.
"""
import json, sys, io
# Fix GBK encoding
if sys.stdout.encoding and sys.stdout.encoding.upper() in ('GBK', 'GB2312', 'CP936'):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
from oni_api import api_get
def get_data(endpoint):
r = api_get(endpoint)
if r.get("success"):
return r.get("data", r)
return None
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 main():
game = get_data('/api/state/game')
resources = get_data('/api/state/resources') or []
buildings = get_data('/api/state/buildings') or []
dups = get_data('/api/state/duplicants') or []
alert_data = get_data('/api/state/alert') or []
research = get_data('/api/state/research') or {}
rdict = {}
if isinstance(resources, list):
for r in resources:
rdict[r.get('name', '')] = r.get('amountKg', 0)
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)
# Building type counts
btypes = {}
for b in buildings if isinstance(buildings, list) else []:
bid = b.get('id', '')
btypes[bid] = btypes.get(bid, 0) + 1
print("=" * 56)
print(" ONI Analysis Report")
print(" ONI Agent — Game Analysis")
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 [])}")
if game:
speed_str = f"{game.get('gameSpeed','?')}x speed" if not game.get('isPaused') else "PAUSED"
print(f" Cycle {game.get('cycle','?')} | {game.get('duplicantCount','?')} dupes | "
f"{game.get('gridWidth','?')}x{game.get('gridHeight','?')} grid | {speed_str}")
print()
if critical:
print(f" [CRITICAL] {len(critical)} issues — act immediately!")
for _, msg in critical:
print(f" ! {msg}")
print()
# 1. Oxygen
print("─── Oxygen ───")
o2 = rdict.get('Oxygen', 0)
algae = rdict.get('Algae', 0)
pw = rdict.get('PollutedWater', 0)
water = rdict.get('Water', 0)
has_elec = 'Electrolyzer' in btypes
has_diff = 'OxygenDiffuser' in btypes
if warns:
print(f" [WARN] {len(warns)} issues")
for _, msg in warns:
print(f" * {msg}")
print()
print(f" O2: {o2:.0f} kg Algae: {algae:.0f} kg Water: {water:.0f} kg")
if o2 < 500:
print(f" ⚠ CRITICAL: Oxygen low! {o2:.0f} kg remaining")
if not has_elec and not has_diff:
print(f" ⚠ No oxygen production. Build OxygenDiffuser or plan SPOM")
elif not has_elec and algae < 1000:
print(f" ⚠ Algae running out ({algae:.0f} kg). Build Electrolyzer (SPOM)")
if water > 50000 and not has_elec:
print(f" Water abundant ({water:.0f} kg). Good time for SPOM")
if pw > 100000:
print(f" Polluted Water: {pw:.0f} kg — sieve into water or use for thimble reed")
if infos:
print(f" [INFO] {len(infos)} notes")
for _, msg in infos:
print(f" i {msg}")
print()
# 2. Food
print(f"\n─── Food ───")
calories = rdict.get('Calories', 0)
dirt = rdict.get('Dirt', 0)
has_farm = any(bid in ('PlanterBox', 'FarmTile') for bid in btypes)
has_grill = 'ElectricGrill' in btypes
print(f" Calories: {calories:.0f} kcal Dirt: {dirt:.0f} kg")
if calories < 200000:
print(f" ⚠ CRITICAL: Food shortage! {calories:.0f} kcal")
elif calories < 500000:
print(f" ⚠ Food declining ({calories:.0f} kcal). Build farm")
if not has_farm:
print(f" No farm. Build 5x PlanterBox + plant Mealwood (uses Dirt)")
if not has_grill and has_farm:
print(f" No grill. Build ElectricGrill for better food quality")
if not all_warnings:
print(" Status: All stable")
print()
# 3. Power
print(f"\n─── Power ───")
coal = rdict.get('Carbon', 0)
hydrogen = rdict.get('Hydrogen', 0)
has_manual = 'ManualGenerator' in btypes
has_coal = 'CoalGenerator' in btypes
has_hydro = 'HydrogenGenerator' in btypes
has_solar = 'SolarPanel' in btypes
has_natgas = 'NaturalGasGenerator' in btypes
print(f" Coal: {coal:.0f} kg Hydrogen: {hydrogen:.0f} kg")
gen_list = []
if has_manual: gen_list.append('Manual')
if has_coal: gen_list.append('Coal')
if has_hydro: gen_list.append('Hydrogen')
if has_natgas: gen_list.append('NaturalGas')
if has_solar: gen_list.append('Solar')
print(f" Generators: {', '.join(gen_list) if gen_list else 'None'}")
if coal < 1000 and has_coal:
print(f" ⚠ Coal low ({coal:.0f} kg). Diversify power production")
if not has_hydro and 'Electrolyzer' in btypes:
print(f" Have Electrolyzer but no HydrogenGenerator — wasting H2!")
if not any([has_manual, has_coal, has_hydro, has_natgas, has_solar]):
print(f" ⚠ No power generation! Build ManualGenerator or CoalGenerator")
if suggestions:
print(f" Suggestions ({len(suggestions)}):")
for s in suggestions:
print(f" -> {s}")
print()
# 4. Temperature
print(f"\n─── Temperature ───")
ice = rdict.get('Ice', 0) + rdict.get('CrushedIce', 0) + rdict.get('Snow', 0)
granite = rdict.get('Granite', 0)
igneous = rdict.get('IgneousRock', 0)
print(f" Ice/Snow: {ice:.0f} kg")
if ice > 0:
print(f" Ice available for cooling if melted")
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', '?')}")
# 5. Water
print(f"\n─── Water ───")
salt_water = rdict.get('SaltWater', 0)
brine = rdict.get('Brine', 0)
print(f" Water: {water:.0f} kg Polluted Water: {pw:.0f} kg Salt Water: {salt_water:.0f} kg")
if water < 10000 and pw < 10000:
print(f" ⚠ Low water! Collect from geysers or filter polluted water")
if water < 1000:
print(f" ⚠ CRITICAL: Water nearly empty!")
# 6. Research
print(f"\n─── Research ───")
completed = research.get('completedTechs', [])
print(f" Completed: {len(completed)} techs")
if completed:
print(f" Last: {completed[-1] if completed else 'none'}")
# 7. Alerts
if alert_data:
print(f"\n─── Active Alerts ───")
for a in alert_data:
print(f" [{a.get('severity','?')}] {a.get('title','?')}")
print()
print("=" * 56)
print(" Analysis complete.")
print("=" * 56)
return True
if __name__ == '__main__':
state = get_game_state()
if not print_report(state):
sys.exit(1)
if __name__ == "__main__":
main()

File diff suppressed because it is too large Load Diff

View File

@ -1,163 +1,185 @@
import json
import sys
from oni_api import api_post
#!/usr/bin/env python3
"""
ONI Agent — Blueprint Builder
==============================
Pre-built building modules for one-click deployment.
Usage:
python oni_builder.py list List all blueprints
python oni_builder.py show <name> Show blueprint details
python oni_builder.py build <name> <x> <y> Build blueprint at anchor point
"""
import json, sys
from oni_api import api_post, api_get
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},
'desc': 'Standard Rodriguez SPOM: electrolyzer + H2 generators + gas pumps',
'size': {'w': 8, 'h': 6},
'dig': {'dx': -1, 'dy': -1, 'w': 10, 'h': 8},
'steps': [
{'type': 'build', 'id': 'Electrolyzer', 'x': 3, 'y': 2},
{'type': 'build', 'id': 'GasPump', 'x': 1, 'y': 2},
{'type': 'build', 'id': 'GasPump', 'x': 5, 'y': 2},
{'type': 'build', 'id': 'HydrogenGenerator', 'x': 1, 'y': 0},
{'type': 'build', 'id': 'HydrogenGenerator', 'x': 4, '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},
'desc': 'Compact electrolyzer + 1 hydrogen generator for early game',
'size': {'w': 5, 'h': 4},
'dig': {'dx': -1, 'dy': -1, 'w': 7, 'h': 6},
'steps': [
{'type': 'build', 'id': 'Electrolyzer', 'x': 2, 'y': 1},
{'type': 'build', 'id': 'GasPump', 'x': 1, 'y': 1},
{'type': 'build', '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},
'desc': '2 Lavatories -> Water Purifier closed loop',
'size': {'w': 8, 'h': 4},
'dig': {'dx': -1, 'dy': -1, 'w': 10, 'h': 6},
'steps': [
{'type': 'build', 'id': 'Lavatory', 'x': 1, 'y': 1},
{'type': 'build', 'id': 'Lavatory', 'x': 3, 'y': 1},
{'type': 'build', 'id': 'WaterPurifier', 'x': 6, '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},
'desc': '5 PlanterBox + 1 StorageLocker',
'size': {'w': 6, 'h': 4},
'dig': {'dx': -1, 'dy': -1, 'w': 8, 'h': 6},
'steps': [
{'type': 'build', 'id': 'PlanterBox', 'x': 1, 'y': 1},
{'type': 'build', 'id': 'PlanterBox', 'x': 3, 'y': 1},
{'type': 'build', 'id': 'PlanterBox', 'x': 5, 'y': 1},
{'type': 'build', 'id': 'PlanterBox', 'x': 1, 'y': 3},
{'type': 'build', 'id': 'PlanterBox', 'x': 3, 'y': 3},
{'type': 'build', 'id': 'StorageLocker', 'x': 5, 'y': 3},
],
},
'bedroom': {
'name': 'Standard Bedroom',
'desc': '4 Cots + decor for bedroom room bonus',
'size': {'w': 8, 'h': 4},
'dig': {'dx': -1, 'dy': -1, 'w': 10, 'h': 6},
'steps': [
{'type': 'build', 'id': 'Cot', 'x': 1, 'y': 1},
{'type': 'build', 'id': 'Cot', 'x': 3, 'y': 1},
{'type': 'build', 'id': 'Cot', 'x': 5, 'y': 1},
{'type': 'build', 'id': 'Cot', 'x': 7, 'y': 1},
],
},
'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},
'desc': 'Aquatuner + Steam Turbine cooling loop',
'size': {'w': 8, 'h': 6},
'dig': {'dx': -1, 'dy': -1, 'w': 10, 'h': 8},
'steps': [
{'type': 'build', 'id': 'SteamTurbine', 'x': 1, 'y': 4},
{'type': 'build', 'id': 'SteamTurbine', 'x': 5, 'y': 4},
{'type': 'build', 'id': 'Aquatuner', 'x': 2, '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},
'ranch_hatch': {
'name': 'Hatch Ranch',
'desc': 'RanchStation + Incubator + StorageLocker',
'size': {'w': 10, 'h': 6},
'dig': {'dx': -1, 'dy': -1, 'w': 12, 'h': 8},
'steps': [
{'type': 'build', 'id': 'RanchStation', 'x': 1, 'y': 1},
{'type': 'build', 'id': 'Incubator', 'x': 4, 'y': 1},
{'type': 'build', 'id': 'StorageLocker', 'x': 8, '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']}")
def cmd_list(args):
print(f"Available blueprints ({len(BLUEPRINTS)}):")
for name, bp in sorted(BLUEPRINTS.items()):
print(f" {name:15s} {bp['name']}")
print(f" {'':15s} {bp['desc']}")
print(f" {'':15s} Size: {bp['size']['w']}x{bp['size']['h']} | {len(bp['steps'])} buildings")
print()
def cmd_show(args):
if not args:
print("Usage: show <blueprint_name>", file=sys.stderr)
return
name = args[0]
if name not in BLUEPRINTS:
print(f"Unknown blueprint: {name}", file=sys.stderr)
return
bp = BLUEPRINTS[name]
print(f"Blueprint: {bp['name']}")
print(f" {bp['desc']}")
print(f" Size: {bp['size']['w']}x{bp['size']['h']}")
if bp.get('dig'):
d = bp['dig']
print(f" Dig area: offset ({d['dx']},{d['dy']}) {d['w']}x{d['h']}")
print(f" Buildings ({len(bp['steps'])}):")
for s in bp['steps']:
print(f" {s['id']:25s} at anchor+({s['x']},{s['y']})")
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
def cmd_build(args):
if len(args) < 3:
print("Usage: build <blueprint_name> <anchor_x> <anchor_y>", file=sys.stderr)
return
name, ax, ay = args[0], int(args[1]), int(args[2])
if name not in BLUEPRINTS:
print(f"Unknown blueprint: {name}", file=sys.stderr)
return
bp = BLUEPRINTS[name]
print(f"Applying blueprint: {bp['name']}")
print(f" Origin: ({origin_x}, {origin_y})")
print(f" Size: {bp['size']['width']} x {bp['size']['height']}")
print()
# 1. Dig the area first
if bp.get('dig'):
d = bp['dig']
dx, dy = ax + d['dx'], ay + d['dy']
print(f"Digging area: ({dx},{dy}) {d['w']}x{d['h']}...")
r = api_post('/api/action/dig', {"x": dx, "y": dy, "width": d['w'], "height": d['h']})
if r.get("success"):
info = r.get("data", {})
print(f" Queued {info.get('count', '?')} dig orders")
else:
print(f" Dig warning: {r.get('errorMessage', r.get('error', 'unknown'))}")
# 2. Build each structure
results = []
for step in bp['steps']:
bx, by = ax + step['x'], ay + step['y']
print(f"Building {step['id']} at ({bx},{by})...")
r = api_post('/api/action/build', {"buildingId": step['id'], "x": bx, "y": by})
if r.get("success"):
info = r.get("data", {})
has_mat = info.get('hasAllMaterials', False)
results.append(f"{step['id']} at ({bx},{by}) {'(may wait for materials)' if not has_mat else ''}")
else:
results.append(f"{step['id']} at ({bx},{by}): {r.get('errorMessage', r.get('error', 'unknown'))}")
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}")
print(f"\n--- {bp['name']} build results ---")
for r in results:
print(r)
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)
def main():
if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help", "help"):
print(__doc__)
return
cmd = sys.argv[1]
args = sys.argv[2:]
if cmd == "list":
cmd_list(args)
elif cmd == "show":
cmd_show(args)
elif cmd == "build":
cmd_build(args)
else:
print("Usage: python oni_builder.py <list|build>")
print(f"Unknown: {cmd}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()

View File

@ -1,355 +1,219 @@
#!/usr/bin/env python3
"""
ONI Commander — 高级指令封装
============================
把多个底层 API 调用组合成一条"指挥官指令"AI 一句话就能执行复杂操作。
ONI Commander — High-Level Game Operations
===========================================
Combines multiple low-level API calls into one "commander directive".
Usage:
python oni_commander.py diagnose Full game diagnostic
python oni_commander.py fix_co2 Auto-vent CO2 pockets
python oni_commander.py fix_overload Diagnose power overloads
python oni_commander.py emergency_o2 Emergency oxygen setup
python oni_commander.py expand_base <cx> <cy> <w> <h> Dig expansion area
"""
import json
import sys
import os
import inspect
import json, sys, io
# Fix GBK encoding
if sys.stdout.encoding and sys.stdout.encoding.upper() in ('GBK', 'GB2312', 'CP936'):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
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 ──────────────────────────────────────────────────────────────
from oni_api import api_get, api_post, get_data
def pause_before():
"""High-level ops always pause first."""
api_post('/api/action/pause', {"reason": "High-level operation"})
api_post('/api/action/pause', {"reason": "commander operation"})
def pause_after():
"""Resume after operation."""
def unpause_after():
api_post('/api/action/unpause', {"speed": 1})
def cmd_diagnose(args):
pause_before()
game = get_data('/api/state/game')
resources = get_data('/api/state/resources') or []
buildings = get_data('/api/state/buildings') or []
dups = get_data('/api/state/duplicants') or []
alerts = get_data('/api/state/alert') or []
# ── Command Registry ─────────────────────────────────────────────────────
rdict = {}
for r in resources if isinstance(resources, list) else []:
rdict[r.get('name','')] = r.get('amountKg', 0)
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,
}
btypes = {}
for b in buildings if isinstance(buildings, list) else []:
bid = b.get('id', '')
btypes[bid] = btypes.get(bid, 0) + 1
def cmd_snapshot(args):
"""Take screenshot + view. Usage: snapshot [name]"""
from oni_api import cmd_snapshot as api_snapshot
api_snapshot(args)
print("=" * 60)
print(" ONI Full Diagnostic")
print("=" * 60)
if game:
speed_str = f"{game.get('gameSpeed','?')}x" if not game.get('isPaused') else "PAUSED"
print(f" Cycle {game.get('cycle','?')} | {game.get('duplicantCount','?')} dupes | {speed_str}")
def cmd_camera(args):
"""Move camera. Usage: camera <x> <y> [zoom]"""
from oni_api import cmd_camera as api_camera
api_camera(args)
print(f"\n── Resources ──")
for name in ['Oxygen', 'Water', 'PollutedWater', 'Dirt', 'Carbon', 'Hydrogen']:
v = rdict.get(name, 0)
print(f" {name:20s} {v:>10.1f} kg")
if __name__ == '__main__':
cmd = sys.argv[1] if len(sys.argv) > 1 else 'help'
print(f"\n── Buildings ({len(btypes)} types) ──")
for bid, cnt in sorted(btypes.items()):
print(f" {bid:30s} x{cnt}")
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)
print(f"\n── Dupes ({len(dups)}) ──")
for d in dups if isinstance(dups, list) else []:
print(f" {d.get('name','?'):12s} ({d.get('x',0)},{d.get('y',0)}) hp={d.get('health',0)}")
print(f"\n── Alerts ({len(alerts)}) ──")
for a in alerts:
print(f" [{a.get('severity','?')}] {a.get('title','?')}")
# Suggestions
print(f"\n── Suggestions ──")
o2 = rdict.get('Oxygen', 0)
cal = rdict.get('Calories', 0)
water = rdict.get('Water', 0)
coal = rdict.get('Carbon', 0)
if o2 < 500: print(f" 🔴 O2 CRISIS: {o2:.0f} kg — build SPOM immediately")
elif o2 < 2000: print(f" 🟡 O2 low: {o2:.0f} kg — plan oxygen production")
if cal < 200000: print(f" 🔴 FOOD CRISIS: {cal:.0f} kcal — build farm")
elif cal < 500000: print(f" 🟡 Food low: {cal:.0f} kcal")
if water < 5000: print(f" 🔴 WATER CRISIS: {water:.0f} kg")
if coal < 1000 and 'CoalGenerator' in btypes:
print(f" 🟡 Coal low ({coal:.0f} kg) — diversify power")
unpause_after()
def cmd_fix_co2(args):
"""Find CO2 pockets below base and dig to let it settle."""
pause_before()
game = get_data('/api/state/game')
if not game:
print("Cannot connect")
return
gw, gh = game.get('gridWidth', 256), game.get('gridHeight', 384)
print("Scanning for CO2 pockets...")
# Scan bottom portion of the map for CO2
scan_y = max(0, gh - 40)
r = api_get(f'/api/state/cells/slice?axis=y&index={scan_y}&start=0&end={gw-1}')
if not r.get("success"):
print("Cannot scan area")
unpause_after()
return
cells = r.get("data", {}).get("cells", [])
co2_cells = [c for c in cells if c.get('element') == 'CarbonDioxide' and c.get('isSolid') == False]
if not co2_cells:
print("No accessible CO2 pockets found in scan area")
unpause_after()
return
print(f"Found {len(co2_cells)} CO2 cells. Digging to the right for ventilation...")
for c in co2_cells[:5]:
dig_x = c['x'] + 1
r2 = api_post('/api/action/dig', {"x": dig_x, "y": scan_y, "width": 3, "height": 3})
if r2.get("success"):
print(f" Dig at ({dig_x},{scan_y}) 3x3 — queued")
unpause_after()
def cmd_fix_overload(args):
pause_before()
print("Diagnosing power...")
buildings = get_data('/api/state/buildings') or []
resources = get_data('/api/state/resources') or []
rdict = {}
for r in resources if isinstance(resources, list) else []:
rdict[r.get('name','')] = r.get('amountKg', 0)
btypes = {}
for b in buildings if isinstance(buildings, list) else []:
bid = b.get('id', '')
btypes[bid] = btypes.get(bid, 0) + 1
has_coal = 'CoalGenerator' in btypes
has_hydro = 'HydrogenGenerator' in btypes
has_manual = 'ManualGenerator' in btypes
coal = rdict.get('Carbon', 0)
print(f" Coal: {coal:.0f} kg")
print(f" Generators: Manual={has_manual} Coal={has_coal} Hydrogen={has_hydro}")
print(f" Coal plants: {btypes.get('CoalGenerator', 0)}")
print(f" Total buildings: {len(buildings)}")
if has_coal and coal < 2000:
print(f" ⚠ Low coal — supplement with manual generators")
if not has_hydro and 'Electrolyzer' in btypes:
print(f" ⚠ Wasteful: Electrolyzer running without HydrogenGenerator")
unpause_after()
def cmd_emergency_o2(args):
pause_before()
print("Emergency O2 response...")
resources = get_data('/api/state/resources') or []
buildings = get_data('/api/state/buildings') or []
rdict = {}
for r in resources if isinstance(resources, list) else []:
rdict[r.get('name','')] = r.get('amountKg', 0)
has_diffuser = any(b.get('id') == 'OxygenDiffuser' for b in buildings if isinstance(buildings, list))
algae = rdict.get('Algae', 0)
if has_diffuser:
print(" ✓ OxygenDiffuser exists — ensure it has power and algae")
elif algae > 200:
print(" Building OxygenDiffuser (uses algae)...")
r = api_post('/api/action/build', {"buildingId": "OxygenDiffuser", "x": 30, "y": 30})
if r.get("success"):
print(" → OxygenDiffuser queued")
else:
fn()
print(f" → Build failed: {r.get('errorMessage', r.get('error', 'unknown'))}")
else:
print(" No algae for diffuser. Need SPOM (Electrolyzer)")
water = rdict.get('Water', 0)
if water > 5000:
print(f" Water: {water:.0f} kg — enough for SPOM")
else:
print(f" Water: {water:.0f} kg — insufficient for electrolysis")
unpause_after()
def cmd_expand_base(args):
if len(args) < 4:
print("Usage: expand_base <center_x> <center_y> <width> <height>", file=sys.stderr)
return
cx, cy, w, h = int(args[0]), int(args[1]), int(args[2]), int(args[3])
dig_x, dig_y = cx - w // 2, cy - h // 2
pause_before()
print(f"Expanding: dig ({dig_x},{dig_y}) {w}x{h}")
r = api_post('/api/action/dig', {"x": dig_x, "y": dig_y, "width": w, "height": h})
if r.get("success"):
info = r.get("data", {})
print(f" Queued {info.get('count', '?')} dig orders")
else:
print(f" Error: {r.get('errorMessage', r.get('error', 'unknown'))}")
unpause_after()
def main():
if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help"):
print(__doc__)
return
cmd = sys.argv[1]
args = sys.argv[2:]
cmds = {
"diagnose": cmd_diagnose, "fix_co2": cmd_fix_co2,
"fix_overload": cmd_fix_overload, "emergency_o2": cmd_emergency_o2,
"expand_base": cmd_expand_base,
}
if cmd in cmds:
cmds[cmd](args)
else:
print(f"Unknown: {cmd}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()