""" main.py — 순수 AI 플레이 버전 치트 없이 실제 게임 메커니즘으로 플레이. 걷기/채굴/제작에 실제 시간이 걸리므로 ACTION_DELAY를 설정. 핵심 변경: - 시작 시 플레이어 접속 여부 확인 - game.player → game.players[1] (RCON 호환) - JSON 파싱 실패 시 재시도 (ai_planner 내부) """ import time import os import json from factorio_rcon import FactorioRCON from state_reader import StateReader from context_compressor import ContextCompressor from ai_planner import AIPlanner, OLLAMA_MODEL, OLLAMA_HOST from action_executor import ActionExecutor from agent_last_action_memory import save_last_action_memory, load_last_action_memory from ore_patch_memory import load_ore_patch_memory, compute_distance_sq RCON_HOST = os.getenv("FACTORIO_HOST", "127.0.0.1") RCON_PORT = int(os.getenv("FACTORIO_PORT", "25575")) RCON_PASSWORD = os.getenv("FACTORIO_PASSWORD", "factorio_ai") ACTION_DELAY = 0.2 LOG_FILE = "agent_log.jsonl" # 건물 N개 이상이면 압축 모드로 전환 COMPRESS_THRESHOLD = 50 def check_player(rcon: FactorioRCON) -> bool: """서버에 접속한 플레이어가 있는지 확인하고 안내""" print("\n[초기화] 플레이어 접속 확인 중...") for attempt in range(30): # 최대 30초 대기 result = rcon.check_player() if result: info = rcon.lua(""" local p = game.players[1] if p then rcon.print(p.name .. " @ " .. string.format("%.0f, %.0f", p.position.x, p.position.y)) end """) print(f"[초기화] OK 플레이어 발견: {info}") return True if attempt == 0: print("[초기화] WARN 접속한 플레이어가 없습니다!") print("[초기화] 팩토리오 클라이언트로 이 서버에 접속하세요.") print(f"[초기화] 서버 주소: {RCON_HOST}") print(f"[초기화] 대기 중... (최대 30초)") time.sleep(1) print("[오류] 30초 내에 플레이어가 접속하지 않았습니다.") print("[오류] 팩토리오 클라이언트로 서버에 접속한 후 다시 실행하세요.") return False def _append_memory_sections(summary: str) -> str: """ 압축 모드(중반/후반)에서는 context_compressor 출력만 들어가서 state_reader의 "기억된 광맥/마지막 행동" 섹션이 사라질 수 있다. 이 정보를 프롬프트에 다시 덧붙여 폴백/복구를 가능하게 한다. """ ore_mem = load_ore_patch_memory() last_action = load_last_action_memory() # compressed summary에도 "위치: (x, y)"가 포함되므로, 여기서 플레이어 좌표만 뽑는다. m = None try: import re m = re.search(r"위치:\s*\(\s*(-?\d+)\s*,\s*(-?\d+)\s*\)", summary) except Exception: m = None px = py = None if m: try: px = int(m.group(1)) py = int(m.group(2)) except Exception: px = py = None mem_lines: list[str] = [] mem_lines.append("### 기억된 광맥 (좌표)") if ore_mem: known: list[tuple[int, str, int, int, object]] = [] 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): if px is not None and py is not None: dist = int(compute_distance_sq(px, py, tx, ty) ** 0.5) else: dist = 10**18 known.append((dist, ore, tx, ty, patch.get("count", "?"))) known.sort(key=lambda x: x[0]) for dist, ore, tx, ty, cnt in known[:10]: if dist >= 10**18: mem_lines.append(f"- {ore}: ({tx},{ty}) [거리: ~?타일] count={cnt}") else: mem_lines.append(f"- {ore}: ({tx},{ty}) [거리: ~{dist}타일] count={cnt}") else: mem_lines.append("- 없음") mem_lines.append("") mem_lines.append("### 마지막 행동(기억)") if last_action and isinstance(last_action, dict) and last_action.get("action"): act = last_action.get("action", "") success = last_action.get("success", None) msg = last_action.get("message", "") mem_lines.append(f"- action={act} success={success} message={msg}") params = last_action.get("params", {}) if params: try: mem_lines.append(f"- params={json.dumps(params, ensure_ascii=False)}") except Exception: pass else: mem_lines.append("- 없음") return summary + "\n\n" + "\n".join(mem_lines) def run(): print("=" * 60) print(" 팩토리오 순수 AI 에이전트 (치트 없음)") print(" - 실제 걷기 / 실제 채굴 / 실제 제작 / 건설 거리 제한") print(f" - LLM: Ollama {OLLAMA_MODEL}") print(f" - Ollama host: {OLLAMA_HOST}") print("=" * 60) with FactorioRCON(RCON_HOST, RCON_PORT, RCON_PASSWORD) as rcon: # ── 플레이어 접속 확인 ── if not check_player(rcon): return reader = StateReader(rcon) compressor = ContextCompressor(rcon) planner = AIPlanner() executor = ActionExecutor(rcon) total_actions = 0 queue: list[dict] = [] entity_count = 0 while True: try: if not queue: print(f"\n{'='*50}") print(f"[재계획] 총 {total_actions}개 실행 완료") # ── 압축 레벨 자동 선택 ── if entity_count < COMPRESS_THRESHOLD: print("[상태] 초반 모드 - 상세 상태 수집") state = reader.get_full_state() summary = reader.summarize_for_ai(state) entity_count = sum( v for v in state.get("buildings", {}).values() if isinstance(v, int) ) elif entity_count < 200: print(f"[상태] 중반 모드 - 구역 압축 (건물 {entity_count}개)") summary = compressor.get_compressed_state(detail_level=1) summary = _append_memory_sections(summary) else: print(f"[상태] 후반 모드 - 글로벌 압축 (건물 {entity_count}개)") summary = compressor.get_compressed_state(detail_level=0) summary = _append_memory_sections(summary) global_info = compressor._get_global_summary() entity_count = global_info.get("total_entities", entity_count) # ── AI 계획 ── queue = planner.decide(summary) if not queue: print("[경고] AI가 행동 반환 안 함. 10초 후 재시도") time.sleep(10) continue # ── 행동 실행 ── action = queue.pop(0) total_actions += 1 act = action.get("action", "") reason = action.get("reason", "") print(f"\n[{total_actions}] {act}") print(f" 이유: {reason}") print(f" params: {json.dumps(action.get('params',{}), ensure_ascii=False)}") success, message = executor.execute(action) 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: print(" -> 실패. 재계획 요청") queue.clear() _log(total_actions, action, success, message) time.sleep(ACTION_DELAY) except KeyboardInterrupt: print(f"\n종료 (총 {total_actions}개 실행)") break except Exception as e: print(f"[오류] {e}") import traceback traceback.print_exc() time.sleep(5) def _log(step, action, success, message): with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(json.dumps({ "step": step, "action": action, "success": success, "message": message, "timestamp": time.time() }, ensure_ascii=False) + "\n") if __name__ == "__main__": run()