Adds camera_proxy support

Author: fillefilip8

custom_components/hass_agent_notifier/notify.py

@@ -14,11 +14,11 @@
 
 import logging
 from typing import Any
-
-from functools import partial
+from contextlib import suppress
 
 import requests
 import voluptuous as vol
+import re
 
 from homeassistant.components.notify import (
     ATTR_TITLE_DEFAULT,
@@ -30,6 +30,8 @@
 
 from homeassistant.components import media_source
 
+from homeassistant.helpers.network import NoURLAvailableError, get_url
+
 from http import HTTPStatus
 
 from homeassistant.const import (
@@ -41,6 +43,8 @@
 
 _LOGGER = logging.getLogger(__name__)
 
+CAMERA_PROXY_REGEX = re.compile(r"\/api\/camera_proxy\/camera\.(.*)")
+
 
 def get_service(hass, config, discovery_info=None):
     """Get the HASS Agent notification service."""
@@ -59,37 +63,57 @@ def __init__(self, hass, resource):
         self._resource = resource
         self._hass = hass
 
-    def send(self, url, data):
+    def send_request(self, url, data):
+        """Sends the json request"""
         return requests.post(url, json=data, timeout=10)
 
     async def async_send_message(self, message: str, **kwargs: Any):
         """Send the message to the provided resource."""
-        _LOGGER.debug("Preparing notification ..")
+        _LOGGER.debug("Preparing notification")
 
         title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
         data = kwargs.get(ATTR_DATA, None)
 
         image = data.get("image", None)
+
         if image is not None:
-            if media_source.is_media_source_id(image):
+            new_url = None
+
+            camera_proxy_match = CAMERA_PROXY_REGEX.match(image)
+
+            if camera_proxy_match is not None:
+                camera = self.hass.states.get(f"camera.{camera_proxy_match.group(1)}")
+
+                if camera is not None:
+                    external_url = None
+                    with suppress(NoURLAvailableError):  # external_url not configured
+                        external_url = get_url(self.hass, allow_internal=False)
+
+                    if external_url is not None:
+                        access_token = camera.attributes["access_token"]
+                        new_url = f"{external_url}{image}?token={access_token}"
+
+            elif media_source.is_media_source_id(image):
                 sourced_media = await media_source.async_resolve_media(self.hass, image)
                 sourced_media = media_source.async_process_play_media_url(
                     self.hass, sourced_media.url
                 )
+                new_url = sourced_media
 
-                data.update({"image": sourced_media})
+            if new_url is not None:
+                data.update({"image": new_url})
 
         payload = {"message": message, "title": title, "data": data}
 
-        _LOGGER.debug("Sending notification ..")
+        _LOGGER.debug("Sending notification")
 
         try:
 
             response = await self.hass.async_add_executor_job(
-                self.send, self._resource, payload
+                self.send_request, self._resource, payload
             )
 
-            _LOGGER.debug("Checking result ..")
+            _LOGGER.debug("Checking result")
 
             if response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR:
                 _LOGGER.error(
@@ -146,5 +170,5 @@ async def async_send_message(self, message: str, **kwargs: Any):
                     "Unknown response %d: %s", response.status_code, response.reason
                 )
 
-        except Exception as e:
-            _LOGGER.debug("Error sending message: %s", e)
+        except Exception as ex:
+            _LOGGER.debug("Error sending message: %s", ex)