diff --git a/homeassistant/components/syncthru/sensor.py b/homeassistant/components/syncthru/sensor.py index 4abe0e41136..3f4c802e62d 100644 --- a/homeassistant/components/syncthru/sensor.py +++ b/homeassistant/components/syncthru/sensor.py @@ -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