Refactor Syncthru sensor platform (#142704)

This commit is contained in:
Joost Lekkerkerker 2025-04-11 08:12:59 +02:00 committed by GitHub
parent f519b20495
commit 56c4121eb2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -2,9 +2,13 @@
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any, cast
from pysyncthru import SyncThru, SyncthruState
from homeassistant.components.sensor import SensorEntity
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME, PERCENTAGE
from homeassistant.core import HomeAssistant
@ -15,17 +19,6 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import SyncthruCoordinator, device_identifiers
from .const import DOMAIN
COLORS = ["black", "cyan", "magenta", "yellow"]
DRUM_COLORS = COLORS
TONER_COLORS = COLORS
TRAYS = range(1, 6)
OUTPUT_TRAYS = range(6)
DEFAULT_MONITORED_CONDITIONS = []
DEFAULT_MONITORED_CONDITIONS.extend([f"toner_{key}" for key in TONER_COLORS])
DEFAULT_MONITORED_CONDITIONS.extend([f"drum_{key}" for key in DRUM_COLORS])
DEFAULT_MONITORED_CONDITIONS.extend([f"tray_{key}" for key in TRAYS])
DEFAULT_MONITORED_CONDITIONS.extend([f"output_tray_{key}" for key in OUTPUT_TRAYS])
SYNCTHRU_STATE_HUMAN = {
SyncthruState.INVALID: "invalid",
SyncthruState.OFFLINE: "unreachable",
@ -37,6 +30,85 @@ SYNCTHRU_STATE_HUMAN = {
}
@dataclass(frozen=True, kw_only=True)
class SyncThruSensorDescription(SensorEntityDescription):
"""Describes a SyncThru sensor entity."""
value_fn: Callable[[SyncThru], str | None]
extra_state_attributes_fn: Callable[[SyncThru], dict[str, str | int]] | None = None
def get_toner_entity_description(color: str) -> SyncThruSensorDescription:
"""Get toner entity description for a specific color."""
return SyncThruSensorDescription(
key=f"toner_{color}",
name=f"Toner {color}",
native_unit_of_measurement=PERCENTAGE,
value_fn=lambda printer: printer.toner_status().get(color, {}).get("remaining"),
extra_state_attributes_fn=lambda printer: printer.toner_status().get(color, {}),
)
def get_drum_entity_description(color: str) -> SyncThruSensorDescription:
"""Get drum entity description for a specific color."""
return SyncThruSensorDescription(
key=f"drum_{color}",
name=f"Drum {color}",
native_unit_of_measurement=PERCENTAGE,
value_fn=lambda printer: printer.drum_status().get(color, {}).get("remaining"),
extra_state_attributes_fn=lambda printer: printer.drum_status().get(color, {}),
)
def get_input_tray_entity_description(tray: str) -> SyncThruSensorDescription:
"""Get input tray entity description for a specific tray."""
return SyncThruSensorDescription(
key=f"tray_{tray}",
name=f"Tray {tray}",
value_fn=(
lambda printer: printer.input_tray_status().get(tray, {}).get("newError")
or "Ready"
),
extra_state_attributes_fn=(
lambda printer: printer.input_tray_status().get(tray, {})
),
)
def get_output_tray_entity_description(tray: int) -> SyncThruSensorDescription:
"""Get output tray entity description for a specific tray."""
return SyncThruSensorDescription(
key=f"output_tray_{tray}",
name=f"Output Tray {tray}",
value_fn=(
lambda printer: printer.output_tray_status().get(tray, {}).get("status")
or "Ready"
),
extra_state_attributes_fn=(
lambda printer: cast(
dict[str, str | int], printer.output_tray_status().get(tray, {})
)
),
)
SENSOR_TYPES: tuple[SyncThruSensorDescription, ...] = (
SyncThruSensorDescription(
key="active_alerts",
name="Active Alerts",
value_fn=lambda printer: printer.raw().get("GXI_ACTIVE_ALERT_TOTAL"),
),
SyncThruSensorDescription(
key="main",
name="",
value_fn=lambda printer: SYNCTHRU_STATE_HUMAN[printer.device_status()],
extra_state_attributes_fn=lambda printer: {
"display_text": printer.device_status_details(),
},
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -45,7 +117,7 @@ async def async_setup_entry(
"""Set up from config entry."""
coordinator: SyncthruCoordinator = hass.data[DOMAIN][config_entry.entry_id]
printer: SyncThru = coordinator.data
printer = coordinator.data
supp_toner = printer.toner_status(filter_supported=True)
supp_drum = printer.drum_status(filter_supported=True)
@ -53,40 +125,39 @@ async def async_setup_entry(
supp_output_tray = printer.output_tray_status()
name: str = config_entry.data[CONF_NAME]
entities: list[SyncThruSensor] = [
SyncThruMainSensor(coordinator, name),
SyncThruActiveAlertSensor(coordinator, name),
entities: list[SyncThruSensorDescription] = [
get_toner_entity_description(color) for color in supp_toner
]
entities.extend(SyncThruTonerSensor(coordinator, name, key) for key in supp_toner)
entities.extend(SyncThruDrumSensor(coordinator, name, key) for key in supp_drum)
entities.extend(
SyncThruInputTraySensor(coordinator, name, key) for key in supp_tray
)
entities.extend(
SyncThruOutputTraySensor(coordinator, name, int_key)
for int_key in supp_output_tray
)
entities.extend(get_drum_entity_description(color) for color in supp_drum)
entities.extend(get_input_tray_entity_description(key) for key in supp_tray)
entities.extend(get_output_tray_entity_description(key) for key in supp_output_tray)
async_add_entities(entities)
async_add_entities(
SyncThruSensor(coordinator, name, description)
for description in SENSOR_TYPES + tuple(entities)
)
class SyncThruSensor(CoordinatorEntity[SyncthruCoordinator], SensorEntity):
"""Implementation of an abstract Samsung Printer sensor platform."""
_attr_icon = "mdi:printer"
entity_description: SyncThruSensorDescription
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
def __init__(
self,
coordinator: SyncthruCoordinator,
name: str,
entity_description: SyncThruSensorDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = entity_description
self.syncthru = coordinator.data
self._attr_name = name
self._id_suffix = ""
@property
def unique_id(self):
"""Return unique ID for the sensor."""
serial = self.syncthru.serial_number()
return f"{serial}{self._id_suffix}" if serial else None
self._attr_name = f"{name} {entity_description.name}".strip()
serial_number = coordinator.data.serial_number()
assert serial_number is not None
self._attr_unique_id = f"{serial_number}_{entity_description.key}"
@property
def device_info(self) -> DeviceInfo | None:
@ -97,146 +168,14 @@ class SyncThruSensor(CoordinatorEntity[SyncthruCoordinator], SensorEntity):
identifiers=identifiers,
)
class SyncThruMainSensor(SyncThruSensor):
"""Implementation of the main sensor, conducting the actual polling.
It also shows the detailed state and presents
the displayed current status message.
"""
_attr_entity_registry_enabled_default = False
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._id_suffix = "_main"
@property
def native_value(self) -> str | int | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.syncthru)
@property
def native_value(self):
"""Set state to human readable version of syncthru status."""
return SYNCTHRU_STATE_HUMAN[self.syncthru.device_status()]
@property
def extra_state_attributes(self):
"""Show current printer display text."""
return {
"display_text": self.syncthru.device_status_details(),
}
class SyncThruTonerSensor(SyncThruSensor):
"""Implementation of a Samsung Printer toner sensor platform."""
_attr_native_unit_of_measurement = PERCENTAGE
def __init__(self, coordinator: SyncthruCoordinator, name: str, color: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Toner {color}"
self._color = color
self._id_suffix = f"_toner_{color}"
@property
def extra_state_attributes(self):
"""Show all data returned for this toner."""
return self.syncthru.toner_status().get(self._color, {})
@property
def native_value(self):
"""Show amount of remaining toner."""
return self.syncthru.toner_status().get(self._color, {}).get("remaining")
class SyncThruDrumSensor(SyncThruSensor):
"""Implementation of a Samsung Printer drum sensor platform."""
_attr_native_unit_of_measurement = PERCENTAGE
def __init__(self, coordinator: SyncthruCoordinator, name: str, color: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Drum {color}"
self._color = color
self._id_suffix = f"_drum_{color}"
@property
def extra_state_attributes(self):
"""Show all data returned for this drum."""
return self.syncthru.drum_status().get(self._color, {})
@property
def native_value(self):
"""Show amount of remaining drum."""
return self.syncthru.drum_status().get(self._color, {}).get("remaining")
class SyncThruInputTraySensor(SyncThruSensor):
"""Implementation of a Samsung Printer input tray sensor platform."""
def __init__(
self, coordinator: SyncthruCoordinator, name: str, number: str
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Tray {number}"
self._number = number
self._id_suffix = f"_tray_{number}"
@property
def extra_state_attributes(self):
"""Show all data returned for this input tray."""
return self.syncthru.input_tray_status().get(self._number, {})
@property
def native_value(self):
"""Display ready unless there is some error, then display error."""
tray_state = (
self.syncthru.input_tray_status().get(self._number, {}).get("newError")
)
if tray_state == "":
tray_state = "Ready"
return tray_state
class SyncThruOutputTraySensor(SyncThruSensor):
"""Implementation of a Samsung Printer output tray sensor platform."""
def __init__(
self, coordinator: SyncthruCoordinator, name: str, number: int
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Output Tray {number}"
self._number = number
self._id_suffix = f"_output_tray_{number}"
@property
def extra_state_attributes(self):
"""Show all data returned for this output tray."""
return self.syncthru.output_tray_status().get(self._number, {})
@property
def native_value(self):
"""Display ready unless there is some error, then display error."""
tray_state = (
self.syncthru.output_tray_status().get(self._number, {}).get("status")
)
if tray_state == "":
tray_state = "Ready"
return tray_state
class SyncThruActiveAlertSensor(SyncThruSensor):
"""Implementation of a Samsung Printer active alerts sensor platform."""
def __init__(self, coordinator: SyncthruCoordinator, name: str) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, name)
self._attr_name = f"{name} Active Alerts"
self._id_suffix = "_active_alerts"
@property
def native_value(self):
"""Show number of active alerts."""
return self.syncthru.raw().get("GXI_ACTIVE_ALERT_TOTAL")
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
if self.entity_description.extra_state_attributes_fn:
return self.entity_description.extra_state_attributes_fn(self.syncthru)
return None