Python: define WorkoutState with the FIT intensity values.

python/README.md

@@ -33,7 +33,10 @@ newtons, written when set.
 - Oarlock summary fields 11–18
 - Dual-oarlock 200–211
 - In-stroke axis metadata 90–92, curve arrays, curve summaries, `.instroke.json`
-- WorkoutState enumeration (the draft has no value table; ID 9 is an opaque UINT8)
+
+`WorkoutState` (ID 9) uses the FIT intensity values: 0 Active, 1 Rest, 2 Warmup,
+3 Cooldown, 4 Recovery, 5 Interval, 6 Other. An unknown value is read as Other.
+When a lap's `intensity` disagrees with a record, the lap wins.
 
 Heart rate is the native `heart_rate` field. When no sensor is connected, omit
 it rather than write 0. `cycle_length16` is the distance the boat travels in

python/src/rowing_data/__init__.py

@@ -7,6 +7,7 @@
     PROTOCOL_VERSION,
     STANDARD_VERSION,
     StrokeState,
+    WorkoutState,
 )
 from .fields import (
     DEVELOPER_FIELDS,
@@ -34,6 +35,7 @@
     "RowingSession",
     "STANDARD_VERSION",
     "StrokeState",
+    "WorkoutState",
     "decode",
     "encode",
     "field_by_id",

python/src/rowing_data/constants.py

@@ -49,3 +49,15 @@ class StrokeState(IntEnum):
     DRIVE = 2
     DWELL = 3
     RECOVERY = 4
+
+
+class WorkoutState(IntEnum):
+    """Record developer field ID 9, the native FIT intensity values."""
+
+    ACTIVE = 0
+    REST = 1
+    WARMUP = 2
+    COOLDOWN = 3
+    RECOVERY = 4
+    INTERVAL = 5
+    OTHER = 6

python/src/rowing_data/fields.py

@@ -116,7 +116,10 @@ def _native(
         BaseType.UINT8,
         1,
         "",
-        notes="Opaque UINT8; the draft has no enum table.",
+        notes=(
+            "FIT intensity enum: 0 Active, 1 Rest, 2 Warmup, 3 Cooldown, "
+            "4 Recovery, 5 Interval, 6 Other."
+        ),
     ),
     _dev(10, "RecordingStrategy", MessageType.SESSION, BaseType.UINT8, 1, ""),
     _dev(94, "SlipThreshold", MessageType.SESSION, BaseType.UINT16, 1, "N"),

python/src/rowing_data/model.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 
-from .constants import RecordingStrategy, StrokeState
+from .constants import RecordingStrategy, StrokeState, WorkoutState
 from .strokes import resolve_stroke_rate, stroke_counts
 
 
@@ -32,7 +32,7 @@ class Record:
     average_drive_force: float | None = None
     peak_drive_force: float | None = None
     average_boat_speed_mps: float | None = None
-    workout_state: int | None = None
+    workout_state: WorkoutState | None = None
     stroke_work_j: int | None = None
     stroke_rate: float | None = None
     stroke_state: StrokeState | None = None
@@ -45,12 +45,19 @@ def resolved_stroke_rate(self) -> float | None:
             fractional_cadence=self.fractional_cadence,
         )
 
+    def resolved_workout_state(self, lap: Lap | None) -> WorkoutState | None:
+        """Lap intensity wins when the lap carries one."""
+        if lap is not None and lap.intensity is not None:
+            return lap.intensity
+        return self.workout_state
+
 
 @dataclass(frozen=True, slots=True)
 class Lap:
     start_time: datetime
     total_elapsed_s: float | None = None
     total_distance_m: float | None = None
+    intensity: WorkoutState | None = None
 
 
 @dataclass(frozen=True, slots=True)

python/src/rowing_data/reader.py

@@ -13,6 +13,7 @@
     PROTOCOL_VERSION,
     RecordingStrategy,
     StrokeState,
+    WorkoutState,
 )
 from .fields import FieldDef, field_by_id, field_by_name
 from .model import ATTR_BY_FIELD_ID, Lap, Record, RowingSession
@@ -63,21 +64,23 @@ def read_fit(path: str | Path) -> RowingSession:
         wash_threshold_n = _session_developer_int(message, our_fields, "WashThreshold")
         break
 
-    laps = _laps(messages)
+    laps, lap_issues = _laps(messages)
+    issues.extend(lap_issues)
     records_list: list[Record] = []
     for message in messages:
         if message.name != "record":
             continue
-        record, state_issue = _record(message, our_fields, laps)
+        record, record_issues = _record(message, our_fields, laps)
         records_list.append(record)
-        if state_issue is not None:
-            issues.append(
+        if record_issues:
+            issues.extend(
                 Issue(
-                    state_issue.level,
-                    state_issue.code,
-                    state_issue.message,
+                    item.level,
+                    item.code,
+                    item.message,
                     len(records_list) - 1,
                 )
+                for item in record_issues
             )
     records = tuple(records_list)
     return RowingSession(
@@ -234,8 +237,9 @@ def _as_datetime(value: object) -> datetime | None:
     return None
 
 
-def _laps(messages: list) -> list[Lap]:
+def _laps(messages: list) -> tuple[list[Lap], list[Issue]]:
     laps: list[Lap] = []
+    issues: list[Issue] = []
     for message in messages:
         if message.name != "lap":
             continue
@@ -246,15 +250,49 @@ def _laps(messages: list) -> list[Lap]:
             continue
         elapsed = message.get_value("total_elapsed_time")
         distance = message.get_value("total_distance")
+        intensity, issue = _as_workout_state(message.get_value("intensity"))
+        if issue is not None:
+            issues.append(issue)
         laps.append(
             Lap(
                 start_time=start,
                 total_elapsed_s=None if elapsed is None else float(elapsed),
                 total_distance_m=None if distance is None else float(distance),
+                intensity=intensity,
             )
         )
     laps.sort(key=lambda lap: lap.start_time)
-    return laps
+    return laps, issues
+
+
+def _as_workout_state(value: object) -> tuple[WorkoutState | None, Issue | None]:
+    if value is None:
+        return None, None
+    if isinstance(value, str):
+        try:
+            return WorkoutState[value.strip().upper()], None
+        except KeyError:
+            return WorkoutState.OTHER, Issue(
+                "warning",
+                "workout_state",
+                f"WorkoutState {value!r} is unknown; treating as Other",
+            )
+    try:
+        number = int(value)
+    except (TypeError, ValueError):
+        return WorkoutState.OTHER, Issue(
+            "warning",
+            "workout_state",
+            f"WorkoutState {value!r} is unknown; treating as Other",
+        )
+    try:
+        return WorkoutState(number), None
+    except ValueError:
+        return WorkoutState.OTHER, Issue(
+            "warning",
+            "workout_state",
+            f"WorkoutState {number} is above 6; treating as Other",
+        )
 
 
 def _lap_index(timestamp: datetime, laps: list[Lap]) -> int | None:
@@ -271,7 +309,7 @@ def _lap_index(timestamp: datetime, laps: list[Lap]) -> int | None:
 
 def _record(
     message, our_fields: dict[str, FieldDef], laps: list[Lap]
-) -> tuple[Record, Issue | None]:
+) -> tuple[Record, list[Issue]]:
     timestamp = _as_datetime(message.get_value("timestamp"))
     if timestamp is None:
         raise ValueError("record message is missing timestamp")
@@ -312,7 +350,13 @@ def _record(
             continue
         kwargs[attr] = decoded
 
-    return Record(**kwargs), stroke_issue  # type: ignore[arg-type]
+    record_issues = [stroke_issue] if stroke_issue is not None else []
+    if kwargs.get("workout_state") is not None:
+        parsed, state_issue = _as_workout_state(kwargs["workout_state"])
+        kwargs["workout_state"] = parsed
+        if state_issue is not None:
+            record_issues.append(state_issue)
+    return Record(**kwargs), record_issues  # type: ignore[arg-type]
 
 
 def _as_stroke_state(value: object) -> tuple[StrokeState | None, Issue | None]:

python/src/rowing_data/validation.py

@@ -71,6 +71,22 @@ def _record_issues(record: Record, index: int, session: RowingSession) -> list[I
         except CodecError as exc:
             issues.append(Issue("error", "overflow", str(exc), index))
 
+    lap = _lap_for(record, session)
+    if (
+        lap is not None
+        and lap.intensity is not None
+        and record.workout_state is not None
+        and record.workout_state != lap.intensity
+    ):
+        issues.append(
+            Issue(
+                "warning",
+                "workout_state",
+                "WorkoutState disagrees with lap intensity; the lap wins",
+                index,
+            )
+        )
+
     if (
         record.stroke_state is not None
         and session.recording_strategy is not RecordingStrategy.TIME_SAMPLED
@@ -111,3 +127,11 @@ def _record_issues(record: Record, index: int, session: RowingSession) -> list[I
                 )
             )
     return issues
+
+
+def _lap_for(record: Record, session: RowingSession):
+    if record.lap_index is None:
+        return None
+    if record.lap_index < 0 or record.lap_index >= len(session.laps):
+        return None
+    return session.laps[record.lap_index]

python/src/rowing_data/writer.py

@@ -22,6 +22,7 @@
     Event,
     EventType,
     FileType,
+    Intensity,
     Manufacturer,
     Sport,
     SubSport,
@@ -33,6 +34,7 @@
     CYCLE_LENGTH16_SCALE,
     PROTOCOL_VERSION,
     RecordingStrategy,
+    WorkoutState,
 )
 from .fields import BaseType, FieldDef, field_by_id, field_by_name
 from .model import (
@@ -132,6 +134,9 @@ def write_fit(session: RowingSession, path: str | Path) -> None:
                 lap_msg.total_timer_time = lap.total_elapsed_s
             if lap.total_distance_m is not None:
                 lap_msg.total_distance = lap.total_distance_m
+            lap_msg.intensity = Intensity(
+                WorkoutState.ACTIVE if lap.intensity is None else lap.intensity
+            )
             lap_msg.sport = Sport.ROWING
             builder.add(lap_msg)
     else:
@@ -142,6 +147,7 @@ def write_fit(session: RowingSession, path: str | Path) -> None:
         lap_msg.total_elapsed_time = elapsed_s
         lap_msg.total_timer_time = elapsed_s
         lap_msg.total_distance = total_distance
+        lap_msg.intensity = Intensity.ACTIVE
         lap_msg.sport = Sport.ROWING
         builder.add(lap_msg)
 

python/tests/test_model.py

@@ -6,7 +6,8 @@
 
 import pytest
 
-from rowing_data.constants import RecordingStrategy
+from rowing_data import validate
+from rowing_data.constants import RecordingStrategy, WorkoutState
 from rowing_data.model import Lap, Record, RowingSession
 
 
@@ -30,6 +31,31 @@ def test_session_start_and_stroke_counts() -> None:
     assert session.recording_strategy is RecordingStrategy.GPS_UPDATE
 
 
+def test_workout_state_enum_matches_fit_intensity() -> None:
+    assert WorkoutState.ACTIVE == 0
+    assert WorkoutState.REST == 1
+    assert WorkoutState.WARMUP == 2
+    assert WorkoutState.COOLDOWN == 3
+    assert WorkoutState.RECOVERY == 4
+    assert WorkoutState.INTERVAL == 5
+    assert WorkoutState.OTHER == 6
+
+
+def test_lap_intensity_wins_when_workout_state_disagrees() -> None:
+    record = Record(
+        timestamp=_ts(0),
+        workout_state=WorkoutState.ACTIVE,
+        lap_index=0,
+    )
+    lap = Lap(start_time=_ts(0), intensity=WorkoutState.REST)
+    session = RowingSession(records=(record,), laps=(lap,))
+    assert record.resolved_workout_state(lap) is WorkoutState.REST
+    issues = validate(session)
+    assert any(
+        issue.code == "workout_state" and issue.level == "warning" for issue in issues
+    )
+
+
 def test_empty_session_without_start_time_raises() -> None:
     session = RowingSession(records=())
     with pytest.raises(ValueError, match="no records"):