3d118fe649
- `mine_resource` 함수에서 채굴 상태를 매 루프마다 반복 설정하던 로직을 제거하여, 우클릭 유지와 유사한 동작으로 최적화 - GLM API 호출 실패 시 예외 처리 로직을 개선하여, 다양한 오류 원인을 로그로 출력하고 상태 요약 기반의 폴백 로직 추가 - 새로운 연결 검사 스크립트 추가 및 README.md에 GLM API 연결 문제 디버깅 섹션 추가 - 관련 문서 및 주의사항 업데이트
662 lines
30 KiB
Python
662 lines
30 KiB
Python
"""
|
|
action_executor.py — 순수 AI 플레이 버전
|
|
|
|
핵심 수정:
|
|
- mine_resource: 실패한 타일 좌표를 기억하고 Lua에 전달해서 제외
|
|
- 매 라운드마다 다른 타일을 찾도록 보장
|
|
"""
|
|
import time
|
|
from factorio_rcon import FactorioRCON
|
|
from ore_patch_memory import load_ore_patch_memory, save_ore_patch_memory, update_ore_patch_memory
|
|
|
|
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, "캐릭터 없음"
|
|
success, message = result
|
|
|
|
# stone-furnace 배치 직후 자동 부트스트랩(연료+광석 투입)
|
|
# - AI가 insert_to_entity를 누락해도 제련이 시작되게 보정
|
|
if act == "place_entity" and success:
|
|
furnace_name = params.get("name")
|
|
if furnace_name == "stone-furnace":
|
|
rx, ry = params.get("x"), params.get("y")
|
|
parsed = self._extract_coords_from_message(message)
|
|
if parsed is not None:
|
|
rx, ry = parsed
|
|
boot_ok, boot_msg = self._auto_bootstrap_furnace(
|
|
furnace_name=furnace_name,
|
|
x=int(rx),
|
|
y=int(ry),
|
|
reason=action.get("reason", "") or "",
|
|
)
|
|
message = f"{message} / bootstrap: {'OK' if boot_ok else 'FAIL'} {boot_msg}"
|
|
|
|
return success, message
|
|
except TypeError as e: return False, f"파라미터 오류: {e}"
|
|
except Exception as e: return False, f"실행 오류: {e}"
|
|
|
|
def _extract_coords_from_message(self, message: str) -> tuple[int, int] | None:
|
|
"""
|
|
place_entity 성공 메시지에서 실제로 지어진 좌표를 파싱.
|
|
예) "stone-furnace 배치 (-91, -68)" / "stone-furnace 배치 (-91,-68)"
|
|
"""
|
|
import re
|
|
|
|
if not isinstance(message, str):
|
|
return None
|
|
m = re.search(r"\(\s*(-?\d+(?:\.\d+)?)\s*,\s*(-?\d+(?:\.\d+)?)\s*\)", message)
|
|
if not m:
|
|
return None
|
|
try:
|
|
x = int(round(float(m.group(1))))
|
|
y = int(round(float(m.group(2))))
|
|
return x, y
|
|
except Exception:
|
|
return None
|
|
|
|
def _auto_bootstrap_furnace(
|
|
self,
|
|
*,
|
|
furnace_name: str,
|
|
x: int,
|
|
y: int,
|
|
reason: str = "",
|
|
) -> tuple[bool, str]:
|
|
"""
|
|
stone-furnace 배치 직후:
|
|
- coal(연료) 투입
|
|
- iron-ore 또는 copper-ore 투입(가능한 것 우선)
|
|
|
|
반환값은 현재 전략에서 성공/실패를 상위에 강제하지 않지만,
|
|
디버깅을 위해 메시지를 남긴다.
|
|
"""
|
|
if furnace_name != "stone-furnace":
|
|
return False, "unsupported_furnace"
|
|
|
|
coal_have = self._get_item_count("coal")
|
|
if coal_have <= 0:
|
|
return False, "no_coal"
|
|
|
|
ore_have_iron = self._get_item_count("iron-ore")
|
|
ore_have_copper = self._get_item_count("copper-ore")
|
|
|
|
# AI가 reason에 어떤 ore를 의도했는지 우선 반영(있을 때만).
|
|
preferred_ore: str | None = None
|
|
if "iron-ore" in reason:
|
|
preferred_ore = "iron-ore"
|
|
elif "copper-ore" in reason:
|
|
preferred_ore = "copper-ore"
|
|
|
|
if preferred_ore == "iron-ore" and ore_have_iron > 0:
|
|
ore = "iron-ore"
|
|
elif preferred_ore == "copper-ore" and ore_have_copper > 0:
|
|
ore = "copper-ore"
|
|
elif ore_have_iron > 0:
|
|
ore = "iron-ore"
|
|
elif ore_have_copper > 0:
|
|
ore = "copper-ore"
|
|
else:
|
|
return False, "no_ore"
|
|
|
|
# insert_to_entity는 내부적으로 inv.remove를 수행하므로,
|
|
# count는 "최대 50, 현재 보유량"으로 제한해 과잉/오차를 줄인다.
|
|
coal_count = min(50, coal_have)
|
|
ore_count = min(50, self._get_item_count(ore))
|
|
|
|
ok1, msg1 = self.insert_to_entity(x, y, item="coal", count=coal_count)
|
|
ok2, msg2 = self.insert_to_entity(x, y, item=ore, count=ore_count)
|
|
|
|
return ok1 and ok2, f"bootstrap coal+ore: {msg1}; {msg2}"
|
|
|
|
# ── 탐색 ─────────────────────────────────────────────────────────
|
|
def explore(
|
|
self,
|
|
direction: str = "east",
|
|
max_steps: int = 200,
|
|
wanted_ores: list[str] | None = None,
|
|
) -> 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}}}")
|
|
wanted_ores = wanted_ores or []
|
|
# JSON에서 single string으로 올 수도 있으니 방어
|
|
if isinstance(wanted_ores, str):
|
|
wanted_ores = [wanted_ores]
|
|
|
|
wanted_lua = "local wanted = {}\n"
|
|
for name in wanted_ores:
|
|
safe = str(name).replace("\\", "\\\\").replace('"', '\\"')
|
|
wanted_lua += f'wanted["{safe}"] = true\n'
|
|
|
|
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 + wanted_lua + """
|
|
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 = {}
|
|
local best_x, best_y = pos.x, pos.y
|
|
local best_d2 = math.huge
|
|
local wanted_active = wanted and next(wanted) ~= nil
|
|
|
|
for _, e in ipairs(res) do
|
|
local n = e.name
|
|
counts[n] = (counts[n] or 0) + 1
|
|
|
|
local relevant = true
|
|
if wanted_active then
|
|
relevant = wanted[n] and true or false
|
|
end
|
|
if relevant then
|
|
local dx = e.position.x - pos.x
|
|
local dy = e.position.y - pos.y
|
|
local d2 = dx*dx + dy*dy
|
|
if d2 < best_d2 then
|
|
best_d2 = d2
|
|
best_x = e.position.x
|
|
best_y = e.position.y
|
|
end
|
|
end
|
|
end
|
|
|
|
-- wanted_ores가 지정된 경우: 그 중 하나라도 있으면 FOUND, 아니면 UNWANTED 반환
|
|
if wanted_active then
|
|
for n, _ in pairs(wanted) do
|
|
if counts[n] and counts[n] > 0 then
|
|
local parts = {}
|
|
for name, count in pairs(counts) do parts[#parts+1] = name .. "=" .. count end
|
|
return "FOUND:" .. string.format("%.0f,%.0f", best_x, best_y) .. "|" .. table.concat(parts, ",")
|
|
end
|
|
end
|
|
return "UNWANTED:" .. string.format("%.0f,%.0f", pos.x, pos.y)
|
|
end
|
|
|
|
local parts = {}
|
|
for name, count in pairs(counts) do parts[#parts+1] = name .. "=" .. count end
|
|
return "FOUND:" .. string.format("%.0f,%.0f", best_x, best_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("|")
|
|
# FOUND: "tx,ty|name=count,name2=count" 형태
|
|
try:
|
|
loc = parts[0]
|
|
lx, ly = loc.split(",", 1)
|
|
tile_x = int(float(lx))
|
|
tile_y = int(float(ly))
|
|
except Exception:
|
|
tile_x, tile_y = None, None
|
|
|
|
counts: dict[str, int] = {}
|
|
if len(parts) > 1 and parts[1]:
|
|
for seg in parts[1].split(","):
|
|
if "=" not in seg:
|
|
continue
|
|
k, v = seg.split("=", 1)
|
|
try:
|
|
counts[k] = int(v)
|
|
except Exception:
|
|
continue
|
|
|
|
ores_to_store = wanted_ores if wanted_ores else list(counts.keys())
|
|
if tile_x is not None and tile_y is not None and ores_to_store:
|
|
mem = load_ore_patch_memory()
|
|
updated = False
|
|
for ore in ores_to_store:
|
|
c = counts.get(ore)
|
|
if c is None or c <= 0:
|
|
continue
|
|
mem = update_ore_patch_memory(mem, ore, tile_x, tile_y, count=c)
|
|
updated = True
|
|
if updated:
|
|
save_ore_patch_memory(mem)
|
|
|
|
return True, f"자원 발견! 위치({parts[0]}), 자원: {parts[1] if len(parts)>1 else ''}"
|
|
if result.startswith("UNWANTED:"):
|
|
# 원하는 자원이 아니면 계속 걷기
|
|
continue
|
|
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 mine_resource(self, ore: str = "iron-ore", count: int = 10) -> tuple[bool, str]:
|
|
"""1. 가까운 광석 찾기 (실패한 좌표 제외)
|
|
2. 그 위치로 걸어가기
|
|
3. 채굴 시도
|
|
4. 실패 시 해당 좌표를 제외 목록에 추가 → 반복"""
|
|
before_count = self._get_item_count(ore)
|
|
target_count = before_count + count
|
|
total_mined = 0
|
|
failed_positions: set[tuple[int, int]] = set() # Lua exclude 키와 동일한 정수 타일 좌표만 저장
|
|
|
|
for round_num in range(15): # 최대 15번 시도
|
|
# 1. Lua에서 가장 가까운 광석 찾기 (실패한 좌표 제외)
|
|
# 제외 좌표를 Lua 테이블로 전달
|
|
exclude_lua = "local exclude = {}\n"
|
|
for i, (fx, fy) in enumerate(failed_positions):
|
|
exclude_lua += f'exclude["{fx},{fy}"] = true\n'
|
|
|
|
find_result = self.rcon.lua(P + f"""
|
|
{exclude_lua}
|
|
local res = p.surface.find_entities_filtered{{position = p.position, radius = 250, name = "{ore}"}}
|
|
if #res == 0 then rcon.print("NOT_FOUND") return end
|
|
-- 거리순 정렬
|
|
local pos = p.position
|
|
table.sort(res, 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)
|
|
-- 제외 목록에 없는 가장 가까운 광석 찾기
|
|
for _, e in ipairs(res) do
|
|
-- Lua 키를 "정수 타일"로 통일 (반올림/절삭 방식 차이로 exclude가 빗나가는 문제 방지)
|
|
local tx = math.floor(e.position.x + 0.5)
|
|
local ty = math.floor(e.position.y + 0.5)
|
|
local key = string.format("%d,%d", tx, ty)
|
|
if not exclude[key] then
|
|
-- mining_state는 타일 정수가 아니라, 엔티티의 "정확한 e.position"을 주는 게 안정적
|
|
-- 반환 형식: "<tx,ty>|<ex,ey>"
|
|
rcon.print(string.format("%d,%d|%.3f,%.3f", tx, ty, e.position.x, e.position.y))
|
|
return
|
|
end
|
|
end
|
|
rcon.print("ALL_EXCLUDED")
|
|
""")
|
|
if not find_result or find_result in ("NO_PLAYER", "NO_CHARACTER"):
|
|
break
|
|
if find_result == "NOT_FOUND":
|
|
if total_mined > 0:
|
|
return True, f"{ore} {total_mined}개 채굴 (주변 광석 소진)"
|
|
return False, f"반경 250 내 {ore} 없음 - explore로 다른 광맥 찾기"
|
|
if find_result == "ALL_EXCLUDED":
|
|
if total_mined > 0:
|
|
return True, f"{ore} {total_mined}개 채굴 (접근 가능 타일 모두 시도)"
|
|
return False, f"{ore} 근처 타일 {len(failed_positions)}개 모두 접근 불가 - 다른 위치로 이동 필요"
|
|
|
|
try:
|
|
find_result = find_result.strip()
|
|
key_part, pos_part = find_result.split("|", 1)
|
|
key_xy = key_part.split(",")
|
|
ox, oy = int(key_xy[0]), int(key_xy[1]) # move 타겟(정수 타일)
|
|
|
|
pos_xy = pos_part.split(",")
|
|
mine_x, mine_y = float(pos_xy[0]), float(pos_xy[1]) # mining 타겟(엔티티 정확 좌표)
|
|
except:
|
|
return False, f"좌표 파싱 실패: {find_result}"
|
|
|
|
# 2. 광석 위치로 걸어가기
|
|
print(f" [채굴] 광석({mine_x:.2f},{mine_y:.2f})으로 이동... (시도 {round_num+1}, 제외: {len(failed_positions)}개)")
|
|
ok, msg = self.move(mine_x, mine_y)
|
|
if not ok:
|
|
print(f" [채굴] 이동 실패: {msg}")
|
|
failed_positions.add((ox, oy))
|
|
continue
|
|
|
|
# 2.5. 이동 후 실제로 해당 광석이 있는지 재확인
|
|
# Factorio는 자원이 "타일"이 아니라 "엔티티"라서, 좌표 미세오차/경계로 인해
|
|
# move 후에도 p.position 근처에 자원이 없으면 채굴을 시도하지 않고 다음 후보로 넘어간다.
|
|
near_lua = P + f"""
|
|
local res = p.surface.find_entities_filtered{{position = p.position, radius = 1.2, name = "{ore}"}}
|
|
if res and #res > 0 then rcon.print("YES") else rcon.print("NO") end
|
|
"""
|
|
has_ore = self.rcon.lua(near_lua).strip() == "YES"
|
|
if not has_ore:
|
|
failed_positions.add((ox, oy))
|
|
continue
|
|
|
|
# 3. 현재 위치에서 채굴 시도
|
|
# 우클릭 “유지”: mining_state를 매 루프마다 다시 쓰면 릴리즈 후 재클릭처럼 동작해
|
|
# 진행이 끊기고 RCON도 과도하게 호출된다. 한 번만 설정하고 틱 동안 그대로 둔다.
|
|
self.rcon.lua(P + f"""
|
|
p.update_selected_entity({{x = {mine_x}, y = {mine_y}}})
|
|
p.mining_state = {{mining = true, position = {{x = {mine_x}, y = {mine_y}}}}}
|
|
""")
|
|
stall_count = 0
|
|
last_item = self._get_item_count(ore)
|
|
mined_this_tile = False
|
|
|
|
for tick in range(300): # 최대 30초
|
|
time.sleep(0.1)
|
|
|
|
if tick % 8 == 7:
|
|
current = self._get_item_count(ore)
|
|
if current >= target_count:
|
|
self.rcon.lua(P + "p.mining_state = {mining = false}")
|
|
total_mined = current - before_count
|
|
return True, f"{ore} {total_mined}개 채굴 완료"
|
|
|
|
# 아이템이 아직 안 늘었더라도, 채굴 진행률이 올라가기 시작하면 “채굴 시작됨”으로 간주
|
|
prog_raw = self.rcon.lua(P + "rcon.print(tostring(p.character_mining_progress or 0))")
|
|
try:
|
|
prog = float(prog_raw)
|
|
except:
|
|
prog = 0.0
|
|
|
|
if current > last_item:
|
|
stall_count = 0
|
|
last_item = current
|
|
mined_this_tile = True
|
|
else:
|
|
stall_count += 1
|
|
if prog > 0.02:
|
|
mined_this_tile = True
|
|
|
|
# 진행률이 0에 가까운데도 오랫동안 아이템이 안 늘면, 그 타일은 접근 불가로 보고 중단
|
|
if stall_count >= 20 and prog <= 0.02:
|
|
break
|
|
|
|
# mined_this_tile=false인 경우(진행 시작 전)만 stall_count로 종료
|
|
# mined_this_tile=true인 경우에는 items가 늦게 증가해도 루프 끝까지 기다림
|
|
|
|
self.rcon.lua(P + "p.mining_state = {mining = false}")
|
|
total_mined = self._get_item_count(ore) - before_count
|
|
|
|
if not mined_this_tile:
|
|
# 이 타일에서 한 개도 못 캤음 → 접근 불가
|
|
failed_positions.add((ox, oy))
|
|
print(f" [채굴] ({ox:.0f},{oy:.0f}) 접근 불가 -> 제외 목록 추가 (총 {len(failed_positions)}개)")
|
|
else:
|
|
# 채굴 됐지만 타일 고갈
|
|
print(f" [채굴] ({ox:.0f},{oy:.0f}) 고갈/중단. 현재 {total_mined}개 채굴됨")
|
|
failed_positions.add((ox, oy)) # 고갈된 것도 제외
|
|
|
|
if total_mined >= count:
|
|
return True, f"{ore} {total_mined}개 채굴 완료"
|
|
|
|
# 모든 라운드 후
|
|
if total_mined > 0:
|
|
return True, f"{ore} {total_mined}개 채굴 (목표 {count}개 중 일부)"
|
|
return False, f"{ore} 채굴 실패 - {len(failed_positions)}개 타일 접근 불가"
|
|
|
|
# ── 제작/배치/삽입/레시피/연구/대기 ──────────────────────────────
|
|
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 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
|
|
|
|
-- (x,y)가 자원 패치 위/겹침 등으로 배치 불가인 경우가 있어,
|
|
-- 인접 타일을 can_place_entity로 먼저 확인 후 성공 좌표를 사용한다.
|
|
local ox, oy = {x}, {y}
|
|
local candidates = {{
|
|
{{ox, oy}},
|
|
{{ox+1, oy}}, {{ox-1, oy}}, {{ox, oy+1}}, {{ox, oy-1}},
|
|
{{ox+1, oy+1}}, {{ox-1, oy-1}}, {{ox+1, oy-1}}, {{ox-1, oy+1}}
|
|
}}
|
|
|
|
-- 배치 아이템이 없어도, 이미 같은 엔티티가 있으면 재사용으로 처리
|
|
for _, pos in ipairs(candidates) do
|
|
local cx, cy = pos[1], pos[2]
|
|
local existing = p.surface.find_entities_filtered{{
|
|
position = {{cx, cy}},
|
|
radius = 0.5,
|
|
name = "{name}",
|
|
}}
|
|
if existing and #existing > 0 then
|
|
rcon.print(string.format("REUSED:%.0f,%.0f", cx, cy))
|
|
return
|
|
end
|
|
end
|
|
|
|
-- 요청 좌표가 기존 엔티티와 약간 떨어져 있을 수 있으므로,
|
|
-- 더 넓은 반경 내에서도 같은 엔티티가 있으면 재사용 처리한다.
|
|
do
|
|
local nearby = p.surface.find_entities_filtered{{
|
|
position = {{ox, oy}},
|
|
radius = 3,
|
|
name = "{name}",
|
|
}}
|
|
if nearby and #nearby > 0 then
|
|
local best = nearby[1]
|
|
local best_d2 = math.huge
|
|
for _, e in ipairs(nearby) do
|
|
if e and e.valid then
|
|
local dx = e.position.x - ox
|
|
local dy = e.position.y - oy
|
|
local d2 = dx*dx + dy*dy
|
|
if d2 < best_d2 then
|
|
best = e
|
|
best_d2 = d2
|
|
end
|
|
end
|
|
end
|
|
rcon.print(string.format("REUSED:%.0f,%.0f", best.position.x, best.position.y))
|
|
return
|
|
end
|
|
end
|
|
|
|
-- 여기까지 왔는데 배치할 엔티티가 없다면,
|
|
-- 그 때서야 인벤토리에 아이템이 있는지 확인한다.
|
|
local inv = p.get_main_inventory()
|
|
local have = inv.get_item_count("{name}")
|
|
if have < 1 then rcon.print("NO_ITEM") return end
|
|
|
|
for _, pos in ipairs(candidates) do
|
|
local cx, cy = pos[1], pos[2]
|
|
local can = p.surface.can_place_entity{{name="{name}", position={{cx, cy}}, direction={lua_dir}}}
|
|
if can then
|
|
p.cursor_stack.set_stack{{name="{name}", count=1}}
|
|
local built = p.build_from_cursor{{position = {{cx, cy}}, direction = {lua_dir}}}
|
|
p.cursor_stack.clear()
|
|
if built then
|
|
rcon.print(string.format("OK:%.0f,%.0f", cx, cy))
|
|
return
|
|
end
|
|
end
|
|
end
|
|
|
|
rcon.print("BLOCKED")
|
|
""")
|
|
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("OK:"):
|
|
coords = result.split(":", 1)[1]
|
|
return True, f"{name} 배치 ({coords})"
|
|
elif result.startswith("REUSED:"):
|
|
coords = result.split(":", 1)[1]
|
|
return True, f"{name} 재사용 ({coords})"
|
|
elif result == "NO_ITEM": return False, f"인벤토리에 {name} 없음"
|
|
elif result.startswith("TOO_FAR"): return False, f"너무 멀음 - move 먼저"
|
|
elif result == "BLOCKED": return False, f"배치 불가"
|
|
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, positions = 0, 0, []
|
|
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, "벨트 설치 실패"
|
|
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 and e.burner then
|
|
local fi = e.burner.inventory
|
|
if fi and fi.can_insert({{name="{item}", count=1}}) then
|
|
-- can_insert 이후에만 inv에서 제거해서, 미삽입/미스매치로 인한 아이템 증발을 방지
|
|
local removed = inv.remove({{name="{item}", count=actual}})
|
|
if removed > 0 then
|
|
local inserted_count = fi.insert({{name="{item}", count=removed}})
|
|
if inserted_count and inserted_count > 0 then inserted = true break end
|
|
end
|
|
end
|
|
end
|
|
if not inserted and e.valid 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
|
|
local inserted_count = ti.insert({{name="{item}", count=removed}})
|
|
if inserted_count and inserted_count > 0 then inserted = true end
|
|
end
|
|
break
|
|
end
|
|
end
|
|
end
|
|
if inserted then break end
|
|
end
|
|
rcon.print(inserted and "OK" or "NOT_FOUND")
|
|
""")
|
|
if not result or result in ("NO_PLAYER", "NO_CHARACTER"): return False, "플레이어 없음"
|
|
if result == "OK": return True, f"({x},{y})에 {item} 삽입"
|
|
elif result == "TOO_FAR": return False, "너무 멀음"
|
|
elif result == "NO_ITEM": return False, f"{item} 없음"
|
|
return False, "건물 없음"
|
|
|
|
def set_recipe(self, x: int, y: int, recipe: str) -> tuple[bool, str]:
|
|
result = self.rcon.lua(P + f"""
|
|
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 result == "OK": return True, f"레시피: {recipe}"
|
|
return False, "조립기 없음"
|
|
|
|
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 result == "OK": return True, f"{tech} 연구 시작"
|
|
elif result == "ALREADY_DONE": return True, f"{tech} 완료"
|
|
return False, result or "실패"
|
|
|
|
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
|