factorio-ai-agent/ore_patch_memory.py

"""
ore_patch_memory.py

에이전트가 "발견"한 광맥(자원 엔티티) 좌표를 파일로 기억하기 위한 유틸.

메모리 포맷:
- key: ore name (예: "iron-ore", "stone", "coal")
- value: patch 좌표들의 리스트
  - { tile_x, tile_y, last_seen, count? }
"""

from __future__ import annotations

import json
import os
import time
from typing import Any


ORE_PATCH_CACHE_FILE = "ore_patch_memory.json"
BASE_DIR = os.path.dirname(os.path.abspath(__file__))


def _cache_path(cache_dir: str | None = None) -> str:
    base = cache_dir or BASE_DIR
    return os.path.join(base, ORE_PATCH_CACHE_FILE)


def load_ore_patch_memory(cache_dir: str | None = None) -> dict[str, list[dict[str, Any]]]:
    path = _cache_path(cache_dir)
    try:
        if not os.path.exists(path):
            return {}
        with open(path, "r", encoding="utf-8") as f:
            raw = f.read().strip()
            if not raw:
                return {}
            data = json.loads(raw)
            if not isinstance(data, dict):
                return {}

            # 과거(단일 좌표) 포맷 호환: ore -> dict
            converted: dict[str, list[dict[str, Any]]] = {}
            for ore, v in data.items():
                if isinstance(v, list):
                    converted[ore] = [x for x in v if isinstance(x, dict)]
                elif isinstance(v, dict):
                    converted[ore] = [v]
            return converted
    except Exception:
        return {}


def save_ore_patch_memory(
    memory: dict[str, list[dict[str, Any]]],
    cache_dir: str | None = None,
) -> None:
    path = _cache_path(cache_dir)
    try:
        if not isinstance(memory, dict):
            return
        with open(path, "w", encoding="utf-8") as f:
            json.dump(memory, f, ensure_ascii=False)
    except Exception:
        # 메모리는 부가 기능이므로 실패해도 전체 동작이 깨지면 안 됨
        pass


def update_ore_patch_memory(
    ore_patch_memory: dict[str, list[dict[str, Any]]],
    ore: str,
    tile_x: int,
    tile_y: int,
    *,
    last_seen: float | None = None,
    count: int | None = None,
    max_patches_per_ore: int = 8,
    dedupe_if_same_tile: bool = True,
) -> dict[str, list[dict[str, Any]]]:
    """
    memory[ore]의 patch 리스트에 (tile_x, tile_y)를 추가/갱신한다.

    - 같은 타일이면 last_seen 갱신(+ count가 있으면 count도 더 큰 값을 우선)
    - 다른 타일이면 리스트에 추가하고 최근(last_seen) 기준으로 max_patches_per_ore까지 유지
    """
    if not isinstance(ore_patch_memory, dict):
        ore_patch_memory = {}
    if not ore:
        return ore_patch_memory

    ore = str(ore)
    tile_x = int(tile_x)
    tile_y = int(tile_y)
    now = time.time() if last_seen is None else float(last_seen)

    patches = ore_patch_memory.get(ore)
    if not isinstance(patches, list):
        patches = []
    else:
        patches = [p for p in patches if isinstance(p, dict)]

    # 같은 타일이면 해당 patch를 업데이트한다.
    updated = False
    for p in patches:
        px = p.get("tile_x")
        py = p.get("tile_y")
        if not dedupe_if_same_tile:
            continue
        if isinstance(px, int) and isinstance(py, int) and px == tile_x and py == tile_y:
            p["last_seen"] = now
            if count is not None:
                prev_count = p.get("count")
                if prev_count is None or not isinstance(prev_count, int):
                    p["count"] = int(count)
                else:
                    p["count"] = int(count) if int(count) >= int(prev_count) else prev_count
            updated = True
            break

    if not updated:
        patch: dict[str, Any] = {
            "tile_x": tile_x,
            "tile_y": tile_y,
            "last_seen": now,
        }
        if count is not None:
            patch["count"] = int(count)
        patches.append(patch)

    # 최신 순으로 자르고 다시 저장
    patches.sort(key=lambda x: float(x.get("last_seen", 0.0)), reverse=True)
    ore_patch_memory[ore] = patches[: int(max_patches_per_ore)]

    return ore_patch_memory


def compute_distance_sq(px: float, py: float, tile_x: int, tile_y: int) -> float:
    dx = float(tile_x) - float(px)
    dy = float(tile_y) - float(py)
    return dx * dx + dy * dy