change log to dataset & field

app.py

@@ -17,36 +17,59 @@ HF_DATA_TOKEN = os.getenv("HF_DATA_TOKEN")
 app = FastAPI(title="MuscleCare FastAPI Server")
 
 # ----- 모델 -----
-class LogUploadPayload(BaseModel):
-    user_id: str
-    session_id: str
-    measure_date: str
-    rms: float
-    freq: float
-    fatigue: float
-    rms_base: Optional[float] = None
-    freq_base: Optional[float] = None
-    user_emb: Optional[List[float]] = Field(default=None, description="length=12")
-    mode: str
-    window_count: int
-    measurement_count: int
+class DatasetItem(BaseModel):
+    user_id: int
+    session_id: Optional[str] = None
+    window_id: int
+    window_start_ms: int
+    window_end_ms: int
+    timestamp_utc: Optional[str] = None
+
+    acc_x_mean: Optional[float] = None
+    acc_y_mean: Optional[float] = None
+    acc_z_mean: Optional[float] = None
+    gyro_x_mean: Optional[float] = None
+    gyro_y_mean: Optional[float] = None
+    gyro_z_mean: Optional[float] = None
+    linacc_x_mean: Optional[float] = None
+    linacc_y_mean: Optional[float] = None
+    linacc_z_mean: Optional[float] = None
+    gravity_x_mean: Optional[float] = None
+    gravity_y_mean: Optional[float] = None
+    gravity_z_mean: Optional[float] = None
+
+    acc_x_std: Optional[float] = None
+    acc_y_std: Optional[float] = None
+    acc_z_std: Optional[float] = None
+    gyro_x_std: Optional[float] = None
+    gyro_y_std: Optional[float] = None
+    gyro_z_std: Optional[float] = None
+
+    rms_acc: Optional[float] = None
+    rms_gyro: Optional[float] = None
+    mean_freq_acc: Optional[float] = None
+    mean_freq_gyro: Optional[float] = None
+    entropy_acc: Optional[float] = None
+    entropy_gyro: Optional[float] = None
+    jerk_mean: Optional[float] = None
+    jerk_std: Optional[float] = None
+    stability_index: Optional[float] = None
 
-class BatchLogItem(BaseModel):
-    user_id: str
-    session_id: str
-    measure_date: str
-    rms: float
-    freq: float
-    fatigue: float
     rms_base: Optional[float] = None
     freq_base: Optional[float] = None
-    user_emb: Optional[List[float]] = Field(default=None, description="length=12")
-    mode: str
-    window_count: int
-    measurement_count: int
+    user_emb: Optional[List[float]] = Field(default=None, description="length=12 vector")
+
+    fatigue_prev: Optional[float] = None
+    fatigue: Optional[float] = None
+    fatigue_level: Optional[int] = None
+
+    quality_flag: Optional[int] = 1
+    window_size_ms: Optional[int] = 2000
+    overlap_rate: Optional[float] = 0.5
 
-class BatchLogsPayload(BaseModel):
-    batch_data: List[BatchLogItem]
+
+class DatasetBatchPayload(BaseModel):
+    batch_data: List[DatasetItem]
 
 
 # ----- 엔드포인트 -----
@@ -60,7 +83,7 @@ def root():
         "endpoints": {
             "health": "/health (빠른 체크)",
             "docs": "/docs",
-            "upload_logs": "/upload_logs (개별 로그 데이터)",
+            "upload_dataset": "/upload_dataset (배치 데이터 업로드)",
             "user_dataset": "/user_dataset/{user_id}"
         }
     }
@@ -78,85 +101,9 @@ def health():
     except Exception as e:
         return {"ok": False, "error": str(e)}
 
-@app.post("/upload_logs")
-async def upload_logs(payload: LogUploadPayload):
-    """개별 로그 데이터를 Hugging Face Hub로 푸시"""
-    try:
-        # Hugging Face 환경변수 확인
-        hf_repo_id = os.getenv("HF_DATA_REPO_ID")
-        hf_token = os.getenv("HF_DATA_TOKEN")
-        
-        if not hf_repo_id or not hf_token:
-            raise HTTPException(status_code=500, detail="Hugging Face 설정이 필요합니다 (HF_DATA_REPO_ID, HF_DATA_TOKEN)")
-
-        # 단일 레코드 생성
-        record = {
-            "session_id": payload.session_id,
-            "measure_date": payload.measure_date,
-            "rms": payload.rms,
-            "freq": payload.freq,
-            "fatigue": payload.fatigue,
-            "rms_base": payload.rms_base,
-            "freq_base": payload.freq_base,
-            "user_emb": payload.user_emb,
-            "mode": payload.mode,
-            "window_count": payload.window_count,
-            "measurement_count": payload.measurement_count,
-            "timestamp": datetime.now().isoformat()
-        }
-
-        # 현재 repo에 있는 데이터 불러오기
-        try:
-            existing = load_dataset(hf_repo_id, token=hf_token)
-            print(f"📂 기존 데이터 로드 완료")
-        except Exception:
-            existing = DatasetDict()
-            print("📂 기존 repo 없음 → 새로 생성")
-
-        # 사용자 데이터 처리
-        user_id = payload.user_id
-        try:
-            # 새 데이터 처리
-            new_df = pd.DataFrame([record])
-            new_dataset = df_to_dataset(new_df)
-
-            if user_id in existing:
-                # 기존 데이터와 병합
-                old_df = existing[user_id].to_pandas()
-                merged_df = pd.concat([old_df, new_df], ignore_index=True)
-                existing[user_id] = df_to_dataset(merged_df)
-                print(f"📊 {user_id}: 기존 데이터와 병합 ({len(old_df)} + 1 = {len(merged_df)}개 레코드)")
-            else:
-                existing[user_id] = new_dataset
-                print(f"📊 {user_id}: 신규 데이터 추가 (1개 레코드)")
-
-            # 데이터 푸시
-            existing.push_to_hub(hf_repo_id, token=hf_token, private=True)
-            print(f"✅ {user_id} 데이터 푸시 완료")
-
-            return {
-                "user_id": user_id,
-                "status": "success",
-                "new_rows": 1,
-                "filename": f"{user_id}.parquet",
-                "repo_id": hf_repo_id,
-                "message": f"Log uploaded successfully for user {user_id}"
-            }
-                
-        except Exception as e:
-            print(f"❌ {user_id} 처리 실패: {e}")
-            raise HTTPException(status_code=500, detail=f"데이터 처리 실패: {str(e)}")
-
-    except HTTPException:
-        raise
-    except Exception as e:
-        print(f"❌ 로그 업로드 실패: {e}")
-        raise HTTPException(status_code=500, detail=f"로그 업로드 실패: {str(e)}")
-
-
-@app.post("/upload_batch_logs")
-async def upload_batch_logs(payload: BatchLogsPayload):
-    """배치 로그 데이터를 Hugging Face Hub에 병렬 아닌 일괄 반영 (스키마 정규화 포함)"""
+@app.post("/upload_dataset")
+async def upload_dataset(payload: DatasetBatchPayload):
+    """배치 데이터셋을 Hugging Face Hub에 일괄 반영 (스키마 정규화 포함)"""
     try:
         hf_repo_id = os.getenv("HF_DATA_REPO_ID")
         hf_token = os.getenv("HF_DATA_TOKEN")
@@ -165,9 +112,48 @@ async def upload_batch_logs(payload: BatchLogsPayload):
 
         # 새 스키마 정의
         target_cols = [
-            "session_id", "measure_date", "rms", "freq", "fatigue",
-            "rms_base", "freq_base", "user_emb", "mode", "window_count",
-            "measurement_count", "timestamp"
+            "user_id",
+            "session_id",
+            "window_id",
+            "window_start_ms",
+            "window_end_ms",
+            "timestamp_utc",
+            "acc_x_mean",
+            "acc_y_mean",
+            "acc_z_mean",
+            "gyro_x_mean",
+            "gyro_y_mean",
+            "gyro_z_mean",
+            "linacc_x_mean",
+            "linacc_y_mean",
+            "linacc_z_mean",
+            "gravity_x_mean",
+            "gravity_y_mean",
+            "gravity_z_mean",
+            "acc_x_std",
+            "acc_y_std",
+            "acc_z_std",
+            "gyro_x_std",
+            "gyro_y_std",
+            "gyro_z_std",
+            "rms_acc",
+            "rms_gyro",
+            "mean_freq_acc",
+            "mean_freq_gyro",
+            "entropy_acc",
+            "entropy_gyro",
+            "jerk_mean",
+            "jerk_std",
+            "stability_index",
+            "rms_base",
+            "freq_base",
+            "user_emb",
+            "fatigue_prev",
+            "fatigue",
+            "fatigue_level",
+            "quality_flag",
+            "window_size_ms",
+            "overlap_rate",
         ]
 
         # 기존 데이터 로드
@@ -196,8 +182,8 @@ async def upload_batch_logs(payload: BatchLogsPayload):
                 df["user_emb"] = df["user_emb"].apply(_parse_emb)
 
             # 타임스탬프 없으면 추가
-            if "timestamp" not in df.columns:
-                df["timestamp"] = datetime.now().isoformat()
+            if "timestamp_utc" not in df.columns or df["timestamp_utc"].isnull().all():
+                df["timestamp_utc"] = datetime.now().isoformat()
 
             # 타겟 컬럼 세트로 맞추기
             for c in target_cols:
@@ -210,22 +196,10 @@ async def upload_batch_logs(payload: BatchLogsPayload):
         # payload를 사용자별로 그룹화
         user_groups: dict[str, list[dict]] = {}
         for item in payload.batch_data:
-            # 레코드 생성
-            rec = {
-                "session_id": item.session_id,
-                "measure_date": item.measure_date,
-                "rms": item.rms,
-                "freq": item.freq,
-                "fatigue": item.fatigue,
-                "rms_base": item.rms_base,
-                "freq_base": item.freq_base,
-                "user_emb": item.user_emb,
-                "mode": item.mode,
-                "window_count": item.window_count,
-                "measurement_count": item.measurement_count,
-                "timestamp": datetime.now().isoformat()
-            }
-            user_groups.setdefault(item.user_id, []).append(rec)
+            rec = item.model_dump()
+            if not rec.get("timestamp_utc"):
+                rec["timestamp_utc"] = datetime.now().isoformat()
+            user_groups.setdefault(str(item.user_id), []).append(rec)
 
         results = {}
 
@@ -271,8 +245,8 @@ async def upload_batch_logs(payload: BatchLogsPayload):
     except HTTPException:
         raise
     except Exception as e:
-        print(f"❌ 배치 로그 업로드 실패: {e}")
-        raise HTTPException(status_code=500, detail=f"배치 로그 업로드 실패: {str(e)}")
+        print(f"❌ 배치 데이터셋 업로드 실패: {e}")
+        raise HTTPException(status_code=500, detail=f"배치 데이터셋 업로드 실패: {str(e)}")
 
 def df_to_dataset(df):
     """DataFrame을 Dataset으로 변환"""