Add context_compressor.py - 계층적 컨텍스트 압축 모듈

context_compressor.py

@@ -0,0 +1,383 @@
+"""
+context_compressor.py — 계층적 컨텍스트 압축
+
+중반부 이후 건물이 수백 개가 되면 전체 상태를 프롬프트에
+담을 수 없음. 이 모듈은 공장을 '구역(zone)'으로 나눠
+AI가 소화할 수 있는 크기로 압축한다.
+
+압축 전: 건물 수백 개 * 좌표/상태 → 토큰 수천 개
+압축 후: 구역 요약 10줄 + 병목/문제 목록 → 토큰 200개 이하
+"""
+import json
+import math
+from factorio_rcon import FactorioRCON
+
+
+# ── 구역 설정 ────────────────────────────────────────────────────
+# 공장을 이 크기의 격자로 나눔 (타일 단위)
+ZONE_SIZE = 64
+
+
+class ContextCompressor:
+    """
+    게임 상태를 AI가 소화 가능한 크기로 압축.
+
+    계층 구조:
+      Level 0 (raw)   : 모든 건물 좌표 + 상태 (토큰 수천)
+      Level 1 (zone)  : 구역별 건물 수 + 상태 요약 (토큰 ~300)
+      Level 2 (global): 전체 공장 핵심 지표 + 문제 목록 (토큰 ~100)
+    """
+
+    def __init__(self, rcon: FactorioRCON):
+        self.rcon = rcon
+
+    # ── 공개 API ────────────────────────────────────────────────
+
+    def get_compressed_state(self, detail_level: int = 1) -> str:
+        """
+        detail_level:
+          0 = 글로벌 요약만 (후반 대규모 공장, 토큰 ~80)
+          1 = 구역별 요약 (중반, 토큰 ~250)  ← 기본값
+          2 = 구역 요약 + 문제 구역 드릴다운 (토큰 ~400)
+        """
+        global_summary = self._get_global_summary()
+        problems       = self._detect_problems()
+        player         = self._get_player_info()
+
+        if detail_level == 0:
+            return self._format_global(global_summary, problems, player)
+
+        zones = self._get_zone_summaries()
+
+        if detail_level == 1:
+            return self._format_level1(global_summary, zones, problems, player)
+
+        # level 2: 문제 구역은 드릴다운
+        drilldown = self._drilldown_problem_zones(zones, problems)
+        return self._format_level2(global_summary, zones, problems, drilldown, player)
+
+    # ── 글로벌 지표 ─────────────────────────────────────────────
+
+    def _get_global_summary(self) -> dict:
+        lua = """
+local surface = game.player.surface
+local force   = game.player.force
+local center  = game.player.position
+
+-- 전체 건물 수 집계
+local counts = {}
+local entity_types = {
+  "burner-mining-drill","electric-mining-drill",
+  "stone-furnace","steel-furnace","electric-furnace",
+  "assembling-machine-1","assembling-machine-2","assembling-machine-3",
+  "transport-belt","fast-transport-belt","express-transport-belt",
+  "inserter","fast-inserter","filter-inserter",
+  "small-electric-pole","medium-electric-pole","big-electric-pole",
+  "steam-engine","solar-panel","accumulator",
+  "boiler","offshore-pump","pipe",
+  "lab","rocket-silo","radar",
+  "gun-turret","laser-turret","wall","gate",
+  "oil-refinery","chemical-plant","pumpjack"
+}
+
+local total = 0
+for _, name in ipairs(entity_types) do
+  local found = surface.find_entities_filtered{
+    area={{center.x-500,center.y-500},{center.x+500,center.y+500}},
+    name=name
+  }
+  if #found > 0 then
+    counts[name] = #found
+    total = total + #found
+  end
+end
+
+-- 전력 현황
+local stats = force.get_entity_production_statistics(surface)
+local power_produced = 0
+local power_consumed = 0
+
+-- 연구 진행
+local current_tech = "none"
+local tech_progress = 0
+if force.current_research then
+  current_tech = force.current_research.name
+  tech_progress = math.floor(force.research_progress * 100)
+end
+
+-- 완료된 연구 수
+local researched = 0
+for _, t in pairs(force.technologies) do
+  if t.researched then researched = researched + 1 end
+end
+
+rcon.print(game.table_to_json({
+  total_entities = total,
+  counts = counts,
+  current_research = current_tech,
+  research_progress = tech_progress,
+  researched_count = researched,
+  evolution = math.floor(force.evolution_factor * 100)
+}))
+"""
+        raw = self.rcon.lua(lua)
+        return json.loads(raw) if raw else {}
+
+    # ── 구역 요약 ───────────────────────────────────────────────
+
+    def _get_zone_summaries(self) -> list[dict]:
+        """공장을 ZONE_SIZE×ZONE_SIZE 격자로 나눠 구역별 요약"""
+        lua = f"""
+local surface = game.player.surface
+local center  = game.player.position
+local Z = {ZONE_SIZE}
+
+-- 활성 구역만 탐색 (플레이어 주변 ±512)
+local radius = 512
+local zones  = {{}}
+
+local function zone_key(x, y)
+  return math.floor(x/Z) .. "," .. math.floor(y/Z)
+end
+
+-- 모든 엔티티를 구역으로 분류
+local all_entities = surface.find_entities_filtered{{
+  area={{{{center.x-radius, center.y-radius}},
+         {{center.x+radius, center.y+radius}}}},
+  force = "player"
+}}
+
+for _, e in ipairs(all_entities) do
+  local key = zone_key(e.position.x, e.position.y)
+  if not zones[key] then
+    zones[key] = {{
+      key = key,
+      x   = math.floor(e.position.x/Z)*Z,
+      y   = math.floor(e.position.y/Z)*Z,
+      entities = 0,
+      miners = 0, furnaces = 0, assemblers = 0,
+      belts = 0, inserters = 0, poles = 0,
+      turrets = 0, labs = 0, pumps = 0
+    }}
+  end
+  local z = zones[key]
+  z.entities = z.entities + 1
+  local n = e.name
+  if n:find("mining%-drill") then z.miners = z.miners + 1
+  elseif n:find("furnace") then z.furnaces = z.furnaces + 1
+  elseif n:find("assembling") then z.assemblers = z.assemblers + 1
+  elseif n:find("belt") then z.belts = z.belts + 1
+  elseif n:find("inserter") then z.inserters = z.inserters + 1
+  elseif n:find("pole") then z.poles = z.poles + 1
+  elseif n:find("turret") then z.turrets = z.turrets + 1
+  elseif n == "lab" then z.labs = z.labs + 1
+  elseif n == "pumpjack" or n == "offshore-pump" then z.pumps = z.pumps + 1
+  end
+end
+
+-- 배열로 변환 (entities > 0인 것만)
+local result = {{}}
+for _, z in pairs(zones) do
+  if z.entities > 2 then
+    result[#result+1] = z
+  end
+end
+rcon.print(game.table_to_json(result))
+"""
+        raw = self.rcon.lua(lua)
+        return json.loads(raw) if raw else []
+
+    # ── 문제 감지 ───────────────────────────────────────────────
+
+    def _detect_problems(self) -> list[str]:
+        lua = """
+local surface = game.player.surface
+local force   = game.player.force
+local center  = game.player.position
+local problems = {}
+
+-- 1. 연료 없는 채굴기/제련소
+local burner_entities = surface.find_entities_filtered{
+  area={{center.x-300,center.y-300},{center.x+300,center.y+300}},
+  type={"mining-drill","furnace"}
+}
+local no_fuel = 0
+for _, e in ipairs(burner_entities) do
+  if e.burner then
+    local fi = e.burner.inventory
+    if fi and fi.is_empty() then
+      no_fuel = no_fuel + 1
+    end
+  end
+end
+if no_fuel > 0 then
+  problems[#problems+1] = "연료 부족: " .. no_fuel .. "개 건물이 연료 없음"
+end
+
+-- 2. 꽉 찬 삽입기 (병목)
+local inserters = surface.find_entities_filtered{
+  area={{center.x-300,center.y-300},{center.x+300,center.y+300}},
+  type="inserter"
+}
+local stuck = 0
+for _, ins in ipairs(inserters) do
+  if ins.held_stack and ins.held_stack.valid_for_read then
+    stuck = stuck + 1
+  end
+end
+if stuck > 5 then
+  problems[#problems+1] = "벨트 병목: 삽입기 " .. stuck .. "개가 아이템 들고 멈춤"
+end
+
+-- 3. 전력 부족
+local consumers = surface.find_entities_filtered{
+  area={{center.x-400,center.y-400},{center.x+400,center.y+400}},
+  type={"assembling-machine","lab","electric-mining-drill","electric-furnace"}
+}
+local no_power = 0
+for _, e in ipairs(consumers) do
+  if e.energy < e.electric_buffer_size * 0.1 then
+    no_power = no_power + 1
+  end
+end
+if no_power > 0 then
+  problems[#problems+1] = "전력 부족: " .. no_power .. "개 건물 전력 10% 미만"
+end
+
+-- 4. 자원 고갈 임박한 광맥
+local drills = surface.find_entities_filtered{
+  area={{center.x-300,center.y-300},{center.x+300,center.y+300}},
+  type="mining-drill"
+}
+local depleting = 0
+for _, d in ipairs(drills) do
+  local ore = surface.find_entities_filtered{
+    area={{d.position.x-3, d.position.y-3},
+          {d.position.x+3, d.position.y+3}},
+    type="resource"
+  }
+  if #ore > 0 and ore[1].amount < 5000 then
+    depleting = depleting + 1
+  end
+end
+if depleting > 0 then
+  problems[#problems+1] = "자원 고갈 임박: " .. depleting .. "개 채굴기 구역 5000 미만"
+end
+
+rcon.print(game.table_to_json(problems))
+"""
+        raw = self.rcon.lua(lua)
+        try:
+            return json.loads(raw) if raw else []
+        except Exception:
+            return []
+
+    # ── 드릴다운 (문제 구역 상세) ────────────────────────────────
+
+    def _drilldown_problem_zones(self, zones: list, problems: list) -> str:
+        if not problems or not zones:
+            return ""
+        # 엔티티가 가장 많은 상위 2개 구역 드릴다운
+        top = sorted(zones, key=lambda z: z.get("entities", 0), reverse=True)[:2]
+        lines = ["#### 주요 구역 상세"]
+        for z in top:
+            lines.append(
+                f"  구역({z['x']},{z['y']}): "
+                f"채굴기{z.get('miners',0)} 제련소{z.get('furnaces',0)} "
+                f"조립기{z.get('assemblers',0)} 벨트{z.get('belts',0)} "
+                f"삽입기{z.get('inserters',0)} 전선주{z.get('poles',0)}"
+            )
+        return "\n".join(lines)
+
+    # ── 플레이어 정보 ────────────────────────────────────────────
+
+    def _get_player_info(self) -> dict:
+        lua = """
+local p = game.player
+local inv = p.get_main_inventory()
+local contents = inv.get_contents()
+
+-- 핵심 아이템만 추출
+local key_items = {
+  "iron-plate","copper-plate","steel-plate","iron-ore","copper-ore","coal",
+  "stone","stone-furnace","burner-mining-drill","electric-mining-drill",
+  "transport-belt","inserter","burner-inserter","fast-inserter",
+  "small-electric-pole","medium-electric-pole","pipe",
+  "offshore-pump","boiler","steam-engine","assembling-machine-1",
+  "assembling-machine-2","lab","iron-gear-wheel",
+  "automation-science-pack","logistic-science-pack","chemical-science-pack",
+  "military-science-pack","production-science-pack","utility-science-pack"
+}
+local inv_summary = {}
+for _, name in ipairs(key_items) do
+  local c = contents[name] or 0
+  if c > 0 then inv_summary[name] = c end
+end
+
+rcon.print(game.table_to_json({
+  x = math.floor(p.position.x),
+  y = math.floor(p.position.y),
+  inventory = inv_summary
+}))
+"""
+        raw = self.rcon.lua(lua)
+        return json.loads(raw) if raw else {}
+
+    # ── 포맷터 ──────────────────────────────────────────────────
+
+    def _format_global(self, g: dict, problems: list, player: dict) -> str:
+        lines = [
+            "## 공장 상태 (글로벌 요약)",
+            f"위치: ({player.get('x','?')}, {player.get('y','?')})",
+            f"전체 건물 수: {g.get('total_entities', 0)}개",
+            f"진화도: {g.get('evolution', 0)}%",
+            f"완료 연구: {g.get('researched_count', 0)}개",
+            f"진행 중 연구: {g.get('current_research','없음')} ({g.get('research_progress',0)}%)",
+        ]
+        if problems:
+            lines.append("\n## 감지된 문제")
+            for p in problems:
+                lines.append(f"  ! {p}")
+        lines += self._format_inventory(player.get("inventory", {}))
+        return "\n".join(lines)
+
+    def _format_level1(self, g: dict, zones: list, problems: list, player: dict) -> str:
+        lines = [self._format_global(g, problems, player), "\n## 구역별 요약"]
+        for z in sorted(zones, key=lambda x: x.get("entities", 0), reverse=True)[:8]:
+            role = self._guess_zone_role(z)
+            lines.append(
+                f"  [{z['x']:+d},{z['y']:+d}] {role} | "
+                f"채굴{z.get('miners',0)} 제련{z.get('furnaces',0)} "
+                f"조립{z.get('assemblers',0)} 벨트{z.get('belts',0)}"
+            )
+        return "\n".join(lines)
+
+    def _format_level2(self, g, zones, problems, drilldown, player) -> str:
+        base = self._format_level1(g, zones, problems, player)
+        if drilldown:
+            base += "\n\n" + drilldown
+        return base
+
+    def _format_inventory(self, inv: dict) -> list[str]:
+        if not inv:
+            return ["\n인벤토리: 비어있음"]
+        lines = ["\n## 인벤토리 (핵심 아이템)"]
+        for name, count in sorted(inv.items(), key=lambda x: -x[1])[:20]:
+            lines.append(f"  {name}: {count}")
+        return lines
+
+    def _guess_zone_role(self, z: dict) -> str:
+        miners    = z.get("miners", 0)
+        furnaces  = z.get("furnaces", 0)
+        assemblers= z.get("assemblers", 0)
+        turrets   = z.get("turrets", 0)
+        labs      = z.get("labs", 0)
+        pumps     = z.get("pumps", 0)
+
+        if turrets > 3:          return "방어구역"
+        if labs > 0:             return "연구실"
+        if pumps > 0:            return "정유/전력"
+        if assemblers > miners:  return "조립라인"
+        if furnaces > miners:    return "제련구역"
+        if miners > 0:           return "채굴구역"
+        return "기타"