Add db manager and ha to components

Author: springfall2008

apps/predbat/components.py

@@ -13,9 +13,37 @@
 from ohme import OhmeAPI
 from octopus import OctopusAPI
 from web import WebInterface
+from ha import HAInterface
+from db_manager import DatabaseManager
+import os
 
 COMPONENT_LIST = {
-    "web": {"class": WebInterface, "name": "Web Interface", "args": {"port": {"required": False, "config": "web_port", "default": 5052}}},
+    "db": {
+        "class": DatabaseManager,
+        "name": "Database Manager",
+        "args": {
+            "db_enable": {"required": True, "config": "db_enable"},
+            "db_days": {"required": False, "config": "db_days", "default": 30},
+        }
+    },
+    "ha": {
+        "class": HAInterface,
+        "name": "Home Assistant Interface",
+        "args": {
+            "ha_url": {"required": False, "config": "ha_url", "default": "http://supervisor/core"},
+            "ha_key": {"required": False, "config": "ha_key", "default": os.environ.get("SUPERVISOR_TOKEN", None)},
+            "db_enable": {"required": False, "config": "db_enable", "default": False},
+            "db_mirror_ha": {"required": False, "config": "db_mirror_ha", "default": False},
+            "db_primary": {"required": False, "config": "db_primary", "default": False},
+        }
+    },
+    "web": {
+        "class": WebInterface,
+        "name": "Web Interface",
+        "args": {
+            "port": {"required": False, "config": "web_port", "default": 5052}
+        }
+    },
     "gecloud": {
         "class": GECloudDirect,
         "name": "GivEnergy Cloud Direct",
@@ -101,9 +129,11 @@ def start(self):
                     raise ValueError(f"{component_info['name']} API failed to start")
 
     async def stop(self):
-        for component_name, component in self.components.items():
+        for component_name, component_info in reversed(list(self.components.items())):
+            component = self.components[component_name]
             if component:
                 self.log(f"Stopping {component_name} interface")
+                print(component_name, component)
                 await component.stop()
                 self.component_tasks[component_name] = None
                 self.components[component_name] = None
@@ -114,17 +144,20 @@ async def stop(self):
 
     async def select_event(self, entity_id, value):
         for component_name, component in self.components.items():
-            if component and COMPONENT_LIST[component_name].get("event_filter", "") in entity_id:
+            event_filter = COMPONENT_LIST[component_name].get("event_filter", None)
+            if component and event_filter and (event_filter in entity_id):
                 await component.select_event(entity_id, value)
 
     async def switch_event(self, entity_id, service):
         for component_name, component in self.components.items():
-            if component and COMPONENT_LIST[component_name].get("event_filter", "") in entity_id:
+            event_filter = COMPONENT_LIST[component_name].get("event_filter", None)
+            if component and event_filter and (event_filter in entity_id):
                 await component.switch_event(entity_id, service)
 
     async def number_event(self, entity_id, value):
         for component_name, component in self.components.items():
-            if component and COMPONENT_LIST[component_name].get("event_filter", "") in entity_id:
+            event_filter = COMPONENT_LIST[component_name].get("event_filter", None)
+            if component and event_filter and (event_filter in entity_id):
                 await component.number_event(entity_id, value)
 
     def is_all_alive(self):
@@ -135,9 +168,14 @@ def is_alive(self, name):
         """Check if a single component is alive"""
         if name not in self.components:
             return False
+        if not self.components[name]:
+            # Disabled components can be ignored
+            return True        
         if not self.component_tasks[name] or not self.component_tasks[name].is_alive():
             return False
-        return self.components[name].is_alive()
+        if not self.components[name].is_alive():
+            return False
+        return True
 
     def get_active(self):
         active_components = [name for name, comp in self.components.items() if comp]

apps/predbat/db_manager.py

@@ -16,7 +16,7 @@
 
 
 class DatabaseManager:
-    def __init__(self, base, db_days):
+    def __init__(self, db_enable, db_days, base):
         self.base = base
         self.log = base.log
         self.db_days = db_days
@@ -27,6 +27,7 @@ def __init__(self, base, db_days):
         self.sync_event = threading.Event()
         self.async_event = asyncio.Event()
         self.return_event = threading.Event()
+        self.api_started = False
 
     def bridge_event(self, loop):
         """
@@ -39,6 +40,24 @@ def bridge_event(self, loop):
             self.sync_event.clear()  # Clear the event to allow the loop to continue
             loop.call_soon_threadsafe(self.async_event.set)
 
+    def is_alive(self):
+        """Check if the database manager is alive"""
+        return self.api_started
+    
+    def wait_api_started(self):
+        """
+        Wait for the API to start
+        """
+        self.log("DBManager: Waiting for API to start")
+        count = 0
+        while not self.api_started and count < 240:
+            time.sleep(1)
+            count += 1
+        if not self.api_started:
+            self.log("Warn: DBManager: Failed to start")
+            return False
+        return True
+
     async def start(self):
         """
         Initialize the database and clean up old data
@@ -52,6 +71,7 @@ async def start(self):
         threading.Thread(target=self.bridge_event, args=(loop,), daemon=True).start()
 
         self.log("db_manager: Started")
+        self.api_started = True
 
         while not self.stop_thread:
             if not self.db_queue:
@@ -87,6 +107,7 @@ async def start(self):
 
         self.db_engine._close()
         self.log("db_manager: Stopped")
+        self.api_started = False
 
     def send_via_ipc(self, command, info, expect_response=False):
         """
@@ -114,14 +135,14 @@ def send_via_ipc(self, command, info, expect_response=False):
                 return None
         return None
 
-    def stop(self):
+    async def stop(self):
         """
         Close the database connection
         """
         self.stop_thread = True
-        queue_id = self.queue_id
-        self.queue_id += 1
         self.send_via_ipc("stop", {}, expect_response=False)
+        self.api_started = False
+        self.log("db_manager: stop command sent")
 
     def get_state_db(self, entity_id):
         """

apps/predbat/ha.py

@@ -17,6 +17,7 @@
 import requests
 import traceback
 import threading
+import time
 from config import TIME_FORMAT_HA, TIMEOUT, TIME_FORMAT_HA_TZ
 
 from db_manager import DatabaseManager  # database functions have been refactored to DatabaseManager
@@ -50,37 +51,51 @@ class HAInterface:
     """
     Direct interface to Home Assistant
     """
+    def is_active(self):
+        return self.is_alive()
 
-    def is_running(self):
+    def is_alive(self):
+        if not self.api_started:
+            return False
         if self.websocket_task:
             if not self.websocket_task.is_alive():
                 return False
             if not self.websocket_active:
                 return False
-        if self.db_manager_task:
-            if not self.db_manager_task.is_alive():
-                return False
         return True
 
-    def __init__(self, base):
+    def wait_api_started(self):
+        """
+        Wait for the API to start
+        """
+        self.log("HAInterface: Waiting for API to start")
+        count = 0
+        while not self.api_started and count < 240:
+            time.sleep(1)
+            count += 1
+        if not self.api_started:
+            self.log("Warn: HAInterface: Failed to start")
+            return False
+        return True
+
+    def __init__(self, ha_url, ha_key, db_enable, db_mirror_ha, db_primary, base):
         """
         Initialize the interface to Home Assistant.
         """
-        self.ha_url = base.args.get("ha_url", "http://supervisor/core")
-        self.ha_key = base.args.get("ha_key", os.environ.get("SUPERVISOR_TOKEN", None))
-        self.db_enable = base.args.get("db_enable", False)
-        self.db_days = base.args.get("db_days", 30)
-        self.db_mirror_ha = base.args.get("db_mirror_ha", False)
-        self.db_primary = base.args.get("db_primary", False)
+        self.ha_url = ha_url
+        self.ha_key = ha_key
+        self.db_enable = db_enable
+        self.db_mirror_ha = db_mirror_ha
+        self.db_primary = db_primary
         self.db_manager = None
-        self.db_manager_task = None
         self.db_mirror_list = {}
         self.db = None
         self.db_cursor = None
         self.websocket_active = False
         self.websocket_task = None
         self.api_errors = 0
         self.stop_thread = False
+        self.api_started = False
 
         self.base = base
         self.log = base.log
@@ -93,43 +108,39 @@ def __init__(self, base):
                 raise ValueError
             else:
                 self.log("Info: Using SQL Lite database as primary data source, no HA interface available")
-        else:
-            self.log("Getting slug id from HA supervisor")
 
+        if self.ha_key:
             # Get the current addon info, but suppress warning message if the API call fails as non-HAOS installs won't have supervisor running
             res = self.api_call("/addons/self/info", core=False, silent=True)
             if res:
                 # get add-on slug name which is the actual directory name under /addon_configs that /config is mounted to
                 self.slug = res["data"]["slug"]
                 self.log("Info: Add-on slug is {}".format(self.slug))
 
-            self.log("Checking Predbat can talk to HA")
-
             check = self.api_call("/api/services")
             if not check:
                 self.log("Warn: Unable to connect directly to Home Assistant at {}, please check your configuration of ha_url/ha_key".format(self.ha_url))
                 self.ha_key = None
                 raise ValueError
-            else:
-                self.log("Info: Connected to Home Assistant at {}".format(self.ha_url))
-
-                self.log("Creating websocket")
-
-                self.websocket_task = self.base.create_task(self.socketLoop())
-                self.websocket_active = True
-                self.log("Info: Web Socket task started")
 
         if self.db_enable:
             # Open the SQL Lite database called predbat.db using the DatabaseManager
-            self.log("Info: Opening database")
-            self.db_manager = DatabaseManager(self.base, self.db_days)
-            self.db_manager_task = self.base.create_task(self.db_manager.start())
+            self.db_manager = self.base.components.get_component("db")
+
+        # API Has started
+        self.base.ha_interface = self # Set pointer back to ourselves as other components require this one
+        self.api_started = True
+
+    async def start(self):
+        """ Async start not required
+        """
+        self.log("Info: Starting HA interface")
+        self.websocket_active = True
+        await self.socketLoop()
 
-    def stop(self):
+    async def stop(self):
         self.stop_thread = True
-        if self.db_manager:
-            self.log("Info: Stopping database manager")
-            self.db_manager.stop()
+        self.api_started = False
 
     def get_slug(self):
         """

apps/predbat/octopus.py

@@ -416,7 +416,7 @@ async def save_octopus_cache(self):
         with open(self.cache_file, "w") as f:
             yaml.dump(octopus_cache, f)
 
-    def stop(self):
+    async def stop(self):
         self.stop_api = True
 
     def get_tariff(self, tariff_type):

apps/predbat/predbat.py

@@ -70,7 +70,6 @@
 )
 from prediction import reset_prediction_globals
 from utils import minutes_since_yesterday, dp1, dp2, dp3
-from ha import HAInterface
 from predheat import PredHeat
 from octopus import Octopus
 from energydataservice import Energidataservice
@@ -1278,8 +1277,6 @@ def is_running(self):
 
         if not self.ha_interface:
             return False
-        if not self.ha_interface.is_running():
-            return False
 
         if self.components:
             if not self.components.is_all_alive():
@@ -1318,19 +1315,14 @@ def initialize(self):
         try:
             self.reset()
             self.update_time(print=False)
-            self.log("Starting HA interface")
-            try:
-                self.ha_interface = HAInterface(self)
-            except ValueError as e:
-                self.log("Error: Exception raised {}".format(e))
-                self.log("Error: " + traceback.format_exc())
-                self.record_status("Error: Exception raised {}".format(e), debug=traceback.format_exc())
-                raise e
 
             # Start all sub-components
             self.components = Components(self)
             self.components.start()
 
+            if not self.ha_interface:
+                raise ValueError("HA interface not found")
+
             # Initialize plugin system and discover plugins
             self.init_plugin_system()
 
@@ -1396,8 +1388,6 @@ async def terminate(self):
         self.stop_thread = True
         if self.components:
             await self.components.stop()
-        if self.ha_interface:
-            self.ha_interface.stop()
 
         await asyncio.sleep(0)
         if hasattr(self, "pool"):

apps/predbat/web.py

@@ -472,7 +472,7 @@ def get_status_html(self, status, version):
         if status and (("Warn:" in status) or ("Error:" in status)):
             text += "<tr><td>Status</td><td bgcolor=#ff7777>{}</td></tr>\n".format(status)
         elif not is_running:
-            text += "<tr><td colspan='2' bgcolor='#ff7777'>Predbat has errors</td></tr>\n"
+            text += "<tr><td colspan='2' bgcolor='#ff7777'>{} (with errors)</td></tr>\n".format(status)
         else:
             text += "<tr><td>Status</td><td>{}</td></tr>\n".format(status)
         text += "<tr><td>Last Updated</td><td>{}</td></tr>\n".format(last_updated)