""" generate_sample_data.py — RodentMealScope 합성 섭식 이벤트 로그 생성기 자동 급이기(BioDAQ 류)에서 얻는 원자료를 모사한다. 실제 설치류 섭식의 특징을 반영한다: - 식사(meal)는 군집(burst)을 이룬다: 식사 내부 짧은 간격 + 식사 사이 긴 간격 → 로그-생존곡선이 이봉성(bimodal)을 띠게 됨 (Tolkamp 변곡점법 검증용) - 일주기성: 야간(dark active phase)에 섭식이 집중 - 군간 차이: * Control : 정상 섭식 * Drug-A : 식사 크기 감소형 (식욕억제제 모사) * Drug-B : 식사 빈도 감소형 - 일부 노이즈/흘림 이벤트 삽입 (QC 검증용) 재현성: RANDOM_SEED 고정. 실행하면 data/ 하위에 CSV가 생성된다. 용도: 참고용·연구용 (synthetic data, for research reference only) """ import os import numpy as np import pandas as pd RANDOM_SEED = 20260518 N_DAYS = 5 # 관측 일수 LIGHTS_ON_HOUR = 7 # 점등 시각 (ZT0) LIGHT_PHASE_HOURS = 12 # 12:12 LD START_DATE = pd.Timestamp("2026-05-12 07:00:00") OUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") # 군 정의: (군명, 개체수, 식사크기배수, 식사빈도배수, 개체ID 접두) # 접두는 군마다 고유해야 함 (개체ID 충돌 방지) GROUPS = [ ("Control", 8, 1.00, 1.00, "CTL"), ("Drug-A", 8, 0.62, 1.05, "DGA"), # 식사 크기 감소형 ("Drug-B", 8, 0.98, 0.60, "DGB"), # 식사 빈도 감소형 ] # 기본 섭식 파라미터 BASE_MEALS_PER_DAY = 9.0 # 대조군 1일 식사 횟수 기대값 DARK_MEAL_FRACTION = 0.78 # 야간 식사 비율 WITHIN_MEAL_EVENT_MEAN = 6 # 식사당 이벤트(bout) 수 기대값 WITHIN_MEAL_GAP_S = (2, 30) # 식사 내부 이벤트 간격 범위(초) MIN_INTERMEAL_GAP_MIN = 12.0 # 식사 사이 최소 간격(분) — 이봉성 보장 EVENT_INTAKE_G = (0.005, 0.045) # 이벤트당 섭취량 범위(g) def _meal_times_for_day(rng, day_start, n_meals): """ 하루치 식사 시작시각을 생성. 야간(ZT12-24)에 가중하고, 식사 사이에 최소 간격(MIN_INTERMEAL_GAP_MIN)을 보장하여 간격 분포가 이봉성을 띠도록 한다. """ raw = [] for _ in range(n_meals): if rng.random() < DARK_MEAL_FRACTION: # dark active phase: ZT 12~24 -> 실제시각 19:00~익일07:00 zt = rng.uniform(12.0, 24.0) else: zt = rng.uniform(0.0, 12.0) t = day_start + pd.Timedelta(hours=zt) raw.append(t) raw = sorted(raw) # 너무 가까운 식사는 뒤로 밀어 최소 간격 확보 min_gap = pd.Timedelta(minutes=MIN_INTERMEAL_GAP_MIN) times = [] for t in raw: if times and t - times[-1] < min_gap: t = times[-1] + min_gap + pd.Timedelta(minutes=rng.uniform(0, 25)) times.append(t) return times def generate(): rng = np.random.default_rng(RANDOM_SEED) rows = [] event_counter = 0 for grp, n_animals, size_mult, freq_mult, prefix in GROUPS: for a in range(n_animals): animal_id = f"{prefix}{a + 1:02d}" for d in range(N_DAYS): day_start = START_DATE + pd.Timedelta(days=d) # 그날 식사 횟수 (포아송, 군 배수 적용) lam = BASE_MEALS_PER_DAY * freq_mult n_meals = max(1, rng.poisson(lam)) meal_starts = _meal_times_for_day(rng, day_start, n_meals) for ms in meal_starts: # 식사당 이벤트 수 n_ev = max(1, rng.poisson(WITHIN_MEAL_EVENT_MEAN)) # 식사 크기 목표 (군 배수 적용) t = ms for e in range(n_ev): amt = rng.uniform(*EVENT_INTAKE_G) * size_mult rows.append({ "animal_id": animal_id, "group": grp, "timestamp": t, "intake_amount": round(float(amt), 5), "bout": f"B{event_counter}", }) event_counter += 1 gap = rng.uniform(*WITHIN_MEAL_GAP_S) t = t + pd.Timedelta(seconds=gap) df = pd.DataFrame(rows).sort_values(["animal_id", "timestamp"]).reset_index(drop=True) # --- 노이즈/QC 검증용 이벤트 삽입 --- noise_rows = [] # 흘림(spillage): 비정상적으로 큰 단일 섭취량 for _ in range(25): i = rng.integers(0, len(df)) noise_rows.append({ "animal_id": df.loc[i, "animal_id"], "group": df.loc[i, "group"], "timestamp": df.loc[i, "timestamp"] + pd.Timedelta(minutes=rng.uniform(1, 30)), "intake_amount": round(float(rng.uniform(2.5, 6.0)), 5), "bout": f"NOISE_SPILL_{_}", }) # 이중계수(double count): 1초 이내 중복 이벤트 for _ in range(20): i = rng.integers(0, len(df)) noise_rows.append({ "animal_id": df.loc[i, "animal_id"], "group": df.loc[i, "group"], "timestamp": df.loc[i, "timestamp"] + pd.Timedelta(milliseconds=int(rng.uniform(100, 800))), "intake_amount": round(float(df.loc[i, "intake_amount"]), 5), "bout": f"NOISE_DUP_{_}", }) df = pd.concat([df, pd.DataFrame(noise_rows)], ignore_index=True) df = df.sort_values(["animal_id", "timestamp"]).reset_index(drop=True) df["timestamp"] = df["timestamp"].dt.strftime("%Y-%m-%d %H:%M:%S") os.makedirs(OUT_DIR, exist_ok=True) # 1) 메인 합성 로그 (RodentMealScope 공통 스키마) main_path = os.path.join(OUT_DIR, "sample_feeder_events.csv") df.to_csv(main_path, index=False) # 2) FED3 스타일 export 예시 (pellet 기반) fed3 = df[df["group"] == "Control"].copy().head(800) fed3 = fed3.rename(columns={"timestamp": "MM:DD:YYYY hh:mm:ss", "animal_id": "Device"}) fed3["PelletCount"] = (fed3["intake_amount"] / 0.02).round().clip(lower=1).astype(int) fed3["Event"] = "Pellet" fed3 = fed3[["MM:DD:YYYY hh:mm:ss", "Device", "PelletCount", "Event"]] fed3_path = os.path.join(OUT_DIR, "sample_fed3_export.csv") fed3.to_csv(fed3_path, index=False) # 3) lickometer 스타일 export 예시 lick = df[df["group"] == "Drug-A"].copy().head(800) lick = lick.rename(columns={"timestamp": "time", "animal_id": "subject"}) lick["licks"] = (lick["intake_amount"] / 0.0015).round().clip(lower=1).astype(int) lick = lick[["time", "subject", "group", "licks"]] lick_path = os.path.join(OUT_DIR, "sample_lickometer_export.csv") lick.to_csv(lick_path, index=False) print(f"[OK] 메인 로그 : {main_path} ({len(df)} events)") print(f"[OK] FED3 예시 : {fed3_path} ({len(fed3)} events)") print(f"[OK] lickometer 예시 : {lick_path} ({len(lick)} events)") print(f"개체수: {sum(g[1] for g in GROUPS)}, 군: {[g[0] for g in GROUPS]}, " f"관측일수: {N_DAYS}") return main_path if __name__ == "__main__": generate()