Python: write optional SlipThreshold and WashThreshold on the session.

python/README.md

@@ -17,6 +17,9 @@ application UUID `89e86158-6d47-5c98-9d46-7d29437f27b9`.
 §5.1 still requires native `cadence` when rate is known; this library writes
 and reads both `cadence` and ID 93.
 
+`SlipThreshold` (94) and `WashThreshold` (95) are optional session fields, in
+newtons, written when set.
+
 **Not implemented** (later work, or blocked on draft defects):
 
 - Oarlock summary fields 11–18

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"),
 )

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"),

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]:

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))
 

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),
 )

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"