From 8858d52b1cd6ccebe7e1a41052b1c8cb928a74bd Mon Sep 17 00:00:00 2001 From: gihyeon Date: Wed, 25 Mar 2026 10:23:15 +0900 Subject: [PATCH] =?UTF-8?q?Add=20action=5Fexecutor.py=20-=20=EC=99=84?= =?UTF-8?q?=EC=A0=84=20=EC=9E=90=EC=9C=A8=20=EC=97=90=EC=9D=B4=EC=A0=84?= =?UTF-8?q?=ED=8A=B8=EC=9A=A9=20=EC=8B=A4=ED=96=89=EA=B8=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- action_executor.py | 282 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 282 insertions(+) create mode 100644 action_executor.py diff --git a/action_executor.py b/action_executor.py new file mode 100644 index 0000000..d02601a --- /dev/null +++ b/action_executor.py @@ -0,0 +1,282 @@ +""" +action_executor.py — 완전 자율 에이전트용 실행기 + +AI가 계획한 모든 행동을 실제 팩토리오에서 실행. +성공/실패 여부와 이유를 반환해서 AI 피드백 루프에 활용. +""" +import time +import json +from factorio_rcon import FactorioRCON + + +class ActionExecutor: + def __init__(self, rcon: FactorioRCON): + self.rcon = rcon + + def execute(self, action: dict) -> tuple[bool, str]: + """ + 행동 실행 후 (성공여부, 메시지) 반환 + AI 피드백 루프에서 사용 + """ + act = action.get("action", "") + params = action.get("params", {}) + + handlers = { + "move": self.move, + "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: + return handler(**params) + except TypeError as e: + return False, f"파라미터 오류: {e}" + except Exception as e: + return False, f"실행 오류: {e}" + + # ── 이동 ──────────────────────────────────────────────────── + def move(self, x: int, y: int) -> tuple[bool, str]: + self.rcon.lua(f"game.player.teleport({{x={x}, y={y}}})") + time.sleep(0.2) + return True, f"({x}, {y})로 이동" + + # ── 자원 채굴 ─────────────────────────────────────────────── + def mine_resource(self, ore: str = "iron-ore", count: int = 50) -> tuple[bool, str]: + self.rcon.lua(f'game.player.insert({{name="{ore}", count={count}}})') + return True, f"{ore} {count}개 획득" + + # ── 아이템 제작 ───────────────────────────────────────────── + def craft_item(self, item: str, count: int = 1) -> tuple[bool, str]: + # 제작 가능 여부 확인 후 인벤토리에 지급 + lua = f""" +local ok = pcall(function() + game.player.insert({{name="{item}", count={count}}}) +end) +rcon.print(ok and "ok" or "fail") +""" + result = self.rcon.lua(lua) + if "ok" in result: + return True, f"{item} {count}개 제작" + return False, f"{item} 제작 실패 (아이템 이름 확인 필요)" + + # ── 단일 건물 배치 ────────────────────────────────────────── + 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") + + lua = f""" +local surface = game.player.surface +local force = game.player.force +local inv = game.player.get_main_inventory() + +-- 인벤토리 확인 +local have = inv.get_item_count("{name}") +if have < 1 then + rcon.print("NO_ITEM:" .. have) + return +end + +-- 이미 해당 위치에 건물이 있는지 확인 +local existing = surface.find_entity("{name}", {{{x}, {y}}}) +if existing then + rcon.print("ALREADY_EXISTS") + return +end + +-- 배치 +inv.remove({{name="{name}", count=1}}) +local e = surface.create_entity{{ + name = "{name}", + position = {{{x}, {y}}}, + direction = {lua_dir}, + force = force, + raise_built = true +}} + +if e then + rcon.print("OK:" .. e.name) +else + rcon.print("BLOCKED") +end +""" + result = self.rcon.lua(lua) + + if result.startswith("OK:"): + return True, f"{name} 배치 완료 ({x},{y})" + elif result.startswith("NO_ITEM"): + count = result.split(":")[1] if ":" in result else "0" + return False, f"인벤토리에 {name} 없음 (보유: {count}개) — craft_item 먼저" + elif result == "ALREADY_EXISTS": + return True, f"{name}이 이미 ({x},{y})에 있음 — 건너뜀" + elif result == "BLOCKED": + return False, f"({x},{y}) 위치가 막혀있음 — 다른 좌표 시도" + else: + return False, f"알 수 없는 결과: {result}" + + # ── 벨트 라인 ─────────────────────────────────────────────── + def place_belt_line( + self, + from_x: int, from_y: int, + to_x: int, to_y: int + ) -> tuple[bool, str]: + lua = f""" +local surface = game.player.surface +local force = game.player.force +local inv = game.player.get_main_inventory() +local placed = 0 +local failed = 0 + +local x1, y1 = {from_x}, {from_y} +local x2, y2 = {to_x}, {to_y} + +-- 수평 이동 +if x1 ~= x2 then + local step = x1 < x2 and 1 or -1 + local dir = x1 < x2 and defines.direction.east or defines.direction.west + for bx = x1, x2, step do + if inv.get_item_count("transport-belt") > 0 then + inv.remove({{name="transport-belt", count=1}}) + local e = surface.create_entity{{ + name="transport-belt", position={{bx, y1}}, + direction=dir, force=force, raise_built=true + }} + if e then placed = placed + 1 else failed = failed + 1 end + else + failed = failed + 1 + end + end +end + +-- 수직 이동 +if y1 ~= y2 then + local step = y1 < y2 and 1 or -1 + local dir = y1 < y2 and defines.direction.south or defines.direction.north + for by = y1 + (x1 ~= x2 and 1 or 0), y2, step do + if inv.get_item_count("transport-belt") > 0 then + inv.remove({{name="transport-belt", count=1}}) + local e = surface.create_entity{{ + name="transport-belt", position={{x2, by}}, + direction=dir, force=force, raise_built=true + }} + if e then placed = placed + 1 else failed = failed + 1 end + else + failed = failed + 1 + end + end +end + +rcon.print("placed=" .. placed .. " failed=" .. failed) +""" + result = self.rcon.lua(lua) + if "placed=" in result: + p = result.split("placed=")[1].split(" ")[0] + f = result.split("failed=")[1] if "failed=" in result else "0" + if int(f) > 0: + return False, f"벨트 {p}개 설치, {f}개 실패 (벨트 부족 또는 지형 막힘)" + return True, f"벨트 {p}개 설치 완료" + return False, f"벨트 설치 실패: {result}" + + # ── 건물에 아이템 삽입 ────────────────────────────────────── + def insert_to_entity( + self, + x: int, y: int, + item: str = "coal", + count: int = 50 + ) -> tuple[bool, str]: + lua = f""" +local surface = game.player.surface +-- 해당 위치 근처 모든 엔티티 검색 +local entities = surface.find_entities_filtered{{ + area = {{{{{x}-1, {y}-1}}, {{{x}+1, {y}+1}}}} +}} + +local inserted = false +for _, e in ipairs(entities) do + if e.valid and e.get_fuel_inventory and e.get_fuel_inventory() then + local fi = e.get_fuel_inventory() + fi.insert({{name="{item}", count={count}}}) + inserted = true + elseif e.valid and e.get_inventory then + local inv = e.get_inventory(defines.inventory.chest) + if inv then + inv.insert({{name="{item}", count={count}}}) + inserted = true + end + end +end + +rcon.print(inserted and "OK" or "NOT_FOUND") +""" + result = self.rcon.lua(lua) + if result == "OK": + return True, f"({x},{y}) 건물에 {item} {count}개 삽입" + return False, f"({x},{y})에서 건물을 찾을 수 없음" + + # ── 조립기 레시피 설정 ────────────────────────────────────── + def set_recipe(self, x: int, y: int, recipe: str) -> tuple[bool, str]: + lua = f""" +local e = game.player.surface.find_entities_filtered{{ + area = {{{{{x}-1, {y}-1}}, {{{x}+1, {y}+1}}}}, + type = "assembling-machine" +}}[1] + +if e then + e.set_recipe("{recipe}") + rcon.print("OK") +else + rcon.print("NOT_FOUND") +end +""" + result = self.rcon.lua(lua) + if result == "OK": + return True, f"({x},{y}) 조립기 레시피: {recipe}" + return False, f"({x},{y})에 조립기 없음" + + # ── 연구 시작 ─────────────────────────────────────────────── + def start_research(self, tech: str = "automation") -> tuple[bool, str]: + lua = f""" +local force = game.player.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 +""" + result = self.rcon.lua(lua) + 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)) # 최대 30초 + return True, f"{seconds}초 대기 완료"