Use attrs instead of properties for ipp (#52270)
* use attrs instead of properties for ipp * Update __init__.py * Create entity.py * Update __init__.py * Create coordinator.py * Update coordinator.py * Update __init__.py * Update entity.py * Update sensor.py * Update sensor.py * Update __init__.py * Update __init__.py * Update coordinator.py * Update entity.py * Update entity.py * Update entity.py * Update sensor.py * Update sensor.pypull/52307/head
Chris Talkington
GitHub
parent
8c37dc5613
commit
6a528acafe
|
|
@ -1,40 +1,17 @@
|
|||
"""The Internet Printing Protocol (IPP) integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from pyipp import IPP, IPPError, Printer as IPPPrinter
|
||||
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
ATTR_NAME,
|
||||
CONF_HOST,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
UpdateFailed,
|
||||
)
|
||||
|
||||
from .const import (
|
||||
ATTR_IDENTIFIERS,
|
||||
ATTR_MANUFACTURER,
|
||||
ATTR_MODEL,
|
||||
ATTR_SOFTWARE_VERSION,
|
||||
CONF_BASE_PATH,
|
||||
DOMAIN,
|
||||
)
|
||||
from .const import CONF_BASE_PATH, DOMAIN
|
||||
from .coordinator import IPPDataUpdateCoordinator
|
||||
|
||||
PLATFORMS = [SENSOR_DOMAIN]
|
||||
SCAN_INTERVAL = timedelta(seconds=60)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -68,92 +45,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.entry_id)
|
||||
return unload_ok
|
||||
|
||||
|
||||
class IPPDataUpdateCoordinator(DataUpdateCoordinator[IPPPrinter]):
|
||||
"""Class to manage fetching IPP data from single endpoint."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
*,
|
||||
host: str,
|
||||
port: int,
|
||||
base_path: str,
|
||||
tls: bool,
|
||||
verify_ssl: bool,
|
||||
) -> None:
|
||||
"""Initialize global IPP data updater."""
|
||||
self.ipp = IPP(
|
||||
host=host,
|
||||
port=port,
|
||||
base_path=base_path,
|
||||
tls=tls,
|
||||
verify_ssl=verify_ssl,
|
||||
session=async_get_clientsession(hass, verify_ssl),
|
||||
)
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
update_interval=SCAN_INTERVAL,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> IPPPrinter:
|
||||
"""Fetch data from IPP."""
|
||||
try:
|
||||
return await self.ipp.printer()
|
||||
except IPPError as error:
|
||||
raise UpdateFailed(f"Invalid response from API: {error}") from error
|
||||
|
||||
|
||||
class IPPEntity(CoordinatorEntity):
|
||||
"""Defines a base IPP entity."""
|
||||
|
||||
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._enabled_default = enabled_default
|
||||
self._entry_id = entry_id
|
||||
self._icon = icon
|
||||
self._name = name
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name of the entity."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def icon(self) -> str:
|
||||
"""Return the mdi icon of the entity."""
|
||||
return self._icon
|
||||
|
||||
@property
|
||||
def entity_registry_enabled_default(self) -> bool:
|
||||
"""Return if the entity should be enabled when first added to the entity registry."""
|
||||
return self._enabled_default
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return device information about this IPP device."""
|
||||
if self._device_id is None:
|
||||
return None
|
||||
|
||||
return {
|
||||
ATTR_IDENTIFIERS: {(DOMAIN, self._device_id)},
|
||||
ATTR_NAME: self.coordinator.data.info.name,
|
||||
ATTR_MANUFACTURER: self.coordinator.data.info.manufacturer,
|
||||
ATTR_MODEL: self.coordinator.data.info.model,
|
||||
ATTR_SOFTWARE_VERSION: self.coordinator.data.info.version,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,55 @@
|
|||
"""Coordinator for The Internet Printing Protocol (IPP) integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from pyipp import IPP, IPPError, Printer as IPPPrinter
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=60)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IPPDataUpdateCoordinator(DataUpdateCoordinator[IPPPrinter]):
|
||||
"""Class to manage fetching IPP data from single endpoint."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
*,
|
||||
host: str,
|
||||
port: int,
|
||||
base_path: str,
|
||||
tls: bool,
|
||||
verify_ssl: bool,
|
||||
) -> None:
|
||||
"""Initialize global IPP data updater."""
|
||||
self.ipp = IPP(
|
||||
host=host,
|
||||
port=port,
|
||||
base_path=base_path,
|
||||
tls=tls,
|
||||
verify_ssl=verify_ssl,
|
||||
session=async_get_clientsession(hass, verify_ssl),
|
||||
)
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
update_interval=SCAN_INTERVAL,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> IPPPrinter:
|
||||
"""Fetch data from IPP."""
|
||||
try:
|
||||
return await self.ipp.printer()
|
||||
except IPPError as error:
|
||||
raise UpdateFailed(f"Invalid response from API: {error}") from error
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
"""Entities for The Internet Printing Protocol (IPP) integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.const import ATTR_NAME
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import (
|
||||
ATTR_IDENTIFIERS,
|
||||
ATTR_MANUFACTURER,
|
||||
ATTR_MODEL,
|
||||
ATTR_SOFTWARE_VERSION,
|
||||
DOMAIN,
|
||||
)
|
||||
from .coordinator import IPPDataUpdateCoordinator
|
||||
|
||||
|
||||
class IPPEntity(CoordinatorEntity):
|
||||
"""Defines a base IPP entity."""
|
||||
|
||||
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:
|
||||
"""Return device information about this IPP device."""
|
||||
if self._device_id is None:
|
||||
return None
|
||||
|
||||
return {
|
||||
ATTR_IDENTIFIERS: {(DOMAIN, self._device_id)},
|
||||
ATTR_NAME: self.coordinator.data.info.name,
|
||||
ATTR_MANUFACTURER: self.coordinator.data.info.manufacturer,
|
||||
ATTR_MODEL: self.coordinator.data.info.model,
|
||||
ATTR_SOFTWARE_VERSION: self.coordinator.data.info.version,
|
||||
}
|
||||
|
|
@ -11,7 +11,6 @@ from homeassistant.core import HomeAssistant
|
|||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.util.dt import utcnow
|
||||
|
||||
from . import IPPDataUpdateCoordinator, IPPEntity
|
||||
from .const import (
|
||||
ATTR_COMMAND_SET,
|
||||
ATTR_INFO,
|
||||
|
|
@ -24,6 +23,8 @@ from .const import (
|
|||
ATTR_URI_SUPPORTED,
|
||||
DOMAIN,
|
||||
)
|
||||
from .coordinator import IPPDataUpdateCoordinator
|
||||
from .entity import IPPEntity
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
|
|
@ -69,12 +70,9 @@ class IPPSensor(IPPEntity, SensorEntity):
|
|||
unit_of_measurement: str | None = None,
|
||||
) -> None:
|
||||
"""Initialize IPP sensor."""
|
||||
self._unit_of_measurement = unit_of_measurement
|
||||
self._key = key
|
||||
self._unique_id = None
|
||||
|
||||
if unique_id is not None:
|
||||
self._unique_id = f"{unique_id}_{key}"
|
||||
self._attr_unique_id = f"{unique_id}_{key}"
|
||||
self._attr_unit_of_measurement = unit_of_measurement
|
||||
|
||||
super().__init__(
|
||||
entry_id=entry_id,
|
||||
|
|
@ -85,16 +83,6 @@ class IPPSensor(IPPEntity, SensorEntity):
|
|||
enabled_default=enabled_default,
|
||||
)
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return the unique ID for this sensor."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self) -> str:
|
||||
"""Return the unit this state is expressed in."""
|
||||
return self._unit_of_measurement
|
||||
|
||||
|
||||
class IPPMarkerSensor(IPPSensor):
|
||||
"""Defines an IPP marker sensor."""
|
||||
|
|
|
|||
Loading…
Reference in New Issue