make_sample_data.py

# -*- coding: utf-8 -*-
"""
make_sample_data.py
===================
**합성(synthetic) 샘플 CSV 생성기.**

실제 센서 CSV가 아직 없을 때, 전체 파이프라인(train/evaluate/predict)을 곧바로
실행·검증할 수 있도록 그럴듯한 7일치 데이터를 만들어 sample.csv 로 저장한다.

생성 컬럼: timestamp, temp, humidity, pressure, lux, soil1, soil2, soil3, soil4
- 1분 간격 시계열 (7일 = 10080행)
- 하루 주기(sin/cos)를 따르는 온도/습도/조도
- 관수(물주기) 후 서서히 마르는 패턴의 토양수분
- 약간의 잡음과 일부 결측치(전처리 검증용)를 포함

주의: 이 스크립트는 순수하게 CSV(데이터)만 생성한다.
      하드웨어/제어/통신 코드는 일절 포함하지 않는다.
"""

import numpy as np
import pandas as pd

from prepare_data import (
    TS_COL, TEMP_COL, HUMIDITY_COL, PRESSURE_COL, LUX_COL, SOIL_COLS,
    DEFAULT_CSV_PATH,
)

# 생성 파라미터
DAYS = 7                       # 생성할 일수
MINUTES_PER_DAY = 24 * 60      # 하루 분 수
N_ROWS = DAYS * MINUTES_PER_DAY  # 총 행 수 (10080)
START_TIME = "2026-06-01 00:00:00"  # 시작 시각
MISSING_RATE = 0.005           # 결측치 비율 (0.5%) — 전처리 로직 검증용
RANDOM_SEED = 42               # 재현성


def make_sample_dataframe(seed=RANDOM_SEED):
    """7일치 합성 데이터를 담은 DataFrame을 생성해 반환한다."""
    rng = np.random.default_rng(seed)

    # 1분 간격 타임스탬프
    timestamps = pd.date_range(start=START_TIME, periods=N_ROWS, freq="1min")

    # 하루 주기를 만들기 위한 위상 (0~2π가 하루에 한 바퀴)
    minute_of_day = timestamps.hour * 60 + timestamps.minute
    day_phase = 2 * np.pi * minute_of_day / MINUTES_PER_DAY

    # --- 온도: 낮에 높고 새벽에 낮음 (평균 24도, 진폭 ±5도) ---
    temp = 24.0 + 5.0 * np.sin(day_phase - np.pi / 2) + rng.normal(0, 0.3, N_ROWS)

    # --- 습도: 온도와 반대로 움직임 (평균 60%, 진폭 ±15%) ---
    humidity = 60.0 - 15.0 * np.sin(day_phase - np.pi / 2) + rng.normal(0, 1.0, N_ROWS)
    humidity = np.clip(humidity, 20, 95)

    # --- 기압: 거의 일정 + 완만한 변동 (모델 입력엔 안 쓰지만 컬럼은 채움) ---
    pressure = 1013.0 + 3.0 * np.sin(day_phase / 3) + rng.normal(0, 0.2, N_ROWS)

    # --- 조도: 낮에만 높음(밤엔 0). 일출~일몰 구간만 양수 ---
    lux_raw = np.sin(day_phase - np.pi / 2)
    lux = np.where(lux_raw > 0, lux_raw, 0.0) * 1000.0 + rng.normal(0, 5.0, N_ROWS)
    lux = np.clip(lux, 0, None)

    # --- 토양수분: 관수 후 서서히 증발하는 톱니 패턴 ---
    # 하루 2번(아침/저녁) 관수한다고 가정하고, 관수 시 +값, 이후 완만히 감소.
    soil_base = np.empty(N_ROWS)
    level = 55.0  # 시작 수분 레벨(%)
    for i in range(N_ROWS):
        mod = minute_of_day[i]
        # 06:00, 18:00 경에 관수 -> 수분 급상승
        if mod == 6 * 60 or mod == 18 * 60:
            level = min(level + 25.0, 90.0)
        else:
            # 증발 + 식물 흡수로 분당 소량 감소 (온도 높을수록 더 빨리 마름)
            evap = 0.02 + 0.004 * max(temp[i] - 24.0, 0.0)
            level = max(level - evap, 15.0)
        soil_base[i] = level

    # 센서 4개: 같은 추세 + 센서별 오프셋/잡음 (현실적 다양성)
    soil_data = {}
    for k, col in enumerate(SOIL_COLS):
        offset = rng.normal(0, 2.0)  # 센서별 설치 위치 차이
        noise = rng.normal(0, 0.5, N_ROWS)
        soil_data[col] = np.clip(soil_base + offset + noise, 0, 100)

    # DataFrame 조립
    df = pd.DataFrame({
        TS_COL: timestamps,
        TEMP_COL: np.round(temp, 2),
        HUMIDITY_COL: np.round(humidity, 2),
        PRESSURE_COL: np.round(pressure, 2),
        LUX_COL: np.round(lux, 1),
    })
    for col in SOIL_COLS:
        df[col] = np.round(soil_data[col], 2)

    # --- 일부 결측치 삽입 (전처리 ffill/보간 로직 검증용) ---
    # timestamp 제외한 수치 컬럼에 무작위로 NaN을 뿌린다.
    value_cols = [TEMP_COL, HUMIDITY_COL, PRESSURE_COL, LUX_COL] + SOIL_COLS
    for col in value_cols:
        mask = rng.random(N_ROWS) < MISSING_RATE
        df.loc[mask, col] = np.nan

    return df


def main(path=DEFAULT_CSV_PATH):
    df = make_sample_dataframe()
    df.to_csv(path, index=False)
    n_missing = int(df.isna().sum().sum())
    print(f"[make_sample_data] '{path}' 생성 완료")
    print(f"  - 행 수      : {len(df)}  ({DAYS}일치, 1분 간격)")
    print(f"  - 컬럼       : {list(df.columns)}")
    print(f"  - 결측치 수  : {n_missing} (전처리 검증용)")
    print(f"  - 기간       : {df[TS_COL].iloc[0]} ~ {df[TS_COL].iloc[-1]}")


if __name__ == "__main__":
    main()