From 4d1ed1cb8f74f200b72c0d66ba64ded0ae6f189e Mon Sep 17 00:00:00 2001 From: Raman Gupta <7243222+raman325@users.noreply.github.com> Date: Mon, 10 May 2021 01:39:40 -0400 Subject: [PATCH] Handle firmware events better --- test/model/test_node.py | 41 +++++++++++++++- zwave_js_server/model/firmware.py | 79 +++++++++++++++++++++++++++++++ zwave_js_server/model/node.py | 12 +++++ 3 files changed, 131 insertions(+), 1 deletion(-) create mode 100644 zwave_js_server/model/firmware.py diff --git a/test/model/test_node.py b/test/model/test_node.py index 9ed8e9f09..65a3e0ae8 100644 --- a/test/model/test_node.py +++ b/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 diff --git a/zwave_js_server/model/firmware.py b/zwave_js_server/model/firmware.py new file mode 100644 index 000000000..f722c0d72 --- /dev/null +++ b/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") diff --git a/zwave_js_server/model/node.py b/zwave_js_server/model/node.py index db8cf0cde..f5e059eeb 100644 --- a/zwave_js_server/model/node.py +++ b/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) + )