Fix one-zone telnet power refresh handling

denonavr/api.py

@@ -468,6 +468,7 @@ class DenonAVRTelnetApi:
     host: str = attr.ib(converter=str, default="localhost")
     timeout: float = attr.ib(converter=float, default=2.0)
     is_denon: bool = attr.ib(converter=bool, default=True)
+    power_query_command: str = attr.ib(converter=str, default="ZM?")
     _connection_enabled: bool = attr.ib(default=False)
     _last_message_time: float = attr.ib(default=-1.0)
     _connect_lock: asyncio.Lock = attr.ib(default=attr.Factory(asyncio.Lock))
@@ -555,7 +556,7 @@ async def _async_trigger_updates(self) -> None:
         """Trigger update of all attributes."""
         commands = [
             # Critical State Info
-            "ZM?",  # Main Zone Power
+            self.power_query_command,  # Main Zone Power
             "SI?",  # Select INPUT source
             "MV?",  # MASTER VOLUME
             "MU?",  # Mute

denonavr/foundation.py

@@ -46,6 +46,7 @@
     HDMI_OUTPUT_MAP_REVERSE,
     ILLUMINATION_MAP,
     ILLUMINATION_MAP_REVERSE,
+    ALL_ZONES,
     MAIN_ZONE,
     POWER_STATES,
     SETTINGS_MENU_STATES,
@@ -253,7 +254,12 @@ def __attrs_post_init__(self) -> None:
 
     def _power_callback(self, zone: str, event: str, parameter: str) -> None:
         """Handle a power change event."""
-        if self.zone == zone and parameter in POWER_STATES:
+        if parameter not in POWER_STATES:
+            return
+
+        if self.zone == zone or (
+            self.zone == MAIN_ZONE and event == "PW" and zone == ALL_ZONES
+        ):
             self._power = parameter
 
     def _settings_menu_callback(self, zone: str, event: str, parameter: str) -> None:
@@ -437,9 +443,12 @@ async def async_setup(self) -> None:
                 # ZM events do not always work when the receiver has only one zone
                 # In this case it is safe to turn the entire device on and off
                 power_event = "PW"
+                self.telnet_api.power_query_command = "PW?"
                 self.telnet_commands = self.telnet_commands._replace(
                     command_power_on="PWON", command_power_standby="PWSTANDBY"
                 )
+            elif self.zone == MAIN_ZONE:
+                self.telnet_api.power_query_command = "ZM?"
             self.telnet_api.register_callback(power_event, self._power_callback)
 
             self.telnet_api.register_callback("MN", self._settings_menu_callback)

denonavr/input.py

@@ -27,6 +27,7 @@
     CHANGE_INPUT_MAPPING,
     DENON_ATTR_SETATTR,
     HDTUNER_SOURCES,
+    ALL_ZONES,
     MAIN_ZONE,
     NETAUDIO_PLAYING,
     NETAUDIO_SOURCES,
@@ -203,6 +204,8 @@ def setup(self) -> None:
             power_event = "Z2"
         elif self._device.zone == ZONE3:
             power_event = "Z3"
+        elif self._device.zone == MAIN_ZONE and self._device.zones == 1:
+            power_event = "PW"
         self._device.telnet_api.register_callback(power_event, self._power_callback)
         self._device.telnet_api.register_callback("SI", self._input_callback)
         self._device.telnet_api.register_callback("NSE", self._netaudio_callback)
@@ -232,7 +235,9 @@ def _input_callback(self, zone: str, event: str, parameter: str) -> None:
 
     def _power_callback(self, zone: str, event: str, parameter: str) -> None:
         """Handle a power change event."""
-        if self._device.zone != zone:
+        if self._device.zone != zone and not (
+            self._device.zone == MAIN_ZONE and event == "PW" and zone == ALL_ZONES
+        ):
             return
 
         if parameter != POWER_ON:

tests/test_denonavr.py

@@ -16,7 +16,7 @@
 
 import denonavr
 from denonavr.api import DenonAVRTelnetApi, DenonAVRTelnetProtocol
-from denonavr.const import SOUND_MODE_MAPPING
+from denonavr.const import SOUND_MODE_MAPPING, STATE_OFF, STATE_ON
 from denonavr.exceptions import AvrNetworkError, AvrTimoutError
 
 FAKE_IP = "10.0.0.0"
@@ -154,6 +154,26 @@ def custom_matcher(self, request: httpx.Request, *args, **kwargs):
     def _callback(self, zone, event, parameter):
         self.future.set_result(True)
 
+    @staticmethod
+    def _command_response(command: str) -> bytes:
+        """Return a telnet response that confirms a command."""
+        response_map = {
+            "PW?": "PWSTANDBY",
+            "ZM?": "ZMOFF",
+            "SI?": "SISAT/CBL",
+            "MV?": "MV50",
+            "MU?": "MUOFF",
+            "Z2?": "Z2OFF",
+            "Z2MU?": "Z2MUOFF",
+            "Z3?": "Z3OFF",
+            "Z3MU?": "Z3MUOFF",
+            "MS?": "MSSTEREO",
+        }
+        response = response_map.get(
+            command, command.replace("?", "").replace(" ", "") + "X"
+        )
+        return f"{response}\r".encode("utf-8")
+
     @pytest.mark.asyncio
     @pytest.mark.httpx_mock(can_send_already_matched_responses=True)
     async def test_receiver_type(self, httpx_mock: HTTPXMock):
@@ -562,6 +582,85 @@ def create_conn(proto_lambda, host, port):
             await self.future
             assert self.denon.power == "OFF"
 
+    @pytest.mark.asyncio
+    @pytest.mark.httpx_mock(can_send_already_matched_responses=True)
+    async def test_one_zone_power_state_updates_from_pw_events(
+        self, httpx_mock: HTTPXMock
+    ):
+        """Check that one-zone receivers derive state updates from PW events."""
+        httpx_mock.add_callback(self.custom_matcher)
+        self.testing_receiver = "AVR-1713"
+        self.denon = denonavr.DenonAVR(FAKE_IP)
+        await self.denon.async_setup()
+        await self.denon.async_update()
+        # pylint: disable=protected-access
+        assert self.denon._device.zones == 1
+        # pylint: disable=protected-access
+        self.denon.input._input_func = None
+
+        # pylint: disable=protected-access
+        self.denon._device.telnet_api._process_event("PWON")
+        assert self.denon.power == "ON"
+        assert self.denon.state == STATE_ON
+
+        # pylint: disable=protected-access
+        self.denon._device.telnet_api._process_event("PWSTANDBY")
+        assert self.denon.state == STATE_OFF
+
+    @pytest.mark.asyncio
+    @pytest.mark.httpx_mock(can_send_already_matched_responses=True)
+    async def test_one_zone_telnet_connect_refreshes_power_before_source(
+        self, httpx_mock: HTTPXMock
+    ):
+        """Check that reconnect queries power before source for one-zone receivers."""
+        sent_commands = []
+        transport = mock.Mock(is_closing=lambda: False)
+        protocol = DenonAVRTelnetProtocol(None, None)
+
+        def write_side_effect(data: bytes):
+            command = data.decode("utf-8").strip()
+            sent_commands.append(command)
+            protocol.data_received(self._command_response(command))
+
+        transport.write.side_effect = write_side_effect
+
+        def create_conn(proto_lambda, host, port):
+            proto = proto_lambda()
+            # pylint: disable=protected-access
+            protocol._on_message = proto._on_message
+            # pylint: disable=protected-access
+            protocol._on_connection_lost = proto._on_connection_lost
+            proto.connection_made(transport)
+            protocol.connection_made(transport)
+            return [transport, proto]
+
+        httpx_mock.add_callback(self.custom_matcher)
+        self.testing_receiver = "AVR-1713"
+        self.denon = denonavr.DenonAVR(FAKE_IP)
+        # pylint: disable=protected-access
+        self.denon._device.telnet_api._send_confirmation_timeout = 0.01
+        await self.denon.async_setup()
+        await self.denon.async_update()
+        # Simulate a stale power state from before the reconnect.
+        # pylint: disable=protected-access
+        self.denon._device._power = "ON"
+        # pylint: disable=protected-access
+        self.denon.input._state = STATE_OFF
+
+        with mock.patch("asyncio.get_event_loop", new_callable=mock.Mock) as debug_mock:
+            debug_mock.return_value.create_connection = mock.AsyncMock(
+                side_effect=create_conn
+            )
+            await self.denon.async_telnet_connect()
+            for _ in range(20):
+                if "SI?" in sent_commands:
+                    break
+                await asyncio.sleep(0)
+
+        assert sent_commands[0] == "PW?"
+        assert self.denon.power == "STANDBY"
+        assert self.denon.state == STATE_OFF
+
     @pytest.mark.asyncio
     @pytest.mark.httpx_mock(can_send_already_matched_responses=True)
     async def test_volume_min(self, httpx_mock: HTTPXMock):