Add page count sensors to IPP integration

homeassistant/components/ipp/coordinator.py

@@ -1,10 +1,12 @@
 """Coordinator for The Internet Printing Protocol (IPP) integration."""
 
+from dataclasses import dataclass
 from datetime import timedelta
 import logging
-from typing import override
+from typing import Any, override
 
 from pyipp import IPP, IPPError, Printer as IPPPrinter
+from pyipp.enums import IppOperation
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
@@ -16,12 +18,35 @@
 
 SCAN_INTERVAL = timedelta(seconds=60)
 
+# Integer page-count attributes returned by Get-Printer-Attributes
+PAGE_COUNT_INT_ATTRIBUTES = (
+    "printer-impressions-completed",
+    "printer-pages-completed",
+    "printer-media-sheets-completed",
+)
+
+# Collection page-count attributes — dicts of monochrome/full-color sub-counters
+PAGE_COUNT_COLLECTION_ATTRIBUTES = ("printer-impressions-completed-col",)
+
+REQUESTED_PAGE_COUNT_ATTRIBUTES = (
+    *PAGE_COUNT_INT_ATTRIBUTES,
+    *PAGE_COUNT_COLLECTION_ATTRIBUTES,
+)
+
 _LOGGER = logging.getLogger(__name__)
 
 type IPPConfigEntry = ConfigEntry[IPPDataUpdateCoordinator]
 
 
-class IPPDataUpdateCoordinator(DataUpdateCoordinator[IPPPrinter]):
+@dataclass
+class IPPData:
+    """Data fetched from an IPP printer."""
+
+    printer: IPPPrinter
+    page_counts: dict[str, int]
+
+
+class IPPDataUpdateCoordinator(DataUpdateCoordinator[IPPData]):
     """Class to manage fetching IPP data from single endpoint."""
 
     config_entry: IPPConfigEntry
@@ -47,9 +72,51 @@ def __init__(self, hass: HomeAssistant, config_entry: IPPConfigEntry) -> None:
         )
 
     @override
-    async def _async_update_data(self) -> IPPPrinter:
+    async def _async_update_data(self) -> IPPData:
         """Fetch data from IPP."""
         try:
-            return await self.ipp.printer()
+            printer = await self.ipp.printer()
         except IPPError as error:
             raise UpdateFailed(f"Invalid response from API: {error}") from error
+
+        # Page counts are fetched via a separate request for now. Once pyipp PR #715
+        # (https://github.com/ctalkington/python-ipp/pull/715) is merged, page
+        # counters will be included in printer.counters by default and this extra
+        # request can be removed.
+        previous_page_counts = self.data.page_counts if self.data else {}
+        page_counts = await self._async_fetch_page_counts(previous_page_counts)
+
+        return IPPData(printer=printer, page_counts=page_counts)
+
+    async def _async_fetch_page_counts(
+        self, previous_page_counts: dict[str, int]
+    ) -> dict[str, int]:
+        """Fetch page count attributes from the printer."""
+        try:
+            response = await self.ipp.execute(
+                IppOperation.GET_PRINTER_ATTRIBUTES,
+                {
+                    "operation-attributes-tag": {
+                        "requested-attributes": REQUESTED_PAGE_COUNT_ATTRIBUTES,
+                    },
+                },
+            )
+        except IPPError, TimeoutError:
+            _LOGGER.debug(
+                "Failed to fetch page count attributes from printer", exc_info=True
+            )
+            return previous_page_counts
+
+        parsed: dict[str, Any] = next(iter(response.get("printers") or []), {})
+        page_counts: dict[str, int] = {}
+
+        for attr in PAGE_COUNT_INT_ATTRIBUTES:
+            if (value := parsed.get(attr)) is not None:
+                page_counts[attr] = value
+
+        # pyipp parses collection attributes into dicts of member name to value
+        for attr in PAGE_COUNT_COLLECTION_ATTRIBUTES:
+            for sub_key, sub_value in parsed.get(attr, {}).items():
+                page_counts[f"{attr}/{sub_key}"] = sub_value
+
+        return page_counts

homeassistant/components/ipp/diagnostics.py

@@ -20,5 +20,6 @@ async def async_get_config_entry_diagnostics(
             },
             "unique_id": config_entry.unique_id,
         },
-        "data": coordinator.data.as_dict(),
+        "data": coordinator.data.printer.as_dict(),
+        "page_counts": coordinator.data.page_counts,
     }

homeassistant/components/ipp/entity.py

@@ -24,12 +24,13 @@ def __init__(
         self.entity_description = description
 
         self._attr_unique_id = f"{coordinator.device_id}_{description.key}"
+        printer = self.coordinator.data.printer
         self._attr_device_info = DeviceInfo(
             identifiers={(DOMAIN, coordinator.device_id)},
-            manufacturer=self.coordinator.data.info.manufacturer,
-            model=self.coordinator.data.info.model,
-            name=self.coordinator.data.info.name,
-            serial_number=self.coordinator.data.info.serial,
-            sw_version=self.coordinator.data.info.version,
-            configuration_url=self.coordinator.data.info.more_info,
+            manufacturer=printer.info.manufacturer,
+            model=printer.info.model,
+            name=printer.info.name,
+            serial_number=printer.info.serial,
+            sw_version=printer.info.version,
+            configuration_url=printer.info.more_info,
         )

homeassistant/components/ipp/icons.json

@@ -1,9 +1,24 @@
 {
   "entity": {
     "sensor": {
+      "impressions_completed": {
+        "default": "mdi:file-document-outline"
+      },
+      "impressions_completed_full_color": {
+        "default": "mdi:file-document-outline"
+      },
+      "impressions_completed_monochrome": {
+        "default": "mdi:file-document-outline"
+      },
       "marker": {
         "default": "mdi:water"
       },
+      "media_sheets_completed": {
+        "default": "mdi:file-document-outline"
+      },
+      "pages_completed": {
+        "default": "mdi:file-document-outline"
+      },
       "printer": {
         "default": "mdi:printer"
       },

homeassistant/components/ipp/sensor.py

@@ -41,6 +41,13 @@ class IPPSensorEntityDescription(SensorEntityDescription):
     attributes_fn: Callable[[Printer], dict[Any, StateType]] = lambda _: {}
 
 
+@dataclass(frozen=True, kw_only=True)
+class IPPPageCountSensorEntityDescription(SensorEntityDescription):
+    """Describes IPP page count sensor entity."""
+
+    ipp_attribute: str
+
+
 def _get_marker_attributes_fn(
     marker_index: int, attributes_fn: Callable[[Marker], dict[Any, StateType]]
 ) -> Callable[[Printer], dict[Any, StateType]]:
@@ -81,6 +88,44 @@ def _get_marker_value_fn(
     ),
 )
 
+PAGE_COUNT_SENSORS: tuple[IPPPageCountSensorEntityDescription, ...] = (
+    IPPPageCountSensorEntityDescription(
+        key="pages_completed",
+        translation_key="pages_completed",
+        state_class=SensorStateClass.TOTAL_INCREASING,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        ipp_attribute="printer-pages-completed",
+    ),
+    IPPPageCountSensorEntityDescription(
+        key="impressions_completed",
+        translation_key="impressions_completed",
+        state_class=SensorStateClass.TOTAL_INCREASING,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        ipp_attribute="printer-impressions-completed",
+    ),
+    IPPPageCountSensorEntityDescription(
+        key="media_sheets_completed",
+        translation_key="media_sheets_completed",
+        state_class=SensorStateClass.TOTAL_INCREASING,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        ipp_attribute="printer-media-sheets-completed",
+    ),
+    IPPPageCountSensorEntityDescription(
+        key="impressions_completed_monochrome",
+        translation_key="impressions_completed_monochrome",
+        state_class=SensorStateClass.TOTAL_INCREASING,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        ipp_attribute="printer-impressions-completed-col/monochrome",
+    ),
+    IPPPageCountSensorEntityDescription(
+        key="impressions_completed_full_color",
+        translation_key="impressions_completed_full_color",
+        state_class=SensorStateClass.TOTAL_INCREASING,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        ipp_attribute="printer-impressions-completed-col/full-color",
+    ),
+)
+
 
 async def async_setup_entry(
     hass: HomeAssistant,
@@ -97,7 +142,7 @@ async def async_setup_entry(
         for description in PRINTER_SENSORS
     ]
 
-    for index, marker in enumerate(coordinator.data.markers):
+    for index, marker in enumerate(coordinator.data.printer.markers):
         sensors.append(
             IPPSensor(
                 coordinator,
@@ -123,6 +168,12 @@ async def async_setup_entry(
             )
         )
 
+    sensors.extend(
+        IPPPageCountSensor(coordinator, description)
+        for description in PAGE_COUNT_SENSORS
+        if description.ipp_attribute in coordinator.data.page_counts
+    )
+
     async_add_entities(sensors, True)
 
 
@@ -135,10 +186,24 @@ class IPPSensor(IPPEntity, SensorEntity):
     @override
     def extra_state_attributes(self) -> dict[str, Any]:
         """Return the state attributes of the entity."""
-        return self.entity_description.attributes_fn(self.coordinator.data)
+        return self.entity_description.attributes_fn(self.coordinator.data.printer)
 
     @property
     @override
     def native_value(self) -> StateType | datetime:
         """Return the state of the sensor."""
-        return self.entity_description.value_fn(self.coordinator.data)
+        return self.entity_description.value_fn(self.coordinator.data.printer)
+
+
+class IPPPageCountSensor(IPPEntity, SensorEntity):
+    """Defines an IPP page count sensor."""
+
+    entity_description: IPPPageCountSensorEntityDescription
+
+    @property
+    @override
+    def native_value(self) -> StateType:
+        """Return the state of the sensor."""
+        return self.coordinator.data.page_counts.get(
+            self.entity_description.ipp_attribute
+        )

homeassistant/components/ipp/strings.json

@@ -34,6 +34,26 @@
   },
   "entity": {
     "sensor": {
+      "impressions_completed": {
+        "name": "Impressions completed",
+        "unit_of_measurement": "impressions"
+      },
+      "impressions_completed_full_color": {
+        "name": "Color impressions completed",
+        "unit_of_measurement": "[%key:component::ipp::entity::sensor::impressions_completed::unit_of_measurement%]"
+      },
+      "impressions_completed_monochrome": {
+        "name": "Monochrome impressions completed",
+        "unit_of_measurement": "[%key:component::ipp::entity::sensor::impressions_completed::unit_of_measurement%]"
+      },
+      "media_sheets_completed": {
+        "name": "Media sheets completed",
+        "unit_of_measurement": "sheets"
+      },
+      "pages_completed": {
+        "name": "Pages completed",
+        "unit_of_measurement": "pages"
+      },
       "printer": {
         "state": {
           "idle": "[%key:common::state::idle%]",

tests/components/ipp/conftest.py

@@ -81,6 +81,19 @@ def mock_ipp(mock_printer: Printer) -> Generator[MagicMock]:
     ) as ipp_mock:
         client = ipp_mock.return_value
         client.printer.return_value = mock_printer
+        client.execute.return_value = {
+            "printers": [
+                {
+                    "printer-pages-completed": 1234,
+                    "printer-impressions-completed": 2468,
+                    "printer-media-sheets-completed": 1234,
+                    "printer-impressions-completed-col": {
+                        "monochrome": 1500,
+                        "full-color": 968,
+                    },
+                }
+            ],
+        }
         yield client
 
 

tests/components/ipp/snapshots/test_diagnostics.ambr

@@ -97,5 +97,12 @@
       }),
       'unique_id': 'cfe92100-67c4-11d4-a45f-f8d027761251',
     }),
+    'page_counts': dict({
+      'printer-impressions-completed': 2468,
+      'printer-impressions-completed-col/full-color': 968,
+      'printer-impressions-completed-col/monochrome': 1500,
+      'printer-media-sheets-completed': 1234,
+      'printer-pages-completed': 1234,
+    }),
   })
 # ---