Read panel software version, firmware and UUID from mDNS (v1.16.0)

README.md

@@ -34,8 +34,25 @@ asyncio.run(main())
 - **Entity definitions** for sensors, binary sensors, and switches
 - **FLOAT32 decoding** with big endian (ABCD) byte order
 - **Type-safe dataclasses** for entity definitions
+- **mDNS device information**: panel software version, controller firmware and UUID from the controller's advertisement
 - **Monotonic clamping with reset detection** for the energy totals: sub-kWh jitter is clamped, while a drop of more than 1 kWh that persists for 3 consecutive reads is accepted as a counter reset (logged as a warning); `clear_monotonic_cache()` forgets all baselines
 
+## Device information over mDNS
+
+The controller advertises a `_workstation._tcp` mDNS service whose TXT record holds the software version shown on the panel, the controller firmware and a stable UUID. Register 77 (`async_get_software_version()`) reads 0.0 on recent firmware, so use this for the version instead:
+
+```python
+from zeroconf.asyncio import AsyncZeroconf
+from python_qube_heatpump import async_get_device_info
+
+aiozc = AsyncZeroconf()  # in Home Assistant: zeroconf.async_get_async_instance(hass)
+info = await async_get_device_info("192.168.1.100", aiozc)
+if info:
+    print(info.software_version, info.controller_firmware, info.uuid)
+```
+
+The lookup returns `None` when no record is found within the timeout (default 5 s), for example when mDNS is not forwarded between VLANs. `parse_device_info(properties)` parses TXT properties you already have, such as those from a zeroconf discovery.
+
 ## Entity Definitions
 
 The library provides pre-defined entity definitions for all Qube Modbus registers:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "python-qube-heatpump"
-version = "1.15.0"
+version = "1.16.0"
 authors = [
   { name="MattieGit", email="6250046+MattieGit@users.noreply.github.com" },
 ]
@@ -18,6 +18,7 @@ classifiers = [
 ]
 dependencies = [
     "pymodbus>=3.11.0",
+    "zeroconf>=0.132.0",
 ]
 
 [project.optional-dependencies]

src/python_qube_heatpump/__init__.py

@@ -17,6 +17,7 @@
     SENSORS,
     SWITCHES,
 )
+from .mdns import QubeDeviceInfo, async_get_device_info, parse_device_info
 from .models import QubeState
 from .network import async_get_mac_address
 
@@ -41,4 +42,8 @@
     "resolve_status",
     # Network
     "async_get_mac_address",
+    # mDNS device information
+    "QubeDeviceInfo",
+    "async_get_device_info",
+    "parse_device_info",
 ]

src/python_qube_heatpump/mdns.py

@@ -0,0 +1,141 @@
+"""Device information from the Qube's mDNS advertisement.
+
+The Qube's Carel controller advertises a ``_workstation._tcp`` service whose
+TXT record carries the panel software version (``ProjectRelease``), the
+controller firmware (``FWRelease``) and a stable ``Uuid``. The software
+version register (77) returns 0.0 on recent firmware, so this is the only
+reliable source of the version shown on the panel.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from collections.abc import Mapping
+from dataclasses import dataclass
+import logging
+import socket
+
+from zeroconf import ServiceStateChange, Zeroconf
+from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf
+
+_LOGGER = logging.getLogger(__name__)
+
+SERVICE_TYPE = "_workstation._tcp.local."
+CAREL_VENDOR_ID = "000A5C"
+
+# Per-service resolve timeout (ms) and overall lookup timeout (s)
+_RESOLVE_TIMEOUT_MS = 1500
+DEFAULT_LOOKUP_TIMEOUT = 5.0
+
+
+@dataclass(frozen=True)
+class QubeDeviceInfo:
+    """Device information advertised by the Qube over mDNS."""
+
+    uuid: str
+    software_version: str | None
+    controller_firmware: str | None
+    project_name: str | None
+
+
+def _decode(value: str | bytes | None) -> str | None:
+    """Decode a TXT value; empty values become None."""
+    if isinstance(value, bytes):
+        value = value.decode("utf-8", errors="replace")
+    return value or None
+
+
+def parse_device_info(
+    properties: Mapping[str, str | None] | Mapping[bytes, bytes | None],
+) -> QubeDeviceInfo | None:
+    """Parse the TXT properties of a Qube ``_workstation._tcp`` service.
+
+    Accepts both raw zeroconf properties (bytes) and decoded properties
+    (str, as Home Assistant passes them). Keys are matched case-insensitively.
+
+    Returns None when the record does not come from a Carel controller or
+    has no Uuid.
+    """
+    txt: dict[str, str | None] = {}
+    for key, value in properties.items():
+        name = _decode(key)
+        if name is not None:
+            txt[name.lower()] = _decode(value)
+
+    vendor = txt.get("vendor")
+    uuid = txt.get("uuid")
+    if vendor is None or vendor.upper() != CAREL_VENDOR_ID or uuid is None:
+        return None
+
+    return QubeDeviceInfo(
+        uuid=uuid,
+        software_version=txt.get("projectrelease"),
+        controller_firmware=txt.get("fwrelease"),
+        project_name=txt.get("projectname"),
+    )
+
+
+def _resolve_ip(host: str) -> str | None:
+    """Resolve a hostname to an IPv4 address."""
+    try:
+        return socket.gethostbyname(host)
+    except OSError:
+        return None
+
+
+def _matches_host(info: AsyncServiceInfo, host: str, ip: str | None) -> bool:
+    """Return True if the resolved service belongs to the given host."""
+    if ip is not None and ip in info.parsed_addresses():
+        return True
+    server = (info.server or "").lower().rstrip(".")
+    return server == host.lower().rstrip(".")
+
+
+async def async_get_device_info(
+    host: str,
+    aiozc: AsyncZeroconf,
+    timeout: float = DEFAULT_LOOKUP_TIMEOUT,
+) -> QubeDeviceInfo | None:
+    """Look up the mDNS device information of the Qube at ``host``.
+
+    Browses ``_workstation._tcp`` services and returns the first Carel
+    record whose address (or mDNS hostname) matches ``host``. The caller
+    owns ``aiozc``; Home Assistant integrations must pass the shared
+    instance from ``zeroconf.async_get_async_instance``.
+
+    Returns None if no matching record is found within ``timeout`` seconds,
+    for example when mDNS is not forwarded between VLANs.
+    """
+    ip = await asyncio.to_thread(_resolve_ip, host)
+    names: asyncio.Queue[str] = asyncio.Queue()
+    seen: set[str] = set()
+
+    def _on_change(
+        zeroconf: Zeroconf,
+        service_type: str,
+        name: str,
+        state_change: ServiceStateChange,
+    ) -> None:
+        if state_change is not ServiceStateChange.Removed and name not in seen:
+            seen.add(name)
+            names.put_nowait(name)
+
+    browser = AsyncServiceBrowser(aiozc.zeroconf, [SERVICE_TYPE], handlers=[_on_change])
+    try:
+        async with asyncio.timeout(timeout):
+            while True:
+                name = await names.get()
+                info = AsyncServiceInfo(SERVICE_TYPE, name)
+                if not await info.async_request(aiozc.zeroconf, _RESOLVE_TIMEOUT_MS):
+                    continue
+                if not _matches_host(info, host, ip):
+                    continue
+                if (device := parse_device_info(info.properties)) is not None:
+                    return device
+    except TimeoutError:
+        _LOGGER.debug("No mDNS device information found for %s", host)
+        return None
+    finally:
+        with contextlib.suppress(Exception):
+            await browser.async_cancel()

tests/test_mdns.py

@@ -0,0 +1,183 @@
+"""Test mDNS device information lookup."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from zeroconf import ServiceStateChange
+
+from python_qube_heatpump.mdns import (
+    SERVICE_TYPE,
+    QubeDeviceInfo,
+    async_get_device_info,
+    parse_device_info,
+)
+
+# TXT record captured from a Qube (fw 4.1.00)
+QUBE_TXT = {
+    b"Vendor": b"000A5C",
+    b"MachineType": b"312",
+    b"HWCode": b"344",
+    b"InterfaceType": b"11",
+    b"Uuid": b"000100000007B5EA",
+    b"FWRelease": b"v5.1.007",
+    b"ToolRelease": b"5.1.9",
+    b"ProjectRelease": b"4.1.00",
+    b"ProjectName": b"DEQSIHPB000CR",
+    b"OemId": b"",
+    b"MachineCode": None,
+}
+
+QUBE_INFO = QubeDeviceInfo(
+    uuid="000100000007B5EA",
+    software_version="4.1.00",
+    controller_firmware="v5.1.007",
+    project_name="DEQSIHPB000CR",
+)
+
+
+def test_parse_bytes_properties():
+    """Raw zeroconf properties are parsed."""
+    assert parse_device_info(QUBE_TXT) == QUBE_INFO
+
+
+def test_parse_str_properties_case_insensitive():
+    """Decoded properties with different key case are parsed."""
+    props = {
+        "vendor": "000a5c",
+        "UUID": "000100000007B5EA",
+        "projectrelease": "4.1.00",
+    }
+    assert parse_device_info(props) == QubeDeviceInfo(
+        uuid="000100000007B5EA",
+        software_version="4.1.00",
+        controller_firmware=None,
+        project_name=None,
+    )
+
+
+@pytest.mark.parametrize(
+    "props",
+    [
+        {b"Vendor": b"001122", b"Uuid": b"X"},
+        {b"Vendor": b"000A5C"},
+        {b"Vendor": b"000A5C", b"Uuid": b""},
+        {b"Uuid": b"X"},
+        {},
+    ],
+)
+def test_parse_rejects_non_qube(props):
+    """Records from other vendors or without a Uuid are ignored."""
+    assert parse_device_info(props) is None
+
+
+def _service(name: str, addresses: list[str], server: str, props: dict):
+    """Build a fake resolved AsyncServiceInfo."""
+    info = MagicMock()
+    info.name = name
+    info.server = server
+    info.properties = props
+    info.parsed_addresses.return_value = addresses
+    info.async_request = AsyncMock(return_value=True)
+    return info
+
+
+def _patch_zeroconf(services: dict):
+    """Patch the browser to announce ``services`` and the resolver to return them."""
+
+    def browser(zc, types, handlers):
+        assert types == [SERVICE_TYPE]
+        for name in services:
+            handlers[0](
+                zeroconf=zc,
+                service_type=SERVICE_TYPE,
+                name=name,
+                state_change=ServiceStateChange.Added,
+            )
+        mock = MagicMock()
+        mock.async_cancel = AsyncMock()
+        return mock
+
+    return (
+        patch("python_qube_heatpump.mdns.AsyncServiceBrowser", side_effect=browser),
+        patch(
+            "python_qube_heatpump.mdns.AsyncServiceInfo",
+            side_effect=lambda type_, name: services[name],
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_lookup_matches_ip():
+    """The Carel record whose address matches the host is returned."""
+    services = {
+        "PC._workstation._tcp.local.": _service(
+            "PC", ["192.168.1.10"], "pc.local.", {b"Vendor": b"000A5C", b"Uuid": b"Y"}
+        ),
+        "Qube._workstation._tcp.local.": _service(
+            "Qube", ["192.168.5.208"], "Qube.local.", QUBE_TXT
+        ),
+    }
+    browser, info = _patch_zeroconf(services)
+    with (
+        browser,
+        info,
+        patch("python_qube_heatpump.mdns._resolve_ip", return_value="192.168.5.208"),
+    ):
+        result = await async_get_device_info("192.168.5.208", MagicMock())
+
+    assert result == QUBE_INFO
+
+
+@pytest.mark.asyncio
+async def test_lookup_matches_mdns_hostname():
+    """A .local host that does not resolve via DNS matches on the SRV server."""
+    services = {
+        "Qube._workstation._tcp.local.": _service(
+            "Qube", ["192.168.5.208"], "Qube.local.", QUBE_TXT
+        ),
+    }
+    browser, info = _patch_zeroconf(services)
+    with (
+        browser,
+        info,
+        patch("python_qube_heatpump.mdns._resolve_ip", return_value=None),
+    ):
+        result = await async_get_device_info("qube.local", MagicMock())
+
+    assert result == QUBE_INFO
+
+
+@pytest.mark.asyncio
+async def test_lookup_skips_unresolved_and_non_qube():
+    """Services that fail to resolve or are not Carel are skipped until timeout."""
+    unresolved = _service("A", ["192.168.5.208"], "a.local.", QUBE_TXT)
+    unresolved.async_request = AsyncMock(return_value=False)
+    services = {
+        "A._workstation._tcp.local.": unresolved,
+        "B._workstation._tcp.local.": _service(
+            "B", ["192.168.5.208"], "b.local.", {b"Vendor": b"001122"}
+        ),
+    }
+    browser, info = _patch_zeroconf(services)
+    with (
+        browser,
+        info,
+        patch("python_qube_heatpump.mdns._resolve_ip", return_value="192.168.5.208"),
+    ):
+        result = await async_get_device_info("192.168.5.208", MagicMock(), timeout=0.05)
+
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_lookup_times_out_without_services():
+    """No advertisement (e.g. mDNS blocked) returns None after the timeout."""
+    browser, info = _patch_zeroconf({})
+    with (
+        browser,
+        info,
+        patch("python_qube_heatpump.mdns._resolve_ip", return_value="10.0.0.5"),
+    ):
+        result = await async_get_device_info("10.0.0.5", MagicMock(), timeout=0.05)
+
+    assert result is None