Skip to content

Commit

Permalink
Improve typing
Browse files Browse the repository at this point in the history
  • Loading branch information
natekspencer committed Aug 2, 2022
1 parent f208136 commit 30365e9
Show file tree
Hide file tree
Showing 17 changed files with 315 additions and 209 deletions.
3 changes: 2 additions & 1 deletion pylintrc
Original file line number Diff line number Diff line change
Expand Up @@ -13,4 +13,5 @@ persistent=no
disable=
line-too-long,
too-few-public-methods,
too-many-arguments
too-many-arguments,
too-many-public-methods
10 changes: 6 additions & 4 deletions vivintpy/account.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,8 @@ def __init__(
"""Initialize an account."""
self.__connected = False
self.__load_devices = False
self.__pubnub: PubNubAsyncio = None
self.__pubnub_listener: VivintPubNubSubscribeListener = None
self.__pubnub: PubNubAsyncio | None = None
self.__pubnub_listener: VivintPubNubSubscribeListener | None = None
self.vivintskyapi = VivintSkyApi(
username=username,
password=password,
Expand All @@ -48,7 +48,7 @@ def __init__(
self.systems: list[System] = []

@property
def connected(self):
def connected(self) -> bool:
"""Return True if connected."""
return self.__connected

Expand Down Expand Up @@ -94,6 +94,7 @@ async def __pubnub_unsubscribe_all(self) -> None:
The pubnub code doesn't properly wait for the unsubscribe event to finish or to
be canceled, so we have to manually do it by finding the coroutine in asyncio.
"""
assert self.__pubnub
self.__pubnub.unsubscribe_all()
tasks = [
task
Expand Down Expand Up @@ -129,7 +130,8 @@ async def refresh(self, authuser_data: dict = None) -> None:
# is this an existing account_system?
system = first_or_none(
self.systems,
lambda system: system.id == system_data[SystemAttribute.PANEL_ID],
lambda system, system_data=system_data: system.id # type: ignore
== system_data[SystemAttribute.PANEL_ID],
)
if system:
await system.refresh()
Expand Down
30 changes: 14 additions & 16 deletions vivintpy/devices/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

def get_device_class(device_type: str) -> Type[VivintDevice]:
"""Map a device_type string to the class that implements that device."""
from . import UnknownDevice
# pylint: disable=import-outside-toplevel
from .camera import Camera
from .door_lock import DoorLock
from .garage_door import GarageDoor
Expand All @@ -30,9 +30,9 @@ def get_device_class(device_type: str) -> Type[VivintDevice]:
DeviceType.CAMERA: Camera,
DeviceType.DOOR_LOCK: DoorLock,
DeviceType.GARAGE_DOOR: GarageDoor,
DeviceType.MULTILEVEL_SWITCH: MultilevelSwitch,
DeviceType.MULTI_LEVEL_SWITCH: MultilevelSwitch,
DeviceType.THERMOSTAT: Thermostat,
DeviceType.TOUCH_PANEL: VivintDevice,
DeviceType.PANEL: VivintDevice,
DeviceType.WIRELESS_SENSOR: WirelessSensor,
}

Expand All @@ -46,15 +46,15 @@ def __init__(self, data: dict, alarm_panel: AlarmPanel = None) -> None:
"""Initialize a device."""
super().__init__(data)
self.alarm_panel = alarm_panel
self._manufacturer = None
self._model = None
self._manufacturer: str | None = None
self._model: str | None = None
self._capabilities = (
{
CapabilityCategoryType(capability_category.get(Attribute.TYPE)): [
CapabilityType(capability)
for capability in capability_category.get(Attribute.CAPABILITY)
]
for capability_category in data.get(Attribute.CAPABILITY_CATEGORY)
for capability_category in data.get(Attribute.CAPABILITY_CATEGORY, [])
}
if data.get(Attribute.CAPABILITY_CATEGORY)
else None
Expand All @@ -66,9 +66,9 @@ def __repr__(self) -> str:
return f"<{self.__class__.__name__} {self.id}, {self.name}>"

@property
def id(self) -> int:
def id(self) -> int: # pylint: disable=invalid-name
"""Device's id."""
return self.data[Attribute.ID]
return int(self.data[Attribute.ID])

@property
def is_valid(self) -> bool:
Expand Down Expand Up @@ -99,7 +99,7 @@ def capabilities(
@property
def device_type(self) -> DeviceType:
"""Return the device type."""
return DeviceType(self.data[Attribute.TYPE])
return DeviceType(self.data.get(Attribute.TYPE))

@property
def has_battery(self) -> bool:
Expand Down Expand Up @@ -136,7 +136,7 @@ def model(self) -> str | None:
@property
def panel_id(self) -> int:
"""Return the id of the panel this device is associated to."""
return self.data.get(Attribute.PANEL_ID)
return int(self.data[Attribute.PANEL_ID])

@property
def parent(self) -> VivintDevice | None:
Expand All @@ -152,7 +152,7 @@ def serial_number(self) -> str | None:
return serial_number

@property
def software_version(self) -> str:
def software_version(self) -> str | None:
"""Return the software version of this device, if any."""
# panels
current_software_version = self.data.get(Attribute.CURRENT_SOFTWARE_VERSION)
Expand All @@ -175,10 +175,10 @@ def vivintskyapi(self) -> VivintSkyApi:
assert self.alarm_panel, """no alarm panel set for this device"""
return self.alarm_panel.system.vivintskyapi

def get_zwave_details(self):
def get_zwave_details(self) -> None:
"""Get Z-Wave details."""
if self.data.get("zpd") is None:
return None
return

result = get_zwave_device_info(
self.data.get("manid"),
Expand All @@ -200,8 +200,6 @@ def get_zwave_details(self):
else:
self._model = "Unknown"

return [self._manufacturer, self._model]

def emit(self, event_name: str, data: dict) -> None:
"""Add device data and then send to parent."""
if data.get(DEVICE) is None:
Expand All @@ -217,7 +215,7 @@ class BypassTamperDevice(VivintDevice):
def is_bypassed(self) -> bool:
"""Return True if the device is bypassed."""
return (
self.data.get(Attribute.BYPASSED, ZoneBypass.UNBYPASSED)
int(self.data.get(Attribute.BYPASSED, ZoneBypass.UNBYPASSED))
!= ZoneBypass.UNBYPASSED
)

Expand Down
54 changes: 33 additions & 21 deletions vivintpy/devices/alarm_panel.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from ..const import PubNubMessageAttribute, PubNubOperatorAttribute, SystemAttribute
from ..enums import ArmedState, DeviceType
from ..exceptions import VivintSkyApiError
from ..utils import add_async_job, first_or_none
from ..utils import add_async_job, first_or_none, send_deprecation_warning
from ..vivintskyapi import VivintSkyApi
from . import VivintDevice, get_device_class

Expand All @@ -35,11 +35,11 @@ def __init__(self, data: dict, system: System):
self.__parse_data(data=data, init=True)

# store a reference to the physical panel device
self.__panel_credentials = None
self.__panel_credentials: dict = {}
self.__panel = first_or_none(
self.devices,
lambda device: DeviceType(device.data.get(Attribute.TYPE))
== DeviceType.TOUCH_PANEL,
== DeviceType.PANEL,
)

@property
Expand All @@ -50,20 +50,20 @@ def vivintskyapi(self) -> VivintSkyApi:
@property
def id(self) -> int:
"""Panel's id."""
return self.data[Attribute.PANEL_ID]
return int(self.data[Attribute.PANEL_ID])

@property
def name(self) -> str:
"""Panel's name."""
return self.system.name

@property
def manufacturer(self):
def manufacturer(self) -> str:
"""Return Vivint as the manufacturer of this panel."""
return "Vivint"

@property
def model(self):
def model(self) -> str:
"""Return the model of the physical panel."""
return (
"Sky Control"
Expand All @@ -72,32 +72,38 @@ def model(self):
)

@property
def software_version(self) -> str:
def software_version(self) -> str | None:
"""Return the software version of the panel."""
return self.__panel.software_version if self.__panel else None

@property
def partition_id(self) -> int:
"""Panel's partition id."""
return self.data[Attribute.PARTITION_ID]
return int(self.data[Attribute.PARTITION_ID])

@property
def is_disarmed(self) -> bool:
"""Return True if alarm is disarmed."""
return self.get_armed_state() == ArmedState.DISARMED
return self.state == ArmedState.DISARMED

@property
def is_armed_away(self) -> bool:
"""Return True if alarm is in armed away state."""
return self.get_armed_state() == ArmedState.ARMED_AWAY
return self.state == ArmedState.ARMED_AWAY

@property
def is_armed_stay(self) -> bool:
"""Return True if alarm is in armed stay state."""
return self.get_armed_state() == ArmedState.ARMED_STAY
return self.state == ArmedState.ARMED_STAY

def get_armed_state(self):
@property
def state(self) -> ArmedState:
"""Return the panel's armed state."""
return ArmedState(self.data.get(Attribute.STATE)) # type: ignore

def get_armed_state(self) -> Any:
"""Return the panel's arm state."""
send_deprecation_warning("method get_armed_state", "property state")
return self.data[Attribute.STATE]

async def set_armed_state(self, state: int) -> None:
Expand Down Expand Up @@ -130,7 +136,7 @@ def get_devices(
device_types: set[Type[VivintDevice]] = None,
) -> list[VivintDevice]:
"""Get a list of associated devices."""
devices: list[VivintDevice] = None
devices: list[VivintDevice] = []

if device_types:
devices = [
Expand All @@ -141,7 +147,7 @@ def get_devices(

return devices

def refresh(self, data: dict[str, Any], new_device: bool = False) -> None:
def refresh(self, data: dict, new_device: bool = False) -> None:
"""Refresh the alarm panel."""
if not new_device:
self.update_data(data, override=True)
Expand Down Expand Up @@ -181,7 +187,8 @@ def handle_pubnub_message(self, message: dict) -> None:
add_async_job(self.handle_new_device, device_id)
else:
device = first_or_none(
self.devices, lambda device: device.id == device_id
self.devices,
lambda device, device_id=device_id: device.id == device_id, # type: ignore
)
if not device:
_LOGGER.debug(
Expand All @@ -193,7 +200,9 @@ def handle_pubnub_message(self, message: dict) -> None:
# for the sake of consistency, we also need to update the panel's raw data
raw_device_data = first_or_none(
self.data[Attribute.DEVICES],
lambda raw_device_data: raw_device_data["_id"]
lambda raw_device_data, device_data=device_data: raw_device_data[ # type: ignore
"_id"
]
== device_data["_id"],
)

Expand All @@ -207,12 +216,14 @@ def handle_pubnub_message(self, message: dict) -> None:
self.emit(DEVICE_DELETED, {"device": device})
else:
device.handle_pubnub_message(device_data)
assert raw_device_data
raw_device_data.update(device_data)

async def handle_new_device(self, device_id: int) -> None:
"""Handle a new device."""
try:
device = first_or_none(self.devices, lambda device: device.id == device_id)
assert device
while not device.is_valid:
await asyncio.sleep(1)
if device.id in self.unregistered_devices:
Expand All @@ -224,30 +235,31 @@ async def handle_new_device(self, device_id: int) -> None:
except VivintSkyApiError:
_LOGGER.error("Error getting new device data for device %s", device_id)

def __parse_data(self, data: dict[str, Any], init: bool = False) -> None:
def __parse_data(self, data: dict, init: bool = False) -> None:
"""Parse the alarm panel data."""
for device_data in data[Attribute.DEVICES]:
device: VivintDevice = None
device: VivintDevice | None = None
if not init:
device = first_or_none(
self.devices,
lambda device: device.id == device_data[Attribute.ID],
lambda device, device_data=device_data: device.id # type: ignore
== device_data[Attribute.ID],
)
if device:
device.update_data(device_data, override=True)
else:
self.__parse_device_data(device_data=device_data)

if data.get(Attribute.UNREGISTERED):
self.unregistered_devices: dict[int, tuple] = {
self.unregistered_devices = {
device[Attribute.ID]: (
device[Attribute.NAME],
DeviceType(device[Attribute.TYPE]),
)
for device in data[Attribute.UNREGISTERED]
}

def __parse_device_data(self, device_data: dict[str, Any]) -> None:
def __parse_device_data(self, device_data: dict) -> None:
"""Parse device data and optionally emit a device discovered event."""
device_class = get_device_class(device_data[Attribute.TYPE])
device = device_class(device_data, self)
Expand Down
Loading

0 comments on commit 30365e9

Please sign in to comment.