103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
from io import BytesIO
|
|
|
|
import fitdecode
|
|
|
|
from backend.app.models.activity import Activity, DataPoint
|
|
|
|
|
|
def parse_fit_file(
|
|
file_content: bytes,
|
|
rider_id: uuid.UUID,
|
|
file_path: str,
|
|
) -> tuple[Activity, list[DataPoint]]:
|
|
"""Parse a .FIT file and return an Activity with its DataPoints."""
|
|
data_points: list[DataPoint] = []
|
|
session_data: dict = {}
|
|
|
|
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)
|
|
|
|
start_time = data_points[0].timestamp if data_points else datetime.now(timezone.utc)
|
|
end_time = data_points[-1].timestamp if data_points else start_time
|
|
duration = int((end_time - start_time).total_seconds()) if data_points else 0
|
|
|
|
activity = Activity(
|
|
rider_id=rider_id,
|
|
name=session_data.get("sport", "Ride"),
|
|
activity_type=session_data.get("sub_sport", "road"),
|
|
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
|
|
|
|
|
|
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."""
|
|
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"),
|
|
}
|
|
|
|
|
|
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)
|