feat(sound): add WHO 16 media transport commands and empirical conformance fixtures

OWNd/message/sound.py

@@ -29,6 +29,14 @@ def __init__(self, data: str) -> None:
         self._environment = self._zone[1] if self._is_routing_event else None
         self._routed_source = self._zone[2] if self._is_routing_event else None
         self._volume: int | None = None
+        self._frequency_khz: int | None = None
+        self._track: int | None = None
+        self._rds_text: str | None = None
+        self._is_source_busy: bool = False
+        self._track_step_forward: int | None = None
+        self._track_step_backward: int | None = None
+        self._is_seek_up: bool = False
+        self._is_seek_down: bool = False
 
         if self._is_source_event:
             subject = f"Audio Source {self._source_id}"
@@ -43,6 +51,25 @@ def __init__(self, data: str) -> None:
             self._human_readable_log = f"{subject} is switched ON."
         elif self._state in (10, 13):
             self._human_readable_log = f"{subject} is switched OFF."
+        elif self._state == 100:
+            self._is_source_busy = True
+            self._human_readable_log = f"{subject} is BUSY."
+        elif self._state == 5000:
+            self._is_seek_up = True
+            self._human_readable_log = f"{subject} seek forward."
+        elif self._state == 5100:
+            self._is_seek_down = True
+            self._human_readable_log = f"{subject} seek backward."
+        elif self._state is not None and 6001 <= self._state <= 6015:
+            self._track_step_forward = self._state - 6000
+            self._human_readable_log = (
+                f"{subject} advance track/station by {self._track_step_forward} step(s)."
+            )
+        elif self._state is not None and 6101 <= self._state <= 6115:
+            self._track_step_backward = self._state - 6100
+            self._human_readable_log = (
+                f"{subject} return track/station by {self._track_step_backward} step(s)."
+            )
         elif self._state is not None:
             self._human_readable_log = f"{subject} received command: {self._state}."
         elif self._dimension == 1 and self._dimension_value:
@@ -53,6 +80,47 @@ def __init__(self, data: str) -> None:
             self._human_readable_log = (
                 f"Audio zone {self._zone} volume is set to {self._volume}."
             )
+        elif self._dimension == 6 and self._dimension_value:
+            try:
+                self._frequency_khz = int(self._dimension_value[-1])
+            except ValueError:
+                return
+            if self._frequency_khz <= 0:
+                self._frequency_khz = None
+                return
+            self._human_readable_log = (
+                f"{subject} frequency is {self._frequency_khz} kHz."
+            )
+        elif self._dimension == 7 and self._dimension_value:
+            try:
+                self._track = int(self._dimension_value[-1])
+            except ValueError:
+                return
+            if self._track < 1:
+                self._track = None
+                return
+            self._human_readable_log = (
+                f"{subject} station/track is {self._track}."
+            )
+        elif self._dimension == 8 and self._dimension_value:
+            # Dimension 8 carries the RDS PS (Program Service) name as exactly 8 decimal
+            # ASCII character codes per the WHO 16 specification. Frames with other
+            # lengths are ignored as malformed bus artifacts.
+            if len(self._dimension_value) != 8:
+                return
+            chars = []
+            for code in self._dimension_value:
+                try:
+                    val = int(code)
+                    if 32 <= val <= 126:
+                        chars.append(chr(val))
+                except ValueError:
+                    continue
+            text = "".join(chars).strip()
+            self._rds_text = text if text else None
+            self._human_readable_log = (
+                f"{subject} RDS text: '{self._rds_text or ''}'."
+            )
 
     @property
     def is_on(self) -> bool:
@@ -94,6 +162,38 @@ def zone(self) -> str:
     def volume(self) -> int | None:
         return self._volume
 
+    @property
+    def frequency_khz(self) -> int | None:
+        return self._frequency_khz
+
+    @property
+    def track(self) -> int | None:
+        return self._track
+
+    @property
+    def rds_text(self) -> str | None:
+        return self._rds_text
+
+    @property
+    def is_source_busy(self) -> bool:
+        return self._is_source_busy
+
+    @property
+    def track_step_forward(self) -> int | None:
+        return self._track_step_forward
+
+    @property
+    def track_step_backward(self) -> int | None:
+        return self._track_step_backward
+
+    @property
+    def is_seek_up(self) -> bool:
+        return self._is_seek_up
+
+    @property
+    def is_seek_down(self) -> bool:
+        return self._is_seek_down
+
 
 
 class OWNAVCommand(OWNCommand):
@@ -215,6 +315,121 @@ def set_volume(cls, where: str | int, volume: int | str) -> OWNSoundCommand:
         )
         return message
 
+    @classmethod
+    def next_track(cls, where: str | int, step: int = 1) -> OWNSoundCommand:
+        """Advance track or station forward (1..15 steps)."""
+        s = int(step)
+        if not 1 <= s <= 15:
+            raise ValueError(f"step must be between 1 and 15, got {step}")
+        what = 6000 + s
+        message = cls(f"*16*{what}*{where}##")
+        message._human_readable_log = (
+            f"Advancing track/station on {where} by {s} step(s)."
+        )
+        return message
+
+    @classmethod
+    def previous_track(cls, where: str | int, step: int = 1) -> OWNSoundCommand:
+        """Return track or station backward (1..15 steps)."""
+        s = int(step)
+        if not 1 <= s <= 15:
+            raise ValueError(f"step must be between 1 and 15, got {step}")
+        what = 6100 + s
+        message = cls(f"*16*{what}*{where}##")
+        message._human_readable_log = (
+            f"Returning track/station on {where} by {s} step(s)."
+        )
+        return message
+
+    @classmethod
+    def seek_up(cls, where: str | int) -> OWNSoundCommand:
+        """Seek forward to next receivable FM station."""
+        message = cls(f"*16*5000*{where}##")
+        message._human_readable_log = f"Seeking forward on {where}."
+        return message
+
+    @classmethod
+    def seek_down(cls, where: str | int) -> OWNSoundCommand:
+        """Seek backward to previous receivable FM station."""
+        message = cls(f"*16*5100*{where}##")
+        message._human_readable_log = f"Seeking backward on {where}."
+        return message
+
+    @classmethod
+    def select_track(cls, where: str | int, track: int) -> OWNSoundCommand:
+        """Select stored station or CD/track number.
+
+        Note:
+            For tuner sources (radio), the stored station preset range is 1–5
+            per the OpenWebNet specification (or up to 15 on extended tuners such
+            as F500N). For CD or external media sources, track numbers can be
+            greater (>= 1).
+        """
+        t = int(track)
+        if t < 1:
+            raise ValueError(f"track must be greater than or equal to 1, got {track}")
+        message = cls(f"*#16*{where}*#7*{t}##")
+        message._human_readable_log = (
+            f"Selecting station/track {t} on {where}."
+        )
+        return message
+
+    @classmethod
+    def request_track(cls, where: str | int) -> OWNSoundCommand:
+        """Request currently playing station or track number."""
+        message = cls(f"*#16*{where}*7##")
+        message._human_readable_log = f"Requesting station/track of {where}."
+        return message
+
+    @classmethod
+    def set_frequency(
+        cls, where: str | int, kilohertz: int | float
+    ) -> OWNSoundCommand:
+        """Tune tuner source to frequency in kHz (leading 0 parameter).
+
+        The value is sent without zero padding: an F500N on an MH200N
+        accepted `*#16*101*#6*0*96200##` (MyHOME#427).
+
+        Examples:
+            `107000` tunes to 107.00 MHz FM, `96200` to 96.20 MHz.
+        """
+        khz = int(round(float(kilohertz)))
+        if khz <= 0:
+            raise ValueError(f"frequency in kHz must be positive, got {kilohertz}")
+        message = cls(f"*#16*{where}*#6*0*{khz}##")
+        message._human_readable_log = (
+            f"Tuning {where} to {khz} kHz."
+        )
+        return message
+
+    @classmethod
+    def request_frequency(cls, where: str | int) -> OWNSoundCommand:
+        """Request tuned frequency."""
+        message = cls(f"*#16*{where}*6##")
+        message._human_readable_log = f"Requesting frequency of {where}."
+        return message
+
+    @classmethod
+    def start_rds(cls, where: str | int) -> OWNSoundCommand:
+        """Start autonomous RDS station name reporting."""
+        message = cls(f"*16*101*{where}##")
+        message._human_readable_log = f"Starting RDS reporting on {where}."
+        return message
+
+    @classmethod
+    def stop_rds(cls, where: str | int) -> OWNSoundCommand:
+        """Stop autonomous RDS station name reporting."""
+        message = cls(f"*16*102*{where}##")
+        message._human_readable_log = f"Stopping RDS reporting on {where}."
+        return message
+
+    @classmethod
+    def request_rds(cls, where: str | int) -> OWNSoundCommand:
+        """Request RDS station text."""
+        message = cls(f"*#16*{where}*8##")
+        message._human_readable_log = f"Requesting RDS text of {where}."
+        return message
+
 
 register_event_parser(16, OWNSoundEvent)
 register_command_parser(16, OWNSoundCommand)

tests/test_sound.py

@@ -155,3 +155,123 @@ def test_non_routing_events_have_no_environment(frame: str) -> None:
     assert event.environment is None
     assert event.routed_source is None
     assert event.zone == event.where
+
+
+def test_who16_sound_source_commands() -> None:
+    """Verify WHO 16 sound source command builders and range validation."""
+    assert str(OWNSoundCommand.next_track("101")) == "*16*6001*101##"
+    assert str(OWNSoundCommand.next_track(102, 5)) == "*16*6005*102##"
+    assert str(OWNSoundCommand.previous_track("101")) == "*16*6101*101##"
+    assert str(OWNSoundCommand.previous_track(102, 3)) == "*16*6103*102##"
+    assert str(OWNSoundCommand.seek_up("101")) == "*16*5000*101##"
+    assert str(OWNSoundCommand.seek_down("101")) == "*16*5100*101##"
+    assert str(OWNSoundCommand.select_track("101", 4)) == "*#16*101*#7*4##"
+    assert str(OWNSoundCommand.request_track("101")) == "*#16*101*7##"
+    assert str(OWNSoundCommand.set_frequency("101", 107000)) == "*#16*101*#6*0*107000##"
+    # Below 100 MHz the value is not zero-padded (F500N capture, MyHOME#427).
+    assert str(OWNSoundCommand.set_frequency("101", 96200)) == "*#16*101*#6*0*96200##"
+    assert str(OWNSoundCommand.request_frequency("101")) == "*#16*101*6##"
+    assert str(OWNSoundCommand.start_rds("101")) == "*16*101*101##"
+    assert str(OWNSoundCommand.stop_rds("101")) == "*16*102*101##"
+    assert str(OWNSoundCommand.request_rds("101")) == "*#16*101*8##"
+
+    with pytest.raises(ValueError, match="step must be between 1 and 15"):
+        OWNSoundCommand.next_track("101", 0)
+    with pytest.raises(ValueError, match="step must be between 1 and 15"):
+        OWNSoundCommand.next_track("101", 16)
+    with pytest.raises(ValueError, match="step must be between 1 and 15"):
+        OWNSoundCommand.previous_track("101", 0)
+    with pytest.raises(ValueError, match="step must be between 1 and 15"):
+        OWNSoundCommand.previous_track("101", 16)
+    with pytest.raises(ValueError, match="track must be greater than or equal to 1"):
+        OWNSoundCommand.select_track("101", 0)
+    with pytest.raises(ValueError, match="track must be greater than or equal to 1"):
+        OWNSoundCommand.select_track("101", -1)
+    with pytest.raises(ValueError, match="frequency in kHz must be positive"):
+        OWNSoundCommand.set_frequency("101", 0)
+    with pytest.raises(ValueError, match="frequency in kHz must be positive"):
+        OWNSoundCommand.set_frequency("101", -107000)
+
+
+def test_who16_sound_source_events() -> None:
+    """Verify WHO 16 sound source event parsing and state extraction."""
+    # Source busy indicator
+    busy = OWNMessage.parse("*16*100*101##")
+    assert isinstance(busy, OWNSoundEvent)
+    assert busy.is_source_busy is True
+    assert busy.source_id == "1"
+    assert "BUSY" in busy.human_readable_log
+
+    # Track advance steps
+    step_fwd = OWNMessage.parse("*16*6001*101##")
+    assert isinstance(step_fwd, OWNSoundEvent)
+    assert step_fwd.track_step_forward == 1
+    assert "advance track/station by 1 step(s)" in step_fwd.human_readable_log
+
+    step_fwd5 = OWNMessage.parse("*16*6005*101##")
+    assert isinstance(step_fwd5, OWNSoundEvent)
+    assert step_fwd5.track_step_forward == 5
+
+    # Track return steps
+    step_back = OWNMessage.parse("*16*6101*101##")
+    assert isinstance(step_back, OWNSoundEvent)
+    assert step_back.track_step_backward == 1
+    assert "return track/station by 1 step(s)" in step_back.human_readable_log
+
+    # Hardware seek
+    seek_up = OWNMessage.parse("*16*5000*101##")
+    assert isinstance(seek_up, OWNSoundEvent)
+    assert seek_up.is_seek_up is True
+
+    seek_down = OWNMessage.parse("*16*5100*101##")
+    assert isinstance(seek_down, OWNSoundEvent)
+    assert seek_down.is_seek_down is True
+
+    # Frequency report
+    freq = OWNMessage.parse("*#16*101*6*0*107000##")
+    assert isinstance(freq, OWNSoundEvent)
+    assert freq.frequency_khz == 107000
+    assert "frequency is 107000 kHz" in freq.human_readable_log
+
+    # Stored station/track report
+    trk = OWNMessage.parse("*#16*101*7*0*3##")
+    assert isinstance(trk, OWNSoundEvent)
+    assert trk.track == 3
+    assert "station/track is 3" in trk.human_readable_log
+
+    # RDS text report
+    rds = OWNMessage.parse("*#16*101*8*82*65*68*73*79*32*32*49##")
+    assert isinstance(rds, OWNSoundEvent)
+    assert rds.rds_text == "RADIO  1"
+    assert "RADIO  1" in rds.human_readable_log
+
+    # Malformed dimension error handling
+    bad_freq = OWNSoundEvent("*#16*101*6*##")
+    assert bad_freq.frequency_khz is None
+
+    bad_freq_neg = OWNSoundEvent("*#16*101*6*0*0##")
+    assert bad_freq_neg.frequency_khz is None
+
+    bad_trk = OWNSoundEvent("*#16*101*7*##")
+    assert bad_trk.track is None
+
+    bad_trk_zero = OWNSoundEvent("*#16*101*7*0*0##")
+    assert bad_trk_zero.track is None
+
+    bad_rds = OWNSoundEvent("*#16*101*8*82*65*68*73*79*32*32*##")
+    assert bad_rds.rds_text == "RADIO"
+
+    zero_rds = OWNSoundEvent("*#16*101*8*82*65*68*73*79*32*32*0##")
+    assert zero_rds.rds_text == "RADIO"
+
+    # Non-8-code RDS payloads are ignored as malformed frames
+    short_rds = OWNSoundEvent("*#16*101*8*82*65*68##")
+    assert short_rds.rds_text is None
+
+    long_rds = OWNSoundEvent("*#16*101*8*82*65*68*73*79*32*32*49*50##")
+    assert long_rds.rds_text is None
+
+    # Blank RDS payload
+    blank_rds = OWNSoundEvent("*#16*101*8*32*32*32*32*32*32*32*32##")
+    assert blank_rds.rds_text is None
+    assert "RDS text: ''" in blank_rds.human_readable_log