Add type hints in FibaroController (#101494)

This commit is contained in:
rappenze 2023-10-06 10:13:30 +02:00 committed by GitHub
parent 67dfd1a86b
commit d009ff8b01
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -8,6 +8,7 @@ from typing import Any
from pyfibaro.fibaro_client import FibaroClient from pyfibaro.fibaro_client import FibaroClient
from pyfibaro.fibaro_device import DeviceModel from pyfibaro.fibaro_device import DeviceModel
from pyfibaro.fibaro_room import RoomModel
from pyfibaro.fibaro_scene import SceneModel from pyfibaro.fibaro_scene import SceneModel
from requests.exceptions import HTTPError from requests.exceptions import HTTPError
@ -86,14 +87,14 @@ class FibaroController:
# Whether to import devices from plugins # Whether to import devices from plugins
self._import_plugins = config[CONF_IMPORT_PLUGINS] self._import_plugins = config[CONF_IMPORT_PLUGINS]
self._room_map = None # Mapping roomId to room object self._room_map: dict[int, RoomModel] # Mapping roomId to room object
self._device_map = None # Mapping deviceId to device object self._device_map: dict[int, DeviceModel] # Mapping deviceId to device object
self.fibaro_devices: dict[Platform, list[DeviceModel]] = defaultdict( self.fibaro_devices: dict[Platform, list[DeviceModel]] = defaultdict(
list list
) # List of devices by entity platform ) # List of devices by entity platform
# All scenes # All scenes
self._scenes: list[SceneModel] = [] self._scenes: list[SceneModel] = []
self._callbacks: dict[Any, Any] = {} # Update value callbacks by deviceId self._callbacks: dict[int, list[Any]] = {} # Update value callbacks by deviceId
self.hub_serial: str # Unique serial number of the hub self.hub_serial: str # Unique serial number of the hub
self.hub_name: str # The friendly name of the hub self.hub_name: str # The friendly name of the hub
self.hub_software_version: str self.hub_software_version: str
@ -101,7 +102,7 @@ class FibaroController:
# Device infos by fibaro device id # Device infos by fibaro device id
self._device_infos: dict[int, DeviceInfo] = {} self._device_infos: dict[int, DeviceInfo] = {}
def connect(self): def connect(self) -> bool:
"""Start the communication with the Fibaro controller.""" """Start the communication with the Fibaro controller."""
connected = self._client.connect() connected = self._client.connect()
@ -138,15 +139,15 @@ class FibaroController:
except Exception as ex: except Exception as ex:
raise FibaroConnectFailed from ex raise FibaroConnectFailed from ex
def enable_state_handler(self): def enable_state_handler(self) -> None:
"""Start StateHandler thread for monitoring updates.""" """Start StateHandler thread for monitoring updates."""
self._client.register_update_handler(self._on_state_change) self._client.register_update_handler(self._on_state_change)
def disable_state_handler(self): def disable_state_handler(self) -> None:
"""Stop StateHandler thread used for monitoring updates.""" """Stop StateHandler thread used for monitoring updates."""
self._client.unregister_update_handler() self._client.unregister_update_handler()
def _on_state_change(self, state): def _on_state_change(self, state: Any) -> None:
"""Handle change report received from the HomeCenter.""" """Handle change report received from the HomeCenter."""
callback_set = set() callback_set = set()
for change in state.get("changes", []): for change in state.get("changes", []):
@ -177,12 +178,12 @@ class FibaroController:
for callback in self._callbacks[item]: for callback in self._callbacks[item]:
callback() callback()
def register(self, device_id, callback): def register(self, device_id: int, callback: Any) -> None:
"""Register device with a callback for updates.""" """Register device with a callback for updates."""
self._callbacks.setdefault(device_id, []) self._callbacks.setdefault(device_id, [])
self._callbacks[device_id].append(callback) self._callbacks[device_id].append(callback)
def get_children(self, device_id): def get_children(self, device_id: int) -> list[DeviceModel]:
"""Get a list of child devices.""" """Get a list of child devices."""
return [ return [
device device
@ -190,7 +191,7 @@ class FibaroController:
if device.parent_fibaro_id == device_id if device.parent_fibaro_id == device_id
] ]
def get_children2(self, device_id, endpoint_id): def get_children2(self, device_id: int, endpoint_id: int) -> list[DeviceModel]:
"""Get a list of child devices for the same endpoint.""" """Get a list of child devices for the same endpoint."""
return [ return [
device device
@ -199,7 +200,7 @@ class FibaroController:
and (not device.has_endpoint_id or device.endpoint_id == endpoint_id) and (not device.has_endpoint_id or device.endpoint_id == endpoint_id)
] ]
def get_siblings(self, device): def get_siblings(self, device: DeviceModel) -> list[DeviceModel]:
"""Get the siblings of a device.""" """Get the siblings of a device."""
if device.has_endpoint_id: if device.has_endpoint_id:
return self.get_children2( return self.get_children2(
@ -209,7 +210,7 @@ class FibaroController:
return self.get_children(self._device_map[device.fibaro_id].parent_fibaro_id) return self.get_children(self._device_map[device.fibaro_id].parent_fibaro_id)
@staticmethod @staticmethod
def _map_device_to_platform(device: Any) -> Platform | None: def _map_device_to_platform(device: DeviceModel) -> Platform | None:
"""Map device to HA device type.""" """Map device to HA device type."""
# Use our lookup table to identify device type # Use our lookup table to identify device type
platform: Platform | None = None platform: Platform | None = None
@ -248,7 +249,7 @@ class FibaroController:
if device.parent_fibaro_id <= 1: if device.parent_fibaro_id <= 1:
return return
master_entity: Any | None = None master_entity: DeviceModel | None = None
if device.parent_fibaro_id == 1: if device.parent_fibaro_id == 1:
master_entity = device master_entity = device
else: else:
@ -271,7 +272,7 @@ class FibaroController:
via_device=(DOMAIN, self.hub_serial), via_device=(DOMAIN, self.hub_serial),
) )
def get_device_info(self, device: Any) -> DeviceInfo: def get_device_info(self, device: DeviceModel) -> DeviceInfo:
"""Get the device info by fibaro device id.""" """Get the device info by fibaro device id."""
if device.fibaro_id in self._device_infos: if device.fibaro_id in self._device_infos:
return self._device_infos[device.fibaro_id] return self._device_infos[device.fibaro_id]
@ -289,7 +290,7 @@ class FibaroController:
"""Return list of scenes.""" """Return list of scenes."""
return self._scenes return self._scenes
def _read_devices(self): def _read_devices(self) -> None:
"""Read and process the device list.""" """Read and process the device list."""
devices = self._client.read_devices() devices = self._client.read_devices()
self._device_map = {} self._device_map = {}