Handle SDNO digital mode and speed up probe tests

Author: balloob

README.md

@@ -175,7 +175,7 @@ await receiver.set_digital_input(DigitalInputMode.AUTO)
 await receiver.set_digital_input(DigitalInputMode.HDMI)
 await receiver.set_digital_input(DigitalInputMode.DIGITAL)
 await receiver.set_digital_input(DigitalInputMode.ANALOG)
-mode = await receiver.query_digital_input()  # DigitalInputMode enum
+mode = await receiver.query_digital_input()  # DigitalInputMode enum or None ("NO")
 ```
 
 Legacy models also support `PCM`, `DTS`, `RF`, `EXT_IN_1`, `EXT_IN_2`.

src/denon_rs232/__init__.py

@@ -20,7 +20,7 @@
 BAUD_RATE = 9600
 COMMAND_TIMEOUT = 2.0  # seconds to wait for a response
 MULTI_RESPONSE_DELAY = 0.3  # seconds to wait for multi-response query results
-PROBE_TIMEOUT = 0.5  # seconds to wait for each probe attempt
+PROBE_TIMEOUT = 0.8  # seconds to wait for each probe attempt
 CR = b"\r"
 
 # Volume range constants (dB)
@@ -526,8 +526,11 @@ async def set_digital_input(self, mode: DigitalInputMode) -> None:
         """Set digital input mode."""
         await self._send_command("SD", mode.value)
 
-    async def query_digital_input(self) -> DigitalInputMode:
-        return DigitalInputMode(await self._query("SD"))
+    async def query_digital_input(self) -> DigitalInputMode | None:
+        param = await self._query("SD")
+        if param == "NO":
+            return None
+        return DigitalInputMode(param)
 
     # -- Video select commands --
 
@@ -638,7 +641,9 @@ async def zone3_set_volume(self, db: float) -> None:
 
     # -- Probing --
 
-    async def probe_sources(self) -> frozenset[InputSource]:
+    async def probe_sources(
+        self, timeout: float | None = None
+    ) -> frozenset[InputSource]:
         """Probe which input sources the receiver supports.
 
         Tries setting each input source and checks if the receiver accepts it.
@@ -650,6 +655,9 @@ async def probe_sources(self) -> frozenset[InputSource]:
         if not self._connected:
             raise ConnectionError("Not connected")
 
+        if timeout is None:
+            timeout = PROBE_TIMEOUT
+
         original = self._state.input_source
         available: set[InputSource] = set()
 
@@ -660,7 +668,7 @@ async def probe_sources(self) -> frozenset[InputSource]:
         for source in InputSource:
             if source == original:
                 continue
-            resp = await self._send_and_wait("SI", source.value)
+            resp = await self._send_and_wait("SI", source.value, timeout=timeout)
             if resp == source.value:
                 available.add(source)
 
@@ -844,11 +852,15 @@ def _process_message(self, message: str) -> None:
             changed = self._process_ps_param(param)
 
         elif prefix == "SD":
-            try:
-                self._state.digital_input = DigitalInputMode(param)
+            if param == "NO":
+                self._state.digital_input = None
                 changed = True
-            except ValueError:
-                _LOGGER.warning("Unknown digital input mode: %s", param)
+            else:
+                try:
+                    self._state.digital_input = DigitalInputMode(param)
+                    changed = True
+                except ValueError:
+                    _LOGGER.warning("Unknown digital input mode: %s", param)
 
         elif prefix == "SV":
             if param in ("SOURCE", "OFF"):

tests/conftest.py

@@ -11,7 +11,7 @@
 
 # Speed up tests by reducing delays
 denon_rs232.MULTI_RESPONSE_DELAY = 0.01
-denon_rs232.PROBE_TIMEOUT = 0.1
+denon_rs232.PROBE_TIMEOUT = 0.01
 
 # Default responses for all query prefixes during startup.
 DEFAULT_QUERY_RESPONSES: dict[str, list[str]] = {

tests/test_denon_rs232.py

@@ -626,6 +626,17 @@ async def respond():
     assert result == DigitalInputMode.AUTO
 
 
+async def test_query_digital_input_no_returns_none(receiver, mock_serial):
+    async def respond():
+        await asyncio.sleep(0.05)
+        mock_serial.inject_response("SDNO")
+
+    task = asyncio.create_task(respond())
+    result = await receiver.query_digital_input()
+    await task
+    assert result is None
+
+
 async def test_query_video_select(receiver, mock_serial):
     async def respond():
         await asyncio.sleep(0.05)
@@ -872,6 +883,18 @@ async def test_digital_input_event(receiver, mock_serial):
     assert states[-1].digital_input == DigitalInputMode.AUTO
 
 
+async def test_digital_input_no_event(receiver, mock_serial, caplog):
+    """SD NO should clear digital_input to None without warning."""
+    states: list[DenonState] = []
+    receiver.subscribe(lambda s: states.append(s))
+
+    mock_serial.inject_response("SDNO")
+    await asyncio.sleep(0.1)
+
+    assert states[-1].digital_input is None
+    assert "Unknown digital input mode: NO" not in caplog.text
+
+
 # -- Event tests: video select --