From 42a9221d91b67aa85ba379d662173973b485c3a4 Mon Sep 17 00:00:00 2001 From: Sander Roosendaal Date: Wed, 7 Oct 2026 09:35:17 +0200 Subject: [PATCH] Python: write optional SlipThreshold and WashThreshold on the session. The force thresholds are device settings, so they are stored once rather than on every stroke. Co-authored-by: Cursor --- python/README.md | 3 +++ python/src/rowing_data/fields.py | 2 ++ python/src/rowing_data/model.py | 8 ++++++++ python/src/rowing_data/reader.py | 21 ++++++++++++++++++++ python/src/rowing_data/writer.py | 33 ++++++++++++++++++++++++++++---- python/tests/test_fields.py | 2 ++ python/tests/test_roundtrip.py | 13 +++++++++++++ 7 files changed, 78 insertions(+), 4 deletions(-) diff --git a/python/README.md b/python/README.md index 2d4bfaa..7e07a51 100644 --- a/python/README.md +++ b/python/README.md @@ -18,6 +18,9 @@ That is conformance **Levels 1–2** as in §7, plus StrokeRate (93). §7’s Le bullet list omits 93; §5.1 still requires native `cadence` when rate is known, so this library writes and reads ID 93. +`SlipThreshold` (94) and `WashThreshold` (95) are optional session fields, in +newtons, written when set. + **Not implemented** (Level 3–4, or blocked on draft defects): - Oarlock summary fields 11–18 diff --git a/python/src/rowing_data/fields.py b/python/src/rowing_data/fields.py index f8d2720..6bb5c5a 100644 --- a/python/src/rowing_data/fields.py +++ b/python/src/rowing_data/fields.py @@ -139,6 +139,8 @@ def _native( notes="Opaque UINT8; the draft has no enum table.", ), _dev(10, "RecordingStrategy", MessageType.SESSION, BaseType.UINT8, 1, ""), + _dev(94, "SlipThreshold", MessageType.SESSION, BaseType.UINT16, 1, "N"), + _dev(95, "WashThreshold", MessageType.SESSION, BaseType.UINT16, 1, "N"), _dev(19, "StrokeWork", MessageType.RECORD, BaseType.UINT16, 1, "J"), _dev(93, "StrokeRate", MessageType.RECORD, BaseType.UINT16, 100, "spm"), ) diff --git a/python/src/rowing_data/model.py b/python/src/rowing_data/model.py index 81fb6dc..4a21aba 100644 --- a/python/src/rowing_data/model.py +++ b/python/src/rowing_data/model.py @@ -60,6 +60,8 @@ class RowingSession: recording_strategy: RecordingStrategy = RecordingStrategy.UNKNOWN laps: tuple[Lap, ...] = field(default_factory=tuple) start_time: datetime | None = None + slip_threshold_n: int | None = None + wash_threshold_n: int | None = None # Warnings collected by read_fit (invalid RecordingStrategy, pre-v1.2 scales). read_issues: tuple = () @@ -92,6 +94,12 @@ def stroke_counts(self) -> list[int]: ATTR_BY_FIELD_ID = dict(RECORD_DEVELOPER_ATTRS) +# Optional session developer fields. RecordingStrategy is always written. +SESSION_DEVELOPER_ATTRS: Sequence[tuple[int, str]] = ( + (94, "slip_threshold_n"), + (95, "wash_threshold_n"), +) + # Record attributes that map to native FIT fields in spec §4. NATIVE_RECORD_ATTRS: Sequence[tuple[str, str]] = ( ("distance_m", "distance"), diff --git a/python/src/rowing_data/reader.py b/python/src/rowing_data/reader.py index bec34d1..b96ac5b 100644 --- a/python/src/rowing_data/reader.py +++ b/python/src/rowing_data/reader.py @@ -42,6 +42,8 @@ def read_fit(path: str | Path) -> RowingSession: strategy = RecordingStrategy.UNKNOWN start_time: datetime | None = None + slip_threshold_n: int | None = None + wash_threshold_n: int | None = None for message in messages: if message.name != "session": continue @@ -51,6 +53,8 @@ def read_fit(path: str | Path) -> RowingSession: strategy, strategy_issue = _recording_strategy(message, our_fields) if strategy_issue is not None: issues.append(strategy_issue) + slip_threshold_n = _session_developer_int(message, our_fields, "SlipThreshold") + wash_threshold_n = _session_developer_int(message, our_fields, "WashThreshold") break laps = _laps(messages) @@ -64,6 +68,8 @@ def read_fit(path: str | Path) -> RowingSession: recording_strategy=strategy, laps=tuple(laps), start_time=start_time, + slip_threshold_n=slip_threshold_n, + wash_threshold_n=wash_threshold_n, read_issues=tuple(issues), ) @@ -136,6 +142,21 @@ def _our_developer_fields( return by_name, issues +def _session_developer_int( + message, our_fields: dict[str, FieldDef], name: str +) -> int | None: + spec = our_fields.get(name) + if spec is None: + return None + field = message.get(name) + if field is None or field.raw_value is None: + return None + if isinstance(field.raw_value, list | tuple): + return None + decoded = decode(spec, int(field.raw_value)) + return None if decoded is None else int(decoded) + + def _recording_strategy( message, our_fields: dict[str, FieldDef] ) -> tuple[RecordingStrategy, Issue | None]: diff --git a/python/src/rowing_data/writer.py b/python/src/rowing_data/writer.py index 3ab34dc..27e688b 100644 --- a/python/src/rowing_data/writer.py +++ b/python/src/rowing_data/writer.py @@ -30,7 +30,12 @@ from .codec import encode from .constants import APPLICATION_ID, CYCLE_LENGTH16_SCALE from .fields import BaseType, FieldDef, field_by_id, field_by_name -from .model import RECORD_DEVELOPER_ATTRS, Record, RowingSession +from .model import ( + RECORD_DEVELOPER_ATTRS, + SESSION_DEVELOPER_ATTRS, + Record, + RowingSession, +) from .strokes import native_cadence_parts _DEV_INDEX = 0 @@ -68,6 +73,7 @@ def write_fit(session: RowingSession, path: str | Path) -> None: total_distance = _last_distance(session) record_fields = _record_developer_fields_used(session) + session_fields = _optional_session_fields(session) builder = FitFileBuilder(auto_define=True, min_string_size=64) file_id = FileIdMessage() @@ -93,9 +99,12 @@ def write_fit(session: RowingSession, path: str | Path) -> None: event_start.timestamp = start_ms builder.add(event_start) - _add_developer_definitions(builder, record_fields) + _add_developer_definitions(builder, record_fields, session_fields) - session_msg = SessionMessage(developer_fields=[_recording_strategy_field(session)]) + session_dev = [_recording_strategy_field(session)] + for field, value in session_fields: + session_dev.append(_developer_field(field, value)) + session_msg = SessionMessage(developer_fields=session_dev) session_msg.message_index = 0 session_msg.timestamp = start_ms session_msg.start_time = start_ms @@ -164,14 +173,30 @@ def _record_developer_fields_used(session: RowingSession) -> tuple[FieldDef, ... return tuple(used) +def _optional_session_fields( + session: RowingSession, +) -> tuple[tuple[FieldDef, int], ...]: + used: list[tuple[FieldDef, int]] = [] + for field_id, attr in SESSION_DEVELOPER_ATTRS: + value = getattr(session, attr) + if value is None: + continue + used.append((field_by_id(field_id), int(value))) + return tuple(used) + + def _add_developer_definitions( - builder: FitFileBuilder, record_fields: tuple[FieldDef, ...] + builder: FitFileBuilder, + record_fields: tuple[FieldDef, ...], + session_fields: tuple[tuple[FieldDef, int], ...] = (), ) -> None: dev_id = DeveloperDataIdMessage() dev_id.application_id = APPLICATION_ID dev_id.developer_data_index = _DEV_INDEX builder.add(dev_id) builder.add(_field_description(field_by_id(10))) + for field, _value in session_fields: + builder.add(_field_description(field)) for field in record_fields: builder.add(_field_description(field)) diff --git a/python/tests/test_fields.py b/python/tests/test_fields.py index 658f8b0..2c75a29 100644 --- a/python/tests/test_fields.py +++ b/python/tests/test_fields.py @@ -30,6 +30,8 @@ (8, "AverageBoatSpeed", MessageType.RECORD, BaseType.UINT16, 255, "m/s", False), (9, "WorkoutState", MessageType.RECORD, BaseType.UINT8, 1, "", False), (10, "RecordingStrategy", MessageType.SESSION, BaseType.UINT8, 1, "", False), + (94, "SlipThreshold", MessageType.SESSION, BaseType.UINT16, 1, "N", False), + (95, "WashThreshold", MessageType.SESSION, BaseType.UINT16, 1, "N", False), (19, "StrokeWork", MessageType.RECORD, BaseType.UINT16, 1, "J", False), (93, "StrokeRate", MessageType.RECORD, BaseType.UINT16, 100, "spm", False), ) diff --git a/python/tests/test_roundtrip.py b/python/tests/test_roundtrip.py index 2d334be..63941cb 100644 --- a/python/tests/test_roundtrip.py +++ b/python/tests/test_roundtrip.py @@ -29,6 +29,19 @@ def test_roundtrip_stroke_boundary(tmp_path: Path) -> None: assert validate(loaded) == [] +def test_roundtrip_slip_and_wash_thresholds(tmp_path: Path) -> None: + original = RowingSession( + records=(Record(timestamp=datetime(2026, 1, 1, tzinfo=UTC)),), + slip_threshold_n=80, + wash_threshold_n=120, + ) + path = tmp_path / "thresholds.fit" + write_fit(original, path) + loaded = read_fit(path) + assert loaded.slip_threshold_n == 80 + assert loaded.wash_threshold_n == 120 + + def test_roundtrip_gps_update(tmp_path: Path) -> None: original = gps_update_session() path = tmp_path / "gps.fit"