Spaces:
Running
Running
| import fitdecode | |
| import logging | |
| from typing import List, Dict, Any, Optional | |
| from pathlib import Path | |
| from datetime import datetime | |
| from .distance import compute_total_distance | |
| logger = logging.getLogger(__name__) | |
| def parse_fit_file(path: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| Parses a FIT file and returns a normalized dictionary of run data. | |
| """ | |
| records: List[Dict[str, Any]] = [] | |
| start_time: Optional[datetime] = None | |
| sport = "Running" # Default to running as it is the most common use case | |
| try: | |
| with fitdecode.FitReader(path) as fit: | |
| for frame in fit: | |
| if frame.frame_type == fitdecode.FIT_FRAME_DATA: | |
| if frame.name == "record": | |
| # Extract record fields safely | |
| def get_val(f, name): | |
| return f.get_value(name) if f.has_field(name) else None | |
| # Standard FIT SDK field names | |
| res_lat = get_val(frame, "position_lat") | |
| res_lon = get_val(frame, "position_long") | |
| lat = None | |
| lon = None | |
| if res_lat is not None and res_lon is not None: | |
| # Semicircles to degrees conversion | |
| lat = float(res_lat) * (180.0 / 2**31) | |
| lon = float(res_lon) * (180.0 / 2**31) | |
| # Extract metrics | |
| alt = get_val(frame, "enhanced_altitude") or get_val(frame, "altitude") | |
| dist = get_val(frame, "distance") | |
| hr = get_val(frame, "heart_rate") | |
| cad = get_val(frame, "cadence") | |
| speed = get_val(frame, "enhanced_speed") or get_val(frame, "speed") | |
| timestamp = get_val(frame, "timestamp") | |
| if timestamp: | |
| # Create a flat dictionary record | |
| record = { | |
| "time": timestamp, | |
| "lat": lat, | |
| "lon": lon, | |
| "altitude_m": float(alt) if alt is not None else None, | |
| "distance_m": float(dist) if dist is not None else None, | |
| "hr_bpm": int(hr) if hr is not None else None, | |
| "cadence_rpm": int(cad) if cad is not None else None, | |
| "speed_m_s": float(speed) if speed is not None else None, | |
| } | |
| records.append(record) | |
| elif frame.name == "session": | |
| if frame.has_field("sport") and frame.get_value("sport"): | |
| sport = frame.get_value("sport") | |
| if frame.has_field("start_time") and frame.get_value("start_time"): | |
| st = frame.get_value("start_time") | |
| if isinstance(st, datetime): | |
| start_time = st | |
| except Exception as e: | |
| logger.error(f"Failed to parse FIT file {path}: {e}", exc_info=True) | |
| return None | |
| if not records: | |
| logger.warning(f"No valid 'record' messages with timestamps found in {path}") | |
| return None | |
| # Normalization | |
| if start_time is None and records: | |
| start_time = records[0]["time"] | |
| sport_str = str(sport).lower() | |
| # Strict Running-Only Domain | |
| if "running" not in sport_str: | |
| logger.info(f"Skipping FIT activity: sport={sport_str} (not running)") | |
| return None | |
| normalized_sport = "Running" | |
| # Aggregate metrics | |
| total_dist = compute_total_distance(records) or 0.0 | |
| # Reject zero-distance | |
| if total_dist <= 0: | |
| logger.info(f"Skipping FIT activity {Path(path).stem}: zero distance") | |
| return None | |
| total_duration = 0.0 | |
| if len(records) > 1: | |
| dt = records[-1]["time"] - records[0]["time"] | |
| total_duration = float(dt.total_seconds()) | |
| return { | |
| "id": Path(path).stem, | |
| "sport": normalized_sport, | |
| "start_time": start_time, | |
| "total_distance_m": total_dist, | |
| "total_duration_s": total_duration, | |
| "records": records, | |
| "source_path": str(path), | |
| } | |