mirror of
https://github.com/home-assistant/core.git
synced 2025-07-25 14:17:45 +00:00
Move inkbird coordinator logic into coordinator.py (#142517)
* Move inkbird coordinator logic into coordinator.py Not a functional change, one to one relocation * Move inkbird coordinator logic into coordinator.py Not a functional change, one to one copy * Move inkbird coordinator logic into coordinator.py Not a functional change, one to one copy
This commit is contained in:
J. Nick Koston
GitHub
parent
3f2975e93f
commit
626935ee14
@ -2,106 +2,17 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
from inkbird_ble import INKBIRDBluetoothDeviceData
|
||||
|
||||
from inkbird_ble import INKBIRDBluetoothDeviceData, SensorUpdate
|
||||
|
||||
from homeassistant.components.bluetooth import (
|
||||
BluetoothScanningMode,
|
||||
BluetoothServiceInfo,
|
||||
BluetoothServiceInfoBleak,
|
||||
async_ble_device_from_address,
|
||||
)
|
||||
from homeassistant.components.bluetooth.active_update_processor import (
|
||||
ActiveBluetoothProcessorCoordinator,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.event import async_track_time_interval
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import CONF_DEVICE_TYPE, DOMAIN
|
||||
from .coordinator import INKBIRDActiveBluetoothProcessorCoordinator
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
FALLBACK_POLL_INTERVAL = timedelta(seconds=180)
|
||||
|
||||
|
||||
class INKBIRDActiveBluetoothProcessorCoordinator(ActiveBluetoothProcessorCoordinator):
|
||||
"""Coordinator for INKBIRD Bluetooth devices."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
data: INKBIRDBluetoothDeviceData,
|
||||
) -> None:
|
||||
"""Initialize the INKBIRD Bluetooth processor coordinator."""
|
||||
self._data = data
|
||||
self._entry = entry
|
||||
address = entry.unique_id
|
||||
assert address is not None
|
||||
entry.async_on_unload(
|
||||
async_track_time_interval(
|
||||
hass, self._async_schedule_poll, FALLBACK_POLL_INTERVAL
|
||||
)
|
||||
)
|
||||
super().__init__(
|
||||
hass=hass,
|
||||
logger=_LOGGER,
|
||||
address=address,
|
||||
mode=BluetoothScanningMode.ACTIVE,
|
||||
update_method=self._async_on_update,
|
||||
needs_poll_method=self._async_needs_poll,
|
||||
poll_method=self._async_poll_data,
|
||||
)
|
||||
|
||||
async def _async_poll_data(
|
||||
self, last_service_info: BluetoothServiceInfoBleak
|
||||
) -> SensorUpdate:
|
||||
"""Poll the device."""
|
||||
return await self._data.async_poll(last_service_info.device)
|
||||
|
||||
@callback
|
||||
def _async_needs_poll(
|
||||
self, service_info: BluetoothServiceInfoBleak, last_poll: float | None
|
||||
) -> bool:
|
||||
return (
|
||||
not self.hass.is_stopping
|
||||
and self._data.poll_needed(service_info, last_poll)
|
||||
and bool(
|
||||
async_ble_device_from_address(
|
||||
self.hass, service_info.device.address, connectable=True
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
@callback
|
||||
def _async_on_update(self, service_info: BluetoothServiceInfo) -> SensorUpdate:
|
||||
"""Handle update callback from the passive BLE processor."""
|
||||
update = self._data.update(service_info)
|
||||
if (
|
||||
self._entry.data.get(CONF_DEVICE_TYPE) is None
|
||||
and self._data.device_type is not None
|
||||
):
|
||||
device_type_str = str(self._data.device_type)
|
||||
self.hass.config_entries.async_update_entry(
|
||||
self._entry,
|
||||
data={**self._entry.data, CONF_DEVICE_TYPE: device_type_str},
|
||||
)
|
||||
return update
|
||||
|
||||
@callback
|
||||
def _async_schedule_poll(self, _: datetime) -> None:
|
||||
"""Schedule a poll of the device."""
|
||||
if self._last_service_info and self._async_needs_poll(
|
||||
self._last_service_info, self._last_poll
|
||||
):
|
||||
self._debounced_poll.async_schedule_call()
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up INKBIRD BLE device from a config entry."""
|
||||
|
||||
100
homeassistant/components/inkbird/coordinator.py
Normal file
100
homeassistant/components/inkbird/coordinator.py
Normal file
@ -0,0 +1,100 @@
|
||||
"""The INKBIRD Bluetooth integration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
|
||||
from inkbird_ble import INKBIRDBluetoothDeviceData, SensorUpdate
|
||||
|
||||
from homeassistant.components.bluetooth import (
|
||||
BluetoothScanningMode,
|
||||
BluetoothServiceInfo,
|
||||
BluetoothServiceInfoBleak,
|
||||
async_ble_device_from_address,
|
||||
)
|
||||
from homeassistant.components.bluetooth.active_update_processor import (
|
||||
ActiveBluetoothProcessorCoordinator,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.event import async_track_time_interval
|
||||
|
||||
from .const import CONF_DEVICE_TYPE
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
FALLBACK_POLL_INTERVAL = timedelta(seconds=180)
|
||||
|
||||
|
||||
class INKBIRDActiveBluetoothProcessorCoordinator(ActiveBluetoothProcessorCoordinator):
|
||||
"""Coordinator for INKBIRD Bluetooth devices."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
entry: ConfigEntry,
|
||||
data: INKBIRDBluetoothDeviceData,
|
||||
) -> None:
|
||||
"""Initialize the INKBIRD Bluetooth processor coordinator."""
|
||||
self._data = data
|
||||
self._entry = entry
|
||||
address = entry.unique_id
|
||||
assert address is not None
|
||||
entry.async_on_unload(
|
||||
async_track_time_interval(
|
||||
hass, self._async_schedule_poll, FALLBACK_POLL_INTERVAL
|
||||
)
|
||||
)
|
||||
super().__init__(
|
||||
hass=hass,
|
||||
logger=_LOGGER,
|
||||
address=address,
|
||||
mode=BluetoothScanningMode.ACTIVE,
|
||||
update_method=self._async_on_update,
|
||||
needs_poll_method=self._async_needs_poll,
|
||||
poll_method=self._async_poll_data,
|
||||
)
|
||||
|
||||
async def _async_poll_data(
|
||||
self, last_service_info: BluetoothServiceInfoBleak
|
||||
) -> SensorUpdate:
|
||||
"""Poll the device."""
|
||||
return await self._data.async_poll(last_service_info.device)
|
||||
|
||||
@callback
|
||||
def _async_needs_poll(
|
||||
self, service_info: BluetoothServiceInfoBleak, last_poll: float | None
|
||||
) -> bool:
|
||||
return (
|
||||
not self.hass.is_stopping
|
||||
and self._data.poll_needed(service_info, last_poll)
|
||||
and bool(
|
||||
async_ble_device_from_address(
|
||||
self.hass, service_info.device.address, connectable=True
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
@callback
|
||||
def _async_on_update(self, service_info: BluetoothServiceInfo) -> SensorUpdate:
|
||||
"""Handle update callback from the passive BLE processor."""
|
||||
update = self._data.update(service_info)
|
||||
if (
|
||||
self._entry.data.get(CONF_DEVICE_TYPE) is None
|
||||
and self._data.device_type is not None
|
||||
):
|
||||
device_type_str = str(self._data.device_type)
|
||||
self.hass.config_entries.async_update_entry(
|
||||
self._entry,
|
||||
data={**self._entry.data, CONF_DEVICE_TYPE: device_type_str},
|
||||
)
|
||||
return update
|
||||
|
||||
@callback
|
||||
def _async_schedule_poll(self, _: datetime) -> None:
|
||||
"""Schedule a poll of the device."""
|
||||
if self._last_service_info and self._async_needs_poll(
|
||||
self._last_service_info, self._last_poll
|
||||
):
|
||||
self._debounced_poll.async_schedule_call()
|
||||
@ -12,8 +12,8 @@ from inkbird_ble import (
|
||||
)
|
||||
from sensor_state_data import SensorDeviceClass
|
||||
|
||||
from homeassistant.components.inkbird import FALLBACK_POLL_INTERVAL
|
||||
from homeassistant.components.inkbird.const import CONF_DEVICE_TYPE, DOMAIN
|
||||
from homeassistant.components.inkbird.coordinator import FALLBACK_POLL_INTERVAL
|
||||
from homeassistant.components.sensor import ATTR_STATE_CLASS
|
||||
from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_UNIT_OF_MEASUREMENT
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user