evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bump pySwitchbee to 1.7.19 (#84442)

Browse source 
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Jafar Atili 2023-01-09 21:23:49 +02:00 • committed by GitHub
parent 3bb435c292
commit a8f95c36a6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 37 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

23
homeassistant/components/switchbee/coordinator.py
View file

@ -6,11 +6,11 @@ from collections.abc import Mapping
from datetime import timedelta
import logging
from switchbee.api import CentralUnitPolling, CentralUnitWsRPC
from switchbee.api.central_unit import SwitchBeeError
from switchbee.api.polling import CentralUnitPolling
from switchbee.device import DeviceType, SwitchBeeBaseDevice
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -20,15 +20,15 @@ _LOGGER = logging.getLogger(__name__)
class SwitchBeeCoordinator(DataUpdateCoordinator[Mapping[int, SwitchBeeBaseDevice]]):
"""Class to manage fetching Freedompro data API."""
"""Class to manage fetching SwitchBee data API."""
def __init__(
self,
hass: HomeAssistant,
swb_api: CentralUnitPolling,
swb_api: CentralUnitPolling | CentralUnitWsRPC,
) -> None:
"""Initialize."""
self.api: CentralUnitPolling = swb_api
self.api: CentralUnitPolling | CentralUnitWsRPC = swb_api
self._reconnect_counts: int = 0
self.mac_formatted: str | None = (
None if self.api.mac is None else format_mac(self.api.mac)
@ -38,9 +38,20 @@ class SwitchBeeCoordinator(DataUpdateCoordinator[Mapping[int, SwitchBeeBaseDevic
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=SCAN_INTERVAL_SEC),
update_interval=timedelta(seconds=SCAN_INTERVAL_SEC[type(self.api)]),
)
# Register callback for notification WsRPC
if isinstance(self.api, CentralUnitWsRPC):
self.api.subscribe_updates(self._async_handle_update)
@callback
def _async_handle_update(self, push_data: dict) -> None:
"""Manually update data and notify listeners."""
assert isinstance(self.api, CentralUnitWsRPC)
_LOGGER.debug("Received update: %s", push_data)
self.async_set_updated_data(self.api.devices)
async def _async_update_data(self) -> Mapping[int, SwitchBeeBaseDevice]:
"""Update data via library."""