diff --git a/.gitignore b/.gitignore index f310556..d0432c3 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ run_factorio_ai.ps1 inventory_memory.json +ore_patch_memory.json +agent_last_action_memory.json .cursor/ \ No newline at end of file diff --git a/README.md b/README.md index e910621..fa27b5e 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,7 @@ planner.set_goal( - AI가 "move 먼저 → 작업" 패턴을 학습하도록 프롬프트가 설계되어 있습니다 - `agent_log.jsonl`에 모든 행동과 타임스탬프가 기록됩니다 - `ai_planner.py`는 GLM 응답이 잘리거나(중괄호/대괄호 불일치) 마크다운이 섞여도 JSON 파싱을 복구하도록 `{} / []` 균형 추적과 보정 로직을 사용합니다. +- `ai_planner.py`의 `AIPlanner.decide()`는 `TimeoutError`/`ConnectionError`/`urllib.error.URLError` 같은 GLM HTTP 지연/연결 오류도 재시도 후 폴백(explore)합니다. - `ai_planner.py`의 `_call_glm()`에서 GLM 지연 원인 분석을 위한 타이밍 로그가 출력됩니다. - `total`: 요청 시작~콘텐츠 반환까지 전체 소요 - `http_read`: HTTP 응답 본문 수신까지 소요 @@ -138,6 +139,7 @@ planner.set_goal( - 또한 일부 환경에서 `p.get_main_inventory()`가 없으면 `defines.inventory.character_main`으로 한 번 더 읽습니다. - 캐시 파일이 아직 없으면(초기 실행) fallback도 `{}`가 됩니다. - (중요) 일부 Factorio 버전에서는 `game.table_to_json`이 없을 수 있어서, Lua 내부에 간단 JSON 인코더를 넣어 인벤토리/상태를 안정적으로 가져옵니다. +- `explore` 및 `scan_resources()`로 발견한 광맥(자원 엔티티) 좌표는 `ore_patch_memory.json`에 ore 종류별로 여러 패치 좌표 목록으로 저장되고, 다음 상태 요약에서 AI가 좌표를 재사용(우선 이동)할 수 있게 합니다. - 디버깅용으로 `INV_DEBUG=1`을 켜면, `main inventory` 외에 `cursor_stack`/`armor`/`trash` 총합도 함께 출력합니다(어디에 아이템이 있는지 확인용). - 또한 `main inventory`가 비어 캐시 fallback이 동작하면, 로드된 캐시의 `items/total`도 함께 출력합니다. - `mine_resource`에서 실패한 채굴 타일 제외(`exclude`)는 Lua와 Python 양쪽에서 정수 타일 좌표(`tx, ty`) 키로 통일해, 제외한 좌표가 반복 선택되지 않도록 합니다. @@ -145,7 +147,10 @@ planner.set_goal( - `mine_resource`는 move 후 `p.position` 근처(반경 1.2)에서 실제로 해당 광물 엔티티가 있는지 재확인하고, 없으면 채굴을 시도하지 않고 다음 후보로 넘어갑니다. - `scan_resources()`는 패치 평균 중심(`center_x/y`)만이 아니라 플레이어 기준 가장 가까운 실제 엔티티 좌표(`anchor_x/y`)도 함께 반환하며, `summarize_for_ai()`에서는 앵커를 기준으로 거리/추천을 계산합니다. - `mine_resource`의 초기 “후보 엔티티 탐색” 반경은 패치 중심 오차를 흡수하기 위해 250으로 확장해, 이동 직후 바로 실패하는 케이스를 줄입니다. +- 에이전트를 재시작하더라도 직전에 실행했던 action/result를 `agent_last_action_memory.json`에 저장해, 다음 상태 요약에서 AI가 참고할 수 있게 합니다. +- 저장 기준: 성공/실패 상관없이 가장 마지막으로 실행을 시도한 action 1개만 저장합니다. - `explore`는 `wanted_ores`가 있으면 해당 자원이 발견될 때까지 멈추지 않고 계속 이동해, `iron-ore`처럼 주변에 흔한 자원만 계속 발견되어 진행이 막히는 문제를 줄입니다. +- `place_entity`는 지정 좌표가 `BLOCKED`이면 `surface.can_place_entity`로 주변 `±1 타일`을 먼저 확인해, 가능한 좌표에 배치되도록 완화합니다. - (Cursor) Windows에서 `sessionStart` 훅 실행 중 앱 선택창이 뜨는 경우: - 프로젝트 훅은 `E:/develop/factorio-ai-agent/.cursor/hooks.json` 및 `E:/develop/factorio-ai-agent/.cursor/session-start-hook.ps1`를 PowerShell로 실행하도록 구성되어 있습니다. - Superpowers 플러그인의 `./hooks/session-start`가 bash로 실행되도록 `hooks-cursor.json`을 수정했습니다(필요 시 Cursor 재시작 후 `View -> Output -> Hooks`에서 확인). diff --git a/__pycache__/action_executor.cpython-311.pyc b/__pycache__/action_executor.cpython-311.pyc index baa10d2..b93adef 100644 Binary files a/__pycache__/action_executor.cpython-311.pyc and b/__pycache__/action_executor.cpython-311.pyc differ diff --git a/__pycache__/agent_last_action_memory.cpython-311.pyc b/__pycache__/agent_last_action_memory.cpython-311.pyc new file mode 100644 index 0000000..9e8d85b Binary files /dev/null and b/__pycache__/agent_last_action_memory.cpython-311.pyc differ diff --git a/__pycache__/ai_planner.cpython-311.pyc b/__pycache__/ai_planner.cpython-311.pyc index 87b8ae4..18bfa87 100644 Binary files a/__pycache__/ai_planner.cpython-311.pyc and b/__pycache__/ai_planner.cpython-311.pyc differ diff --git a/__pycache__/context_compressor.cpython-311.pyc b/__pycache__/context_compressor.cpython-311.pyc index f6d3d51..92834c5 100644 Binary files a/__pycache__/context_compressor.cpython-311.pyc and b/__pycache__/context_compressor.cpython-311.pyc differ diff --git a/__pycache__/main.cpython-311.pyc b/__pycache__/main.cpython-311.pyc index b78da32..f622a01 100644 Binary files a/__pycache__/main.cpython-311.pyc and b/__pycache__/main.cpython-311.pyc differ diff --git a/__pycache__/ore_patch_memory.cpython-311.pyc b/__pycache__/ore_patch_memory.cpython-311.pyc new file mode 100644 index 0000000..2ae8178 Binary files /dev/null and b/__pycache__/ore_patch_memory.cpython-311.pyc differ diff --git a/__pycache__/state_reader.cpython-311.pyc b/__pycache__/state_reader.cpython-311.pyc index 5872391..034e35d 100644 Binary files a/__pycache__/state_reader.cpython-311.pyc and b/__pycache__/state_reader.cpython-311.pyc differ diff --git a/action_executor.py b/action_executor.py index 2a2a069..8d0b4ce 100644 --- a/action_executor.py +++ b/action_executor.py @@ -7,6 +7,7 @@ action_executor.py — 순수 AI 플레이 버전 """ import time from factorio_rcon import FactorioRCON +from ore_patch_memory import load_ore_patch_memory, save_ore_patch_memory, update_ore_patch_memory P = """local p = game.players[1] if not p then rcon.print("NO_PLAYER") return end @@ -75,14 +76,37 @@ local ok, data = pcall(function() local res = p.surface.find_entities_filtered{position = pos, radius = 50, type = "resource"} if #res == 0 then return "NONE:" .. string.format("%.0f,%.0f", pos.x, pos.y) end local counts = {} - for _, e in ipairs(res) do counts[e.name] = (counts[e.name] or 0) + 1 end + local best_x, best_y = pos.x, pos.y + local best_d2 = math.huge + local wanted_active = wanted and next(wanted) ~= nil + + for _, e in ipairs(res) do + local n = e.name + counts[n] = (counts[n] or 0) + 1 + + local relevant = true + if wanted_active then + relevant = wanted[n] and true or false + end + if relevant then + local dx = e.position.x - pos.x + local dy = e.position.y - pos.y + local d2 = dx*dx + dy*dy + if d2 < best_d2 then + best_d2 = d2 + best_x = e.position.x + best_y = e.position.y + end + end + end + -- wanted_ores가 지정된 경우: 그 중 하나라도 있으면 FOUND, 아니면 UNWANTED 반환 - if wanted and next(wanted) ~= nil then + if wanted_active then for n, _ in pairs(wanted) do if counts[n] and counts[n] > 0 then local parts = {} for name, count in pairs(counts) do parts[#parts+1] = name .. "=" .. count end - return "FOUND:" .. string.format("%.0f,%.0f", pos.x, pos.y) .. "|" .. table.concat(parts, ",") + return "FOUND:" .. string.format("%.0f,%.0f", best_x, best_y) .. "|" .. table.concat(parts, ",") end end return "UNWANTED:" .. string.format("%.0f,%.0f", pos.x, pos.y) @@ -90,7 +114,7 @@ local ok, data = pcall(function() local parts = {} for name, count in pairs(counts) do parts[#parts+1] = name .. "=" .. count end - return "FOUND:" .. string.format("%.0f,%.0f", pos.x, pos.y) .. "|" .. table.concat(parts, ",") + return "FOUND:" .. string.format("%.0f,%.0f", best_x, best_y) .. "|" .. table.concat(parts, ",") end) if ok then rcon.print(data) else rcon.print("ERROR") end """) @@ -98,6 +122,39 @@ if ok then rcon.print(data) else rcon.print("ERROR") end if result.startswith("FOUND:"): self.rcon.lua(P + "p.walking_state = {walking = false, direction = defines.direction.north}") parts = result.replace("FOUND:", "").split("|") + # FOUND: "tx,ty|name=count,name2=count" 형태 + try: + loc = parts[0] + lx, ly = loc.split(",", 1) + tile_x = int(float(lx)) + tile_y = int(float(ly)) + except Exception: + tile_x, tile_y = None, None + + counts: dict[str, int] = {} + if len(parts) > 1 and parts[1]: + for seg in parts[1].split(","): + if "=" not in seg: + continue + k, v = seg.split("=", 1) + try: + counts[k] = int(v) + except Exception: + continue + + ores_to_store = wanted_ores if wanted_ores else list(counts.keys()) + if tile_x is not None and tile_y is not None and ores_to_store: + mem = load_ore_patch_memory() + updated = False + for ore in ores_to_store: + c = counts.get(ore) + if c is None or c <= 0: + continue + mem = update_ore_patch_memory(mem, ore, tile_x, tile_y, count=c) + updated = True + if updated: + save_ore_patch_memory(mem) + return True, f"자원 발견! 위치({parts[0]}), 자원: {parts[1] if len(parts)>1 else ''}" if result.startswith("UNWANTED:"): # 원하는 자원이 아니면 계속 걷기 @@ -333,13 +390,37 @@ local have = inv.get_item_count("{name}") if have < 1 then rcon.print("NO_ITEM") return end local dist = math.sqrt(({x} - p.position.x)^2 + ({y} - p.position.y)^2) if dist > p.build_distance + 2 then rcon.print("TOO_FAR:" .. string.format("%.1f", dist)) return end -p.cursor_stack.set_stack({{name="{name}", count=1}}) -local built = p.build_from_cursor{{position = {{{x}, {y}}}, direction = {lua_dir}}} -p.cursor_stack.clear() -if built then rcon.print("OK") else rcon.print("BLOCKED") end + +-- (x,y)가 자원 패치 위/겹침 등으로 배치 불가인 경우가 있어, +-- 인접 타일을 can_place_entity로 먼저 확인 후 성공 좌표를 사용한다. +local ox, oy = {x}, {y} +local candidates = {{ + {{ox, oy}}, + {{ox+1, oy}}, {{ox-1, oy}}, {{ox, oy+1}}, {{ox, oy-1}}, + {{ox+1, oy+1}}, {{ox-1, oy-1}}, {{ox+1, oy-1}}, {{ox-1, oy+1}} +}} + +for _, pos in ipairs(candidates) do + local cx, cy = pos[1], pos[2] + local can = p.surface.can_place_entity{{name="{name}", position={{cx, cy}}, direction={lua_dir}}} + if can then + p.cursor_stack.set_stack{{name="{name}", count=1}} + local built = p.build_from_cursor{{position = {{cx, cy}}, direction = {lua_dir}}} + p.cursor_stack.clear() + if built then + rcon.print(string.format("OK:%.0f,%.0f", cx, cy)) + return + end + end +end + +rcon.print("BLOCKED") """) if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, result or "플레이어 없음" if result == "OK": return True, f"{name} 배치 ({x},{y})" + elif result.startswith("OK:"): + coords = result.split(":", 1)[1] + return True, f"{name} 배치 ({coords})" elif result == "NO_ITEM": return False, f"인벤토리에 {name} 없음" elif result.startswith("TOO_FAR"): return False, f"너무 멀음 - move 먼저" elif result == "BLOCKED": return False, f"배치 불가" diff --git a/agent_last_action_memory.py b/agent_last_action_memory.py new file mode 100644 index 0000000..902a208 --- /dev/null +++ b/agent_last_action_memory.py @@ -0,0 +1,46 @@ +""" +agent_last_action_memory.py + +에이전트를 재시작하더라도 직전에 실행했던 action 및 결과를 기억하기 위한 간단 파일 메모리. +""" + +from __future__ import annotations + +import json +import os +from typing import Any + + +LAST_ACTION_CACHE_FILE = "agent_last_action_memory.json" +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) + + +def _cache_path() -> str: + return os.path.join(BASE_DIR, LAST_ACTION_CACHE_FILE) + + +def load_last_action_memory() -> dict[str, Any]: + path = _cache_path() + try: + if not os.path.exists(path): + return {} + with open(path, "r", encoding="utf-8") as f: + raw = f.read().strip() + if not raw: + return {} + data = json.loads(raw) + return data if isinstance(data, dict) else {} + except Exception: + return {} + + +def save_last_action_memory(memory: dict[str, Any]) -> None: + try: + if not isinstance(memory, dict): + return + with open(_cache_path(), "w", encoding="utf-8") as f: + json.dump(memory, f, ensure_ascii=False) + except Exception: + # 메모리는 부가 기능이므로 저장 실패는 전체 동작에 영향 주지 않음 + pass + diff --git a/ai_planner.py b/ai_planner.py index 116cb16..336aacf 100644 --- a/ai_planner.py +++ b/ai_planner.py @@ -28,6 +28,10 @@ SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 치트나 텔레포트 없이, 실제 게임 메커니즘만 사용합니다. 게임 상태와 이전 행동 결과를 분석해서 스스로 판단하고 계획을 세웁니다. +## 재시작/마지막 행동 메모리 +state_reader가 상태 요약에 포함하는 `마지막 행동(기억)` 섹션을 확인하세요. +직전에 실패한 action을 그대로 반복하지 말고, 실패 메시지/상태 변화에 맞춰 원인을 먼저 해결한 뒤 다음 action을 선택하세요. + ## 핵심 제약 사항 (반드시 준수!) 1. **이동은 실제 걷기** — 먼 거리는 시간이 오래 걸림. 불필요한 왕복 최소화 2. **채굴은 자원 패치 근처에서만 가능** — 반드시 자원 위치로 move한 후 mine_resource @@ -68,6 +72,7 @@ SYSTEM_PROMPT = """당신은 팩토리오 게임을 순수하게 플레이하는 - "explore" → {"direction": "east|west|north|south|...", "max_steps": 200, "wanted_ores": ["stone","coal", ...]} (선택) ★ 자원이 보이지 않을 때 반드시 explore 사용! move 대신! ★ `wanted_ores`가 있으면: 해당 자원이 발견될 때까지 계속 걷고, 다른 자원(예: iron-ore)만 계속 발견되더라도 즉시 멈추지 말 것 + ★ 상태 요약에 "기억된 광맥" 좌표가 있으면, 그 좌표로 먼저 이동(move)해 채굴(mine_resource)을 시도 ★ 방향으로 걸으면서 반경 50타일 자원 스캔, 발견 즉시 멈춤 ★ 장애물 자동 감지. 막히면 다른 방향 시도 ★ 한 방향 실패 시 다음 방향 (east→north→south→west) @@ -149,11 +154,15 @@ class AIPlanner: raw = self._call_glm(user_message, attempt=attempt) plan = self._parse_json(raw) break - except (ValueError, json.JSONDecodeError) as e: + except (ValueError, json.JSONDecodeError, TimeoutError, ConnectionError, urllib.error.URLError) as e: if attempt < 2: - print(f"[경고] JSON 파싱 실패 (시도 {attempt+1}/3), 재시도...") + print( + f"[경고] GLM 처리 실패 (시도 {attempt+1}/3): " + f"{type(e).__name__} 재시도..." + ) + time.sleep(1 + attempt * 2) continue - print(f"[오류] JSON 파싱 3회 실패. 기본 탐색 행동 사용.") + print(f"[오류] GLM 처리 3회 실패. 기본 탐색 행동 사용.") plan = { "thinking": "API 응답 파싱 실패로 기본 탐색 수행", "current_goal": "주변 탐색", diff --git a/docs/plan.md b/docs/plan.md index 821720a..482e371 100644 --- a/docs/plan.md +++ b/docs/plan.md @@ -112,3 +112,96 @@ - `README.md` - 인벤토리 캐시 동작과 파일명(`inventory_memory.json`)을 설명 +--- + +## 발견된 광맥 좌표 기억(메모리) 추가 계획 + +### 문제 관찰 +- `explore`로 광맥을 발견해도, 재실행/재계획/초기화 상황에서 동일 좌표를 다시 추천받지 못해 + 불필요한 탐색이 반복될 수 있음. + +### 변경 목표 +1. 자원 엔티티(광맥)의 좌표를 `ore_patch_memory.json`에 ore 종류별로 여러 패치 좌표로 저장 +2. 다음 상태 요약에서 “기억된 광맥” 섹션으로 AI에게 제공 +3. LLM이 가능하면 기억된 좌표로 먼저 이동해 `move -> mine_resource`를 수행하도록 유도 + +### 구현 범위 +- `action_executor.py` + - `explore` 성공 시 발견한 광맥 좌표를 파일에 갱신 +- `state_reader.py` + - `scan_resources()` 결과로도 광맥 좌표를 `ore_patch_memory.json`에 갱신 + - `summarize_for_ai()`에 “기억된 광맥” 출력 추가 +- `ai_planner.py` + - SYSTEM_PROMPT에 기억된 광맥 우선 이동 가이드 추가 + +### README 업데이트 +- `ore_patch_memory.json` 파일과 동작 방식 설명 추가 + +--- + +## 에이전트 재시작 시 마지막 행동 기억(메모리) 추가 계획 + +### 문제 관찰 +- 코드를 수정하면 보통 에이전트를 재시작해야 하고, + 이때 `ai_planner.py`의 `feedback_log`/직전 실패 정보가 초기화되어 같은 시행착오가 반복될 수 있음. + +### 변경 목표 +1. 재시작 시에도 “직전에 실행했던 action”과 결과(success/message)를 파일에 저장 +2. 다음 실행의 상태 요약(`state_reader.summarize_for_ai()`)에 “기억된 마지막 행동” 섹션을 포함 +3. LLM이 마지막 행동이 실패였다면 같은 행동을 즉시 반복하지 않도록 유도 + +### 저장 기준(확정) +- 성공/실패 상관없이 **가장 마지막으로 실행을 시도한 action 1개**만 저장한다. + +### 구현 범위 +- `main.py` + - 행동 실행 직후 action/result를 `agent_last_action_memory.json`에 저장 +- `state_reader.py` + - 재시작 시 파일을 로드해 상태 요약에 포함 +- `ai_planner.py` + - SYSTEM_PROMPT에 “마지막 행동 실패 재시도 금지/원인 해결 우선” 가이드 추가 + +### README 업데이트 +- `agent_last_action_memory.json` 존재/동작 설명 추가 + +--- + +## `place_entity` BLOCKED 완화 계획 (인접 타일 탐색) + +### 문제 관찰 +- 로그에서 `stone-furnace` 배치가 `FAIL 배치 불가`(내부적으로 `BLOCKED`)로 반복됨 +- `mine_resource`가 자원 엔티티 근처(종종 자원 패치 타일)로 이동한 뒤, 같은 좌표에 건물을 배치하려고 하면 + Factorio의 `can_place_entity` 조건에 걸려 막힐 수 있음 + +### 변경 목표 +1. `action_executor.py`의 `place_entity`에서 먼저 `surface.can_place_entity`로 배치 가능 여부를 검사 +2. 현재 좌표 `(x,y)`가 불가능하면, `(x,y)` 주변 `±1 타일` 후보(대각 포함)로 순차 시도 +3. 실제로 배치된 좌표를 결과 메시지에 포함해 이후 계획이 더 안정적으로 반복되게 함 + +### 구현 범위 +- `action_executor.py` + - `place_entity` 로직을 (x,y) 실패 시 인접 타일 후보로 확장 + +### README 업데이트 +- `place_entity`가 `BLOCKED` 시 인접 타일을 자동 탐색하도록 동작 설명 추가 + +--- + +## GLM HTTP read timeout 대응 계획 (예외 포함 재시도) + +### 문제 관찰 +- 로그에 `TimeoutError: The read operation timed out` 가 발생 +- 현재 `ai_planner.py`는 JSON 파싱 실패만 재시도하고, 네트워크 타임아웃/연결 오류는 별도 재시도 경로가 약함 + +### 변경 목표 +1. `AIPlanner.decide()`의 재시도 예외 범위를 `TimeoutError`, `ConnectionError`, `urllib.error.URLError`까지 확장 +2. 해당 오류가 발생하면 동일한 “plan 응답 받기” 재시도로 복구 시도 +3. 3회 연속 실패 시에는 기존과 동일하게 `explore` 기본 행동으로 폴백 + +### 구현 범위 +- `ai_planner.py` + - `decide()`의 `except` 절 범위 확장 및 경고 로그 보강 + +### README 업데이트 +- GLM read timeout/연결 오류 발생 시 재시도 동작을 `주의사항`에 추가 + diff --git a/main.py b/main.py index 2468c69..4256a7c 100644 --- a/main.py +++ b/main.py @@ -17,6 +17,7 @@ from state_reader import StateReader from context_compressor import ContextCompressor from ai_planner import AIPlanner from action_executor import ActionExecutor +from agent_last_action_memory import save_last_action_memory RCON_HOST = os.getenv("FACTORIO_HOST", "127.0.0.1") RCON_PORT = int(os.getenv("FACTORIO_PORT", "25575")) @@ -124,6 +125,18 @@ def run(): status = "OK" if success else "FAIL" print(f" 결과: {status} {message}") + # 재시작 시 직전 행동을 참고할 수 있도록 메모리에 저장 + try: + save_last_action_memory({ + "action": act, + "params": action.get("params", {}), + "success": success, + "message": message, + "timestamp": time.time(), + }) + except Exception: + pass + planner.record_feedback(action, success, message) if not success: diff --git a/ore_patch_memory.py b/ore_patch_memory.py new file mode 100644 index 0000000..50398a4 --- /dev/null +++ b/ore_patch_memory.py @@ -0,0 +1,141 @@ +""" +ore_patch_memory.py + +에이전트가 "발견"한 광맥(자원 엔티티) 좌표를 파일로 기억하기 위한 유틸. + +메모리 포맷: +- key: ore name (예: "iron-ore", "stone", "coal") +- value: patch 좌표들의 리스트 + - { tile_x, tile_y, last_seen, count? } +""" + +from __future__ import annotations + +import json +import os +import time +from typing import Any + + +ORE_PATCH_CACHE_FILE = "ore_patch_memory.json" +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) + + +def _cache_path(cache_dir: str | None = None) -> str: + base = cache_dir or BASE_DIR + return os.path.join(base, ORE_PATCH_CACHE_FILE) + + +def load_ore_patch_memory(cache_dir: str | None = None) -> dict[str, list[dict[str, Any]]]: + path = _cache_path(cache_dir) + try: + if not os.path.exists(path): + return {} + with open(path, "r", encoding="utf-8") as f: + raw = f.read().strip() + if not raw: + return {} + data = json.loads(raw) + if not isinstance(data, dict): + return {} + + # 과거(단일 좌표) 포맷 호환: ore -> dict + converted: dict[str, list[dict[str, Any]]] = {} + for ore, v in data.items(): + if isinstance(v, list): + converted[ore] = [x for x in v if isinstance(x, dict)] + elif isinstance(v, dict): + converted[ore] = [v] + return converted + except Exception: + return {} + + +def save_ore_patch_memory( + memory: dict[str, list[dict[str, Any]]], + cache_dir: str | None = None, +) -> None: + path = _cache_path(cache_dir) + try: + if not isinstance(memory, dict): + return + with open(path, "w", encoding="utf-8") as f: + json.dump(memory, f, ensure_ascii=False) + except Exception: + # 메모리는 부가 기능이므로 실패해도 전체 동작이 깨지면 안 됨 + pass + + +def update_ore_patch_memory( + ore_patch_memory: dict[str, list[dict[str, Any]]], + ore: str, + tile_x: int, + tile_y: int, + *, + last_seen: float | None = None, + count: int | None = None, + max_patches_per_ore: int = 8, + dedupe_if_same_tile: bool = True, +) -> dict[str, list[dict[str, Any]]]: + """ + memory[ore]의 patch 리스트에 (tile_x, tile_y)를 추가/갱신한다. + + - 같은 타일이면 last_seen 갱신(+ count가 있으면 count도 더 큰 값을 우선) + - 다른 타일이면 리스트에 추가하고 최근(last_seen) 기준으로 max_patches_per_ore까지 유지 + """ + if not isinstance(ore_patch_memory, dict): + ore_patch_memory = {} + if not ore: + return ore_patch_memory + + ore = str(ore) + tile_x = int(tile_x) + tile_y = int(tile_y) + now = time.time() if last_seen is None else float(last_seen) + + patches = ore_patch_memory.get(ore) + if not isinstance(patches, list): + patches = [] + else: + patches = [p for p in patches if isinstance(p, dict)] + + # 같은 타일이면 해당 patch를 업데이트한다. + updated = False + for p in patches: + px = p.get("tile_x") + py = p.get("tile_y") + if not dedupe_if_same_tile: + continue + if isinstance(px, int) and isinstance(py, int) and px == tile_x and py == tile_y: + p["last_seen"] = now + if count is not None: + prev_count = p.get("count") + if prev_count is None or not isinstance(prev_count, int): + p["count"] = int(count) + else: + p["count"] = int(count) if int(count) >= int(prev_count) else prev_count + updated = True + break + + if not updated: + patch: dict[str, Any] = { + "tile_x": tile_x, + "tile_y": tile_y, + "last_seen": now, + } + if count is not None: + patch["count"] = int(count) + patches.append(patch) + + # 최신 순으로 자르고 다시 저장 + patches.sort(key=lambda x: float(x.get("last_seen", 0.0)), reverse=True) + ore_patch_memory[ore] = patches[: int(max_patches_per_ore)] + + return ore_patch_memory + + +def compute_distance_sq(px: float, py: float, tile_x: int, tile_y: int) -> float: + dx = float(tile_x) - float(px) + dy = float(tile_y) - float(py) + return dx * dx + dy * dy + diff --git a/state_reader.py b/state_reader.py index 1c21d2b..94da7bf 100644 --- a/state_reader.py +++ b/state_reader.py @@ -11,6 +11,13 @@ RCON을 통해 팩토리오 게임 상태를 읽어오는 모듈 import json import os from factorio_rcon import FactorioRCON +from ore_patch_memory import ( + load_ore_patch_memory, + save_ore_patch_memory, + update_ore_patch_memory, + compute_distance_sq, +) +from agent_last_action_memory import load_last_action_memory P = 'local p = game.players[1] if not p then rcon.print("{}") return end ' @@ -144,6 +151,8 @@ class StateReader: "resources": self.scan_resources(), "buildings": self.get_buildings(), "tech": self.get_research_status(), + "ore_patch_memory": load_ore_patch_memory(), + "last_action_memory": load_last_action_memory(), } def get_player_info(self) -> dict: @@ -400,7 +409,51 @@ if not ok then rcon.print("{}") end """ try: raw = self.rcon.lua(lua) - return json.loads(raw) if raw and raw.startswith("{") else {} + res = json.loads(raw) if raw and raw.startswith("{") else {} + # scan_resources() 결과로도 광맥 좌표를 기억한다. + ore_mem = load_ore_patch_memory() + changed = False + if isinstance(res, dict) and res: + for ore, info in res.items(): + if not isinstance(info, dict): + continue + tx = info.get("anchor_tile_x") + ty = info.get("anchor_tile_y") + cnt = info.get("count") + if isinstance(tx, int) and isinstance(ty, int): + before_patches = ore_mem.get(ore) + before_set: set[tuple[int, int]] = set() + if isinstance(before_patches, list): + for pp in before_patches: + if not isinstance(pp, dict): + continue + bx = pp.get("tile_x") + by = pp.get("tile_y") + if isinstance(bx, int) and isinstance(by, int): + before_set.add((bx, by)) + + ore_mem = update_ore_patch_memory( + ore_mem, + ore, + tx, + ty, + count=cnt if isinstance(cnt, int) else None, + ) + after_patches = ore_mem.get(ore) + after_set: set[tuple[int, int]] = set() + if isinstance(after_patches, list): + for ap in after_patches: + if not isinstance(ap, dict): + continue + ax = ap.get("tile_x") + ay = ap.get("tile_y") + if isinstance(ax, int) and isinstance(ay, int): + after_set.add((ax, ay)) + if before_set != after_set: + changed = True + if changed: + save_ore_patch_memory(ore_mem) + return res except Exception: return {} @@ -452,6 +505,8 @@ if not ok then rcon.print("{}") end inv = state.get("inventory", {}) res = state.get("resources", {}) bld = state.get("buildings", {}) + ore_mem = state.get("ore_patch_memory", {}) or {} + last_action = state.get("last_action_memory", {}) or {} lines = [ "## 현재 게임 상태", @@ -496,6 +551,47 @@ if not ok then rcon.print("{}") end else: lines.append("- 반경 500타일 내 자원 없음 — 더 멀리 탐색 필요") + lines += ["", "### 기억된 광맥 (좌표)"] + if ore_mem: + px = p.get('x', 0) + py = p.get('y', 0) + known = [] + for ore, patches in ore_mem.items(): + if not isinstance(patches, list): + continue + for patch in patches: + if not isinstance(patch, dict): + continue + tx = patch.get("tile_x") + ty = patch.get("tile_y") + if isinstance(tx, int) and isinstance(ty, int): + dist = int(compute_distance_sq(px, py, tx, ty) ** 0.5) + known.append((ore, patch, dist)) + known.sort(key=lambda x: x[2]) + for ore, info, dist in known[:10]: + lines.append( + f"- {ore}: ({info.get('tile_x')},{info.get('tile_y')}) " + f"[거리: ~{dist}타일] count={info.get('count','?')}" + ) + else: + lines.append("- 없음") + + lines += ["", "### 마지막 행동(기억)"] + if last_action and isinstance(last_action, dict) and last_action.get("action"): + # state_summary 크기 방지를 위해 필드만 요약 + act = last_action.get("action", "") + params = last_action.get("params", {}) + success = last_action.get("success", None) + msg = last_action.get("message", "") + lines.append(f"- action={act} success={success} message={msg}") + if params: + try: + lines.append(f"- params={json.dumps(params, ensure_ascii=False)}") + except Exception: + pass + else: + lines.append("- 없음") + lines += ["", "### 건설된 건물"] if bld: for name, count in sorted(bld.items(), key=lambda x: -x[1])[:10]: