First Version

Author: fillefilip8

custom_components/hass.agent/__init__.py

@@ -0,0 +1,203 @@
+"""The HASS.Agent integration."""
+from __future__ import annotations
+import json
+import logging
+import requests
+from homeassistant.helpers import device_registry as dr
+from homeassistant.components.mqtt.models import ReceiveMessage
+from homeassistant.components.mqtt.subscription import (
+    async_prepare_subscribe_topics,
+    async_subscribe_topics,
+    async_unsubscribe_topics,
+)
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ID, CONF_NAME, CONF_URL, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import discovery
+
+from .const import DOMAIN
+
+PLATFORMS: list[Platform] = [Platform.MEDIA_PLAYER]
+
+_logger = logging.getLogger(__name__)
+
+
+def update_device_info(hass: HomeAssistant, entry: ConfigEntry, new_device_info):
+    device_registry = dr.async_get(hass)
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, entry.unique_id)},
+        name=new_device_info["device"]["name"],
+        manufacturer=new_device_info["device"]["manufacturer"],
+        model=new_device_info["device"]["model"],
+        sw_version=new_device_info["device"]["sw_version"],
+    )
+
+
+async def handle_apis_changed(hass: HomeAssistant, entry: ConfigEntry, apis):
+    if apis is not None:
+
+        device_registry = dr.async_get(hass)
+        device = device_registry.async_get_device(
+            identifiers={(DOMAIN, entry.unique_id)}
+        )
+
+        media_player = apis.get("media_player", False)
+        is_media_player_loaded = hass.data[DOMAIN][entry.entry_id]["loaded"][
+            "media_player"
+        ]
+
+        notifications = apis.get("notifications", False)
+
+        is_notifications_loaded = hass.data[DOMAIN][entry.entry_id]["loaded"][
+            "notifications"
+        ]
+
+        if media_player and is_media_player_loaded is False:
+            _logger.debug("loading media_player for device: %s", device.name)
+            await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+            hass.data[DOMAIN][entry.entry_id]["loaded"]["media_player"] = True
+        else:
+            if is_media_player_loaded:
+                _logger.debug(
+                    "unloading media_player for device: %s",
+                    device.name,
+                )
+                await hass.config_entries.async_forward_entry_unload(
+                    entry, Platform.MEDIA_PLAYER
+                )
+
+                hass.data[DOMAIN][entry.entry_id]["loaded"]["media_player"] = False
+
+        if notifications and is_notifications_loaded is False:
+            _logger.debug("loading notify for device: %s", device.name)
+
+            hass.async_create_task(
+                discovery.async_load_platform(
+                    hass,
+                    Platform.NOTIFY,
+                    DOMAIN,
+                    {CONF_ID: entry.entry_id, CONF_NAME: device.name},
+                    {},
+                )
+            )
+            hass.data[DOMAIN][entry.entry_id]["loaded"]["notifications"] = True
+        else:
+            if is_notifications_loaded:
+                _logger.debug("unloading notify for device: %s", device.name)
+                await hass.config_entries.async_unload_platforms(
+                    entry, [Platform.NOTIFY]
+                )
+
+                hass.data[DOMAIN][entry.entry_id]["loaded"]["notifications"] = False
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up HASS.Agent from a config entry."""
+
+    hass.data.setdefault(DOMAIN, {})
+
+    hass.data[DOMAIN].setdefault(
+        entry.entry_id,
+        {
+            "internal_mqtt": {},
+            "apis": {},
+            "mqtt": {},
+            "loaded": {"media_player": False, "notifications": False},
+        },
+    )
+
+    print(entry.options)
+
+    url = entry.data.get(CONF_URL, None)
+
+    if url is not None:
+        def get_device_info():
+            return requests.get(f"{url}/info", timeout=10)
+
+        response = await hass.async_add_executor_job(get_device_info)
+
+        response_json = response.json()
+
+        update_device_info(hass, entry, response_json)
+
+        apis = {
+            "notifications": True,
+            "media_player": False,  # unsupported for the moment
+        }
+
+        hass.async_create_task(handle_apis_changed(hass, entry, apis))
+        hass.data[DOMAIN][entry.entry_id]["apis"] = apis
+
+    else:
+        device_name = entry.data["device"]["name"]
+
+        sub_state = hass.data[DOMAIN][entry.entry_id]["internal_mqtt"]
+
+        def updated(message: ReceiveMessage):
+            payload = json.loads(message.payload)
+            cached = hass.data[DOMAIN][entry.entry_id]["apis"]
+            apis = payload["apis"]
+
+            update_device_info(hass, entry, payload)
+
+            if cached != apis:
+                hass.async_create_task(handle_apis_changed(hass, entry, apis))
+                hass.data[DOMAIN][entry.entry_id]["apis"] = apis
+
+        sub_state = async_prepare_subscribe_topics(
+            hass,
+            sub_state,
+            {
+                f"{entry.unique_id}-apis": {
+                    "topic": f"hass.agent/devices/{device_name}",
+                    "msg_callback": updated,
+                    "qos": 0,
+                }
+            },
+        )
+
+        await async_subscribe_topics(hass, sub_state)
+
+        hass.data[DOMAIN][entry.entry_id]["internal_mqtt"] = sub_state
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+
+    # known issue: notify does not always unload
+
+    loaded = hass.data[DOMAIN][entry.entry_id].get("loaded", None)
+
+    if loaded is not None:
+        notifications = loaded.get("notifications", False)
+
+        media_player = loaded.get("media_player", False)
+
+        if notifications:
+            if unload_ok := await hass.config_entries.async_unload_platforms(
+                entry, [Platform.NOTIFY]
+            ):
+                _logger.debug("unloaded %s for %s", "notify", entry.unique_id)
+
+        if media_player:
+            if unload_ok := await hass.config_entries.async_unload_platforms(
+                entry, [Platform.MEDIA_PLAYER]
+            ):
+                _logger.debug("unloaded %s for %s", "media_player", entry.unique_id)
+    else:
+        _logger.warning("config entry (%s) with has no apis loaded?", entry.entry_id)
+
+    url = entry.data.get(CONF_URL, None)
+    if url is None:
+        async_unsubscribe_topics(
+            hass, hass.data[DOMAIN][entry.entry_id]["internal_mqtt"]
+        )
+
+    hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

custom_components/hass.agent/manifest.json

@@ -0,0 +1,173 @@
+"""Config flow for HASS.Agent"""
+from __future__ import annotations
+import json
+import logging
+import requests
+import voluptuous as vol
+
+from typing import Any
+from homeassistant.components.notify import ATTR_TITLE_DEFAULT
+from homeassistant.core import callback
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, CONF_SSL, CONF_URL
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.service_info.mqtt import MqttServiceInfo
+
+from .const import DOMAIN, CONF_DEFAULT_NOTIFICATION_TITLE
+
+_logger = logging.getLogger(__name__)
+
+
+class OptionsFlowHandler(config_entries.OptionsFlow):
+    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
+    ) -> FlowResult:
+        """Manage the options."""
+        if user_input is not None:
+            user_input[CONF_DEFAULT_NOTIFICATION_TITLE] = user_input[
+                CONF_DEFAULT_NOTIFICATION_TITLE
+            ].strip()
+
+            return self.async_create_entry(title="", data=user_input)
+
+        return self.async_show_form(
+            step_id="init",
+            data_schema=vol.Schema(
+                {
+                    vol.Optional(
+                        CONF_DEFAULT_NOTIFICATION_TITLE,
+                        default=self.config_entry.options.get(
+                            CONF_DEFAULT_NOTIFICATION_TITLE, ATTR_TITLE_DEFAULT
+                        ),
+                    ): str
+                }
+            ),
+        )
+
+
+class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize flow."""
+        self._device_name = ""
+        self._data = {}
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(
+        config_entry: config_entries.ConfigEntry,
+    ) -> config_entries.OptionsFlow:
+        """Create the options flow."""
+        return OptionsFlowHandler(config_entry)
+
+    async def async_step_mqtt(self, discovery_info: MqttServiceInfo) -> FlowResult:
+        """Handle a flow initialized by MQTT discovery."""
+        device_name = discovery_info.topic.split("hass.agent/devices/")[1]
+
+        payload = json.loads(discovery_info.payload)
+
+        serial_number = payload["serial_number"]
+
+        _logger.debug(
+            "found device. Name: %s, Serial Number: %s", device_name, serial_number
+        )
+
+        self._data = {"device": payload["device"], "apis": payload["apis"]}
+
+        for config in self._async_current_entries():
+            if config.unique_id == serial_number:
+                return self.async_abort(reason="already_configured")
+
+        await self.async_set_unique_id(serial_number)
+
+        # "hass.agent/devices/#" is hardcoded in HASS.Agent's manifest
+        assert discovery_info.subscribed_topic == "hass.agent/devices/#"
+
+        self._device_name = device_name
+
+        return await self.async_step_confirm()
+
+    async def async_step_local_api(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+
+        errors = {}
+
+        if user_input is not None:
+            host = user_input[CONF_HOST]
+            port = user_input[CONF_PORT]
+            use_ssl = user_input[CONF_SSL]
+
+            protocol = "https" if use_ssl else "http"
+
+            url = f"{protocol}://{host}:{port}"
+
+            # serial number!
+            try:
+
+                def get_device_info():
+                    return requests.get(f"{url}/info", timeout=10)
+
+                response = await self.hass.async_add_executor_job(get_device_info)
+
+                response_json = response.json()
+
+                await self.async_set_unique_id(response_json["serial_number"])
+
+                return self.async_create_entry(
+                    title=response_json["device"]["name"],
+                    data={CONF_URL: url},
+                    options={CONF_DEFAULT_NOTIFICATION_TITLE: ATTR_TITLE_DEFAULT},
+                )
+
+            except Exception:
+                errors["base"] = "cannot_connect"
+
+        return self.async_show_form(
+            step_id="local_api",
+            data_schema=vol.Schema(
+                # pylint: disable=no-value-for-parameter
+                {
+                    vol.Required(CONF_HOST): str,
+                    vol.Required(CONF_PORT, default=5115): int,
+                    vol.Required(CONF_SSL): bool,
+                }
+            ),
+            errors=errors,
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        return await self.async_step_local_api()
+
+    async def async_step_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Confirm the setup."""
+
+        if user_input is not None:
+            return self.async_create_entry(
+                title=self._device_name,
+                data=self._data,
+                options={CONF_DEFAULT_NOTIFICATION_TITLE: ATTR_TITLE_DEFAULT},
+            )
+
+        placeholders = {CONF_NAME: self._device_name}
+
+        self.context["title_placeholders"] = placeholders
+
+        self._set_confirm_only()
+
+        return self.async_show_form(
+            step_id="confirm",
+            description_placeholders=placeholders,
+        )

custom_components/hass_agent/__init__.py

@@ -0,0 +1,7 @@
+"""Constants for the HASS.Agent integration."""
+
+DOMAIN = "hass_agent"
+
+CONF_ACTION = "action"
+CONF_DEVICE_NAME = "device_name"
+CONF_DEFAULT_NOTIFICATION_TITLE = "default_notification_title"