Davide Scaini
82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
"""Core data models for the extract stage.
|
|
|
|
ParsedActivity is the internal representation produced by parsers.
|
|
It gets fed into metrics computation and the BAS JSON writer.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
# Any timestamp before this is almost certainly an uninitialised sensor value
|
|
# (epoch 0, FIT "no-data" sentinel, RTC not yet synced, etc.).
|
|
_MIN_TIMESTAMP = datetime(2000, 1, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
def strip_bogus_leading_points(points: list["DataPoint"]) -> list["DataPoint"]:
|
|
"""Drop leading points whose timestamp predates the year 2000.
|
|
|
|
FIT files occasionally emit a record with timestamp=0 (or another
|
|
pre-2000 value) as an uninitialised sentinel before the real data
|
|
begins. Keeping such a point as points[0] produces a 1970 start
|
|
time and an absurdly large duration_s.
|
|
"""
|
|
i = 0
|
|
while i < len(points) and points[i].timestamp < _MIN_TIMESTAMP:
|
|
i += 1
|
|
return points[i:]
|
|
|
|
|
|
@dataclass
|
|
class DataPoint:
|
|
"""One measurement sample from a GPS/sensor recording."""
|
|
|
|
timestamp: datetime
|
|
lat: Optional[float] = None
|
|
lon: Optional[float] = None
|
|
elevation_m: Optional[float] = None
|
|
hr_bpm: Optional[int] = None
|
|
cadence_rpm: Optional[int] = None
|
|
# Speed from device (km/h). May be absent; we compute it from GPS if so.
|
|
speed_kmh: Optional[float] = None
|
|
power_w: Optional[int] = None
|
|
temperature_c: Optional[float] = None
|
|
# Cumulative distance from device (metres), if recorded.
|
|
distance_m: Optional[float] = None
|
|
|
|
|
|
@dataclass
|
|
class LapData:
|
|
index: int
|
|
started_at: datetime
|
|
duration_s: Optional[int] = None
|
|
distance_m: Optional[float] = None
|
|
elevation_gain_m: Optional[float] = None
|
|
avg_speed_kmh: Optional[float] = None
|
|
avg_hr_bpm: Optional[int] = None
|
|
avg_power_w: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class ParsedActivity:
|
|
"""Raw activity data as produced by a parser, before metric computation."""
|
|
|
|
points: list[DataPoint]
|
|
sport: str # normalised to BAS sport enum
|
|
started_at: datetime
|
|
source_file: str # basename of original file
|
|
source_hash: str # "sha256:{hex}"
|
|
|
|
sub_sport: Optional[str] = None
|
|
device: Optional[str] = None
|
|
title: Optional[str] = None
|
|
description: Optional[str] = None
|
|
gear: Optional[str] = None
|
|
strava_id: Optional[str] = None
|
|
privacy: Optional[str] = None # "public", "private", or None (caller decides)
|
|
laps: list[LapData] = field(default_factory=list)
|
|
# "barometric" = device has a barometric altimeter (FIT enhanced_altitude present)
|
|
# "gps" = altitude derived from GPS triangulation (GPX, TCX, FIT altitude-only)
|
|
# "unknown" = source not detected (treated as gps for threshold purposes)
|
|
altitude_source: str = "unknown"
|