Skip to content

Commit

Permalink
Use dataclasses where it makes sense to simplify code (#613)
Browse files Browse the repository at this point in the history
* Use dataclasses where it makes sense to simplify code

* more dataclasses

* cleanup

* simplify

* mypy
  • Loading branch information
raman325 authored Mar 22, 2023
1 parent e655ce1 commit 6de977f
Show file tree
Hide file tree
Showing 7 changed files with 260 additions and 489 deletions.
49 changes: 19 additions & 30 deletions zwave_js_server/model/controller/firmware.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Provide a model for Z-Wave controller firmware."""
from __future__ import annotations

from dataclasses import dataclass
from dataclasses import dataclass, field
from enum import IntEnum
from typing import TypedDict

Expand Down Expand Up @@ -59,27 +59,20 @@ class ControllerFirmwareUpdateProgressDataType(TypedDict):
progress: float


@dataclass
class ControllerFirmwareUpdateProgress:
"""Model for a controller firmware update progress data."""

def __init__(self, data: ControllerFirmwareUpdateProgressDataType) -> None:
"""Initialize."""
self.data = data

@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"]
data: ControllerFirmwareUpdateProgressDataType
sent_fragments: int = field(init=False)
total_fragments: int = field(init=False)
progress: float = field(init=False)

@property
def progress(self) -> float:
"""Return progress."""
return float(self.data["progress"])
def __post_init__(self) -> None:
"""Post initialize."""
self.sent_fragments = self.data["sentFragments"]
self.total_fragments = self.data["totalFragments"]
self.progress = float(self.data["progress"])


class ControllerFirmwareUpdateResultDataType(TypedDict):
Expand All @@ -89,19 +82,15 @@ class ControllerFirmwareUpdateResultDataType(TypedDict):
success: bool


@dataclass
class ControllerFirmwareUpdateResult:
"""Model for controller firmware update result data."""

def __init__(self, data: ControllerFirmwareUpdateResultDataType) -> None:
"""Initialize."""
self.data = data

@property
def status(self) -> ControllerFirmwareUpdateStatus:
"""Return the firmware update status."""
return ControllerFirmwareUpdateStatus(self.data["status"])
data: ControllerFirmwareUpdateResultDataType
status: ControllerFirmwareUpdateStatus = field(init=False)
success: bool = field(init=False)

@property
def success(self) -> bool:
"""Return whether the firmware update was successful."""
return self.data["success"]
def __post_init__(self) -> None:
"""Post initialize."""
self.status = ControllerFirmwareUpdateStatus(self.data["status"])
self.success = self.data["success"]
106 changes: 34 additions & 72 deletions zwave_js_server/model/controller/statistics.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Provide a model for the Z-Wave JS controller's statistics."""
from __future__ import annotations

from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, TypedDict

from ..statistics import RouteStatistics, RouteStatisticsDataType
Expand All @@ -21,27 +21,17 @@ class ControllerLifelineRoutesDataType(TypedDict):
class ControllerLifelineRoutes:
"""Represent controller lifeline routes."""

def __init__(
self, client: "Client", data: ControllerLifelineRoutesDataType
) -> None:
"""Initialize controller lifeline routes."""
self.data = data
self._lwr = None
client: "Client"
data: ControllerLifelineRoutesDataType
lwr: RouteStatistics | None = field(init=False, default=None)
nlwr: RouteStatistics | None = field(init=False, default=None)

def __post_init__(self) -> None:
"""Post initialize."""
if lwr := self.data.get("lwr"):
self._lwr = RouteStatistics(client, lwr)
self._nlwr = None
self.lwr = RouteStatistics(self.client, lwr)
if nlwr := self.data.get("nlwr"):
self._nlwr = RouteStatistics(client, nlwr)

@property
def lwr(self) -> RouteStatistics | None:
"""Return the last working route from the controller to this node."""
return self._lwr

@property
def nlwr(self) -> RouteStatistics | None:
"""Return the next to last working route from the controller to this node."""
return self._nlwr
self.nlwr = RouteStatistics(self.client, nlwr)


class ControllerStatisticsDataType(TypedDict):
Expand All @@ -59,12 +49,24 @@ class ControllerStatisticsDataType(TypedDict):
timeoutCallback: int


@dataclass
class ControllerStatistics:
"""Represent a controller statistics update."""

def __init__(self, data: ControllerStatisticsDataType | None = None) -> None:
"""Initialize controller statistics."""
self.data = data or ControllerStatisticsDataType(
data: ControllerStatisticsDataType | None = None
messages_tx: int = field(init=False)
messages_rx: int = field(init=False)
messages_dropped_rx: int = field(init=False)
messages_dropped_tx: int = field(init=False)
nak: int = field(init=False)
can: int = field(init=False)
timeout_ack: int = field(init=False)
timeout_response: int = field(init=False)
timeout_callback: int = field(init=False)

def __post_init__(self) -> None:
"""Post initialize."""
data = self.data or ControllerStatisticsDataType(
CAN=0,
messagesDroppedRX=0,
messagesDroppedTX=0,
Expand All @@ -75,52 +77,12 @@ def __init__(self, data: ControllerStatisticsDataType | None = None) -> None:
timeoutCallback=0,
timeoutResponse=0,
)

@property
def messages_tx(self) -> int:
"""Return number of messages successfully sent to controller."""
return self.data["messagesTX"]

@property
def messages_rx(self) -> int:
"""Return number of messages received by controller."""
return self.data["messagesRX"]

@property
def messages_dropped_rx(self) -> int:
"""Return number of messages from controller that were dropped by host."""
return self.data["messagesDroppedRX"]

@property
def messages_dropped_tx(self) -> int:
"""
Return number of outgoing messages that were dropped.
These messages could not be sent.
"""
return self.data["messagesDroppedTX"]

@property
def nak(self) -> int:
"""Return number of messages that controller did not accept."""
return self.data["NAK"]

@property
def can(self) -> int:
"""Return number of collisions while sending a message to controller."""
return self.data["CAN"]

@property
def timeout_ack(self) -> int:
"""Return number of transmission attempts without an ACK from controller."""
return self.data["timeoutACK"]

@property
def timeout_response(self) -> int:
"""Return number of transmission attempts where controller response timed out."""
return self.data["timeoutResponse"]

@property
def timeout_callback(self) -> int:
"""Return number of transmission attempts where controller callback timed out."""
return self.data["timeoutCallback"]
self.messages_tx = data["messagesTX"]
self.messages_rx = data["messagesRX"]
self.messages_dropped_rx = data["messagesDroppedRX"]
self.messages_dropped_tx = data["messagesDroppedTX"]
self.nak = data["NAK"]
self.can = data["CAN"]
self.timeout_ack = data["timeoutACK"]
self.timeout_response = data["timeoutResponse"]
self.timeout_callback = data["timeoutCallback"]
Loading

0 comments on commit 6de977f

Please sign in to comment.