kcg-monitoring/prediction/algorithms/polygon_builder.py

"""선단/어구그룹 폴리곤 생성기.

프론트엔드 FleetClusterLayer.tsx의 어구그룹 탐지 + convexHull/padPolygon 로직을
Python으로 이관한다. Shapely 라이브러리로 폴리곤 생성.
"""

from __future__ import annotations

import logging
import math
import re
from datetime import datetime, timezone
from typing import Optional

try:
    from shapely.geometry import MultiPoint, Point
    from shapely import wkt as shapely_wkt
    _SHAPELY_AVAILABLE = True
except ImportError:
    _SHAPELY_AVAILABLE = False

from algorithms.location import classify_zone

logger = logging.getLogger(__name__)

# 프론트 FleetClusterLayer.tsx gearGroupMap 패턴과 동일
GEAR_PATTERN = re.compile(r'^(.+?)_\d+_\d+_?$')
MAX_DIST_DEG = 0.15         # ~10NM
STALE_SEC = 3600            # 60분
FLEET_BUFFER_DEG = 0.02
GEAR_BUFFER_DEG = 0.01
MIN_GEAR_GROUP_SIZE = 2     # 최소 어구 수 (비허가 구역 외)

# 수역 내 어구 색상, 수역 외 어구 색상
_COLOR_GEAR_IN_ZONE = '#ef4444'
_COLOR_GEAR_OUT_ZONE = '#f97316'

# classify_zone이 수역 내로 판정하는 zone 값 목록
_IN_ZONE_PREFIXES = ('ZONE_',)


def _is_in_zone(zone_info: dict) -> bool:
    """classify_zone 결과가 특정어업수역 내인지 판별."""
    zone = zone_info.get('zone', '')
    return any(zone.startswith(prefix) for prefix in _IN_ZONE_PREFIXES)


def _cluster_color(seed: int) -> str:
    """프론트 clusterColor(id) 이관 — hsl({(seed * 137) % 360}, 80%, 55%)."""
    h = (seed * 137) % 360
    return f'hsl({h}, 80%, 55%)'


def compute_area_sq_nm(polygon, center_lat: float) -> float:
    """Shapely Polygon의 면적(degrees²) → 제곱 해리 변환.

    1도 위도 ≈ 60 NM, 1도 경도 ≈ 60 * cos(lat) NM
    sq_nm = area_deg2 * 60 * 60 * cos(center_lat_rad)
    """
    area_deg2 = polygon.area
    center_lat_rad = math.radians(center_lat)
    sq_nm = area_deg2 * 60.0 * 60.0 * math.cos(center_lat_rad)
    return round(sq_nm, 4)


def build_group_polygon(
    points: list[tuple[float, float]],
    buffer_deg: float,
) -> tuple[Optional[str], Optional[str], float, float, float]:
    """좌표 목록으로 버퍼 폴리곤을 생성한다.

    Args:
        points: (lon, lat) 좌표 목록 — Shapely (x, y) 순서.
        buffer_deg: 버퍼 크기(도).

    Returns:
        (polygon_wkt, center_wkt, area_sq_nm, center_lat, center_lon)
        — polygon_wkt/center_wkt: ST_GeomFromText에 사용할 WKT 문자열.
        — 좌표가 없거나 Shapely 미설치 시 (None, None, 0.0, 0.0, 0.0).
    """
    if not _SHAPELY_AVAILABLE:
        logger.warning('shapely 미설치 — build_group_polygon 건너뜀')
        return None, None, 0.0, 0.0, 0.0

    if not points:
        return None, None, 0.0, 0.0, 0.0

    if len(points) == 1:
        geom = Point(points[0]).buffer(buffer_deg)
    elif len(points) == 2:
        # LineString → buffer로 Polygon 생성
        from shapely.geometry import LineString
        geom = LineString(points).buffer(buffer_deg)
    else:
        # 3점 이상 → convex_hull → buffer
        geom = MultiPoint(points).convex_hull.buffer(buffer_deg)

    # 중심 계산
    centroid = geom.centroid
    center_lon = centroid.x
    center_lat = centroid.y

    area_sq_nm = compute_area_sq_nm(geom, center_lat)
    polygon_wkt = shapely_wkt.dumps(geom, rounding_precision=6)
    center_wkt = f'POINT({center_lon:.6f} {center_lat:.6f})'

    return polygon_wkt, center_wkt, area_sq_nm, center_lat, center_lon


def detect_gear_groups(
    vessel_dfs: dict,
    vessel_store,
    now: Optional[datetime] = None,
) -> list[dict]:
    """어구 이름 패턴으로 어구그룹을 탐지한다.

    프론트엔드 FleetClusterLayer.tsx gearGroupMap useMemo 로직 이관.

    Args:
        vessel_dfs: {mmsi: DataFrame} — 각 DataFrame은 lat, lon, sog, cog, timestamp 칼럼.
        vessel_store: VesselStore — get_vessel_info(mmsi) → {name, ...}.
        now: 기준 시각 (None이면 UTC now).

    Returns:
        [{parent_name, parent_mmsi, members: [{mmsi, name, lat, lon, sog, cog}]}]
    """
    if now is None:
        now = datetime.now(timezone.utc)

    # 선박명 → mmsi 맵 (모선 탐색용)
    name_to_mmsi: dict[str, str] = {}
    for mmsi, df in vessel_dfs.items():
        if df is None or len(df) == 0:
            continue
        info = vessel_store.get_vessel_info(mmsi)
        name: str = (info or {}).get('name', '') or ''
        name = name.strip()
        if name and not GEAR_PATTERN.match(name):
            name_to_mmsi[name] = mmsi

    # 1단계: 같은 모선명 어구 수집 (60분 이내만)
    raw_groups: dict[str, list[dict]] = {}
    for mmsi, df in vessel_dfs.items():
        if df is None or len(df) == 0:
            continue
        last = df.iloc[-1]
        ts = last.get('timestamp') if hasattr(last, 'get') else last['timestamp']

        # timestamp → datetime 변환
        if isinstance(ts, datetime):
            last_dt = ts if ts.tzinfo is not None else ts.replace(tzinfo=timezone.utc)
        else:
            # pandas Timestamp 또는 숫자(unix seconds)
            try:
                import pandas as pd
                last_dt = pd.Timestamp(ts).to_pydatetime()
                if last_dt.tzinfo is None:
                    last_dt = last_dt.replace(tzinfo=timezone.utc)
            except Exception:
                continue

        age_sec = (now - last_dt).total_seconds()
        if age_sec > STALE_SEC:
            continue

        info = vessel_store.get_vessel_info(mmsi)
        name = (info or {}).get('name', '') or ''
        name = name.strip()

        m = GEAR_PATTERN.match(name)
        if not m:
            continue

        parent_name = m.group(1).strip()
        entry = {
            'mmsi': mmsi,
            'name': name,
            'lat': float(last['lat']),
            'lon': float(last['lon']),
            'sog': float(last.get('sog', 0) if hasattr(last, 'get') else last['sog']),
            'cog': float(last.get('cog', 0) if hasattr(last, 'get') else last['cog']),
        }
        raw_groups.setdefault(parent_name, []).append(entry)

    # 2단계: 거리 기반 서브 클러스터링 (anchor 기준 MAX_DIST_DEG 이내만)
    results: list[dict] = []
    for parent_name, gears in raw_groups.items():
        parent_mmsi = name_to_mmsi.get(parent_name)

        # 기준점(anchor): 모선 있으면 모선 위치, 없으면 첫 어구
        anchor_lat: Optional[float] = None
        anchor_lon: Optional[float] = None

        if parent_mmsi and parent_mmsi in vessel_dfs:
            parent_df = vessel_dfs[parent_mmsi]
            if parent_df is not None and len(parent_df) > 0:
                parent_last = parent_df.iloc[-1]
                anchor_lat = float(parent_last['lat'])
                anchor_lon = float(parent_last['lon'])

        if anchor_lat is None and gears:
            anchor_lat = gears[0]['lat']
            anchor_lon = gears[0]['lon']

        if anchor_lat is None or anchor_lon is None:
            continue

        # MAX_DIST_DEG 이내 어구만 포함
        _anchor_lat: float = anchor_lat
        _anchor_lon: float = anchor_lon
        nearby = [
            g for g in gears
            if abs(g['lat'] - _anchor_lat) <= MAX_DIST_DEG
            and abs(g['lon'] - _anchor_lon) <= MAX_DIST_DEG
        ]

        if not nearby:
            continue

        # members 구성: 어구 목록
        members = [
            {
                'mmsi': g['mmsi'],
                'name': g['name'],
                'lat': g['lat'],
                'lon': g['lon'],
                'sog': g['sog'],
                'cog': g['cog'],
            }
            for g in nearby
        ]

        results.append({
            'parent_name': parent_name,
            'parent_mmsi': parent_mmsi,
            'members': members,
        })

    return results


def build_all_group_snapshots(
    vessel_dfs: dict,
    vessel_store,
    company_vessels: dict[int, list[str]],
    companies: dict[int, dict],
) -> list[dict]:
    """선단(FLEET) + 어구그룹(GEAR) 폴리곤 스냅샷을 생성한다.

    Shapely 미설치 시 빈 리스트를 반환한다.

    Args:
        vessel_dfs: {mmsi: DataFrame}.
        vessel_store: VesselStore — get_vessel_info(mmsi).
        company_vessels: {company_id: [mmsi_list]}.
        companies: {id: {name_cn, name_en}}.

    Returns:
        DB INSERT용 dict 목록.
    """
    if not _SHAPELY_AVAILABLE:
        logger.warning('shapely 미설치 — build_all_group_snapshots 빈 리스트 반환')
        return []

    now = datetime.now(timezone.utc)
    snapshots: list[dict] = []

    # ── FLEET 타입: company_vessels 순회 ──────────────────────────
    for company_id, mmsi_list in company_vessels.items():
        company_info = companies.get(company_id, {})
        group_label = company_info.get('name_cn') or company_info.get('name_en') or str(company_id)

        # 각 선박의 최신 좌표 추출
        points: list[tuple[float, float]] = []
        members: list[dict] = []

        for mmsi in mmsi_list:
            df = vessel_dfs.get(mmsi)
            if df is None or len(df) == 0:
                continue
            last = df.iloc[-1]
            lat = float(last['lat'])
            lon = float(last['lon'])
            sog_val = last.get('sog', 0) if hasattr(last, 'get') else last['sog']
            cog_val = last.get('cog', 0) if hasattr(last, 'get') else last['cog']
            sog = float(sog_val)
            cog = float(cog_val)
            points.append((lon, lat))
            members.append({
                'mmsi': mmsi,
                'name': (vessel_store.get_vessel_info(mmsi) or {}).get('name', ''),
                'lat': lat,
                'lon': lon,
                'sog': sog,
                'cog': cog,
                'role': 'LEADER' if mmsi == mmsi_list[0] else 'MEMBER',
                'isParent': False,
            })

        # 2척 미만은 폴리곤 미생성
        if len(points) < 2:
            continue

        polygon_wkt, center_wkt, area_sq_nm, center_lat, center_lon = build_group_polygon(
            points, FLEET_BUFFER_DEG
        )

        snapshots.append({
            'group_type': 'FLEET',
            'group_key': str(company_id),
            'group_label': group_label,
            'snapshot_time': now,
            'polygon_wkt': polygon_wkt,
            'center_wkt': center_wkt,
            'area_sq_nm': area_sq_nm,
            'member_count': len(members),
            'zone_id': None,
            'zone_name': None,
            'members': members,
            'color': _cluster_color(company_id),
        })

    # ── GEAR 타입: detect_gear_groups 결과 순회 ───────────────────
    gear_groups = detect_gear_groups(vessel_dfs, vessel_store, now=now)

    for group in gear_groups:
        parent_name: str = group['parent_name']
        parent_mmsi: Optional[str] = group['parent_mmsi']
        gear_members: list[dict] = group['members']

        # 수역 분류: anchor(모선 or 첫 어구) 위치 기준
        anchor_lat: Optional[float] = None
        anchor_lon: Optional[float] = None

        if parent_mmsi and parent_mmsi in vessel_dfs:
            parent_df = vessel_dfs.get(parent_mmsi)
            if parent_df is not None and len(parent_df) > 0:
                p_last = parent_df.iloc[-1]
                anchor_lat = float(p_last['lat'])
                anchor_lon = float(p_last['lon'])

        if anchor_lat is None and gear_members:
            anchor_lat = gear_members[0]['lat']
            anchor_lon = gear_members[0]['lon']

        if anchor_lat is None:
            continue

        zone_info = classify_zone(float(anchor_lat), float(anchor_lon))
        in_zone = _is_in_zone(zone_info)
        zone_id = zone_info.get('zone') if in_zone else None
        zone_name = zone_info.get('zone_name') if in_zone else None

        # 비허가(수역 외) 어구: MIN_GEAR_GROUP_SIZE 미만 제외
        if not in_zone and len(gear_members) < MIN_GEAR_GROUP_SIZE:
            continue

        # 폴리곤 points: 어구 좌표 + 모선 좌표
        points = [(g['lon'], g['lat']) for g in gear_members]
        if parent_mmsi and parent_mmsi in vessel_dfs:
            parent_df = vessel_dfs.get(parent_mmsi)
            if parent_df is not None and len(parent_df) > 0:
                p_last = parent_df.iloc[-1]
                p_lat = float(p_last['lat'])
                p_lon = float(p_last['lon'])
                if (p_lon, p_lat) not in points:
                    points.append((p_lon, p_lat))

        polygon_wkt, center_wkt, area_sq_nm, _clat, _clon = build_group_polygon(
            points, GEAR_BUFFER_DEG
        )

        # members JSONB 구성
        members_out: list[dict] = []
        # 모선 먼저
        if parent_mmsi and parent_mmsi in vessel_dfs:
            parent_df = vessel_dfs.get(parent_mmsi)
            if parent_df is not None and len(parent_df) > 0:
                p_last = parent_df.iloc[-1]
                p_sog = float(p_last.get('sog', 0) if hasattr(p_last, 'get') else p_last['sog'])
                p_cog = float(p_last.get('cog', 0) if hasattr(p_last, 'get') else p_last['cog'])
                members_out.append({
                    'mmsi': parent_mmsi,
                    'name': parent_name,
                    'lat': float(p_last['lat']),
                    'lon': float(p_last['lon']),
                    'sog': p_sog,
                    'cog': p_cog,
                    'role': 'PARENT',
                    'isParent': True,
                })
        # 어구 목록
        for g in gear_members:
            members_out.append({
                'mmsi': g['mmsi'],
                'name': g['name'],
                'lat': g['lat'],
                'lon': g['lon'],
                'sog': g['sog'],
                'cog': g['cog'],
                'role': 'GEAR',
                'isParent': False,
            })

        color = _COLOR_GEAR_IN_ZONE if in_zone else _COLOR_GEAR_OUT_ZONE

        snapshots.append({
            'group_type': 'GEAR_IN_ZONE' if in_zone else 'GEAR_OUT_ZONE',
            'group_key': parent_name,
            'group_label': parent_name,
            'snapshot_time': now,
            'polygon_wkt': polygon_wkt,
            'center_wkt': center_wkt,
            'area_sq_nm': area_sq_nm,
            'member_count': len(members_out),
            'zone_id': zone_id,
            'zone_name': zone_name,
            'members': members_out,
            'color': color,
        })

    return snapshots