initial change to support print pages

Author: elad-bar

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 2.0.5
+
+- New feature: Print pages (according to printer's supported jobs) buttons [#161](https://github.com/elad-bar/ha-hpprinter/issues/161)
+
 ## 2.0.4
 
 - Fix update data when printer goes online [#161](https://github.com/elad-bar/ha-hpprinter/issues/161)

custom_components/hpprinter/button.py

@@ -0,0 +1,40 @@
+import logging
+
+from homeassistant.components.button import ButtonEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .common.base_entity import BaseEntity, async_setup_base_entry
+from .common.entity_descriptions import IntegrationButtonEntityDescription
+from .managers.ha_coordinator import HACoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities
+):
+    await async_setup_base_entry(
+        hass,
+        entry,
+        Platform.BUTTON,
+        HAButtonEntity,
+        async_add_entities,
+    )
+
+
+class HAButtonEntity(BaseEntity, ButtonEntity):
+    """Representation of a sensor."""
+
+    def __init__(
+        self,
+        entity_description: IntegrationButtonEntityDescription,
+        coordinator: HACoordinator,
+        device_key: str,
+    ):
+        super().__init__(entity_description, coordinator, device_key)
+
+    async def async_press(self) -> None:
+        """Press the button."""
+        await self.local_coordinator.async_print_job(self._entity_description.key)

custom_components/hpprinter/common/base_entity.py

@@ -94,6 +94,7 @@ def _async_handle_device_created(
 
 
 class BaseEntity(CoordinatorEntity):
+    _entity_description: IntegrationEntityDescription
     _translations: dict
 
     def __init__(
@@ -105,6 +106,7 @@ def __init__(
         super().__init__(coordinator)
 
         self.entity_description = entity_description
+        self._entity_description = entity_description
 
         self._device_key = device_key
         self._device_type = entity_description.device_type

custom_components/hpprinter/common/entity_descriptions.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
 
 from homeassistant.components.binary_sensor import BinarySensorEntityDescription
+from homeassistant.components.button import ButtonEntityDescription
 from homeassistant.components.sensor import SensorEntityDescription
 from homeassistant.const import Platform
 from homeassistant.helpers.entity import EntityDescription
@@ -13,6 +14,13 @@ class IntegrationEntityDescription(EntityDescription):
     exclude: dict | None = None
 
 
+@dataclass(frozen=True, kw_only=True)
+class IntegrationButtonEntityDescription(
+    ButtonEntityDescription, IntegrationEntityDescription
+):
+    platform: Platform | None = Platform.BUTTON
+
+
 @dataclass(frozen=True, kw_only=True)
 class IntegrationBinarySensorEntityDescription(
     BinarySensorEntityDescription, IntegrationEntityDescription

custom_components/hpprinter/managers/ha_config_manager.py

@@ -6,13 +6,15 @@
 
 import aiofiles
 
+from homeassistant.components.button import ButtonDeviceClass
 from homeassistant.config_entries import STORAGE_VERSION, ConfigEntry
 from homeassistant.const import Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import translation
 from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.json import JSONEncoder
 from homeassistant.helpers.storage import Store
+from homeassistant.util import slugify
 
 from ..common.consts import (
     CONFIGURATION_FILE,
@@ -21,9 +23,11 @@
     DEFAULT_NAME,
     DOMAIN,
     DURATION_UNITS,
+    PRINTER_MAIN_DEVICE,
 )
 from ..common.entity_descriptions import (
     IntegrationBinarySensorEntityDescription,
+    IntegrationButtonEntityDescription,
     IntegrationEntityDescription,
     IntegrationSensorEntityDescription,
 )
@@ -333,6 +337,20 @@ def _load_entity_descriptions(self):
 
         self._update_platforms()
 
+    def load_buttons_entity_descriptions(self, job_types: list[str]):
+        for job_type in job_types:
+            entity_description = IntegrationButtonEntityDescription(
+                key=job_type,
+                name=job_type,
+                device_type=PRINTER_MAIN_DEVICE,
+                device_class=ButtonDeviceClass.IDENTIFY,
+                translation_key=slugify(job_type),
+            )
+
+            self._entity_descriptions.append(entity_description)
+
+        self._update_platforms()
+
     def _update_platforms(self):
         for entity_description in self._entity_descriptions:
             if (

custom_components/hpprinter/managers/ha_coordinator.py

@@ -88,7 +88,7 @@ async def initialize(self):
 
         await self._api.initialize()
 
-        await self.async_config_entry_first_refresh()
+        await self.async_request_refresh()
 
     def _load_signal_handlers(self):
         loop = self.hass.loop
@@ -105,7 +105,7 @@ def on_device_discovered(
 
         self.config_entry.async_on_unload(
             async_dispatcher_connect(
-                self.hass, SIGNAL_HA_DEVICE_DISCOVERED, self._on_device_discovered
+                self.hass, SIGNAL_HA_DEVICE_DISCOVERED, on_device_discovered
             )
         )
 
@@ -310,6 +310,9 @@ def get_debug_data(self) -> dict:
 
         return data
 
+    async def async_print_job(self, job: str):
+        await self._api.async_print_job(job)
+
     def get_devices(self) -> dict[str, DeviceInfo]:
         return self._devices
 
@@ -343,6 +346,9 @@ async def _on_device_discovered(
         else:
             self.create_sub_unit_device(device_key, device_data, device_config)
 
+        job_types: list[str] = device_data.get("job_types", [])
+        self.config_manager.load_buttons_entity_descriptions(job_types)
+
         async_dispatcher_send(
             self.hass,
             SIGNAL_HA_DEVICE_CREATED,

custom_components/hpprinter/managers/rest_api.py

@@ -278,22 +278,6 @@ def _extract_data(self, devices: list[dict]):
         for device_key in self._data:
             self.device_data_changed(device_key)
 
-    @staticmethod
-    def _get_device_from_list(
-        data: list[dict], identifier_key: str, device_id
-    ) -> dict | None:
-        data_items = [
-            data_item
-            for data_item in data
-            if data_item.get(identifier_key) == device_id
-        ]
-
-        if data_items:
-            return data_items[0]
-
-        else:
-            return None
-
     def _get_device_config(self, device_type: str) -> dict | None:
         data_configs = [
             data_point
@@ -358,11 +342,23 @@ def _get_device_data(
         for property_key in properties:
             property_details = properties.get(property_key)
             property_path = property_details.get("path")
+            property_type = property_details.get("type")
             options = property_details.get("options")
             validation_warning = property_details.get("validationWarning", False)
-            value = data_item_flat.get(property_path)
 
-            is_valid = True if options is None else str(value).lower() in options
+            if property_type == "list":
+                value = [
+                    data_item_flat.get(item)
+                    for item in data_item_flat
+                    if item.startswith(f"{property_path}.")
+                ]
+
+                is_valid = True
+
+            else:
+                value = data_item_flat.get(property_path)
+
+                is_valid = True if options is None else str(value).lower() in options
 
             if value is not None:
                 if is_valid:
@@ -497,3 +493,6 @@ def device_data_changed(self, device_key: str):
                 device_data,
                 device_config,
             )
+
+    async def async_print_job(self, job: str):
+        _LOGGER.debug(f"Start job, Name: {job}")

custom_components/hpprinter/manifest.json

@@ -9,5 +9,5 @@
   "iot_class": "local_polling",
   "issue_tracker": "https://github.com/elad-bar/ha-hpprinter/issues",
   "requirements": ["xmltodict~=0.13.0", "flatten_json", "defusedxml"],
-  "version": "2.0.4"
+  "version": "2.0.5"
 }

custom_components/hpprinter/parameters/data_points.json

@@ -31,6 +31,23 @@
       }
     }
   },
+  {
+    "name": "PrintJobs",
+    "endpoint": "/DevMgmt/InternalPrintCap.xml",
+    "path": "InternalPrintCap",
+    "device_type": "Main",
+    "interval": "52w",
+    "properties": {
+      "job_url": {
+        "path": "JobUrl"
+      },
+      "job_types": {
+        "path": "JobTypesSupport.JobType",
+        "type": "list",
+        "platform": "button"
+      }
+    }
+  },
   {
     "name": "Consumable",
     "endpoint": "/DevMgmt/ConsumableConfigDyn.xml",

utils/api_test.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import logging
 import os
 import sys
@@ -48,27 +49,26 @@ async def initialize(self):
         self._api = RestAPIv2(hass, self._config_manager)
         await self._api.initialize()
 
-        if self._api.is_online:
-
-            await self._api.update([PRODUCT_MAIN_ENDPOINT])
+        await self._api.update([PRODUCT_MAIN_ENDPOINT])
 
-            main_device = self._api.data.get(PRINTER_MAIN_DEVICE)
-            model = main_device.get(MODEL_PROPERTY)
-            title = f"{model} ({self._api.config_data.hostname})"
+        main_device = self._api.data.get(PRINTER_MAIN_DEVICE)
+        model = main_device.get(MODEL_PROPERTY)
+        title = f"{model} ({self._api.config_data.hostname})"
 
-            print(title)
+        _LOGGER.info(f"Title: {title}")
 
-            for i in range(0, 1):
-                # self._api.config_data.update({
-                #    key: os.environ.get(key) if i % 2 == 0 or key != "host" else "127.0.0.1"
-                #    for key in self._api.config_data.to_dict()
-                #})
+    def _modify_connection(self, index: int):
+        self._api.config_data.update({
+            key: os.environ.get(key) if index % 2 == 0 or key != "host" else "127.0.0.1"
+            for key in self._api.config_data.to_dict()
+        })
 
+    async def update(self, times: int):
+        if self._api.is_online:
+            for i in range(0, times):
                 await self._api.update()
 
-                # print(json.dumps(self._api.data_config, indent=4))
-                # print(json.dumps(self._api.data, indent=4))
-                # print(json.dumps(self._api.data[PRINTER_MAIN_DEVICE], indent=4))
+                _LOGGER.debug(json.dumps(self._api.data[PRINTER_MAIN_DEVICE], indent=4))
 
                 await asyncio.sleep(10)
 
@@ -83,6 +83,7 @@ async def initialize(self):
 
     try:
         loop.run_until_complete(instance.initialize())
+        loop.run_until_complete(instance.update(1))
 
     except KeyboardInterrupt:
         _LOGGER.info("Aborted")