""" action_executor.py — 순수 AI 플레이 버전 핵심 변경: - mine_resource: 접근 불가 시 광맥 주변 4방향으로 이동 후 재시도 - 검색 반경 20→50, 후보 5→10개 - 한 위치에서 실패 → 10칸씩 이동하며 접근 가능 지점 탐색 """ import time import json from factorio_rcon import FactorioRCON P = """local p = game.players[1] if not p then rcon.print("NO_PLAYER") return end if not p.character then rcon.print("NO_CHARACTER") return end """ class ActionExecutor: def __init__(self, rcon: FactorioRCON): self.rcon = rcon def execute(self, action: dict) -> tuple[bool, str]: act = action.get("action", "") params = action.get("params", {}) handlers = { "move": self.move, "explore": self.explore, "mine_resource": self.mine_resource, "craft_item": self.craft_item, "place_entity": self.place_entity, "place_belt_line": self.place_belt_line, "insert_to_entity": self.insert_to_entity, "set_recipe": self.set_recipe, "start_research": self.start_research, "wait": self.wait, } handler = handlers.get(act) if not handler: return False, f"알 수 없는 행동: {act}" try: result = handler(**params) if isinstance(result, tuple) and isinstance(result[1], str): if "NO_PLAYER" in result[1]: return False, "서버에 접속한 플레이어가 없습니다." if "NO_CHARACTER" in result[1]: return False, "플레이어 캐릭터가 없습니다." return result except TypeError as e: return False, f"파라미터 오류: {e}" except Exception as e: return False, f"실행 오류: {e}" # ── 탐색 ───────────────────────────────────────────────────────── def explore(self, direction: str = "east", max_steps: int = 200) -> tuple[bool, str]: dir_map = { "north": "defines.direction.north", "south": "defines.direction.south", "east": "defines.direction.east", "west": "defines.direction.west", "northeast": "defines.direction.northeast", "northwest": "defines.direction.northwest", "southeast": "defines.direction.southeast", "southwest": "defines.direction.southwest", } lua_dir = dir_map.get(direction, "defines.direction.east") self.rcon.lua(P + f"p.walking_state = {{walking = true, direction = {lua_dir}}}") stuck_count, last_pos = 0, None for step in range(max_steps): time.sleep(0.1) if step % 20 == 0: result = self.rcon.lua(P + """ local ok, data = pcall(function() local pos = p.position 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 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, ",") end) if ok then rcon.print(data) else rcon.print("ERROR") end """) if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, result or "플레이어 없음" if result.startswith("FOUND:"): self.rcon.lua(P + "p.walking_state = {walking = false, direction = defines.direction.north}") parts = result.replace("FOUND:", "").split("|") return True, f"자원 발견! 위치({parts[0]}), 자원: {parts[1] if len(parts)>1 else ''}" if result.startswith("NONE:"): pos_str = result.replace("NONE:", "") if last_pos == pos_str: stuck_count += 1 else: stuck_count = 0 last_pos = pos_str if stuck_count >= 3: self.rcon.lua(P + "p.walking_state = {walking = false, direction = defines.direction.north}") return False, f"탐색 중 장애물에 막힘 위치({pos_str}) — 다른 방향 시도" self.rcon.lua(P + "p.walking_state = {walking = false, direction = defines.direction.north}") return False, f"{direction} 방향 탐색 자원 미발견 — 다른 방향 시도" # ── 이동 ───────────────────────────────────────────────────────── def move(self, x: int, y: int) -> tuple[bool, str]: max_ticks, stuck_count, last_dist = 400, 0, 99999 for _ in range(max_ticks): result = self.rcon.lua(P + f""" local tx, ty = {x}, {y} local dx = tx - p.position.x local dy = ty - p.position.y local dist = math.sqrt(dx*dx + dy*dy) if dist < 1.5 then p.walking_state = {{walking = false, direction = defines.direction.north}} rcon.print("ARRIVED:" .. math.floor(p.position.x) .. "," .. math.floor(p.position.y)) return end local angle = math.atan2(dy, dx) local dir if angle >= -0.393 and angle < 0.393 then dir = defines.direction.east elseif angle >= 0.393 and angle < 1.178 then dir = defines.direction.southeast elseif angle >= 1.178 and angle < 1.963 then dir = defines.direction.south elseif angle >= 1.963 and angle < 2.749 then dir = defines.direction.southwest elseif angle >= -1.178 and angle < -0.393 then dir = defines.direction.northeast elseif angle >= -1.963 and angle < -1.178 then dir = defines.direction.north elseif angle >= -2.749 and angle < -1.963 then dir = defines.direction.northwest else dir = defines.direction.west end p.walking_state = {{walking = true, direction = dir}} rcon.print("WALK:" .. string.format("%.1f", dist)) """) if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, result or "플레이어 없음" if result.startswith("ARRIVED"): return True, f"({x}, {y})로 도착" try: dist = float(result.split(":")[1]) if abs(dist - last_dist) < 0.3: stuck_count += 1 else: stuck_count = 0 last_dist = dist except: pass if stuck_count > 30: self.rcon.lua(P + "p.walking_state = {walking = false, direction = defines.direction.north}") return False, f"({x},{y}) 장애물에 막힘 (남은 거리: {last_dist:.0f})" time.sleep(0.1) self.rcon.lua(P + "p.walking_state = {walking = false, direction = defines.direction.north}") return False, f"({x},{y}) 시간 초과 (남은 거리: {last_dist:.0f})" # ── 한 지점에서 채굴 시도 (내부 헬퍼) ──────────────────────────── def _try_mine_at_position(self, ore: str, target_count: int, before_count: int) -> int: """현재 위치에서 반경 50 내 광석 10개를 거리순으로 시도. 채굴 성공한 개수 반환. 0이면 접근 불가.""" find_result = self.rcon.lua(P + f""" local surface = p.surface local pos = p.position local resources = surface.find_entities_filtered{{position = pos, radius = 50, name = "{ore}"}} if #resources == 0 then rcon.print("NOT_FOUND") return end table.sort(resources, function(a, b) local da = (a.position.x - pos.x)^2 + (a.position.y - pos.y)^2 local db = (b.position.x - pos.x)^2 + (b.position.y - pos.y)^2 return da < db end) local parts = {{}} for i = 1, math.min(10, #resources) do local r = resources[i] parts[#parts+1] = string.format("%.1f,%.1f", r.position.x, r.position.y) end rcon.print("FOUND:" .. table.concat(parts, ";")) """) if not find_result or find_result == "NOT_FOUND": return 0 coords_str = find_result.replace("FOUND:", "") coord_list = [] for c in coords_str.split(";"): parts = c.split(",") if len(parts) == 2: coord_list.append((float(parts[0]), float(parts[1]))) for attempt, (rx, ry) in enumerate(coord_list): stall_count = 0 last_item_count = self._get_item_count(ore) for tick in range(150): # 타일당 최대 15초 self.rcon.lua(P + f"p.mining_state = {{mining = true, position = {{{rx}, {ry}}}}}") time.sleep(0.1) if tick % 10 == 9: current = self._get_item_count(ore) if current >= target_count: self.rcon.lua(P + "p.mining_state = {mining = false}") return current - before_count if current > last_item_count: # 채굴 진행 중! 계속 stall_count = 0 last_item_count = current else: stall_count += 1 if stall_count >= 2: # 이 타일 접근 불가 → 다음 타일 break self.rcon.lua(P + "p.mining_state = {mining = false}") return self._get_item_count(ore) - before_count # ── 채굴 (접근 불가 시 광맥 주변 이동 후 재시도) ────────────────── def mine_resource(self, ore: str = "iron-ore", count: int = 10) -> tuple[bool, str]: """현재 위치에서 시도 → 실패 시 광맥 주변 4방향으로 이동하며 접근점 탐색""" before_count = self._get_item_count(ore) target_count = before_count + count # 1차: 현재 위치에서 시도 print(f" [채굴] 현재 위치에서 {ore} 채굴 시도...") mined = self._try_mine_at_position(ore, target_count, before_count) if mined >= count: return True, f"{ore} {mined}개 채굴 완료" if mined > 0: return True, f"{ore} {mined}개 채굴 (일부 — 접근 가능 타일에서 최대한)" # 2차: 실패 시 광맥 주변 4방향으로 이동하며 재시도 # 현재 위치 기준 동서남북 10칸씩 이동 offsets = [(10, 0), (-10, 0), (0, 10), (0, -10), (15, 10), (-15, 10), (10, -15), (-10, -15)] cur_pos = self.rcon.lua(P + 'rcon.print(string.format("%.0f,%.0f", p.position.x, p.position.y))') try: cx, cy = [float(v) for v in cur_pos.split(",")] except: cx, cy = 0, 0 for i, (dx, dy) in enumerate(offsets): tx, ty = int(cx + dx), int(cy + dy) print(f" [채굴] 위치({tx},{ty})로 이동하여 재시도 ({i+1}/{len(offsets)})...") ok, msg = self.move(tx, ty) if not ok: print(f" [채굴] 이동 실패: {msg}") continue before_now = self._get_item_count(ore) target_now = before_count + count # 원래 목표 기준 mined = self._try_mine_at_position(ore, target_now, before_count) total = self._get_item_count(ore) - before_count if total >= count: return True, f"{ore} {total}개 채굴 완료 (위치 이동 {i+1}회)" if total > 0: print(f" [채굴] {total}개 채굴 중... 계속 시도") # 최종 결과 total = self._get_item_count(ore) - before_count if total > 0: return True, f"{ore} {total}개 채굴 (목표 {count}개 중 일부 — 광맥 주변 탐색 완료)" return False, f"{ore} 채굴 실패 — 광맥 주변 8방향 이동 후에도 접근 불가. explore로 다른 광맥 찾기" # ── 제작 ───────────────────────────────────────────────────────── def craft_item(self, item: str, count: int = 1) -> tuple[bool, str]: result = self.rcon.lua(P + f""" local recipe = p.force.recipes["{item}"] if not recipe then rcon.print("NO_RECIPE") return end if not recipe.enabled then rcon.print("LOCKED") return end local crafted = p.begin_crafting{{count={count}, recipe="{item}"}} if crafted > 0 then rcon.print("CRAFTING:" .. crafted) else rcon.print("NO_INGREDIENTS") end """) if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, result or "플레이어 없음" if result.startswith("CRAFTING:"): crafted = int(result.split(":")[1]) time.sleep(min(crafted * 2, 30)) return True, f"{item} {crafted}개 제작 완료" elif result == "NO_RECIPE": return False, f"{item} 레시피 없음" elif result == "LOCKED": return False, f"{item} 레시피 잠김" elif result == "NO_INGREDIENTS": return False, f"{item} 재료 부족" return False, f"제작 실패: {result}" # ── 건물 배치 ──────────────────────────────────────────────────── def place_entity(self, name: str, x: int, y: int, direction: str = "north") -> tuple[bool, str]: dir_map = {"north": "defines.direction.north", "south": "defines.direction.south", "east": "defines.direction.east", "west": "defines.direction.west"} lua_dir = dir_map.get(direction, "defines.direction.north") result = self.rcon.lua(P + f""" local inv = p.get_main_inventory() local have = inv.get_item_count("{name}") if have < 1 then rcon.print("NO_ITEM:" .. have) 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 local existing = p.surface.find_entity("{name}", {{{x}, {y}}}) if existing then rcon.print("ALREADY_EXISTS") return end p.cursor_stack.set_stack({{name="{name}", count=1}}) local built = p.build_from_cursor{{position = {{{x}, {y}}}, direction = {lua_dir}}} if built then p.cursor_stack.clear() rcon.print("OK") else p.cursor_stack.clear() rcon.print("BLOCKED") end """) 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("NO_ITEM"): return False, f"인벤토리에 {name} 없음" elif result.startswith("TOO_FAR"): return False, f"({x},{y})가 너무 멀음 — move 먼저" elif result == "ALREADY_EXISTS": return True, f"{name}이 이미 ({x},{y})에 있음" elif result == "BLOCKED": return False, f"({x},{y}) 배치 불가" return False, f"배치 실패: {result}" # ── 벨트/삽입/레시피/연구/대기 (변경 없음) ─────────────────────── def place_belt_line(self, from_x: int, from_y: int, to_x: int, to_y: int) -> tuple[bool, str]: placed, failed = 0, 0 positions = [] if from_x != to_x: step = 1 if from_x < to_x else -1 d = "east" if step == 1 else "west" for bx in range(from_x, to_x + step, step): positions.append((bx, from_y, d)) if from_y != to_y: step = 1 if from_y < to_y else -1 d = "south" if step == 1 else "north" sy = from_y + (step if from_x != to_x else 0) for by in range(sy, to_y + step, step): positions.append((to_x, by, d)) for bx, by, d in positions: ok, _ = self.move(bx, by) if not ok: failed += 1; continue ok, _ = self.place_entity("transport-belt", bx, by, d) if ok: placed += 1 else: failed += 1 if placed == 0: return False, f"벨트 설치 실패" return True, f"벨트 {placed}개 설치" + (f", {failed}개 실패" if failed else "") def insert_to_entity(self, x: int, y: int, item: str = "coal", count: int = 50) -> tuple[bool, str]: result = self.rcon.lua(P + f""" 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") return end local inv = p.get_main_inventory() local have = inv.get_item_count("{item}") if have < 1 then rcon.print("NO_ITEM") return end local actual = math.min(have, {count}) local entities = p.surface.find_entities_filtered{{position = {{{x},{y}}}, radius = 2}} local inserted = false for _, e in ipairs(entities) do if e.valid then if e.burner then local fi = e.burner.inventory if fi then local removed = inv.remove({{name="{item}", count=actual}}) if removed > 0 then fi.insert({{name="{item}", count=removed}}) inserted = true break end end end if not inserted then for i = 1, 10 do local ti = e.get_inventory(i) if ti and ti.can_insert({{name="{item}", count=1}}) then local removed = inv.remove({{name="{item}", count=actual}}) if removed > 0 then ti.insert({{name="{item}", count=removed}}) inserted = true end break end end end if inserted then break end end end rcon.print(inserted and "OK" or "NOT_FOUND") """) if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, result or "플레이어 없음" if result == "OK": return True, f"({x},{y})에 {item} 삽입 완료" elif result == "TOO_FAR": return False, f"({x},{y})가 너무 멀음" elif result == "NO_ITEM": return False, f"인벤토리에 {item} 없음" return False, f"({x},{y})에 건물 없음" def set_recipe(self, x: int, y: int, recipe: str) -> tuple[bool, str]: result = self.rcon.lua(P + f""" 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") return end local e = p.surface.find_entities_filtered{{position = {{{x},{y}}}, radius = 2, type = "assembling-machine"}}[1] if e then e.set_recipe("{recipe}") rcon.print("OK") else rcon.print("NOT_FOUND") end """) if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, result or "플레이어 없음" if result == "OK": return True, f"({x},{y}) 레시피: {recipe}" elif result == "TOO_FAR": return False, f"너무 멀음" return False, f"조립기 없음" def start_research(self, tech: str = "automation") -> tuple[bool, str]: result = self.rcon.lua(P + f""" local force = p.force local t = force.technologies["{tech}"] if not t then rcon.print("NO_TECH") elseif t.researched then rcon.print("ALREADY_DONE") else force.research_queue_enabled = true force.add_research("{tech}") rcon.print("OK") end """) if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, result or "플레이어 없음" if result == "OK": return True, f"{tech} 연구 시작" elif result == "ALREADY_DONE": return True, f"{tech} 이미 완료" elif result == "NO_TECH": return False, f"{tech} 기술 없음" return False, result def wait(self, seconds: int = 3) -> tuple[bool, str]: time.sleep(min(seconds, 30)) return True, f"{seconds}초 대기 완료" def _get_item_count(self, item: str) -> int: result = self.rcon.lua(P + f'rcon.print(tostring(p.get_main_inventory().get_item_count("{item}")))') try: return int(result) except: return 0