283 lines
9.8 KiB
Python
283 lines
9.8 KiB
Python
"""
|
|
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}초 대기 완료"
|