diff --git a/OWNd/message/sound.py b/OWNd/message/sound.py index 198c37d..6a280df 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,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) diff --git a/tests/test_sound.py b/tests/test_sound.py index 6647cec..44c7239 100644 --- a/tests/test_sound.py +++ b/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