Dateien nach "/" hochladen

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,
+        )

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)

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"
+        ]

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"
+}

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,
+        }