diff --git a/homeassistant/components/ipp/entity.py b/homeassistant/components/ipp/entity.py index 50f81f74bd..af3f6c17c5 100644 --- a/homeassistant/components/ipp/entity.py +++ b/homeassistant/components/ipp/entity.py @@ -13,29 +13,14 @@ class IPPEntity(CoordinatorEntity[IPPDataUpdateCoordinator]): def __init__( self, - *, - entry_id: str, device_id: str, coordinator: IPPDataUpdateCoordinator, - name: str, - icon: str, - enabled_default: bool = True, ) -> None: """Initialize the IPP entity.""" super().__init__(coordinator) self._device_id = device_id - self._entry_id = entry_id - self._attr_name = name - self._attr_icon = icon - self._attr_entity_registry_enabled_default = enabled_default - @property - def device_info(self) -> DeviceInfo | None: - """Return device information about this IPP device.""" - if self._device_id is None: - return None - - return DeviceInfo( + self._attr_device_info = DeviceInfo( identifiers={(DOMAIN, self._device_id)}, manufacturer=self.coordinator.data.info.manufacturer, model=self.coordinator.data.info.model, diff --git a/homeassistant/components/ipp/sensor.py b/homeassistant/components/ipp/sensor.py index 5058f6d10a..fec628e852 100644 --- a/homeassistant/components/ipp/sensor.py +++ b/homeassistant/components/ipp/sensor.py @@ -1,14 +1,22 @@ """Support for IPP sensors.""" from __future__ import annotations +from collections.abc import Callable +from dataclasses import dataclass from datetime import datetime, timedelta from typing import Any -from homeassistant.components.sensor import SensorDeviceClass, SensorEntity +from homeassistant.components.sensor import ( + RestoreSensor, + SensorDeviceClass, + SensorEntity, + SensorEntityDescription, +) from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ATTR_LOCATION, PERCENTAGE -from homeassistant.core import HomeAssistant +from homeassistant.const import ATTR_LOCATION, PERCENTAGE, EntityCategory +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.typing import StateType from homeassistant.util.dt import utcnow from .const import ( @@ -27,6 +35,61 @@ from .coordinator import IPPDataUpdateCoordinator from .entity import IPPEntity +@dataclass +class IPPSensorEntityDescriptionMixin: + """Mixin for required keys.""" + + value_fn: Callable[[Any], StateType | datetime] + + +@dataclass +class IPPSensorEntityDescription( + SensorEntityDescription, IPPSensorEntityDescriptionMixin +): + """Describes IPP sensor entity.""" + + attributes_fn: Callable[[Any], dict[Any, StateType]] = lambda _: {} + + +PRINTER_SENSORS: tuple[IPPSensorEntityDescription, ...] = ( + IPPSensorEntityDescription( + key="printer", + translation_key="printer", + icon="mdi:printer", + device_class=SensorDeviceClass.ENUM, + options=["idle", "printing", "stopped"], + attributes_fn=lambda printer: { + ATTR_INFO: printer.info.printer_info, + ATTR_SERIAL: printer.info.serial, + ATTR_LOCATION: printer.info.location, + ATTR_STATE_MESSAGE: printer.state.message, + ATTR_STATE_REASON: printer.state.reasons, + ATTR_COMMAND_SET: printer.info.command_set, + ATTR_URI_SUPPORTED: printer.info.printer_uri_supported, + }, + value_fn=lambda printer: printer.state.printer_state, + ), + IPPSensorEntityDescription( + key="uptime", + name="Uptime", + icon="mdi:clock-outline", + device_class=SensorDeviceClass.TIMESTAMP, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, + attributes_fn=lambda printer: { + ATTR_INFO: printer.info.printer_info, + ATTR_SERIAL: printer.info.serial, + ATTR_LOCATION: printer.info.location, + ATTR_STATE_MESSAGE: printer.state.message, + ATTR_STATE_REASON: printer.state.reasons, + ATTR_COMMAND_SET: printer.info.command_set, + ATTR_URI_SUPPORTED: printer.info.printer_uri_supported, + }, + value_fn=lambda printer: (utcnow() - timedelta(seconds=printer.info.uptime)), + ), +) + + async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, @@ -41,160 +104,118 @@ async def async_setup_entry( sensors: list[SensorEntity] = [] - sensors.append(IPPPrinterSensor(entry.entry_id, unique_id, coordinator)) - sensors.append(IPPUptimeSensor(entry.entry_id, unique_id, coordinator)) + sensors.extend( + [ + IPPSensor( + unique_id, + coordinator, + description, + ) + for description in PRINTER_SENSORS + ] + ) - for marker_index in range(len(coordinator.data.markers)): + for index, marker in enumerate(coordinator.data.markers): sensors.append( - IPPMarkerSensor(entry.entry_id, unique_id, coordinator, marker_index) + IPPMarkerSensor( + index, + unique_id, + coordinator, + IPPSensorEntityDescription( + key=f"marker_{index}", + name=marker.name, + icon="mdi:water", + native_unit_of_measurement=PERCENTAGE, + attributes_fn=lambda marker: { + ATTR_MARKER_HIGH_LEVEL: marker.high_level, + ATTR_MARKER_LOW_LEVEL: marker.low_level, + ATTR_MARKER_TYPE: marker.marker_type, + }, + value_fn=lambda marker: marker.level, + ), + ) ) - async_add_entities(sensors, True) + async_add_entities(sensors) class IPPSensor(IPPEntity, SensorEntity): """Defines an IPP sensor.""" + entity_description: IPPSensorEntityDescription + + _attr_has_entity_name = True + def __init__( self, - *, + device_id: str, coordinator: IPPDataUpdateCoordinator, - enabled_default: bool = True, - entry_id: str, - unique_id: str, - icon: str, - key: str, - name: str, - unit_of_measurement: str | None = None, - translation_key: str | None = None, + description: IPPSensorEntityDescription, ) -> None: """Initialize IPP sensor.""" - self._key = key - self._attr_unique_id = f"{unique_id}_{key}" - self._attr_native_unit_of_measurement = unit_of_measurement - self._attr_translation_key = translation_key + self.entity_description = description super().__init__( - entry_id=entry_id, - device_id=unique_id, - coordinator=coordinator, - name=name, - icon=icon, - enabled_default=enabled_default, + device_id, + coordinator, ) + self._attr_unique_id = f"{device_id}_{description.key}" -class IPPMarkerSensor(IPPSensor): + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return the state attributes of the entity.""" + return self.entity_description.attributes_fn(self.coordinator.data) + + @property + def native_value(self) -> StateType | datetime: + """Return the state of the sensor.""" + return self.entity_description.value_fn(self.coordinator.data) + + +class IPPMarkerSensor(IPPEntity, RestoreSensor): """Defines an IPP marker sensor.""" + entity_description: IPPSensorEntityDescription + + _attr_has_entity_name = True + def __init__( self, - entry_id: str, - unique_id: str, - coordinator: IPPDataUpdateCoordinator, marker_index: int, + device_id: str, + coordinator: IPPDataUpdateCoordinator, + description: IPPSensorEntityDescription, ) -> None: """Initialize IPP marker sensor.""" + self.entity_description = description self.marker_index = marker_index super().__init__( - coordinator=coordinator, - entry_id=entry_id, - unique_id=unique_id, - icon="mdi:water", - key=f"marker_{marker_index}", - name=( - f"{coordinator.data.info.name} {coordinator.data.markers[marker_index].name}" - ), - unit_of_measurement=PERCENTAGE, + device_id, + coordinator, ) - @property - def extra_state_attributes(self) -> dict[str, Any] | None: - """Return the state attributes of the entity.""" - return { - ATTR_MARKER_HIGH_LEVEL: self.coordinator.data.markers[ - self.marker_index - ].high_level, - ATTR_MARKER_LOW_LEVEL: self.coordinator.data.markers[ - self.marker_index - ].low_level, - ATTR_MARKER_TYPE: self.coordinator.data.markers[ - self.marker_index - ].marker_type, - } + self._attr_unique_id = f"{device_id}_{description.key}" - @property - def native_value(self) -> int | None: - """Return the state of the sensor.""" - level = self.coordinator.data.markers[self.marker_index].level + async def async_added_to_hass(self) -> None: + """Handle entity which will be added.""" + await super().async_added_to_hass() - if level >= 0: - return level + if self.coordinator.data is None: + if ( + last_sensor_data := await self.async_get_last_sensor_data() + ) is not None: + self._attr_native_value = last_sensor_data.native_value - return None - - -class IPPPrinterSensor(IPPSensor): - """Defines an IPP printer sensor.""" - - _attr_device_class = SensorDeviceClass.ENUM - _attr_options = ["idle", "printing", "stopped"] - - def __init__( - self, entry_id: str, unique_id: str, coordinator: IPPDataUpdateCoordinator - ) -> None: - """Initialize IPP printer sensor.""" - super().__init__( - coordinator=coordinator, - entry_id=entry_id, - unique_id=unique_id, - icon="mdi:printer", - key="printer", - name=coordinator.data.info.name, - unit_of_measurement=None, - translation_key="printer", - ) - - @property - def extra_state_attributes(self) -> dict[str, Any] | None: - """Return the state attributes of the entity.""" - return { - ATTR_INFO: self.coordinator.data.info.printer_info, - ATTR_SERIAL: self.coordinator.data.info.serial, - ATTR_LOCATION: self.coordinator.data.info.location, - ATTR_STATE_MESSAGE: self.coordinator.data.state.message, - ATTR_STATE_REASON: self.coordinator.data.state.reasons, - ATTR_COMMAND_SET: self.coordinator.data.info.command_set, - ATTR_URI_SUPPORTED: self.coordinator.data.info.printer_uri_supported, - } - - @property - def native_value(self) -> str: - """Return the state of the sensor.""" - return self.coordinator.data.state.printer_state - - -class IPPUptimeSensor(IPPSensor): - """Defines a IPP uptime sensor.""" - - _attr_device_class = SensorDeviceClass.TIMESTAMP - - def __init__( - self, entry_id: str, unique_id: str, coordinator: IPPDataUpdateCoordinator - ) -> None: - """Initialize IPP uptime sensor.""" - super().__init__( - coordinator=coordinator, - enabled_default=False, - entry_id=entry_id, - unique_id=unique_id, - icon="mdi:clock-outline", - key="uptime", - name=f"{coordinator.data.info.name} Uptime", - ) - - @property - def native_value(self) -> datetime: - """Return the state of the sensor.""" - return utcnow() - timedelta(seconds=self.coordinator.data.info.uptime) + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + if self.coordinator.data is not None: + self._attr_native_value = self.entity_description.value_fn( + self.coordinator.data.markers[self.marker_index] + ) + self._attr_extra_state_attributes = self.entity_description.attributes_fn( + self.coordinator.data.markers[self.marker_index] + ) + self.async_write_ha_state() diff --git a/tests/components/ipp/test_sensor.py b/tests/components/ipp/test_sensor.py index f8dd94ffc7..a07a66ccae 100644 --- a/tests/components/ipp/test_sensor.py +++ b/tests/components/ipp/test_sensor.py @@ -2,38 +2,40 @@ from datetime import datetime from unittest.mock import patch -from homeassistant.components.ipp.const import DOMAIN -from homeassistant.components.sensor import ( - ATTR_OPTIONS as SENSOR_ATTR_OPTIONS, - DOMAIN as SENSOR_DOMAIN, +import pytest + +from homeassistant.components.ipp.const import ( + ATTR_MARKER_HIGH_LEVEL, + ATTR_MARKER_LOW_LEVEL, + ATTR_MARKER_TYPE, ) -from homeassistant.const import ATTR_ICON, ATTR_UNIT_OF_MEASUREMENT, PERCENTAGE -from homeassistant.core import HomeAssistant +from homeassistant.components.sensor import ATTR_OPTIONS +from homeassistant.const import ( + ATTR_ICON, + ATTR_UNIT_OF_MEASUREMENT, + PERCENTAGE, + EntityCategory, +) +from homeassistant.core import HomeAssistant, State from homeassistant.helpers import entity_registry as er from homeassistant.util import dt as dt_util from . import init_integration, mock_connection +from tests.common import mock_restore_cache_with_extra_data from tests.test_util.aiohttp import AiohttpClientMocker +@pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_sensors( - hass: HomeAssistant, aioclient_mock: AiohttpClientMocker + hass: HomeAssistant, + aioclient_mock: AiohttpClientMocker, + entity_registry: er.EntityRegistry, ) -> None: """Test the creation and values of the IPP sensors.""" mock_connection(aioclient_mock) entry = await init_integration(hass, aioclient_mock, skip_setup=True) - registry = er.async_get(hass) - - # Pre-create registry entries for disabled by default sensors - registry.async_get_or_create( - SENSOR_DOMAIN, - DOMAIN, - "cfe92100-67c4-11d4-a45f-f8d027761251_uptime", - suggested_object_id="epson_xp_6000_series_uptime", - disabled_by=None, - ) test_time = datetime(2019, 11, 11, 9, 10, 32, tzinfo=dt_util.UTC) with patch("homeassistant.components.ipp.sensor.utcnow", return_value=test_time): @@ -44,9 +46,9 @@ async def test_sensors( assert state assert state.attributes.get(ATTR_ICON) == "mdi:printer" assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None - assert state.attributes.get(SENSOR_ATTR_OPTIONS) == ["idle", "printing", "stopped"] + assert state.attributes.get(ATTR_OPTIONS) == ["idle", "printing", "stopped"] - entry = registry.async_get("sensor.epson_xp_6000_series") + entry = entity_registry.async_get("sensor.epson_xp_6000_series") assert entry assert entry.translation_key == "printer" @@ -86,9 +88,10 @@ async def test_sensors( assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None assert state.state == "2019-10-26T15:37:00+00:00" - entry = registry.async_get("sensor.epson_xp_6000_series_uptime") + entry = entity_registry.async_get("sensor.epson_xp_6000_series_uptime") assert entry assert entry.unique_id == "cfe92100-67c4-11d4-a45f-f8d027761251_uptime" + assert entry.entity_category == EntityCategory.DIAGNOSTIC async def test_disabled_by_default_sensors( @@ -117,3 +120,67 @@ async def test_missing_entry_unique_id( entity = registry.async_get("sensor.epson_xp_6000_series") assert entity assert entity.unique_id == f"{entry.entry_id}_printer" + + +@pytest.mark.parametrize( + ( + "entity_id", + "restored_state", + "restored_native_value", + "initial_state", + "initial_attributes", + ), + ( + ( + "sensor.epson_xp_6000_series_black_ink", + "43", + 43, + "43", + [ + ATTR_ICON, + ATTR_MARKER_HIGH_LEVEL, + ATTR_MARKER_LOW_LEVEL, + ATTR_MARKER_TYPE, + ], + ), + ), +) +async def test_restore_marker_state( + hass: HomeAssistant, + aioclient_mock: AiohttpClientMocker, + entity_id: str, + restored_state: str, + restored_native_value: str, + initial_state: str, + initial_attributes: list, +) -> None: + """Test sensor restore state.""" + restored_attributes = { + ATTR_ICON: "mdi:water", + ATTR_MARKER_HIGH_LEVEL: 100, + ATTR_MARKER_LOW_LEVEL: 10, + ATTR_MARKER_TYPE: "ink", + } + + fake_state = State( + entity_id, + restored_state, + restored_attributes, + ) + + fake_extra_data = { + "native_value": restored_native_value, + "native_unit_of_measurement": PERCENTAGE, + } + + mock_restore_cache_with_extra_data(hass, ((fake_state, fake_extra_data),)) + + await init_integration(hass, aioclient_mock, conn_error=True) + + assert (state := hass.states.get(entity_id)) + assert state.state == initial_state + for attr in restored_attributes.items(): + if attr in initial_attributes: + assert state.attributes[attr] == restored_attributes[attr] + else: + assert attr not in state.attributes