Merge pull request #2874 from Mikhail5555/HomeAssistantConnector

Home assistant connector

.github/workflows/ci.yml

@@ -78,6 +78,7 @@ jobs:
           ${{ runner.os }}-${{ matrix.python-version }}-${{ matrix.node-version }}-collectstatic-${{ hashFiles('**/*.css', '**/*.js', 'vue/src/*') }}
 
     - name: Django Testing project
+      timeout-minutes: 6
       run: pytest --junitxml=junit/test-results-${{ matrix.python-version }}.xml
 
     - name: Publish Test Results

.gitignore

@@ -54,7 +54,6 @@ docs/_build/
 target/
 
 \.idea/dataSources/
-
 \.idea/dataSources\.xml
 
 \.idea/dataSources\.local\.xml

cookbook/admin.py

@@ -16,7 +16,7 @@
                      ShoppingList, ShoppingListEntry, ShoppingListRecipe, Space, Step, Storage,
                      Supermarket, SupermarketCategory, SupermarketCategoryRelation, Sync, SyncLog,
                      TelegramBot, Unit, UnitConversion, UserFile, UserPreference, UserSpace,
-                     ViewLog)
+                     ViewLog, ConnectorConfig)
 
 
 class CustomUserAdmin(UserAdmin):
@@ -95,6 +95,14 @@ class StorageAdmin(admin.ModelAdmin):
 admin.site.register(Storage, StorageAdmin)
 
 
+class ConnectorConfigAdmin(admin.ModelAdmin):
+    list_display = ('id', 'name', 'type', 'enabled', 'url')
+    search_fields = ('name', 'url')
+
+
+admin.site.register(ConnectorConfig, ConnectorConfigAdmin)
+
+
 class SyncAdmin(admin.ModelAdmin):
     list_display = ('storage', 'path', 'active', 'last_checked')
     search_fields = ('storage__name', 'path')

cookbook/apps.py

@@ -3,6 +3,7 @@
 from django.apps import AppConfig
 from django.conf import settings
 from django.db import OperationalError, ProgrammingError
+from django.db.models.signals import post_save, post_delete
 from django_scopes import scopes_disabled
 
 from recipes.settings import DEBUG
@@ -14,6 +15,12 @@ class CookbookConfig(AppConfig):
     def ready(self):
         import cookbook.signals  # noqa
 
+        if not settings.DISABLE_EXTERNAL_CONNECTORS:
+            from cookbook.connectors.connector_manager import ConnectorManager  # Needs to be here to prevent loading race condition of oauth2 modules in models.py
+            handler = ConnectorManager()
+            post_save.connect(handler, dispatch_uid="connector_manager")
+            post_delete.connect(handler, dispatch_uid="connector_manager")
+
         # if not settings.DISABLE_TREE_FIX_STARTUP:
         #     # when starting up run fix_tree to:
         #     #   a) make sure that nodes are sorted when switching between sort modes

cookbook/connectors/connector.py

@@ -0,0 +1,29 @@
+from abc import ABC, abstractmethod
+
+from cookbook.models import ShoppingListEntry, Space, ConnectorConfig
+
+
+# A Connector is 'destroyed' & recreated each time 'any' ConnectorConfig in a space changes.
+class Connector(ABC):
+    @abstractmethod
+    def __init__(self, config: ConnectorConfig):
+        pass
+
+    @abstractmethod
+    async def on_shopping_list_entry_created(self, space: Space, instance: ShoppingListEntry) -> None:
+        pass
+
+    # This method might not trigger on 'direct' entry updates: https://stackoverflow.com/a/35238823
+    @abstractmethod
+    async def on_shopping_list_entry_updated(self, space: Space, instance: ShoppingListEntry) -> None:
+        pass
+
+    @abstractmethod
+    async def on_shopping_list_entry_deleted(self, space: Space, instance: ShoppingListEntry) -> None:
+        pass
+
+    @abstractmethod
+    async def close(self) -> None:
+        pass
+
+    # TODO: Add Recipes & possibly Meal Place listeners/hooks (And maybe more?)

cookbook/connectors/connector_manager.py

@@ -0,0 +1,179 @@
+import asyncio
+import logging
+import queue
+import threading
+from asyncio import Task
+from dataclasses import dataclass
+from enum import Enum
+from types import UnionType
+from typing import List, Any, Dict, Optional, Type
+
+from django.conf import settings
+from django_scopes import scope
+
+from cookbook.connectors.connector import Connector
+from cookbook.connectors.homeassistant import HomeAssistant
+from cookbook.models import ShoppingListEntry, Space, ConnectorConfig
+
+REGISTERED_CLASSES: UnionType | Type = ShoppingListEntry
+
+
+class ActionType(Enum):
+    CREATED = 1
+    UPDATED = 2
+    DELETED = 3
+
+
+@dataclass
+class Work:
+    instance: REGISTERED_CLASSES | ConnectorConfig
+    actionType: ActionType
+
+
+# The way ConnectionManager works is as follows:
+# 1. On init, it starts a worker & creates a queue for 'Work'
+# 2. Then any time its called, it verifies the type of action (create/update/delete) and if the item is of interest, pushes the Work (non-blocking) to the queue.
+# 3. The worker consumes said work from the queue.
+# 3.1 If the work is of type ConnectorConfig, it flushes its cache of known connectors (per space.id)
+# 3.2 If work is of type REGISTERED_CLASSES, it asynchronously fires of all connectors and wait for them to finish (runtime should depend on the 'slowest' connector)
+# 4. Work is marked as consumed, and next entry of the queue is consumed.
+# Each 'Work' is processed in sequential by the worker, so the throughput is about [workers * the slowest connector]
+class ConnectorManager:
+    _queue: queue.Queue
+    _listening_to_classes = REGISTERED_CLASSES | ConnectorConfig
+
+    def __init__(self):
+        self._queue = queue.Queue(maxsize=settings.EXTERNAL_CONNECTORS_QUEUE_SIZE)
+        self._worker = threading.Thread(target=self.worker, args=(0, self._queue,), daemon=True)
+        self._worker.start()
+
+    # Called by post save & post delete signals
+    def __call__(self, instance: Any, **kwargs) -> None:
+        if not isinstance(instance, self._listening_to_classes) or not hasattr(instance, "space"):
+            return
+
+        action_type: ActionType
+        if "created" in kwargs and kwargs["created"]:
+            action_type = ActionType.CREATED
+        elif "created" in kwargs and not kwargs["created"]:
+            action_type = ActionType.UPDATED
+        elif "origin" in kwargs:
+            action_type = ActionType.DELETED
+        else:
+            return
+
+        try:
+            self._queue.put_nowait(Work(instance, action_type))
+        except queue.Full:
+            logging.info(f"queue was full, so skipping {action_type} of type {type(instance)}")
+            return
+
+    def stop(self):
+        self._queue.join()
+        self._worker.join()
+
+    @staticmethod
+    def worker(worker_id: int, worker_queue: queue.Queue):
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        logging.info(f"started ConnectionManager worker {worker_id}")
+
+        # When multiple workers are used, please make sure the cache is shared across all threads, otherwise it might lead to un-expected behavior.
+        _connectors_cache: Dict[int, List[Connector]] = dict()
+
+        while True:
+            try:
+                item: Optional[Work] = worker_queue.get()
+            except KeyboardInterrupt:
+                break
+
+            if item is None:
+                break
+
+            # If a Connector was changed/updated, refresh connector from the database for said space
+            refresh_connector_cache = isinstance(item.instance, ConnectorConfig)
+
+            space: Space = item.instance.space
+            connectors: Optional[List[Connector]] = _connectors_cache.get(space.id)
+
+            if connectors is None or refresh_connector_cache:
+                if connectors is not None:
+                    loop.run_until_complete(close_connectors(connectors))
+
+                with scope(space=space):
+                    connectors: List[Connector] = list()
+                    for config in space.connectorconfig_set.all():
+                        config: ConnectorConfig = config
+                        if not config.enabled:
+                            continue
+
+                        try:
+                            connector: Optional[Connector] = ConnectorManager.get_connected_for_config(config)
+                        except BaseException:
+                            logging.exception(f"failed to initialize {config.name}")
+                            continue
+
+                        if connector is not None:
+                            connectors.append(connector)
+
+                    _connectors_cache[space.id] = connectors
+
+            if len(connectors) == 0 or refresh_connector_cache:
+                worker_queue.task_done()
+                continue
+
+            loop.run_until_complete(run_connectors(connectors, space, item.instance, item.actionType))
+            worker_queue.task_done()
+
+        logging.info(f"terminating ConnectionManager worker {worker_id}")
+
+        asyncio.set_event_loop(None)
+        loop.close()
+
+    @staticmethod
+    def get_connected_for_config(config: ConnectorConfig) -> Optional[Connector]:
+        match config.type:
+            case ConnectorConfig.HOMEASSISTANT:
+                return HomeAssistant(config)
+            case _:
+                return None
+
+
+async def close_connectors(connectors: List[Connector]):
+    tasks: List[Task] = [asyncio.create_task(connector.close()) for connector in connectors]
+
+    if len(tasks) == 0:
+        return
+
+    try:
+        await asyncio.gather(*tasks, return_exceptions=False)
+    except BaseException:
+        logging.exception("received an exception while closing one of the connectors")
+
+
+async def run_connectors(connectors: List[Connector], space: Space, instance: REGISTERED_CLASSES, action_type: ActionType):
+    tasks: List[Task] = list()
+
+    if isinstance(instance, ShoppingListEntry):
+        shopping_list_entry: ShoppingListEntry = instance
+
+        match action_type:
+            case ActionType.CREATED:
+                for connector in connectors:
+                    tasks.append(asyncio.create_task(connector.on_shopping_list_entry_created(space, shopping_list_entry)))
+            case ActionType.UPDATED:
+                for connector in connectors:
+                    tasks.append(asyncio.create_task(connector.on_shopping_list_entry_updated(space, shopping_list_entry)))
+            case ActionType.DELETED:
+                for connector in connectors:
+                    tasks.append(asyncio.create_task(connector.on_shopping_list_entry_deleted(space, shopping_list_entry)))
+
+    if len(tasks) == 0:
+        return
+
+    try:
+        # Wait for all async tasks to finish, if one fails, the others still continue.
+        await asyncio.gather(*tasks, return_exceptions=False)
+    except BaseException:
+        logging.exception("received an exception from one of the connectors")

cookbook/connectors/homeassistant.py

@@ -0,0 +1,85 @@
+import logging
+from logging import Logger
+
+from homeassistant_api import Client, HomeassistantAPIError, Domain
+
+from cookbook.connectors.connector import Connector
+from cookbook.models import ShoppingListEntry, ConnectorConfig, Space
+
+
+class HomeAssistant(Connector):
+    _domains_cache: dict[str, Domain]
+    _config: ConnectorConfig
+    _logger: Logger
+    _client: Client
+
+    def __init__(self, config: ConnectorConfig):
+        if not config.token or not config.url or not config.todo_entity:
+            raise ValueError("config for HomeAssistantConnector in incomplete")
+
+        self._domains_cache = dict()
+        self._config = config
+        self._logger = logging.getLogger("connector.HomeAssistant")
+        self._client = Client(self._config.url, self._config.token, async_cache_session=False, use_async=True)
+
+    async def on_shopping_list_entry_created(self, space: Space, shopping_list_entry: ShoppingListEntry) -> None:
+        if not self._config.on_shopping_list_entry_created_enabled:
+            return
+
+        item, description = _format_shopping_list_entry(shopping_list_entry)
+
+        todo_domain = self._domains_cache.get('todo')
+        try:
+            if todo_domain is None:
+                todo_domain = await self._client.async_get_domain('todo')
+                self._domains_cache['todo'] = todo_domain
+
+            logging.debug(f"pushing {item} to {self._config.name}")
+            await todo_domain.add_item(entity_id=self._config.todo_entity, item=item)
+        except HomeassistantAPIError as err:
+            self._logger.warning(f"[HomeAssistant {self._config.name}] Received an exception from the api: {err=}, {type(err)=}")
+
+    async def on_shopping_list_entry_updated(self, space: Space, shopping_list_entry: ShoppingListEntry) -> None:
+        if not self._config.on_shopping_list_entry_updated_enabled:
+            return
+        pass
+
+    async def on_shopping_list_entry_deleted(self, space: Space, shopping_list_entry: ShoppingListEntry) -> None:
+        if not self._config.on_shopping_list_entry_deleted_enabled:
+            return
+
+        item, description = _format_shopping_list_entry(shopping_list_entry)
+
+        todo_domain = self._domains_cache.get('todo')
+        try:
+            if todo_domain is None:
+                todo_domain = await self._client.async_get_domain('todo')
+                self._domains_cache['todo'] = todo_domain
+
+            logging.debug(f"deleting {item} from {self._config.name}")
+            await todo_domain.remove_item(entity_id=self._config.todo_entity, item=item)
+        except HomeassistantAPIError as err:
+            self._logger.warning(f"[HomeAssistant {self._config.name}] Received an exception from the api: {err=}, {type(err)=}")
+
+    async def close(self) -> None:
+        await self._client.async_cache_session.close()
+
+
+def _format_shopping_list_entry(shopping_list_entry: ShoppingListEntry):
+    item = shopping_list_entry.food.name
+    if shopping_list_entry.amount > 0:
+        item += f" ({shopping_list_entry.amount:.2f}".rstrip('0').rstrip('.')
+        if shopping_list_entry.unit and shopping_list_entry.unit.base_unit and len(shopping_list_entry.unit.base_unit) > 0:
+            item += f" {shopping_list_entry.unit.base_unit})"
+        elif shopping_list_entry.unit and shopping_list_entry.unit.name and len(shopping_list_entry.unit.name) > 0:
+            item += f" {shopping_list_entry.unit.name})"
+        else:
+            item += ")"
+
+    description = "Imported by TandoorRecipes"
+    if shopping_list_entry.created_by.first_name and len(shopping_list_entry.created_by.first_name) > 0:
+        description += f", created by {shopping_list_entry.created_by.first_name}"
+    else:
+        description += f", created by {shopping_list_entry.created_by.username}"
+
+    return item, description