"""Based on @skalavala work. https://blog.kalavala.net/usps/homeassistant/mqtt/2018/01/12/usps.html Configuration code contribution from @firstof9 https://github.com/firstof9/ """ import datetime import logging from typing import Any from homeassistant.components.sensor import SensorEntity, SensorEntityDescription from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_HOST, CONF_RESOURCES from homeassistant.core import HomeAssistant from homeassistant.helpers.network import NoURLAvailableError, get_url from homeassistant.helpers.update_coordinator import CoordinatorEntity from . import MailAndPackagesConfigEntry from .const import ( AMAZON_EXCEPTION, AMAZON_EXCEPTION_ORDER, AMAZON_HUB, AMAZON_HUB_CODE, AMAZON_ORDER, AMAZON_OTP, AMAZON_OTP_CODE, ATTR_CODE, ATTR_GRID_IMAGE_NAME, ATTR_IMAGE, ATTR_IMAGE_NAME, ATTR_IMAGE_PATH, ATTR_ORDER, ATTR_TRACKING_NUM, ATTR_USPS_IMAGE, CONF_PATH, DOMAIN, IMAGE_SENSORS, SENSOR_TYPES, VERSION, ) _LOGGER = logging.getLogger(__name__) DELIVERED_SUFFIXES = {"delivered"} async def async_setup_entry( hass, entry: MailAndPackagesConfigEntry, async_add_entities, ): """Set up the sensor entities.""" coordinator = entry.runtime_data.coordinator resources = entry.data.get(CONF_RESOURCES, []) sensors = [ PackagesSensor(entry, SENSOR_TYPES[variable], coordinator) for variable in resources if variable in SENSOR_TYPES ] sensors.extend( ImagePathSensors(hass, entry, value, coordinator) for value in IMAGE_SENSORS.values() ) async_add_entities(sensors, False) class PackagesSensor(CoordinatorEntity, SensorEntity): """Representation of a sensor.""" def __init__( self, config: ConfigEntry, sensor_description: SensorEntityDescription, coordinator: Any, ): """Initialize the sensor.""" super().__init__(coordinator) self.entity_description = sensor_description self.coordinator = coordinator self._config = config self._name = sensor_description.name self.type = sensor_description.key self._host = config.data[CONF_HOST] self._unique_id = self._config.entry_id self.data = self.coordinator.data parts = self.type.split("_") if len(parts) > 1: prefix = "_".join(parts[:-1]) if parts[-1] in DELIVERED_SUFFIXES: self._tracking_key = f"{prefix}_delivered_tracking" else: self._tracking_key = f"{prefix}_tracking" else: self._tracking_key = f"{self.type}_tracking" @property def device_info(self) -> dict: """Return device information about the mailbox.""" return { "connections": {(DOMAIN, self._unique_id)}, "name": self._host, "manufacturer": "IMAP E-Mail", "sw_version": VERSION, } @property def unique_id(self) -> str: """Return a unique, Home Assistant friendly identifier for this entity.""" return f"sensor_{self._host}_{self.type}_{self._unique_id}" @property def name(self) -> str: """Return the name of the sensor.""" return self._name @property def native_value(self) -> Any: """Return the state of the sensor.""" if self.coordinator.data is None: return None value = self.coordinator.data.get(self.type) if self.type == "mail_updated": # Safely handle string vs datetime to prevent ValueError if isinstance(value, str): try: value = datetime.datetime.fromisoformat(value) except ValueError: value = datetime.datetime.now(datetime.UTC) elif value is None: value = datetime.datetime.now(datetime.UTC) return value @property def should_poll(self) -> bool: """No need to poll. Coordinator notifies entity of updates.""" return False @property def available(self) -> bool: """Return if entity is available.""" return self.coordinator.data is not None @property def extra_state_attributes(self) -> str | None: """Return device specific state attributes.""" attr = {} data = self.coordinator.data if data is None: return attr if any( sensor in self.type for sensor in ["_delivering", "_delivered", "_packages", "_exception"] ): if tracking := data.get(self._tracking_key): attr[ATTR_TRACKING_NUM] = tracking if "Amazon" in self._name: self._add_amazon_attributes(attr, data) elif self._name == "Mail USPS Mail": if image_name := data.get(ATTR_IMAGE_NAME): attr[ATTR_IMAGE] = image_name return attr def _add_amazon_attributes(self, attr: dict, data: dict) -> None: """Add Amazon specific attributes to the sensor.""" if self.type == AMAZON_EXCEPTION: if order := data.get(AMAZON_EXCEPTION_ORDER, data.get(ATTR_ORDER)): attr[ATTR_ORDER] = order elif order := data.get(AMAZON_ORDER): attr[ATTR_ORDER] = order elif self.type == AMAZON_HUB: if code := data.get(AMAZON_HUB_CODE, data.get(ATTR_CODE)): attr[ATTR_CODE] = code elif self.type == AMAZON_OTP: if code := data.get(AMAZON_OTP_CODE, data.get(ATTR_CODE)): attr[ATTR_CODE] = code class ImagePathSensors(CoordinatorEntity, SensorEntity): """Representation of a sensor.""" def __init__( self, hass: HomeAssistant, config: ConfigEntry, sensor_description: SensorEntityDescription, coordinator: Any, ): """Initialize the sensor.""" super().__init__(coordinator) self.entity_description = sensor_description self.hass = hass self.coordinator = coordinator self._config = config self._name = sensor_description.name self.type = sensor_description.key self._host = config.data[CONF_HOST] self._unique_id = self._config.entry_id @property def device_info(self) -> dict: """Return device information about the mailbox.""" return { "connections": {(DOMAIN, self._unique_id)}, "name": self._host, "manufacturer": "IMAP E-Mail", "sw_version": VERSION, } @property def unique_id(self) -> str: """Return a unique, Home Assistant friendly identifier for this entity.""" return f"sensor_{self._host}_{self.type}_{self._unique_id}" @property def name(self) -> str: """Return the name of the sensor.""" return self._name @property def native_value(self) -> str | None: """Return the state of the sensor.""" if self.coordinator.data is None: return None image = "" the_path = None image = self.coordinator.data.get(ATTR_USPS_IMAGE) grid_image = self.coordinator.data.get(ATTR_GRID_IMAGE_NAME) path = self.coordinator.data.get( ATTR_IMAGE_PATH, self._config.data.get(CONF_PATH), ) if self.type == "usps_mail_image_system_path" and image: _LOGGER.debug("Updating system image path to: %s", path) the_path = self.hass.config.path(path, image) elif self.type == "usps_mail_grid_image_path" and grid_image: _LOGGER.debug("Updating grid image path to: %s", path) the_path = self.hass.config.path(path, grid_image) elif self.type == "usps_mail_image_url" and image: url = self._get_base_url() if url: the_path = f"{url.rstrip('/')}/local/mail_and_packages/{image}" return the_path def _get_base_url(self) -> str | None: """Return the best available base URL for building image links.""" try: return get_url(self.hass, prefer_external=True) except NoURLAvailableError: _LOGGER.debug("No URL available for image link.") return None @property def should_poll(self) -> bool: """No need to poll. Coordinator notifies entity of updates.""" return False @property def available(self) -> bool: """Return if entity is available.""" return self.coordinator.data is not None