mirror of
https://github.com/elisspace/core.git
synced 2026-08-29 15:43:55 +00:00
restore marker sensor data in IPP
This commit is contained in:
Chris Talkington
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user