gihyeon
86af860267
- explore(direction, max_steps): 8방향 걷기 + 매 20틱마다 반경 50 자원 스캔 - 자원 발견 → 즉시 멈추고 위치 + 자원 종류/수량 반환 - 장애물 stuck 감지 → "다른 방향 시도" 메시지 - insert_to_entity: area 대신 position+radius (중괄호 충돌 방지)
394 lines
19 KiB
Python
394 lines
19 KiB
Python
"""
|
|
action_executor.py — 순수 AI 플레이 버전 (치트 없음)
|
|
|
|
핵심 변경:
|
|
- explore 액션 추가: 방향으로 걸으면서 자원 스캔, 발견 시 즉시 멈춤
|
|
- game.player → game.players[1] (RCON 호환)
|
|
"""
|
|
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]:
|
|
"""방향으로 걸으면서 매 20틱마다 반경 50에서 자원을 스캔.
|
|
자원 발견 시 즉시 멈추고 위치 + 자원 정보 반환."""
|
|
|
|
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}}}")
|
|
|
|
scan_interval = 20 # 매 20틱(~2초)마다 스캔
|
|
stuck_count = 0
|
|
last_pos = None
|
|
|
|
for step in range(max_steps):
|
|
time.sleep(0.1)
|
|
|
|
# 매 scan_interval 틱마다 자원 스캔
|
|
if step % scan_interval == 0:
|
|
result = self.rcon.lua(P + """
|
|
local ok, data = pcall(function()
|
|
local surface = p.surface
|
|
local pos = p.position
|
|
local resources = surface.find_entities_filtered{position = pos, radius = 50, type = "resource"}
|
|
if #resources == 0 then
|
|
return "NONE:" .. string.format("%.0f,%.0f", pos.x, pos.y)
|
|
end
|
|
-- 자원 이름별 카운트
|
|
local counts = {}
|
|
for _, e in ipairs(resources) 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}")
|
|
# 파싱: "FOUND:x,y|iron-ore=500,coal=200"
|
|
parts = result.replace("FOUND:", "").split("|")
|
|
pos_str = parts[0]
|
|
res_str = parts[1] if len(parts) > 1 else ""
|
|
return True, f"자원 발견! 위치({pos_str}), 자원: {res_str}"
|
|
|
|
if result.startswith("NONE:"):
|
|
pos_str = result.replace("NONE:", "")
|
|
# stuck 체크
|
|
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} 방향 {max_steps}틱 탐색했으나 자원 미발견 — 다른 방향 시도"
|
|
|
|
# ── 이동 (실제 걷기) ─────────────────────────────────────────────
|
|
def move(self, x: int, y: int) -> tuple[bool, str]:
|
|
"""8방향 walking_state로 실제 걷기. 도착까지 폴링."""
|
|
max_ticks = 400
|
|
stuck_count = 0
|
|
last_dist = 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 (ValueError, IndexError):
|
|
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 mine_resource(self, ore: str = "iron-ore", count: int = 10) -> tuple[bool, str]:
|
|
find_result = self.rcon.lua(P + f"""
|
|
local surface = p.surface
|
|
local resources = surface.find_entities_filtered{{
|
|
position = p.position, radius = 5, name = "{ore}"
|
|
}}
|
|
if #resources == 0 then
|
|
resources = surface.find_entities_filtered{{
|
|
position = p.position, radius = 15, name = "{ore}"
|
|
}}
|
|
end
|
|
if #resources > 0 then
|
|
local r = resources[1]
|
|
rcon.print("FOUND:" .. string.format("%.1f,%.1f", r.position.x, r.position.y))
|
|
else
|
|
rcon.print("NOT_FOUND")
|
|
end
|
|
""")
|
|
if not find_result or find_result in ("NO_PLAYER", "NO_CHARACTER"):
|
|
return False, find_result or "플레이어 없음"
|
|
if find_result == "NOT_FOUND":
|
|
return False, f"근처에 {ore} 없음 — 자원 패치로 move 먼저"
|
|
|
|
parts = find_result.replace("FOUND:", "").split(",")
|
|
rx, ry = float(parts[0]), float(parts[1])
|
|
|
|
before_count = self._get_item_count(ore)
|
|
target_count = before_count + count
|
|
max_ticks = count * 40
|
|
|
|
for _ in range(max_ticks):
|
|
self.rcon.lua(P + f"p.mining_state = {{mining = true, position = {{{rx}, {ry}}}}}")
|
|
time.sleep(0.1)
|
|
current = self._get_item_count(ore)
|
|
if current >= target_count:
|
|
self.rcon.lua(P + "p.mining_state = {mining = false}")
|
|
return True, f"{ore} {current - before_count}개 채굴 완료"
|
|
|
|
self.rcon.lua(P + "p.mining_state = {mining = false}")
|
|
mined = self._get_item_count(ore) - before_count
|
|
if mined > 0:
|
|
return True, f"{ore} {mined}개 채굴 (목표 {count}개 중 일부)"
|
|
return False, f"{ore} 채굴 실패"
|
|
|
|
# ── 제작 ─────────────────────────────────────────────────────────
|
|
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"벨트 설치 실패 ({failed}개 실패)"
|
|
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"({x},{y})가 너무 멀음"
|
|
return False, f"({x},{y})에 조립기 없음"
|
|
|
|
# ── 연구 ─────────────────────────────────────────────────────────
|
|
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
|