sensor_processing.py

"""
센서 데이터 처리 코어
=====================
- 외부 API/스크립트가 DataFrame만 넘기면 처리할 수 있는 진입점 제공
- GUI, CLI, 배치 코드에서 공통으로 재사용 가능
"""

from dataclasses import dataclass
from typing import Callable, List, Optional

import numpy as np
import pandas as pd


LogCallback = Callable[[str], None]
ProgressCallback = Callable[[float, str], None]


@dataclass
class ProcessingConfig:
    """센서 데이터 처리 설정"""

    timestamp_column: str
    value_columns: List[str]
    resample_minutes: int = 2
    target_count: int = 2000
    noise_ratio: float = 0.005
    offset_seconds: int = 10
    random_seed: Optional[int] = None

    def validate(self):
        if not self.timestamp_column:
            raise ValueError("timestamp_column은 필수입니다.")

        if not self.value_columns:
            raise ValueError("최소 1개 이상의 value_columns가 필요합니다.")

        if self.resample_minutes <= 0:
            raise ValueError("resample_minutes는 1 이상이어야 합니다.")

        if self.target_count <= 0:
            raise ValueError("target_count는 1 이상이어야 합니다.")

        if self.noise_ratio < 0:
            raise ValueError("noise_ratio는 0 이상이어야 합니다.")

        if self.offset_seconds <= 0:
            raise ValueError("offset_seconds는 1 이상이어야 합니다.")

        window_seconds = self.resample_minutes * 60
        if self.offset_seconds > window_seconds:
            raise ValueError("offset_seconds는 리샘플링 윈도우보다 클 수 없습니다.")


@dataclass
class ProcessingSummary:
    source_rows: int
    prepared_rows: int
    generated_rows: int
    final_rows: int
    window_seconds: int
    offset_runs: int
    dropped_rows: int
    start_time: pd.Timestamp
    end_time: pd.Timestamp


@dataclass
class ProcessingResult:
    output_df: pd.DataFrame
    summary: ProcessingSummary


def process_sensor_dataframe(
    raw_df: pd.DataFrame,
    config: ProcessingConfig,
    log_callback: Optional[LogCallback] = None,
    progress_callback: Optional[ProgressCallback] = None,
) -> ProcessingResult:
    """1초 단위 센서 데이터를 2분 학습용 데이터로 증강한다."""

    config.validate()

    if raw_df is None or raw_df.empty:
        raise ValueError("입력 데이터가 비어 있습니다.")

    _log(log_callback, "=" * 50)
    _log(log_callback, "데이터 처리 시작")
    _log(log_callback, "입력 행 수: {:,}".format(len(raw_df)))

    prepared_df, dropped_rows = _prepare_input_dataframe(raw_df, config)
    source_rows = len(raw_df)
    prepared_rows = len(prepared_df)

    if prepared_rows < 2:
        raise ValueError("처리 가능한 데이터가 부족합니다. 타임스탬프/수치 컬럼을 확인하세요.")

    window_seconds = config.resample_minutes * 60
    max_offsets = int(window_seconds / config.offset_seconds)
    start_time = prepared_df[config.timestamp_column].min()
    end_time = prepared_df[config.timestamp_column].max()
    total_hours = (end_time - start_time).total_seconds() / 3600.0

    _log(log_callback, "정제 후 행 수: {:,} (제거: {:,})".format(prepared_rows, dropped_rows))
    _log(log_callback, "데이터 시간 범위: {:.2f}시간".format(total_hours))
    _log(log_callback, "윈도우: {}초 | Offset 간격: {}초".format(window_seconds, config.offset_seconds))
    _update_progress(progress_callback, 0, "입력 데이터 정제 완료")

    rng = np.random.default_rng(config.random_seed)
    augmented_frames = _generate_offset_frames(prepared_df, config, rng, log_callback, progress_callback, max_offsets)

    generated_rows = sum(len(frame) for frame in augmented_frames)
    if generated_rows == 0:
        raise ValueError("증강 결과가 비어 있습니다. 입력 데이터 기간과 설정값을 확인하세요.")

    _log(log_callback, "리샘플링 완료: {:,}개 생성".format(generated_rows))

    if generated_rows < config.target_count:
        _log(log_callback, "목표 미달, 추가 노이즈 증강 진행")
        augmented_frames.extend(_expand_with_noise(augmented_frames, config, rng, progress_callback))

    result_df = pd.concat(augmented_frames, ignore_index=True)
    result_df = result_df.sort_values(config.timestamp_column).reset_index(drop=True)
    result_df = result_df.drop_duplicates(subset=[config.timestamp_column], keep="first").reset_index(drop=True)

    _update_progress(progress_callback, 100, "처리 완료")
    _log(log_callback, "최종 데이터 수: {:,}개".format(len(result_df)))

    summary = ProcessingSummary(
        source_rows=source_rows,
        prepared_rows=prepared_rows,
        generated_rows=generated_rows,
        final_rows=len(result_df),
        window_seconds=window_seconds,
        offset_runs=max_offsets,
        dropped_rows=dropped_rows,
        start_time=start_time,
        end_time=end_time,
    )
    return ProcessingResult(output_df=result_df, summary=summary)


def _prepare_input_dataframe(raw_df: pd.DataFrame, config: ProcessingConfig):
    required_columns = [config.timestamp_column] + config.value_columns
    missing_columns = [col for col in required_columns if col not in raw_df.columns]
    if missing_columns:
        raise ValueError("입력 데이터에 필요한 컬럼이 없습니다: {}".format(", ".join(missing_columns)))

    df = raw_df[required_columns].copy()
    df[config.timestamp_column] = pd.to_datetime(df[config.timestamp_column], errors="coerce")

    for col in config.value_columns:
        df[col] = pd.to_numeric(df[col], errors="coerce")

    before_count = len(df)
    df = df.dropna(subset=[config.timestamp_column])
    df = df.dropna(subset=config.value_columns, how="all")
    df = df.sort_values(config.timestamp_column).reset_index(drop=True)

    dropped_rows = before_count - len(df)
    return df, dropped_rows


def _generate_offset_frames(df: pd.DataFrame, config: ProcessingConfig, rng, log_callback, progress_callback, max_offsets: int):
    timestamp_column = config.timestamp_column
    value_columns = config.value_columns
    window_seconds = config.resample_minutes * 60
    indexed_df = df.set_index(timestamp_column)
    start_time = df[timestamp_column].min()
    frames = []

    _update_progress(progress_callback, 5, "리샘플링 시작")

    for offset_index in range(max_offsets):
        offset_seconds = offset_index * config.offset_seconds
        origin_time = start_time + pd.to_timedelta(offset_seconds, unit="s")

        mean_frame = indexed_df[value_columns].resample(
            "{}s".format(window_seconds),
            origin=origin_time,
            label="left",
            closed="left",
        ).mean()

        mean_frame = mean_frame.dropna(how="all")
        if mean_frame.empty:
            _update_progress(
                progress_callback,
                50 * float(offset_index + 1) / float(max_offsets),
                "리샘플링 중...",
            )
            continue

        if config.noise_ratio > 0:
            std_frame = indexed_df[value_columns].resample(
                "{}s".format(window_seconds),
                origin=origin_time,
                label="left",
                closed="left",
            ).std()
            std_frame = std_frame.reindex(mean_frame.index)
            scale_frame = std_frame.where(std_frame > 0, mean_frame.abs() * config.noise_ratio).fillna(0)
            noise_matrix = rng.normal(0, scale_frame.to_numpy() * config.noise_ratio)
            mean_frame.loc[:, value_columns] = mean_frame.to_numpy() + noise_matrix

        frame = mean_frame.reset_index()
        frame[timestamp_column] = frame[timestamp_column] + pd.to_timedelta(window_seconds / 2.0, unit="s")
        frames.append(frame)

        total_rows = sum(len(item) for item in frames)
        progress_value = 50 * float(offset_index + 1) / float(max_offsets)
        _update_progress(progress_callback, progress_value, "리샘플링 중...")

        if total_rows >= config.target_count:
            _log(log_callback, "목표 데이터 수 도달 (Offset {}/{})".format(offset_index + 1, max_offsets))
            break

    return frames


def _expand_with_noise(frames, config: ProcessingConfig, rng, progress_callback):
    base_df = pd.concat(frames, ignore_index=True)
    if base_df.empty:
        return []

    timestamp_column = config.timestamp_column
    value_columns = config.value_columns
    extra_frames = []
    remaining = config.target_count - len(base_df)

    while remaining > 0:
        sample_size = min(len(base_df), remaining)
        sampled_indices = rng.choice(len(base_df), size=sample_size, replace=True)
        sampled_df = base_df.iloc[sampled_indices].copy()
        
        sampled_df[timestamp_column] = sampled_df[timestamp_column] + pd.to_timedelta(
            rng.integers(1, 60, size=sample_size),
            unit="s",
        )

        if config.noise_ratio > 0:
            noise_scale = sampled_df[value_columns].abs().fillna(0) * (config.noise_ratio * 2)
            extra_noise = rng.normal(0, noise_scale.to_numpy())
            sampled_df.loc[:, value_columns] = sampled_df[value_columns].to_numpy() + extra_noise

        extra_frames.append(sampled_df)
        remaining = config.target_count - (len(base_df) + sum(len(frame) for frame in extra_frames))

        created = sum(len(frame) for frame in extra_frames)
        progress_value = 50 + (50 * float(created) / float(max(1, config.target_count - len(base_df))))
        _update_progress(progress_callback, min(progress_value, 100), "추가 증강 중...")

    return extra_frames


def _log(callback: Optional[LogCallback], message: str):
    if callback:
        callback(message)


def _update_progress(callback: Optional[ProgressCallback], value: float, message: str):
    if callback:
        callback(value, message)