Handle firmware update events better

test/model/test_node.py

@@ -1,6 +1,5 @@
 """Test the node model."""
 import json
-from zwave_js_server.model.value import ConfigurationValue
 
 import pytest
 
@@ -14,7 +13,9 @@
 from zwave_js_server.event import Event
 from zwave_js_server.exceptions import UnwriteableValue
 from zwave_js_server.model import node as node_pkg
+from zwave_js_server.model.firmware import FirmwareUpdateStatus
 from zwave_js_server.model.node import Node, NodeStatus
+from zwave_js_server.model.value import ConfigurationValue
 
 from .. import load_fixture
 
@@ -660,3 +661,41 @@ async def test_refresh_values(multisensor_6, uuid4, mock_command):
         "commandClass": 112,
         "messageId": uuid4,
     }
+
+
+async def test_firmware_events(wallmote_central_scene: Node):
+    """Test firmware events."""
+    node = wallmote_central_scene
+
+    event = Event(
+        type="firmware update progress",
+        data={
+            "source": "node",
+            "event": "firmware update progress",
+            "nodeId": 35,
+            "sentFragments": 1,
+            "totalFragments": 10,
+        },
+    )
+
+    node.handle_firmware_update_progress(event)
+    assert event.data["firmware_update_progress"].sent_fragments == 1
+    assert event.data["firmware_update_progress"].total_fragments == 10
+
+    event = Event(
+        type="firmware update finished",
+        data={
+            "source": "node",
+            "event": "firmware update finished",
+            "nodeId": 35,
+            "status": 255,
+            "waitTime": 10,
+        },
+    )
+
+    node.handle_firmware_update_finished(event)
+    assert (
+        event.data["firmware_update_finished"].status
+        == FirmwareUpdateStatus.OK_RESTART_PENDING
+    )
+    assert event.data["firmware_update_finished"].wait_time == 10

zwave_js_server/model/firmware.py

@@ -0,0 +1,79 @@
+"""Provide a model for Z-Wave firmware."""
+from enum import IntEnum
+from typing import TYPE_CHECKING, Optional, TypedDict
+
+if TYPE_CHECKING:
+    from .node import Node
+
+
+class FirmwareUpdateStatus(IntEnum):
+    """Enum with all Firmware update status values.
+
+    https://zwave-js.github.io/node-zwave-js/#/api/node?id=status
+    """
+
+    ERROR_TIMEOUT = -1
+    ERROR_CHECKSUM = 0
+    ERROR_TRANSMISSION_FAILED = 1
+    ERROR_INVALID_MANUFACTURER_ID = 2
+    ERROR_INVALID_FIRMWARE_ID = 3
+    ERROR_INVALID_FIRMWARE_TARGET = 4
+    ERROR_INVALID_HEADER_INFORMATION = 5
+    ERROR_INVALID_HEADER_FORMAT = 6
+    ERROR_INSUFFICIENT_MEMORY = 7
+    ERROR_INVALID_HARDWARE_VERSION = 8
+    OK_WAITING_FOR_ACTIVATION = 253
+    OK_NO_RESTART = 254
+    OK_RESTART_PENDING = 255
+
+
+class FirmwareUpdateProgressDataType(TypedDict):
+    """Represent a firmware update progress event dict type."""
+
+    sentFragments: int  # required
+    totalFragments: int  # required
+
+
+class FirmwareUpdateProgress:
+    """Model for firmware update progress event."""
+
+    def __init__(self, node: "Node", data: FirmwareUpdateProgressDataType) -> None:
+        """Initialize."""
+        self.data = data
+        self.node = node
+
+    @property
+    def sent_fragments(self) -> int:
+        """Return the number of fragments sent to the device so far."""
+        return self.data["sentFragments"]
+
+    @property
+    def total_fragments(self) -> int:
+        """Return the total number of fragments that need to be sent to the device."""
+        return self.data["totalFragments"]
+
+
+class FirmwareUpdateFinishedDataType(TypedDict, total=False):
+    """Represent a firmware update finished event dict type."""
+
+    status: int  # required
+    waitTime: int
+
+
+class FirmwareUpdateFinished:
+    """Model for firmware update finished event."""
+
+    def __init__(self, node: "Node", data: FirmwareUpdateFinishedDataType) -> None:
+        """Initialize."""
+        self.data = data
+        self.node = node
+
+    @property
+    def status(self) -> FirmwareUpdateStatus:
+        """Return the firmware update status."""
+        return FirmwareUpdateStatus(self.data["status"])
+
+    @property
+    def wait_time(self) -> Optional[int]:
+        """Return the wait time in seconds before the device is functional again."""
+        return self.data.get("waitTime")

zwave_js_server/model/node.py

@@ -10,6 +10,12 @@
 from .device_class import DeviceClass, DeviceClassDataType
 from .device_config import DeviceConfig, DeviceConfigDataType
 from .endpoint import Endpoint, EndpointDataType
+from .firmware import (
+    FirmwareUpdateFinished,
+    FirmwareUpdateFinishedDataType,
+    FirmwareUpdateProgress,
+    FirmwareUpdateProgressDataType,
+)
 from .notification import (
     EntryControlNotification,
     EntryControlNotificationDataType,
@@ -546,6 +552,12 @@ def handle_notification(self, event: Event) -> None:
 
     def handle_firmware_update_progress(self, event: Event) -> None:
         """Process a node firmware update progress event."""
+        event.data["firmware_update_progress"] = FirmwareUpdateProgress(
+            self, cast(FirmwareUpdateProgressDataType, event.data)
+        )
 
     def handle_firmware_update_finished(self, event: Event) -> None:
         """Process a node firmware update finished event."""
+        event.data["firmware_update_finished"] = FirmwareUpdateFinished(
+            self, cast(FirmwareUpdateFinishedDataType, event.data)
+        )