118 lines
4.6 KiB
Python
118 lines
4.6 KiB
Python
"""
|
|
main.py — 순수 AI 플레이 버전
|
|
|
|
치트 없이 실제 게임 메커닉으로 플레이.
|
|
걷기/채굴/제작에 실제 시간이 걸리므로 ACTION_DELAY를 줄임.
|
|
"""
|
|
import time
|
|
import os
|
|
import json
|
|
from factorio_rcon import FactorioRCON
|
|
from state_reader import StateReader
|
|
from context_compressor import ContextCompressor
|
|
from ai_planner import AIPlanner
|
|
from action_executor import ActionExecutor
|
|
|
|
RCON_HOST = os.getenv("FACTORIO_HOST", "127.0.0.1")
|
|
RCON_PORT = int(os.getenv("FACTORIO_PORT", "25575"))
|
|
RCON_PASSWORD = os.getenv("FACTORIO_PASSWORD", "factorio_ai")
|
|
ACTION_DELAY = 0.2 # 행동 자체에 시간이 걸리므로 간격 짧게
|
|
LOG_FILE = "agent_log.jsonl"
|
|
|
|
# 건물 N개 이상이면 압축 모드로 전환
|
|
COMPRESS_THRESHOLD = 50
|
|
|
|
|
|
def run():
|
|
print("=" * 60)
|
|
print(" 팩토리오 순수 AI 에이전트 (치트 없음)")
|
|
print(" - 실제 걷기 / 실제 채굴 / 실제 제작 / 건설 거리 제한")
|
|
print("=" * 60)
|
|
|
|
with FactorioRCON(RCON_HOST, RCON_PORT, RCON_PASSWORD) as rcon:
|
|
reader = StateReader(rcon)
|
|
compressor = ContextCompressor(rcon)
|
|
planner = AIPlanner()
|
|
executor = ActionExecutor(rcon)
|
|
|
|
total_actions = 0
|
|
queue: list[dict] = []
|
|
entity_count = 0
|
|
|
|
while True:
|
|
try:
|
|
if not queue:
|
|
print(f"\n{'='*50}")
|
|
print(f"[재계획] 총 {total_actions}개 실행 완료")
|
|
|
|
# ── 압축 레벨 자동 선택 ──────────────────────
|
|
if entity_count < COMPRESS_THRESHOLD:
|
|
print("[상태] 초반 모드 — 상세 상태 수집")
|
|
state = reader.get_full_state()
|
|
summary = reader.summarize_for_ai(state)
|
|
entity_count = sum(
|
|
v for v in state.get("buildings", {}).values()
|
|
if isinstance(v, int)
|
|
)
|
|
elif entity_count < 200:
|
|
print(f"[상태] 중반 모드 — 구역 압축 (건물 {entity_count}개)")
|
|
summary = compressor.get_compressed_state(detail_level=1)
|
|
else:
|
|
print(f"[상태] 후반 모드 — 글로벌 압축 (건물 {entity_count}개)")
|
|
summary = compressor.get_compressed_state(detail_level=0)
|
|
|
|
global_info = compressor._get_global_summary()
|
|
entity_count = global_info.get("total_entities", entity_count)
|
|
|
|
# ── AI 계획 ──────────────────────────────────
|
|
queue = planner.decide(summary)
|
|
if not queue:
|
|
print("[경고] AI가 행동 반환 안 함. 10초 후 재시도")
|
|
time.sleep(10)
|
|
continue
|
|
|
|
# ── 행동 실행 ─────────────────────────────────────
|
|
action = queue.pop(0)
|
|
total_actions += 1
|
|
act = action.get("action", "")
|
|
reason = action.get("reason", "")
|
|
|
|
print(f"\n[{total_actions}] {act}")
|
|
print(f" 이유: {reason}")
|
|
print(f" params: {json.dumps(action.get('params',{}), ensure_ascii=False)}")
|
|
|
|
success, message = executor.execute(action)
|
|
status = "✅" if success else "❌"
|
|
print(f" 결과: {status} {message}")
|
|
|
|
planner.record_feedback(action, success, message)
|
|
|
|
if not success:
|
|
print(" → 실패. 재계획 요청")
|
|
queue.clear()
|
|
|
|
_log(total_actions, action, success, message)
|
|
time.sleep(ACTION_DELAY)
|
|
|
|
except KeyboardInterrupt:
|
|
print(f"\n종료 (총 {total_actions}개 실행)")
|
|
break
|
|
except Exception as e:
|
|
print(f"[오류] {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
time.sleep(5)
|
|
|
|
|
|
def _log(step, action, success, message):
|
|
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps({
|
|
"step": step, "action": action,
|
|
"success": success, "message": message,
|
|
"timestamp": time.time()
|
|
}, ensure_ascii=False) + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|