From 78a96dc064325487bbd0985142d5c6f893b1fa36 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:46:57 +0200 Subject: [PATCH 1/3] feat(sound): add WHO 16 media transport commands and event decoders - Add OWNSoundCommand builders: next_track, previous_track, seek_up, seek_down, select_track, request_track, set_frequency, request_frequency, start_rds, stop_rds, request_rds - Add OWNSoundEvent properties: frequency_khz, track, rds_text, is_source_busy, track_step_forward, track_step_backward, is_seek_up, is_seek_down - Add unit test coverage in tests/test_sound.py (100% statement and branch coverage) --- OWNd/message/sound.py | 186 ++++++++++++++++++++++++++++++++++++++++++ tests/test_sound.py | 92 +++++++++++++++++++++ 2 files changed, 278 insertions(+) diff --git a/OWNd/message/sound.py b/OWNd/message/sound.py index 198c37d..a40a9ea 100644 --- a/OWNd/message/sound.py +++ b/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,36 @@ 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 + 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 + self._human_readable_log = ( + f"{subject} station/track is {self._track}." + ) + elif self._dimension == 8 and self._dimension_value and len(self._dimension_value) == 8: + 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 +151,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 +304,103 @@ 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.""" + t = int(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).""" + khz = int(round(float(kilohertz))) + message = cls(f"*#16*{where}*#6*0*{khz:06d}##") + 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) diff --git a/tests/test_sound.py b/tests/test_sound.py index 6647cec..66b892d 100644 --- a/tests/test_sound.py +++ b/tests/test_sound.py @@ -155,3 +155,95 @@ 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##" + 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) + + +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_trk = OWNSoundEvent("*#16*101*7*##") + assert bad_trk.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" From fe3037d82d8c98b539f270e424c6d3bfae1f8504 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Thu, 8 Oct 2026 11:34:13 +0200 Subject: [PATCH 2/3] fix(sound): address audit observations on WHO 16 dimensions 6, 7, and 8 - Validate track >= 1 in select_track and OWNSoundEvent with spec docstrings on station presets vs. CD tracks - Validate kilohertz > 0 in set_frequency and OWNSoundEvent with usage docstrings - Clarify 8-byte RDS PS constraint in OWNSoundEvent and add tests for short/long/blank RDS frames --- OWNd/message/sound.py | 32 +++++++++++++++++++++++++++++--- tests/test_sound.py | 26 ++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 3 deletions(-) diff --git a/OWNd/message/sound.py b/OWNd/message/sound.py index a40a9ea..c1a798e 100644 --- a/OWNd/message/sound.py +++ b/OWNd/message/sound.py @@ -85,6 +85,9 @@ def __init__(self, data: str) -> None: 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." ) @@ -93,10 +96,18 @@ def __init__(self, data: str) -> None: 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 and len(self._dimension_value) == 8: + 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: @@ -346,8 +357,17 @@ def seek_down(cls, where: str | int) -> OWNSoundCommand: @classmethod def select_track(cls, where: str | int, track: int) -> OWNSoundCommand: - """Select stored station or CD/track number.""" + """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}." @@ -365,8 +385,14 @@ def request_track(cls, where: str | int) -> OWNSoundCommand: def set_frequency( cls, where: str | int, kilohertz: int | float ) -> OWNSoundCommand: - """Tune tuner source to frequency in kHz (leading 0 parameter).""" + """Tune tuner source to frequency in kHz (leading 0 parameter). + + Examples: + `107000` tunes to 107.00 MHz FM. + """ 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:06d}##") message._human_readable_log = ( f"Tuning {where} to {khz} kHz." diff --git a/tests/test_sound.py b/tests/test_sound.py index 66b892d..995e762 100644 --- a/tests/test_sound.py +++ b/tests/test_sound.py @@ -181,6 +181,14 @@ def test_who16_sound_source_commands() -> None: 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: @@ -239,11 +247,29 @@ def test_who16_sound_source_events() -> None: 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 From 585645795fd581d9f33a939490f7de5391ce655a Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Thu, 8 Oct 2026 21:51:24 +0200 Subject: [PATCH 3/3] fix(sound): send WHO 16 tuner frequency without zero padding set_frequency formatted the value as {khz:06d}, so 96.2 MHz became *#16*101*#6*0*096200##. The F500N capture on an MH200N (MyHOME#427) shows the plain *#16*101*#6*0*96200## accepted, and no capture or oracle row uses the padded form. The "6 digits" wording in the WHO 16 spec only comes with a 107000 example, where padding changes nothing. Adds a test below 100 MHz. Co-Authored-By: Claude Opus 5.5 --- OWNd/message/sound.py | 7 +++++-- tests/test_sound.py | 2 ++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/OWNd/message/sound.py b/OWNd/message/sound.py index c1a798e..6a280df 100644 --- a/OWNd/message/sound.py +++ b/OWNd/message/sound.py @@ -387,13 +387,16 @@ def set_frequency( ) -> 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. + `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:06d}##") + message = cls(f"*#16*{where}*#6*0*{khz}##") message._human_readable_log = ( f"Tuning {where} to {khz} kHz." ) diff --git a/tests/test_sound.py b/tests/test_sound.py index 995e762..44c7239 100644 --- a/tests/test_sound.py +++ b/tests/test_sound.py @@ -168,6 +168,8 @@ def test_who16_sound_source_commands() -> None: 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##"