diff --git a/config_flow.py b/config_flow.py new file mode 100644 index 0000000..4b2301d --- /dev/null +++ b/config_flow.py @@ -0,0 +1,95 @@ +"""Config flow for Notification Store.""" + +from __future__ import annotations + +from typing import Any + +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.const import CONF_NAME + +from .const import ( + CONF_SCAN_INTERVAL, + CONF_STORE_PATH, + DEFAULT_NAME, + DEFAULT_SCAN_INTERVAL, + DEFAULT_STORE_PATH, + DOMAIN, + MIN_SCAN_INTERVAL, +) + + +class NotificationStoreConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for Notification Store.""" + + VERSION = 1 + + async def async_step_user(self, user_input: dict[str, Any] | None = None): + """Handle the initial step.""" + await self.async_set_unique_id(DOMAIN) + self._abort_if_unique_id_configured() + + errors: dict[str, str] = {} + + if user_input is not None: + scan_interval = int(user_input.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)) + if scan_interval < MIN_SCAN_INTERVAL: + errors[CONF_SCAN_INTERVAL] = "scan_interval_too_low" + else: + return self.async_create_entry( + title=user_input.get(CONF_NAME, DEFAULT_NAME), + data=user_input, + ) + + schema = vol.Schema( + { + vol.Required(CONF_NAME, default=DEFAULT_NAME): str, + vol.Required(CONF_STORE_PATH, default=DEFAULT_STORE_PATH): str, + vol.Required(CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL): int, + } + ) + + return self.async_show_form( + step_id="user", + data_schema=schema, + errors=errors, + ) + + @staticmethod + def async_get_options_flow(config_entry: config_entries.ConfigEntry): + """Create the options flow.""" + return NotificationStoreOptionsFlow(config_entry) + + +class NotificationStoreOptionsFlow(config_entries.OptionsFlow): + """Handle options for Notification Store.""" + + def __init__(self, config_entry: config_entries.ConfigEntry) -> None: + """Initialize options flow.""" + self.config_entry = config_entry + + async def async_step_init(self, user_input: dict[str, Any] | None = None): + """Manage options.""" + errors: dict[str, str] = {} + current = {**self.config_entry.data, **self.config_entry.options} + + if user_input is not None: + scan_interval = int(user_input.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)) + if scan_interval < MIN_SCAN_INTERVAL: + errors[CONF_SCAN_INTERVAL] = "scan_interval_too_low" + else: + return self.async_create_entry(title="", data=user_input) + + schema = vol.Schema( + { + vol.Required(CONF_STORE_PATH, default=current.get(CONF_STORE_PATH, DEFAULT_STORE_PATH)): str, + vol.Required(CONF_SCAN_INTERVAL, default=current.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)): int, + } + ) + + return self.async_show_form( + step_id="init", + data_schema=schema, + errors=errors, + ) diff --git a/const.py b/const.py new file mode 100644 index 0000000..fdd6422 --- /dev/null +++ b/const.py @@ -0,0 +1,58 @@ +"""Constants for the Notification Store integration.""" + +from __future__ import annotations + +from datetime import timedelta + +from homeassistant.const import Platform + +DOMAIN = "notification_store" +NAME = "Notification Store" + +PLATFORMS: list[Platform] = [Platform.SENSOR] + +DEFAULT_NAME = "Dashboard Notifications" +DEFAULT_SENSOR_NAME = "Items" +DEFAULT_STORE_PATH = "notification_store/notifications.json" +DEFAULT_SCAN_INTERVAL = 30 +MIN_SCAN_INTERVAL = 5 + +CONF_STORE_PATH = "store_path" +CONF_SCAN_INTERVAL = "scan_interval" +CONF_NAME = "name" + +ATTR_NOTIFICATIONS = "notifications" +ATTR_COUNT = "count" +ATTR_TOTAL = "total" +ATTR_LATEST = "latest" +ATTR_LATEST_UNREAD = "latest_unread" + +EVENT_NOTIFICATION_ADDED = "notification_store_notification_added" +EVENT_NOTIFICATION_READ = "notification_store_notification_read" +EVENT_NOTIFICATIONS_CLEARED = "notification_store_notifications_cleared" + +SERVICE_ADD_NOTIFICATION = "add_notification" +SERVICE_MARK_ALL_READ = "mark_all_read" +SERVICE_MARK_ONE_READ = "mark_one_read" +SERVICE_MARK_INDEX_READ = "mark_index_read" +SERVICE_CLEAR_READ = "clear_read" +SERVICE_CLEAR_ALL = "clear_all" +SERVICE_RELOAD = "reload" + +FIELD_TITLE = "title" +FIELD_MESSAGE = "message" +FIELD_LEVEL = "level" +FIELD_TAG = "tag" +FIELD_SOURCE = "source" +FIELD_ICON = "icon" +FIELD_NOTIFICATION_ID = "notification_id" +FIELD_INDEX = "index" +FIELD_ENTRY_ID = "entry_id" + +LEVEL_INFO = "info" +LEVEL_SUCCESS = "success" +LEVEL_WARNING = "warning" +LEVEL_CRITICAL = "critical" +LEVELS = [LEVEL_INFO, LEVEL_SUCCESS, LEVEL_WARNING, LEVEL_CRITICAL] + +DEFAULT_UPDATE_INTERVAL = timedelta(seconds=DEFAULT_SCAN_INTERVAL) diff --git a/coordinator.py b/coordinator.py new file mode 100644 index 0000000..afe4d2a --- /dev/null +++ b/coordinator.py @@ -0,0 +1,54 @@ +"""Coordinator for the Notification Store integration.""" + +from __future__ import annotations + +from datetime import timedelta +from typing import Any + +import logging + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator + +from .const import DOMAIN, NAME + +_LOGGER = logging.getLogger(__name__) +from .store import NotificationStore + + +class NotificationStoreCoordinator(DataUpdateCoordinator[dict[str, Any]]): + """Coordinate notification store updates.""" + + def __init__( + self, + hass: HomeAssistant, + store: NotificationStore, + update_interval: timedelta, + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, + logger=_LOGGER, + name=NAME, + update_interval=update_interval, + always_update=False, + ) + self.store = store + + async def _async_update_data(self) -> dict[str, Any]: + """Fetch data from the notification store.""" + return await self.store.async_load() + + @property + def notifications(self) -> list[dict[str, Any]]: + """Return all notifications.""" + return list((self.data or {}).get("notifications", [])) + + @property + def unread_notifications(self) -> list[dict[str, Any]]: + """Return unread notifications.""" + return [ + item + for item in self.notifications + if item.get("status", "unread") == "unread" + ] diff --git a/manifest.json b/manifest.json new file mode 100644 index 0000000..7413af1 --- /dev/null +++ b/manifest.json @@ -0,0 +1,14 @@ +{ + "domain": "notification_store", + "name": "Notification Store", + "codeowners": [ + "@torsten_brendgen" + ], + "config_flow": true, + "documentation": "https://github.com/torsten-brendgen/notification-store", + "integration_type": "service", + "iot_class": "local_polling", + "issue_tracker": "https://github.com/torsten-brendgen/notification-store/issues", + "requirements": [], + "version": "0.1.3" +} diff --git a/sensor.py b/sensor.py new file mode 100644 index 0000000..3875e03 --- /dev/null +++ b/sensor.py @@ -0,0 +1,91 @@ +"""Sensor platform for Notification Store.""" + +from __future__ import annotations + +from typing import Any + +from homeassistant.components.sensor import SensorEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_NAME +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers import entity_registry as er +from homeassistant.util import slugify +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .const import ( + ATTR_COUNT, + ATTR_LATEST, + ATTR_LATEST_UNREAD, + ATTR_NOTIFICATIONS, + ATTR_TOTAL, + DEFAULT_NAME, + DEFAULT_SENSOR_NAME, + DOMAIN, +) +from .coordinator import NotificationStoreCoordinator + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up the Notification Store sensor.""" + coordinator: NotificationStoreCoordinator = hass.data[DOMAIN]["coordinators"][entry.entry_id] + + device_name = entry.data.get(CONF_NAME, DEFAULT_NAME) + sensor_name = DEFAULT_SENSOR_NAME + unique_id = f"{entry.entry_id}_notifications" + + registry = er.async_get(hass) + current_entity_id = registry.async_get_entity_id("sensor", DOMAIN, unique_id) + target_entity_id = f"sensor.{slugify(device_name)}_{slugify(sensor_name)}" + + if ( + current_entity_id + and current_entity_id != target_entity_id + and not registry.async_is_registered(target_entity_id) + ): + registry.async_update_entity(current_entity_id, new_entity_id=target_entity_id) + + async_add_entities([NotificationStoreSensor(coordinator, entry, unique_id, sensor_name)]) + + +class NotificationStoreSensor(CoordinatorEntity[NotificationStoreCoordinator], SensorEntity): + """Sensor exposing unread notification count.""" + + _attr_icon = "mdi:bell-ring-outline" + _attr_has_entity_name = True + + def __init__(self, coordinator: NotificationStoreCoordinator, entry: ConfigEntry, unique_id: str, sensor_name: str) -> None: + """Initialize the sensor.""" + super().__init__(coordinator) + self.entry = entry + self._attr_unique_id = unique_id + self._attr_name = sensor_name + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, entry.entry_id)}, + name=entry.title, + manufacturer="Torsten Brendgen", + model="Notification Store", + ) + + @property + def native_value(self) -> int: + """Return unread notification count.""" + return len(self.coordinator.unread_notifications) + + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return notification attributes.""" + notifications = self.coordinator.notifications + unread = self.coordinator.unread_notifications + return { + ATTR_COUNT: len(unread), + ATTR_TOTAL: len(notifications), + ATTR_NOTIFICATIONS: unread, + ATTR_LATEST: notifications[0] if notifications else None, + ATTR_LATEST_UNREAD: unread[0] if unread else None, + }