Poll less: skip the stale drain on fresh connections, poll optional paths only when used

.github/mutations/plan.json

@@ -145,5 +145,12 @@
     "find": "    except ValueError, RecursionError:",
     "replace": "    except json.JSONDecodeError:",
     "expect": "tests/test_ssc_client.py::test_a_line_the_parser_refuses_is_skipped"
+  },
+  {
+    "name": "entities stop telling the coordinator which path they read",
+    "path": "custom_components/neumann_kh/entity.py",
+    "find": "        self.coordinator_context = getattr(description, \"ssc_path\", None)",
+    "replace": "        self.coordinator_context = None",
+    "expect": "tests/test_integration_e2e.py::test_every_optional_path_has_an_entity_asking_for_it"
   }
 ]

README.de.md

@@ -270,7 +270,9 @@ Ein Lautsprecher wird alle 30 Sekunden abgefragt. Dabei wird nicht jedes Mal
 alles geholt: Selten wechselnde Werte (Identität, die meisten
 Rückseiten-Einstellungen, Gerätename) werden nur in jedem zehnten Zyklus
 abgefragt, also alle 5 Minuten. Steuerungsmodus und Auto-Standby-Schalter werden
-in jedem Zyklus gelesen.
+in jedem Zyklus gelesen. Werte, die nur standardmäßig deaktivierte Entities
+anzeigen (Ausgang 1 und 2, Ausgangspegel- und Clip-Anzeige, Standby-Countdown),
+werden nur abgefragt, solange eine solche Entity aktiviert ist.
 Abgefragt wird **jeder Wert einzeln** (ein Blattpfad pro SSC-Nachricht), nicht als
 Sammelnachricht und nicht als Container-Abfrage: Eine Container-Abfrage wie
 `{"device":null}` beantwortet das Gerät mit einem Fehler, eine Sammelnachricht

README.md

@@ -260,7 +260,9 @@ Entities that are disabled by default can be enabled manually under
 A loudspeaker is polled every 30 seconds. Not everything is fetched on every
 cycle: rarely changing values (identity, most rear-panel settings, device name)
 are only queried on every tenth cycle, so every 5 minutes. The control mode and
-auto standby switches are read on every cycle. Each value is fetched
+auto standby switches are read on every cycle. Values that only entities
+disabled by default show (outputs 1 and 2, output meter and clip, standby
+countdown) are only polled while such an entity is enabled. Each value is fetched
 **individually** (one leaf path per SSC message), not as a combined
 message and not as a container query: a container query such as
 `{"device":null}` is answered with an error, and a combined message only

custom_components/neumann_kh/const.py

@@ -269,6 +269,23 @@
     PATH_OUT2_MUTE,
 )
 
+# Fast paths only entities disabled by default read. Polled only while such an
+# entity is enabled (see coordinator._fast_paths); on a KH 750 they were 9 of
+# 21 requests per cycle. Kept equal to the descriptions by test_coordinator.
+OPTIONAL_FAST_POLL_PATHS = frozenset(
+    {
+        PATH_METER_OUTPUT_LEVEL,
+        PATH_METER_OUTPUT_CLIP,
+        PATH_OUT1_LEVEL,
+        PATH_OUT1_DELAY,
+        PATH_OUT1_MUTE,
+        PATH_OUT2_LEVEL,
+        PATH_OUT2_DELAY,
+        PATH_OUT2_MUTE,
+        PATH_STANDBY_COUNTDOWN,
+    }
+)
+
 # Rarely changing subwoofer values (only every SLOW_POLL_EVERY_N_CYCLES).
 SUBWOOFER_SLOW_POLL_PATHS = (
     PATH_DIGITAL_BYPASS,

custom_components/neumann_kh/coordinator.py

@@ -31,6 +31,7 @@
     MODELS_WITH_SUBWOOFER_FEATURES,
     NON_SUBWOOFER_POLL_PATHS,
     NON_SUBWOOFER_SLOW_POLL_PATHS,
+    OPTIONAL_FAST_POLL_PATHS,
     PATH_LOGO_BRIGHTNESS,
     PATH_SAVE_SETTINGS,
     POLL_CYCLE_TIMEOUT_SECONDS,
@@ -177,6 +178,22 @@ def __init__(
         # poll that read the path before the request must not use it up.
         self._reread_requests = 0
 
+    def _fast_paths(self) -> list[tuple[str, ...]]:
+        """Fast paths minus the optional ones no entity asked for.
+
+        Until the first successful refresh no entity has been added, so nothing
+        has asked yet and everything is polled; an enabled optional entity
+        would otherwise stay unknown for a whole cycle.
+        """
+        if self.data is None:
+            return list(self._poll_paths)
+        wanted = set(self.async_contexts())
+        return [
+            path
+            for path in self._poll_paths
+            if path not in OPTIONAL_FAST_POLL_PATHS or path in wanted
+        ]
+
     async def _async_update_data(self) -> dict[str, Any]:
         """Query each path individually; a rejected/faulty single path is skipped."""
         include_slow = self._slow_poll_pending or self._cycle_count % SLOW_POLL_EVERY_N_CYCLES == 0
@@ -185,7 +202,7 @@ async def _async_update_data(self) -> dict[str, Any]:
             # Stays set until the slow poll succeeded (see below).
             self._slow_poll_pending = True
 
-        paths = list(self._poll_paths)
+        paths = self._fast_paths()
         if include_slow:
             paths.extend(self._slow_poll_paths)
 

custom_components/neumann_kh/entity.py

@@ -54,6 +54,13 @@ def __init__(self, coordinator: NeumannKHCoordinator, entry: ConfigEntry) -> Non
         )
         self._unique_id_base = serial
 
+    async def async_added_to_hass(self) -> None:
+        # The coordinator polls an optional path only while an entity names it
+        # as its context; the description is only set after __init__ here.
+        description = getattr(self, "entity_description", None)
+        self.coordinator_context = getattr(description, "ssc_path", None)
+        await super().async_added_to_hass()
+
     async def _async_write(
         self, path: tuple[str, ...], value: Any, rejected_key: str = "change_rejected"
     ) -> None:

custom_components/neumann_kh/ssc_client.py

@@ -227,13 +227,14 @@ def is_link_local(host: str) -> bool:
         # Third hex character must be in range 8..b (fe80..febf = /10).
         return prefix[2] in "89ab"
 
-    async def _ensure_connected(self) -> None:
+    async def _ensure_connected(self) -> bool:
+        """Connect if needed; True when this call opened the connection."""
         if self._closed:
             raise SSCConnectionError(
                 "the connection was closed - the config entry is being unloaded"
             )
         if self._writer is not None and not self._writer.is_closing():
-            return
+            return False
         try:
             self._reader, self._writer = await asyncio.wait_for(
                 asyncio.open_connection(
@@ -250,6 +251,7 @@ async def _ensure_connected(self) -> None:
             raise SSCConnectionError(
                 f"Connection to {mask_host(self._connect_host)}:{self._port} failed: {_os_reason(err)}"
             ) from err
+        return True
 
     def _drop_connection(self) -> None:
         """Drops the current connection; the next access reconnects."""
@@ -503,8 +505,9 @@ async def _locked_request(
             # await, so both can be cancelled. Left outside, a cancellation
             # hitting that window would keep a connection the caller believes
             # was dropped - the invariant the v1.15.0 hardening relies on.
-            await self._ensure_connected()
-            await self._discard_stale_lines()
+            # Nothing can be stale on a socket opened for this request.
+            if not await self._ensure_connected():
+                await self._discard_stale_lines()
             await self._send_raw(payload)
             merged = await self._read_lines_until_settled(expect_path)
         except SSCConnectionError, SSCTimeoutError:

tests/README.md

@@ -98,8 +98,9 @@ headroom for the next person to spend.
 | budget out of step | a function got simpler, was renamed, or is gone | Set the entry to the current value, or drop it. |
 
 The measure is Cognitive Complexity (Campbell / SonarSource), implemented in
-`complexity.py` and calibrated against SonarQube Cloud: Sonar reports 35 for
-`_read_lines_until_settled`, and so does this. `COMPLEXITY_LIMIT` is 15,
+`complexity.py` and calibrated against SonarQube Cloud: at calibration Sonar
+reported 35 for `_read_lines_until_settled`, and so did this. (Today it is 33,
+after the reply parsing moved out into `_parse_reply`.) `COMPLEXITY_LIMIT` is 15,
 SonarSource's default.
 
 ## Adding a guard

tests/complexity.py

@@ -3,7 +3,7 @@
 A measuring stick for the budget guard next door, not a general-purpose
 implementation. Calibrated against SonarQube Cloud during the complexity
 round of 1.18.1: Sonar reported 35 for `_read_lines_until_settled`, and this
-scorer returns 35 for it.
+scorer returned 35 for it then (33 since `_parse_reply` was split out).
 
 Three rules:
 

tests/test_budgets.py

@@ -41,7 +41,8 @@
     # Raised to 549 for the LOCAL-mode error (401) every writer has to name.
     # Raised to 556 for _parse_reply: the nesting limit on faulty replies, taken
     # out of the read loop so its complexity went down instead of up.
-    "ssc_client.py": 556,
+    # Raised to 559 to skip the stale drain on a connection opened just now.
+    "ssc_client.py": 559,
 }
 
 # SonarSource's default. Everything above needs a written entry.

tests/test_coordinator.py

@@ -14,7 +14,9 @@
 
 from custom_components.neumann_kh.const import (
     PATH_DEVICE_NAME,
+    PATH_OUT1_LEVEL,
     PATH_OUTPUT_MUTE,
+    PATH_STANDBY_COUNTDOWN,
     SLOW_POLL_PATHS,
 )
 from custom_components.neumann_kh.coordinator import NeumannKHCoordinator
@@ -29,8 +31,10 @@ def __init__(self, values: dict[tuple[str, ...], Any]) -> None:
         self.fail_connection = False
         self.rejected_paths: set[tuple[str, ...]] = set()
         self.priority_waiting = asyncio.Event()
+        self.asked: list[tuple[str, ...]] = []
 
     async def get(self, path: tuple[str, ...]) -> Any:
+        self.asked.append(path)
         if self.fail_connection:
             raise SSCConnectionError("device offline (test)")
         if path in self.rejected_paths:
@@ -400,3 +404,100 @@ async def _timeout(path, value):
     with pytest.raises(HomeAssistantError) as caught:
         await coordinator.async_persist()
     assert caught.value.translation_key == "settings_not_saved"
+
+
+def _kh750(hass, client: FakeClient) -> NeumannKHCoordinator:
+    coord = NeumannKHCoordinator(hass, client, "test", model="KH 750")
+    coord.update_interval = None
+    return coord
+
+
+async def test_paths_only_disabled_entities_read_are_not_polled(hass):
+    """On a KH 750, 9 of 21 fast paths served only entities disabled by default.
+
+    That was about 26 000 requests a day per subwoofer nobody looked at.
+    """
+    client = FakeClient({})
+    coord = _kh750(hass, client)
+    await coord.async_refresh()  # the first refresh polls everything
+    coord.async_add_listener(lambda: None, PATH_OUTPUT_MUTE)
+    client.asked.clear()
+
+    await coord.async_refresh()
+
+    assert PATH_OUT1_LEVEL not in client.asked
+    assert PATH_STANDBY_COUNTDOWN not in client.asked
+    assert PATH_OUTPUT_MUTE in client.asked
+
+
+async def test_an_enabled_optional_entity_gets_its_path_polled(hass):
+    client = FakeClient({PATH_OUT1_LEVEL: -3.0})
+    coord = _kh750(hass, client)
+    coord.async_add_listener(lambda: None, PATH_OUT1_LEVEL)
+
+    await coord.async_refresh()
+
+    assert coord.value(PATH_OUT1_LEVEL) == -3.0
+
+
+async def test_the_first_refresh_polls_every_path(hass):
+    """The first refresh runs before any entity is added - none has asked yet.
+
+    Skipping the optional paths then would leave an enabled entity unknown
+    until the next cycle.
+    """
+    client = FakeClient({})
+    coord = _kh750(hass, client)
+
+    await coord.async_refresh()
+
+    assert PATH_OUT1_LEVEL in client.asked
+
+
+def _descriptions_with_paths(model: str):
+    """Every entity description that carries an SSC path, across the platforms."""
+    import importlib
+
+    found = []
+    for platform in ("sensor", "binary_sensor", "number", "select", "switch", "text"):
+        module = importlib.import_module(f"custom_components.neumann_kh.{platform}")
+        for value in vars(module).values():
+            for item in value if isinstance(value, tuple) else (value,):
+                if getattr(item, "ssc_path", None) and hasattr(item, "key"):
+                    found.append(item)
+    number = importlib.import_module("custom_components.neumann_kh.number")
+    found.append(number._build_output_delay_description(model == "KH 750"))
+    return found
+
+
+@pytest.mark.parametrize("model", ["KH 120 II", "KH 750"])
+def test_the_optional_poll_paths_are_exactly_those_of_disabled_entities(model):
+    """A hand-kept list drifts: an entity enabled by default whose path sits
+    here would stay unknown, and a new disabled one would still be polled."""
+    from custom_components.neumann_kh.const import OPTIONAL_FAST_POLL_PATHS
+    from custom_components.neumann_kh.coordinator import poll_paths_for_model
+
+    fast, _ = poll_paths_for_model(model)
+    descriptions = _descriptions_with_paths(model)
+    enabled = {tuple(d.ssc_path) for d in descriptions if d.entity_registry_enabled_default}
+    disabled = {tuple(d.ssc_path) for d in descriptions if not d.entity_registry_enabled_default}
+
+    expected = (disabled - enabled) & {tuple(path) for path in fast}
+    assert expected == OPTIONAL_FAST_POLL_PATHS & {tuple(path) for path in fast}
+
+
+async def test_the_first_refresh_polls_every_path_even_with_a_listener(hass):
+    """The first-refresh signal must not hang on Home Assistant internals.
+
+    It was "no listener registered yet", read from the coordinator's private
+    _listeners. A listener added before the first refresh then skipped the
+    optional paths, and a rename of that attribute in Home Assistant would
+    fail every cycle. "No data yet" holds exactly until the first success.
+    """
+    client = FakeClient({})
+    coord = _kh750(hass, client)
+    coord.async_add_listener(lambda: None, PATH_OUTPUT_MUTE)
+
+    await coord.async_refresh()
+
+    assert PATH_OUT1_LEVEL in client.asked

tests/test_integration_e2e.py

@@ -554,3 +554,36 @@ async def test_a_backup_from_out5_firmware_restores_onto_out_firmware(
         await _press(hass, entry, "_restore_backup")
 
         assert new_firmware.state["audio"]["out"]["level"] == 90.0
+
+
+async def test_every_entity_tells_the_coordinator_which_path_it_reads(hass, socket_enabled):
+    """The coordinator skips optional paths no entity asked for.
+
+    An entity that registered without its path would see its value vanish on
+    the next cycle - checked against the real platforms, not a fake.
+    """
+    async with _simulator(MODEL_KH_750) as port:
+        entry = await _setup_entry(hass, MODEL_KH_750, port, "SIM0007500")
+        contexts = set(entry.runtime_data.async_contexts())
+
+    assert ("audio", "out", "level") in contexts
+    assert ("ui", "control_mode") in contexts
+
+
+async def test_every_optional_path_has_an_entity_asking_for_it(hass, socket_enabled, monkeypatch):
+    """With every entity enabled, each optional path must be in the contexts.
+
+    An entity reading such a path without registering it - a hard-wired path
+    instead of its description's ssc_path - would stay unknown from the second
+    cycle on, and the description-based drift test cannot see that pattern.
+    """
+    from homeassistant.helpers.entity import Entity
+
+    from custom_components.neumann_kh.const import OPTIONAL_FAST_POLL_PATHS
+
+    monkeypatch.setattr(Entity, "entity_registry_enabled_default", property(lambda self: True))
+    async with _simulator(MODEL_KH_750) as port:
+        entry = await _setup_entry(hass, MODEL_KH_750, port, "SIM0007500")
+        contexts = set(entry.runtime_data.async_contexts())
+
+    assert OPTIONAL_FAST_POLL_PATHS <= contexts, sorted(OPTIONAL_FAST_POLL_PATHS - contexts)

tests/test_ssc_client.py

@@ -366,9 +366,10 @@ async def _hanging_drain(self) -> None:
         in_drain.set()
         await asyncio.sleep(60)  # hold the request inside the drain window
 
-    monkeypatch.setattr(SSCClient, "_discard_stale_lines", _hanging_drain)
-
     client = _client(server.port)
+    # A fresh connection skips the drain, so open it with a first request.
+    await client.get(("device", "name"))
+    monkeypatch.setattr(SSCClient, "_discard_stale_lines", _hanging_drain)
     try:
         task = asyncio.create_task(client.get(("device", "name")))
         await asyncio.wait_for(in_drain.wait(), timeout=5)
@@ -999,3 +1000,28 @@ async def test_an_unaliased_path_is_sent_unchanged(server):
         assert received == [{"audio": {"out": {"delay": None}}}]
     finally:
         await client.close()
+
+
+async def test_a_fresh_connection_skips_the_stale_drain(server, monkeypatch):
+    """Nothing can be stale on a socket that was just opened.
+
+    The drain waits out its window every time it finds nothing, so running it
+    right after connecting was pure dead time on every reconnect.
+    """
+    drains = 0
+    real_drain = SSCClient._discard_stale_lines
+
+    async def _counting_drain(self) -> None:
+        nonlocal drains
+        drains += 1
+        await real_drain(self)
+
+    monkeypatch.setattr(SSCClient, "_discard_stale_lines", _counting_drain)
+    client = _client(server.port)
+    try:
+        await client.get(("device", "name"))
+        assert drains == 0, "drained a socket that was opened for this very request"
+        await client.get(("device", "name"))
+        assert drains == 1, "the reused connection was not drained"
+    finally:
+        await client.close()