feat: Strava support — bulk-export .gz/.tcx import fix + .tcx parser; full Strava API OAuth live sync (activities via streams) with Profile connect UI; colour gym/no-GPS sports red

backend/app/services/fit_parser.py

@@ -406,6 +406,143 @@ def parse_gpx_file(filepath: str) -> dict:
     }
 
 
+def parse_tcx_file(filepath: str) -> dict:
+    """Parse a Garmin Training Center XML (.tcx) activity file.
+
+    Strava bulk exports include older device uploads as .tcx (often gzipped).
+    TCX is namespaced XML; we match by local tag name so the parser is robust to
+    the various TCX namespace declarations in the wild. Output mirrors
+    parse_gpx_file so downstream ingestion is identical."""
+    import xml.etree.ElementTree as ET
+
+    def local(tag: str) -> str:
+        return tag.split("}")[-1] if "}" in tag else tag
+
+    def find(el, name):
+        for child in el.iter():
+            if local(child.tag) == name:
+                return child
+        return None
+
+    def findall(el, name):
+        return [c for c in el.iter() if local(c.tag) == name]
+
+    def child_text(el, name):
+        for c in list(el):
+            if local(c.tag) == name and c.text:
+                return c.text.strip()
+        return None
+
+    tree = ET.parse(filepath)
+    root = tree.getroot()
+
+    activity_el = find(root, "Activity")
+    sport_raw = (activity_el.get("Sport") if activity_el is not None else None) or "Other"
+    sport = {"running": "running", "biking": "cycling",
+             "walking": "walking", "hiking": "hiking"}.get(sport_raw.lower(), sport_raw.lower())
+
+    data_points = []
+    for tp in findall(root, "Trackpoint"):
+        ts_str = child_text(tp, "Time")
+        ts = None
+        if ts_str:
+            try:
+                ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
+                if ts.tzinfo is None:
+                    ts = ts.replace(tzinfo=timezone.utc)
+            except ValueError:
+                ts = None
+
+        lat = lng = None
+        pos = next((c for c in list(tp) if local(c.tag) == "Position"), None)
+        if pos is not None:
+            lat = _safe_float(child_text(pos, "LatitudeDegrees"))
+            lng = _safe_float(child_text(pos, "LongitudeDegrees"))
+
+        hr = None
+        hr_el = next((c for c in list(tp) if local(c.tag) == "HeartRateBpm"), None)
+        if hr_el is not None:
+            hr = _safe_float(child_text(hr_el, "Value"))
+
+        # Speed/Watts live in a TPX extension; search descendants by local name.
+        speed = watts = None
+        for ext in findall(tp, "Speed"):
+            speed = _safe_float(ext.text)
+            break
+        for ext in findall(tp, "Watts"):
+            watts = _safe_float(ext.text)
+            break
+
+        data_points.append({
+            "timestamp": ts.isoformat() if ts else None,
+            "latitude": lat, "longitude": lng,
+            "altitude_m": _safe_float(child_text(tp, "AltitudeMeters")),
+            "heart_rate": hr,
+            "cadence": _safe_float(child_text(tp, "Cadence")),
+            "speed_ms": speed,
+            "power": watts,
+            "temperature_c": None,
+            "distance_m": _safe_float(child_text(tp, "DistanceMeters")),
+        })
+
+    coords = [(p["latitude"], p["longitude"]) for p in data_points if p["latitude"] and p["longitude"]]
+    encoded_polyline = polyline_lib.encode(coords) if coords else None
+    bounding_box = _bounding_box(coords)
+
+    # Distance: prefer the cumulative DistanceMeters from the file; fall back to
+    # haversine over GPS points when absent (some TCX trackpoints omit it).
+    dist_vals = [p["distance_m"] for p in data_points if p["distance_m"] is not None]
+    if dist_vals:
+        total_dist = max(dist_vals)
+    else:
+        total_dist = 0.0
+        prev = None
+        for p in data_points:
+            if p["latitude"] and p["longitude"]:
+                if prev:
+                    R = 6371000
+                    phi1, phi2 = math.radians(prev[0]), math.radians(p["latitude"])
+                    dphi = math.radians(p["latitude"] - prev[0])
+                    dlam = math.radians(p["longitude"] - prev[1])
+                    a = math.sin(dphi/2)**2 + math.cos(phi1)*math.cos(phi2)*math.sin(dlam/2)**2
+                    total_dist += 2 * R * math.asin(math.sqrt(a))
+                prev = (p["latitude"], p["longitude"])
+            p["distance_m"] = total_dist
+
+    uphill, downhill = 0.0, 0.0
+    alts = [p["altitude_m"] for p in data_points if p["altitude_m"] is not None]
+    for i in range(1, len(alts)):
+        diff = alts[i] - alts[i-1]
+        if diff > 0: uphill += diff
+        else: downhill += abs(diff)
+
+    hrs = [p["heart_rate"] for p in data_points if p["heart_rate"]]
+    start_time_str = next((p["timestamp"] for p in data_points if p["timestamp"]), None)
+    last_time_str = next((p["timestamp"] for p in reversed(data_points) if p["timestamp"]), None)
+    start_dt = datetime.fromisoformat(start_time_str) if start_time_str else None
+    end_dt = datetime.fromisoformat(last_time_str) if last_time_str else None
+    duration = (end_dt - start_dt).total_seconds() if (start_dt and end_dt) else None
+    avg_speed = (total_dist / duration) if (total_dist and duration) else None
+
+    return {
+        "name": f"{sport.title()} {start_dt.date() if start_dt else ''}".strip(),
+        "sport_type": sport, "start_time": start_time_str,
+        "distance_m": total_dist or None, "duration_s": duration, "moving_time_s": None,
+        "elevation_gain_m": uphill, "elevation_loss_m": downhill,
+        "avg_heart_rate": (sum(hrs) / len(hrs)) if hrs else None,
+        "max_heart_rate": max(hrs) if hrs else None,
+        "avg_cadence": None, "avg_power": None, "normalized_power": None,
+        "avg_speed_ms": avg_speed,
+        "max_speed_ms": None, "avg_temperature_c": None, "calories": None,
+        "training_stress_score": None, "vo2max_estimate": None,
+        "polyline": encoded_polyline, "bounding_box": bounding_box,
+        "active_spans": _active_spans(data_points),
+        "source_type": "tcx",
+        "rejected_reason": _vehicle_reason(sport, avg_speed, total_dist, duration),
+        "data_points": data_points, "laps": [],
+    }
+
+
 def calculate_hr_zones(data_points: list, user_max_hr: float) -> dict:
     if not user_max_hr or user_max_hr < 100:
         return {}