Collect Stick NodeInfo

plugwise_usb/__init__.py

@@ -117,6 +117,16 @@ def joined_nodes(self) -> int | None:
             return None
         return len(self._network.registry) + 1
 
+    @property
+    def firmware(self) -> str:
+        """Firmware of USB-Stick."""
+        return self._controller.firmware_stick
+
+    @property
+    def hardware(self) -> str:
+        """Hardware of USB-Stick."""
+        return self._controller.hardware_stick
+
     @property
     def mac_stick(self) -> str:
         """MAC address of USB-Stick. Raises StickError is connection is missing."""
@@ -127,6 +137,11 @@ def mac_coordinator(self) -> str:
         """MAC address of the network coordinator (Circle+). Raises StickError is connection is missing."""
         return self._controller.mac_coordinator
 
+    @property
+    def name(self) -> str:
+        """Return name of Stick."""
+        return self._controller.stick_name
+
     @property
     def network_discovered(self) -> bool:
         """Indicate if discovery of network is active. Raises StickError is connection is missing."""

plugwise_usb/connection/__init__.py

@@ -7,9 +7,21 @@
 from typing import Any
 
 from ..api import StickEvent
+from ..constants import UTF8
 from ..exceptions import NodeError, StickError
-from ..messages.requests import PlugwiseRequest, StickInitRequest
-from ..messages.responses import PlugwiseResponse, StickInitResponse
+from ..helpers.util import version_to_model
+from ..messages.requests import (
+    NodeInfoRequest,
+    NodePingRequest,
+    PlugwiseRequest,
+    StickInitRequest,
+)
+from ..messages.responses import (
+    NodeInfoResponse,
+    NodePingResponse,
+    PlugwiseResponse,
+    StickInitResponse,
+)
 from .manager import StickConnectionManager
 from .queue import StickQueue
 
@@ -26,10 +38,13 @@
         self._unsubscribe_stick_event: Callable[[], None] | None = None
         self._init_sequence_id: bytes | None = None
         self._is_initialized = False
+        self._fw_stick: str | None = None
+        self._hw_stick: str | None = None
         self._mac_stick: str | None = None
         self._mac_nc: str | None = None
         self._network_id: int | None = None
         self._network_online = False
+        self.stick_name: str | None = None
 
     @property
     def is_initialized(self) -> bool:
@@ -43,6 +58,16 @@
         """Return connection state from connection manager."""
         return self._manager.is_connected
 
+    @property
+    def firmware_stick(self) -> str | None:
+        """Firmware version of the Stick."""
+        return self._fw_stick
+
+    @property
+    def hardware_stick(self) -> str | None:
+        """Hardware version of the Stick."""
+        return self._hw_stick
+
     @property
     def mac_stick(self) -> str:
         """MAC address of USB-Stick. Raises StickError when not connected."""
@@ -160,16 +185,50 @@
                 + f"' {self._manager.serial_path}'"
             )
         self._mac_stick = init_response.mac_decoded
+        self.stick_name = f"Stick {self._mac_stick[-5:]}"
         self._network_online = init_response.network_online
 
         # Replace first 2 characters by 00 for mac of circle+ node
         self._mac_nc = init_response.mac_network_controller
         self._network_id = init_response.network_id
         self._is_initialized = True
 
+        # Add Stick NodeInfoRequest
+        node_info, _ = await self.get_node_details(self._mac_stick, ping_first=False)
+        if node_info is not None:
+            self._fw_stick = node_info.firmware
+            hardware, _ = version_to_model(node_info.hardware)
+            self._hw_stick = hardware
+
         if not self._network_online:
             raise StickError("Zigbee network connection to Circle+ is down.")
 
+    async def get_node_details(
+        self, mac: str, ping_first: bool
+    ) -> tuple[NodeInfoResponse | None, NodePingResponse | None]:
+        """Return node discovery type."""
+        ping_response: NodePingResponse | None = None
+        if ping_first:
+            # Define ping request with one retry
+            ping_request = NodePingRequest(
+                self.send, bytes(mac, UTF8), retries=1
+            )
+            try:
+                ping_response = await ping_request.send(suppress_node_errors=True)
+            except StickError:
+                return (None, None)
+            if ping_response is None:
+                return (None, None)
+
+        info_request = NodeInfoRequest(
+            self.send, bytes(mac, UTF8), retries=1
+        )
+        try:
+            info_response = await info_request.send()
+        except StickError:
+            return (None, None)
+        return (info_response, ping_response)
+
     async def send(
         self, request: PlugwiseRequest, suppress_node_errors: bool = True
     ) -> PlugwiseResponse | None:

plugwise_usb/messages/properties.py

@@ -227,7 +227,7 @@ def __init__(self, year: int = 0, month: int = 1, minutes: int = 0) -> None:
 
     def deserialize(self, val: bytes) -> None:
         """Convert data into datetime based on timestamp with offset to Y2k."""
-        if val == b"FFFFFFFF":
+        if val in (b"FFFFFFFF", b"00000000"):
             self._value = None
         else:
             CompositeType.deserialize(self, val)
@@ -389,6 +389,9 @@ def serialize(self) -> bytes:
 
     def deserialize(self, val: bytes) -> None:
         """Convert data into integer value based on log address formatted data."""
+        if val == b"00000000":
+            self._value = int(0)
+            return
         Int.deserialize(self, val)
         self._value = (self.value - LOGADDR_OFFSET) // 32
 

plugwise_usb/messages/responses.py

@@ -205,6 +205,7 @@ def _parse_params(self, response: bytes) -> bytes:
             my_val = response[: len(param)]
             param.deserialize(my_val)
             response = response[len(my_val) :]
+
         return response
 
     def __len__(self) -> int:

plugwise_usb/network/__init__.py

@@ -15,11 +15,7 @@
 from ..constants import UTF8
 from ..exceptions import CacheError, MessageError, NodeError, StickError, StickTimeout
 from ..helpers.util import validate_mac
-from ..messages.requests import (
-    CirclePlusAllowJoiningRequest,
-    NodeInfoRequest,
-    NodePingRequest,
-)
+from ..messages.requests import CirclePlusAllowJoiningRequest, NodePingRequest
 from ..messages.responses import (
     NODE_AWAKE_RESPONSE_ID,
     NODE_JOIN_ID,
@@ -289,7 +285,7 @@ def _unsubscribe_to_protocol_events(self) -> None:
             self._unsubscribe_stick_event = None
 
     # endregion
-
+        
     # region - Coordinator
     async def discover_network_coordinator(self, load: bool = False) -> bool:
         """Discover the Zigbee network coordinator (Circle+/Stealth+)."""
@@ -365,32 +361,6 @@ def _create_node_object(
             self._nodes[mac].cache_folder_create = self._cache_folder_create
             self._nodes[mac].cache_enabled = True
 
-    async def get_node_details(
-        self, mac: str, ping_first: bool
-    ) -> tuple[NodeInfoResponse | None, NodePingResponse | None]:
-        """Return node discovery type."""
-        ping_response: NodePingResponse | None = None
-        if ping_first:
-            # Define ping request with one retry
-            ping_request = NodePingRequest(
-                self._controller.send, bytes(mac, UTF8), retries=1
-            )
-            try:
-                ping_response = await ping_request.send(suppress_node_errors=True)
-            except StickError:
-                return (None, None)
-            if ping_response is None:
-                return (None, None)
-
-        info_request = NodeInfoRequest(
-            self._controller.send, bytes(mac, UTF8), retries=1
-        )
-        try:
-            info_response = await info_request.send()
-        except StickError:
-            return (None, None)
-        return (info_response, ping_response)
-
     async def _discover_battery_powered_node(
         self,
         address: int,
@@ -432,7 +402,7 @@ async def _discover_node(
 
         # Node type is unknown, so we need to discover it first
         _LOGGER.debug("Starting the discovery of node %s", mac)
-        node_info, node_ping = await self.get_node_details(mac, ping_first)
+        node_info, node_ping = await self._controller.get_node_details(mac, ping_first)
         if node_info is None:
             return False
         self._create_node_object(mac, address, node_info.node_type)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "plugwise_usb"
-version         = "v0.40.0a29"
+version         = "v0.40.0a31"
 license         = {file = "LICENSE"}
 description     = "Plugwise USB (Stick) module for Python 3."
 readme          = "README.md"

tests/stick_test_data.py

@@ -36,7 +36,20 @@
         + b"01"  # network_is_online
         + b"0098765432101234"  # circle_plus_mac
         + b"4321"  # network_id
-        + b"00",  # unknown2
+        + b"FF",  # unknown2
+    ),
+    b"\x05\x05\x03\x0300230123456789012345A0EC\r\n": (
+        "Node Info of stick 0123456789012345",
+        b"000000C1",  # Success ack
+        b"0024"  # msg_id
+        + b"0123456789012345"  # mac
+        + b"00000000"  # datetime
+        + b"00000000"  # log address 0
+        + b"00"  # relay
+        + b"80"  # hz
+        + b"653907008512"  # hw_ver
+        + b"4E0843A9"  # fw_ver
+        + b"00",  # node_type (Stick)
     ),
     b"\x05\x05\x03\x03002300987654321012341AE2\r\n": (
         "Node Info of network controller 0098765432101234",

tests/test_usb.py

@@ -459,7 +459,10 @@ async def test_stick_connect(self, monkeypatch: pytest.MonkeyPatch) -> None:
         assert await self.test_connected
         await stick.initialize()
         assert stick.mac_stick == "0123456789012345"
+        assert stick.name == "Stick 12345"
         assert stick.mac_coordinator == "0098765432101234"
+        assert stick.firmware == dt(2011, 6, 27, 8, 47, 37, tzinfo=UTC)
+        assert stick.hardware == "070085"
         assert not stick.network_discovered
         assert stick.network_state
         assert stick.network_id == 17185