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)