""" Zepp Data Import Module Imports health data from Zepp/Mi Band/Amazfit export files into HabitForge database. Supports: Activity, Sleep, Heart Rate, Body measurements, and Sport/Workouts. """ import csv import glob import os import datetime from sqlalchemy.orm import Session from sqlalchemy import create_engine from backend.models import DailyHealthMetrics, User, WorkoutSession from backend.models.workout_session import ZEPP_ACTIVITY_TYPES from backend.database import SessionLocal, engine, Base # --- Configuration --- EXPORT_DIR = "export-zepp" USER_ID = 1 # Default user ID since we are single user for now or running locally def parse_date(date_str): """Parse YYYY-MM-DD date string.""" if not date_str: return None try: return datetime.datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: return None def parse_datetime(dt_str): """Parse datetime string with timezone (e.g., 2025-12-28 18:25:54+0000).""" if not dt_str: return None try: # Remove timezone for simplicity clean_str = dt_str.split('+')[0].split('-0')[0] return datetime.datetime.strptime(clean_str, "%Y-%m-%d %H:%M:%S") except ValueError: return None def get_latest_export_dir(base_dir): """Find the numbered directory inside export-zepp.""" subdirs = glob.glob(os.path.join(base_dir, "*")) if not subdirs: return None # Return the most recent one (sorted by name, which includes timestamp) return sorted(subdirs)[-1] def import_activity(session, export_path): """Import daily activity summary (steps, distance, calories).""" activity_dir = os.path.join(export_path, "ACTIVITY") csv_files = glob.glob(os.path.join(activity_dir, "*.csv")) if not csv_files: print("No Activity CSV found.") return count = 0 for csv_file in csv_files: print(f"Processing Activity: {csv_file}") with open(csv_file, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: d_date = parse_date(row.get('date')) if not d_date: continue steps = int(row.get('steps', 0)) distance = float(row.get('distance', 0)) calories = float(row.get('calories', 0)) # Upsert metric = session.query(DailyHealthMetrics).filter_by( user_id=USER_ID, date=d_date ).first() if not metric: metric = DailyHealthMetrics(user_id=USER_ID, date=d_date) session.add(metric) # Overwrite with Zepp data as it's authoritative if steps > 0: metric.step_count = steps if distance > 0: metric.distance_meters = distance if calories > 0: metric.calories_burned = calories count += 1 session.commit() print(f" Imported {count} activity records.") def import_sleep(session, export_path): """Import sleep data with phases (deep, light, REM, awake).""" sleep_dir = os.path.join(export_path, "SLEEP") csv_files = glob.glob(os.path.join(sleep_dir, "*.csv")) count = 0 for csv_file in csv_files: print(f"Processing Sleep: {csv_file}") with open(csv_file, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: d_date = parse_date(row.get('date')) if not d_date: continue try: deep = int(row.get('deepSleepTime', 0)) light = int(row.get('shallowSleepTime', 0)) wake = int(row.get('wakeTime', 0)) rem = int(row.get('REMTime', 0)) except ValueError: continue total_sleep = deep + light + rem metric = session.query(DailyHealthMetrics).filter_by( user_id=USER_ID, date=d_date ).first() if not metric: metric = DailyHealthMetrics(user_id=USER_ID, date=d_date) session.add(metric) if total_sleep > 0: metric.sleep_duration_minutes = total_sleep metric.deep_sleep_minutes = deep metric.light_sleep_minutes = light metric.rem_sleep_minutes = rem metric.awake_duration_minutes = wake count += 1 session.commit() print(f" Imported {count} sleep records.") def import_heart_rate(session, export_path): """Import manual heart rate readings.""" hr_dir = os.path.join(export_path, "HEARTRATE") csv_files = glob.glob(os.path.join(hr_dir, "*.csv")) daily_hr = {} # date -> [hr_values] for csv_file in csv_files: print(f"Processing Heart Rate: {csv_file}") with open(csv_file, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: ts_str = row.get('time') hr_val = int(row.get('heartRate', 0)) if not ts_str or hr_val <= 0: continue try: dt = datetime.datetime.strptime(ts_str.split('+')[0], "%Y-%m-%d %H:%M:%S") d_date = dt.date() if d_date not in daily_hr: daily_hr[d_date] = [] daily_hr[d_date].append(hr_val) except Exception: pass # Aggregate for d_date, values in daily_hr.items(): if not values: continue avg_hr = sum(values) // len(values) min_hr = min(values) max_hr = max(values) metric = session.query(DailyHealthMetrics).filter_by( user_id=USER_ID, date=d_date ).first() if not metric: metric = DailyHealthMetrics(user_id=USER_ID, date=d_date) session.add(metric) metric.avg_heart_rate = avg_hr metric.min_heart_rate = min_hr metric.max_heart_rate = max_hr session.commit() print(f" Imported heart rate data for {len(daily_hr)} days.") def import_heart_rate_auto(session, export_path): """Import automatic heart rate monitoring data.""" hr_dir = os.path.join(export_path, "HEARTRATE_AUTO") csv_files = glob.glob(os.path.join(hr_dir, "*.csv")) if not csv_files: print("No HEARTRATE_AUTO CSV found.") return daily_hr = {} for csv_file in csv_files: print(f"Processing Heart Rate Auto: {csv_file}") with open(csv_file, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: date_str = row.get('date') hr_val = int(row.get('heartRate', 0)) if not date_str or hr_val <= 0: continue try: d_date = parse_date(date_str) if not d_date: continue if d_date not in daily_hr: daily_hr[d_date] = [] daily_hr[d_date].append(hr_val) except Exception: pass # Aggregate and Save for d_date, values in daily_hr.items(): if not values: continue avg_hr = sum(values) // len(values) min_hr = min(values) max_hr = max(values) rhr = min_hr # Resting HR approximation metric = session.query(DailyHealthMetrics).filter_by( user_id=USER_ID, date=d_date ).first() if not metric: metric = DailyHealthMetrics(user_id=USER_ID, date=d_date) session.add(metric) metric.avg_heart_rate = avg_hr metric.min_heart_rate = min_hr metric.max_heart_rate = max_hr metric.resting_heart_rate = rhr session.commit() print(f" Imported auto heart rate data for {len(daily_hr)} days.") def import_body(session, export_path): """Import body measurements (weight, etc.).""" body_dir = os.path.join(export_path, "BODY") csv_files = glob.glob(os.path.join(body_dir, "*.csv")) count = 0 for csv_file in csv_files: print(f"Processing Body: {csv_file}") with open(csv_file, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: ts_str = row.get('time') w_str = row.get('weight', '0') weight = float(w_str) if not ts_str or weight <= 0: continue try: dt = datetime.datetime.strptime(ts_str.split('+')[0], "%Y-%m-%d %H:%M:%S") d_date = dt.date() metric = session.query(DailyHealthMetrics).filter_by( user_id=USER_ID, date=d_date ).first() if not metric: metric = DailyHealthMetrics(user_id=USER_ID, date=d_date) session.add(metric) metric.weight = weight count += 1 except Exception: pass session.commit() print(f" Imported {count} body measurements.") def import_sport(session, export_path): """Import sport/workout sessions.""" sport_dir = os.path.join(export_path, "SPORT") csv_files = glob.glob(os.path.join(sport_dir, "*.csv")) if not csv_files: print("No SPORT CSV found.") return count = 0 for csv_file in csv_files: print(f"Processing Sport: {csv_file}") with open(csv_file, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) # Columns: type, startTime, sportTime(s), maxPace(/meter), minPace(/meter), # distance(m), avgPace(/meter), calories(kcal) for row in reader: try: type_id = int(row.get('type', 0)) start_time_str = row.get('startTime') duration_seconds = int(row.get('sportTime(s)', 0)) distance = float(row.get('distance(m)', 0)) calories = float(row.get('calories(kcal)', 0)) max_pace = float(row.get('maxPace(/meter)', 0)) min_pace = float(row.get('minPace(/meter)', 0)) avg_pace = float(row.get('avgPace(/meter)', 0)) start_time = parse_datetime(start_time_str) if not start_time: continue end_time = start_time + datetime.timedelta(seconds=duration_seconds) activity_name = WorkoutSession.get_activity_name(type_id) # Check if workout already exists (by start time and user) existing = session.query(WorkoutSession).filter_by( user_id=USER_ID, start_time=start_time ).first() if existing: # Update existing existing.activity_type = activity_name existing.activity_type_id = type_id existing.duration_seconds = duration_seconds existing.distance_meters = distance existing.calories = calories existing.avg_pace = avg_pace if avg_pace > 0 else None existing.max_pace = max_pace if max_pace > 0 else None existing.min_pace = min_pace if min_pace > 0 else None else: # Create new workout = WorkoutSession( user_id=USER_ID, start_time=start_time, end_time=end_time, activity_type=activity_name, activity_type_id=type_id, duration_seconds=duration_seconds, distance_meters=distance, calories=calories, avg_pace=avg_pace if avg_pace > 0 else None, max_pace=max_pace if max_pace > 0 else None, min_pace=min_pace if min_pace > 0 else None ) session.add(workout) count += 1 except Exception as e: print(f" Error processing workout: {e}") continue session.commit() print(f" Imported {count} workout sessions.") def calculate_pai_scores(session): """Calculate PAI scores for all days based on activity and heart rate data.""" print("Calculating PAI scores...") metrics = session.query(DailyHealthMetrics).filter_by(user_id=USER_ID).all() for metric in metrics: # Simple PAI calculation based on activity # Real PAI requires continuous HR data, this is an approximation pai = 0.0 # Base PAI from steps (rough approximation) if metric.step_count: pai += metric.step_count / 1000 # 1 PAI per 1000 steps # Bonus from calories burned (beyond baseline) if metric.calories_burned and metric.calories_burned > 0: pai += metric.calories_burned / 50 # 1 PAI per 50 cal # Bonus from elevated heart rate activities if metric.avg_heart_rate and metric.avg_heart_rate > 100: intensity_bonus = (metric.avg_heart_rate - 100) / 10 pai += intensity_bonus metric.pai_score = round(min(pai, 100), 1) # Cap at 100 session.commit() print(f" Updated PAI scores for {len(metrics)} days.") def main(): """Main import function.""" print("=" * 60) print("Starting Zepp Data Import...") print("=" * 60) session = SessionLocal() try: base_export = get_latest_export_dir(EXPORT_DIR) if not base_export: print(f"No export directory found in {EXPORT_DIR}") return print(f"Using export directory: {base_export}\n") # Import all data types import_activity(session, base_export) import_sleep(session, base_export) import_heart_rate(session, base_export) import_heart_rate_auto(session, base_export) import_body(session, base_export) import_sport(session, base_export) # Calculate derived metrics calculate_pai_scores(session) print("\n" + "=" * 60) print("Import completed successfully!") print("=" * 60) except Exception as e: print(f"\nAn error occurred: {e}") import traceback traceback.print_exc() finally: session.close() if __name__ == "__main__": main()