Extract Syncthru coordinator in separate file (#142620)

This commit is contained in:
Joost Lekkerkerker 2025-04-10 21:24:38 +02:00 committed by GitHub
parent bb3c2175bc
commit bf0d2e9bd2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 76 additions and 86 deletions

View File

@ -2,21 +2,15 @@
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from pysyncthru import ConnectionMode, SyncThru, SyncThruAPINotSupported
from pysyncthru import SyncThru, SyncThruAPINotSupported
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL, Platform
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client, device_registry as dr
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.helpers import device_registry as dr
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
from .coordinator import SyncthruCoordinator
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
@ -24,41 +18,9 @@ PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up config entry."""
session = aiohttp_client.async_get_clientsession(hass)
hass.data.setdefault(DOMAIN, {})
printer = SyncThru(
entry.data[CONF_URL], session, connection_mode=ConnectionMode.API
)
async def async_update_data() -> SyncThru:
"""Fetch data from the printer."""
try:
async with asyncio.timeout(10):
await printer.update()
except SyncThruAPINotSupported as api_error:
# if an exception is thrown, printer does not support syncthru
_LOGGER.debug(
"Configured printer at %s does not provide SyncThru JSON API",
printer.url,
exc_info=api_error,
)
raise
# if the printer is offline, we raise an UpdateFailed
if printer.is_unknown_state():
raise UpdateFailed(f"Configured printer at {printer.url} does not respond.")
return printer
coordinator = DataUpdateCoordinator[SyncThru](
hass,
_LOGGER,
config_entry=entry,
name=DOMAIN,
update_method=async_update_data,
update_interval=timedelta(seconds=30),
)
coordinator = SyncthruCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = coordinator
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
if isinstance(coordinator.last_exception, SyncThruAPINotSupported):
# this means that the printer does not support the syncthru JSON API
# and the config should simply be discarded
@ -67,12 +29,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
configuration_url=printer.url,
connections=device_connections(printer),
configuration_url=coordinator.syncthru.url,
connections=device_connections(coordinator.syncthru),
manufacturer="Samsung",
identifiers=device_identifiers(printer),
model=printer.model(),
name=printer.hostname(),
identifiers=device_identifiers(coordinator.syncthru),
model=coordinator.syncthru.model(),
name=coordinator.syncthru.hostname(),
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from pysyncthru import SyncThru, SyncthruState
from pysyncthru import SyncthruState
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -13,12 +13,9 @@ from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import device_identifiers
from . import SyncthruCoordinator, device_identifiers
from .const import DOMAIN
SYNCTHRU_STATE_PROBLEM = {
@ -39,9 +36,7 @@ async def async_setup_entry(
) -> None:
"""Set up from config entry."""
coordinator: DataUpdateCoordinator[SyncThru] = hass.data[DOMAIN][
config_entry.entry_id
]
coordinator: SyncthruCoordinator = hass.data[DOMAIN][config_entry.entry_id]
name: str = config_entry.data[CONF_NAME]
entities = [
@ -52,12 +47,10 @@ async def async_setup_entry(
async_add_entities(entities)
class SyncThruBinarySensor(
CoordinatorEntity[DataUpdateCoordinator[SyncThru]], BinarySensorEntity
):
class SyncThruBinarySensor(CoordinatorEntity[SyncthruCoordinator], BinarySensorEntity):
"""Implementation of an abstract Samsung Printer binary sensor platform."""
def __init__(self, coordinator: DataUpdateCoordinator[SyncThru], name: str) -> None:
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.syncthru = coordinator.data
@ -85,7 +78,7 @@ class SyncThruOnlineSensor(SyncThruBinarySensor):
_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
def __init__(self, coordinator: DataUpdateCoordinator[SyncThru], name: str) -> None:
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._id_suffix = "_online"
@ -101,9 +94,9 @@ class SyncThruProblemSensor(SyncThruBinarySensor):
_attr_device_class = BinarySensorDeviceClass.PROBLEM
def __init__(self, syncthru, name):
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(syncthru, name)
super().__init__(coordinator, name)
self._id_suffix = "_problem"
@property

View File

@ -0,0 +1,44 @@
"""Coordinator for Syncthru integration."""
import asyncio
from datetime import timedelta
import logging
from pysyncthru import ConnectionMode, SyncThru
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class SyncthruCoordinator(DataUpdateCoordinator[SyncThru]):
"""Class to manage fetching Syncthru data."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize the Syncthru coordinator."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=30),
)
self.syncthru = SyncThru(
entry.data[CONF_URL],
async_get_clientsession(hass),
connection_mode=ConnectionMode.API,
)
async def _async_update_data(self) -> SyncThru:
async with asyncio.timeout(10):
await self.syncthru.update()
if self.syncthru.is_unknown_state():
raise UpdateFailed(
f"Configured printer at {self.syncthru.url} does not respond."
)
return self.syncthru

View File

@ -10,12 +10,9 @@ from homeassistant.const import CONF_NAME, PERCENTAGE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import device_identifiers
from . import SyncthruCoordinator, device_identifiers
from .const import DOMAIN
COLORS = ["black", "cyan", "magenta", "yellow"]
@ -47,9 +44,7 @@ async def async_setup_entry(
) -> None:
"""Set up from config entry."""
coordinator: DataUpdateCoordinator[SyncThru] = hass.data[DOMAIN][
config_entry.entry_id
]
coordinator: SyncthruCoordinator = hass.data[DOMAIN][config_entry.entry_id]
printer: SyncThru = coordinator.data
supp_toner = printer.toner_status(filter_supported=True)
@ -75,12 +70,12 @@ async def async_setup_entry(
async_add_entities(entities)
class SyncThruSensor(CoordinatorEntity[DataUpdateCoordinator[SyncThru]], SensorEntity):
class SyncThruSensor(CoordinatorEntity[SyncthruCoordinator], SensorEntity):
"""Implementation of an abstract Samsung Printer sensor platform."""
_attr_icon = "mdi:printer"
def __init__(self, coordinator: DataUpdateCoordinator[SyncThru], name: str) -> None:
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.syncthru = coordinator.data
@ -112,7 +107,7 @@ class SyncThruMainSensor(SyncThruSensor):
_attr_entity_registry_enabled_default = False
def __init__(self, coordinator: DataUpdateCoordinator[SyncThru], name: str) -> None:
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._id_suffix = "_main"
@ -135,9 +130,7 @@ class SyncThruTonerSensor(SyncThruSensor):
_attr_native_unit_of_measurement = PERCENTAGE
def __init__(
self, coordinator: DataUpdateCoordinator[SyncThru], name: str, color: str
) -> None:
def __init__(self, coordinator: SyncthruCoordinator, name: str, color: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Toner {color}"
@ -160,9 +153,7 @@ class SyncThruDrumSensor(SyncThruSensor):
_attr_native_unit_of_measurement = PERCENTAGE
def __init__(
self, coordinator: DataUpdateCoordinator[SyncThru], name: str, color: str
) -> None:
def __init__(self, coordinator: SyncthruCoordinator, name: str, color: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Drum {color}"
@ -184,7 +175,7 @@ class SyncThruInputTraySensor(SyncThruSensor):
"""Implementation of a Samsung Printer input tray sensor platform."""
def __init__(
self, coordinator: DataUpdateCoordinator[SyncThru], name: str, number: str
self, coordinator: SyncthruCoordinator, name: str, number: str
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
@ -212,7 +203,7 @@ class SyncThruOutputTraySensor(SyncThruSensor):
"""Implementation of a Samsung Printer output tray sensor platform."""
def __init__(
self, coordinator: DataUpdateCoordinator[SyncThru], name: str, number: int
self, coordinator: SyncthruCoordinator, name: str, number: int
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
@ -239,7 +230,7 @@ class SyncThruOutputTraySensor(SyncThruSensor):
class SyncThruActiveAlertSensor(SyncThruSensor):
"""Implementation of a Samsung Printer active alerts sensor platform."""
def __init__(self, coordinator: DataUpdateCoordinator[SyncThru], name: str) -> None:
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Active Alerts"

View File

@ -27,7 +27,7 @@ def mock_syncthru() -> Generator[AsyncMock]:
"""Mock the SyncThru class."""
with (
patch(
"homeassistant.components.syncthru.SyncThru",
"homeassistant.components.syncthru.coordinator.SyncThru",
autospec=True,
) as mock_syncthru,
patch(