"""선단/어구그룹 폴리곤 생성기. 프론트엔드 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_store, now: Optional[datetime] = None, ) -> list[dict]: """어구 이름 패턴으로 어구그룹을 탐지한다. 프론트엔드 FleetClusterLayer.tsx gearGroupMap useMemo 로직 이관. 전체 AIS 선박(vessel_store._tracks)에서 어구 패턴을 탐지한다. Args: vessel_store: VesselStore — get_all_latest_positions() + get_vessel_info(). 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) # 전체 선박의 최신 위치 가져오기 all_positions = vessel_store.get_all_latest_positions() # 선박명 → mmsi 맵 (모선 탐색용, 어구 패턴이 아닌 선박만) name_to_mmsi: dict[str, str] = {} for mmsi, pos in all_positions.items(): name = (pos.get('name') or '').strip() if name and not GEAR_PATTERN.match(name): name_to_mmsi[name] = mmsi # 1단계: 같은 모선명 어구 수집 (60분 이내만) raw_groups: dict[str, list[dict]] = {} for mmsi, pos in all_positions.items(): name = (pos.get('name') or '').strip() if not name: continue # staleness 체크 ts = pos.get('timestamp') if ts is not None: if isinstance(ts, datetime): last_dt = ts if ts.tzinfo is not None else ts.replace(tzinfo=timezone.utc) else: 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 m = GEAR_PATTERN.match(name) if not m: continue parent_name = m.group(1).strip() entry = { 'mmsi': mmsi, 'name': name, 'lat': pos['lat'], 'lon': pos['lon'], 'sog': pos.get('sog', 0), 'cog': pos.get('cog', 0), } 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 all_positions: parent_pos = all_positions[parent_mmsi] anchor_lat = parent_pos['lat'] anchor_lon = parent_pos['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_store, company_vessels: dict[int, list[str]], companies: dict[int, dict], ) -> list[dict]: """선단(FLEET) + 어구그룹(GEAR) 폴리곤 스냅샷을 생성한다. Shapely 미설치 시 빈 리스트를 반환한다. Args: vessel_store: VesselStore — get_all_latest_positions() + get_vessel_info(). 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] = [] all_positions = vessel_store.get_all_latest_positions() # ── 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: pos = all_positions.get(mmsi) if not pos: continue lat = pos['lat'] lon = pos['lon'] sog = pos.get('sog', 0) cog = pos.get('cog', 0) points.append((lon, lat)) members.append({ 'mmsi': mmsi, 'name': pos.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_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 all_positions: parent_pos = all_positions[parent_mmsi] anchor_lat = parent_pos['lat'] anchor_lon = parent_pos['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 all_positions: parent_pos = all_positions[parent_mmsi] p_lon, p_lat = parent_pos['lon'], parent_pos['lat'] 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 all_positions: parent_pos = all_positions[parent_mmsi] members_out.append({ 'mmsi': parent_mmsi, 'name': parent_name, 'lat': parent_pos['lat'], 'lon': parent_pos['lon'], 'sog': parent_pos.get('sog', 0), 'cog': parent_pos.get('cog', 0), '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