180 lines
6.3 KiB
Python
180 lines
6.3 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
from io import BytesIO
|
|
|
|
import fitdecode
|
|
|
|
from backend.app.models.activity import Activity, DataPoint
|
|
|
|
|
|
# Map FIT sport enum values to human-readable names
|
|
SPORT_NAMES = {
|
|
"cycling": "Cycling",
|
|
"running": "Running",
|
|
"swimming": "Swimming",
|
|
"training": "Strength",
|
|
"strength_training": "Strength",
|
|
"cardio_training": "Cardio",
|
|
"walking": "Walking",
|
|
"hiking": "Hiking",
|
|
"generic": "Workout",
|
|
}
|
|
|
|
|
|
def parse_fit_file(
|
|
file_content: bytes,
|
|
rider_id: uuid.UUID,
|
|
file_path: str,
|
|
) -> tuple[Activity, list[DataPoint], list[dict]]:
|
|
"""Parse a .FIT file and return an Activity with its DataPoints and exercise sets."""
|
|
data_points: list[DataPoint] = []
|
|
session_data: dict = {}
|
|
exercise_sets: list[dict] = []
|
|
current_exercise: str | None = None
|
|
|
|
with fitdecode.FitReader(BytesIO(file_content)) as fit:
|
|
for frame in fit:
|
|
if not isinstance(frame, fitdecode.FitDataMessage):
|
|
continue
|
|
|
|
if frame.name == "record":
|
|
dp = _parse_record(frame)
|
|
if dp:
|
|
data_points.append(dp)
|
|
|
|
elif frame.name == "session":
|
|
session_data = _parse_session(frame)
|
|
|
|
elif frame.name == "workout":
|
|
# Workout-level info (name etc.)
|
|
wkt_name = _get_field_str(frame, "wkt_name")
|
|
if wkt_name:
|
|
session_data.setdefault("workout_name", wkt_name)
|
|
|
|
elif frame.name == "exercise_title":
|
|
title = _get_field_str(frame, "exercise_name")
|
|
if title:
|
|
current_exercise = title
|
|
|
|
elif frame.name == "set":
|
|
ex_set = _parse_set(frame, current_exercise)
|
|
if ex_set:
|
|
exercise_sets.append(ex_set)
|
|
|
|
# Determine timing
|
|
sport = session_data.get("sport", "")
|
|
sport_name = SPORT_NAMES.get(sport.lower() if sport else "", sport or "Workout")
|
|
|
|
if data_points:
|
|
start_time = data_points[0].timestamp
|
|
end_time = data_points[-1].timestamp
|
|
duration = int((end_time - start_time).total_seconds())
|
|
else:
|
|
# No record data (strength, etc.) — use session timestamps
|
|
start_time = session_data.get("start_time") or datetime.now(timezone.utc)
|
|
elapsed = session_data.get("total_elapsed_time")
|
|
duration = int(elapsed) if elapsed else 0
|
|
end_time = start_time
|
|
|
|
name = session_data.get("workout_name") or sport_name
|
|
|
|
activity = Activity(
|
|
rider_id=rider_id,
|
|
name=name,
|
|
activity_type=session_data.get("sub_sport") or sport or "generic",
|
|
date=start_time,
|
|
duration=duration,
|
|
distance=session_data.get("total_distance"),
|
|
elevation_gain=session_data.get("total_ascent"),
|
|
file_path=file_path,
|
|
)
|
|
|
|
return activity, data_points, exercise_sets
|
|
|
|
|
|
def _parse_record(frame: fitdecode.FitDataMessage) -> DataPoint | None:
|
|
"""Parse a single record message into a DataPoint."""
|
|
timestamp = _get_field(frame, "timestamp")
|
|
if not timestamp:
|
|
return None
|
|
|
|
if isinstance(timestamp, datetime) and timestamp.tzinfo is None:
|
|
timestamp = timestamp.replace(tzinfo=timezone.utc)
|
|
|
|
return DataPoint(
|
|
timestamp=timestamp,
|
|
power=_get_field(frame, "power"),
|
|
heart_rate=_get_field(frame, "heart_rate"),
|
|
cadence=_get_field(frame, "cadence"),
|
|
speed=_get_field(frame, "speed"),
|
|
latitude=_semicircles_to_degrees(_get_field(frame, "position_lat")),
|
|
longitude=_semicircles_to_degrees(_get_field(frame, "position_long")),
|
|
altitude=_get_field(frame, "altitude"),
|
|
temperature=_get_field(frame, "temperature"),
|
|
)
|
|
|
|
|
|
def _parse_session(frame: fitdecode.FitDataMessage) -> dict:
|
|
"""Extract session-level data from FIT session message."""
|
|
start_time = _get_field(frame, "start_time")
|
|
if isinstance(start_time, datetime) and start_time.tzinfo is None:
|
|
start_time = start_time.replace(tzinfo=timezone.utc)
|
|
|
|
return {
|
|
"sport": _get_field_str(frame, "sport"),
|
|
"sub_sport": _get_field_str(frame, "sub_sport"),
|
|
"total_distance": _get_field(frame, "total_distance"),
|
|
"total_ascent": _get_field(frame, "total_ascent"),
|
|
"total_elapsed_time": _get_field(frame, "total_elapsed_time"),
|
|
"total_calories": _get_field(frame, "total_calories"),
|
|
"avg_heart_rate": _get_field(frame, "avg_heart_rate"),
|
|
"max_heart_rate": _get_field(frame, "max_heart_rate"),
|
|
"start_time": start_time,
|
|
}
|
|
|
|
|
|
def _parse_set(frame: fitdecode.FitDataMessage, exercise_name: str | None) -> dict | None:
|
|
"""Parse a set message from a strength/cardio workout."""
|
|
set_type = _get_field_str(frame, "set_type")
|
|
# set_type: 0=active, 1=rest
|
|
if set_type is not None and str(set_type) in ("1", "rest"):
|
|
return None # Skip rest sets
|
|
|
|
repetitions = _get_field(frame, "repetitions")
|
|
weight = _get_field(frame, "weight")
|
|
duration = _get_field(frame, "duration")
|
|
start_time = _get_field(frame, "start_time") or _get_field(frame, "timestamp")
|
|
category = _get_field_str(frame, "exercise_category")
|
|
exercise = _get_field_str(frame, "exercise_name")
|
|
|
|
return {
|
|
"exercise_name": exercise or exercise_name or category or "Unknown",
|
|
"exercise_category": category,
|
|
"repetitions": int(repetitions) if repetitions is not None else None,
|
|
"weight": round(float(weight), 1) if weight is not None else None,
|
|
"duration": round(float(duration), 1) if duration is not None else None,
|
|
"start_time": start_time.isoformat() if isinstance(start_time, datetime) else None,
|
|
}
|
|
|
|
|
|
def _get_field(frame: fitdecode.FitDataMessage, name: str):
|
|
"""Safely get a field value from a FIT frame."""
|
|
try:
|
|
field = frame.get_field(name)
|
|
return field.value if field else None
|
|
except KeyError:
|
|
return None
|
|
|
|
|
|
def _get_field_str(frame: fitdecode.FitDataMessage, name: str) -> str | None:
|
|
"""Get field value as string."""
|
|
val = _get_field(frame, name)
|
|
return str(val) if val is not None else None
|
|
|
|
|
|
def _semicircles_to_degrees(semicircles: int | None) -> float | None:
|
|
"""Convert Garmin semicircles to decimal degrees."""
|
|
if semicircles is None:
|
|
return None
|
|
return semicircles * (180.0 / 2**31)
|