import pandas as pd from pipeline.constants import SOG_STATIONARY_MAX, SOG_FISHING_MAX class BehaviorDetector: """ 속도 기반 3단계 행동 분류 (Yan et al. 2022, Natale et al. 2015) 정박(STATIONARY) / 조업(FISHING) / 항행(SAILING) """ @staticmethod def classify_point(sog: float) -> str: if sog < SOG_STATIONARY_MAX: return 'STATIONARY' elif sog <= SOG_FISHING_MAX: return 'FISHING' else: return 'SAILING' def detect(self, df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df['state'] = df['sog'].apply(self.classify_point) return df @staticmethod def compute_fishing_ratio(df_vessel: pd.DataFrame) -> float: total = len(df_vessel) if total == 0: return 0.0 fishing = (df_vessel['state'] == 'FISHING').sum() return round(fishing / total * 100, 2)