435 lines
15 KiB
Python
435 lines
15 KiB
Python
"""
|
|
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()
|