diff --git a/.cspell.json b/.cspell.json index bfb0d592d..314e8837e 100644 --- a/.cspell.json +++ b/.cspell.json @@ -14,6 +14,7 @@ "ignorePaths": [ "**/*.json", "**/*.yaml", + ".gitignore", ], "import": [ "@cspell/dict-en-gb/cspell-ext.json" diff --git a/.gitignore b/.gitignore index 0557b91c3..3646ef968 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,88 @@ batpred.code-workspace .vscode/settings.json + +# System files +.DS_Store +.DS_Store? +._* +.Spotlight-V100 +.Trashes +ehthumbs.db +Thumbs.db + +# Python cache files +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# Test and coverage files +coverage/ +.coverage +.pytest_cache/ +.tox/ +.nox/ +.coverage +htmlcov/ +.mypy_cache/ +.dmypy.json +dmypy.json + +# Generated test files +test_*.html +test_*.py +*_test.html +*_test.py + +# Debug and log files +*.log +*.txt.actual.json +*.yaml.actual.json +*.bin +profile.txt +simulation_state.bin + +# Generated HTML files +plan_*.html +*.html_10.html + +# Generated PNG files +*.png + +# Build artifacts +a.out +prediction_server +standalone_ha + +# Temporary and backup files +*.tmp +*.bak +*.swp +*.swo +*~ +*.mark + +# Development artifacts +apps.mod.yaml +apps.yaml.mark +comparisons.yaml +it +deploy +; diff --git a/apps/predbat/components.py b/apps/predbat/components.py new file mode 100644 index 000000000..5de8618ee --- /dev/null +++ b/apps/predbat/components.py @@ -0,0 +1,183 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2024 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +# fmt off +# pylint: disable=consider-using-f-string +# pylint: disable=line-too-long +# pylint: disable=attribute-defined-outside-init + + +from gecloud import GECloudDirect +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 = { + "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", + "event_filter": "predbat_gecloud_", + "args": { + "api_key": { + "required": True, + "config": "ge_cloud_key", + }, + "automatic": { + "required": False, + "default": False, + "config": "ge_cloud_automatic", + }, + }, + }, + "octopus": { + "class": OctopusAPI, + "name": "Octopus Energy Direct", + "event_filter": "predbat_octopus_", + "args": { + "key": { + "required": True, + "config": "octopus_api_key", + }, + "account_id": { + "required": True, + "config": "octopus_api_account", + }, + }, + }, + "ohme": { + "class": OhmeAPI, + "name": "Ohme Charger", + "event_filter": "predbat_ohme_", + "args": { + "email": { + "required": True, + "config": "ohme_login", + }, + "password": { + "required": True, + "config": "ohme_password", + }, + "ohme_automatic_octopus_intelligent": { + "required": False, + "config": "ohme_automatic_octopus_intelligent", + }, + }, + }, +} + + +class Components: + def __init__(self, base): + self.components = {} + self.component_tasks = {} + self.base = base + self.log = base.log + + def start(self): + for component_name, component_info in COMPONENT_LIST.items(): + have_all_args = True + self.components[component_name] = None + self.component_tasks[component_name] = None + + # Check required arguments + arg_dict = {} + for arg, arg_info in component_info["args"].items(): + required = arg_info.get("required", False) + default = arg_info.get("default", None) + if required and self.base.get_arg(arg_info["config"], None, indirect=False) is None: + have_all_args = False + else: + arg_dict[arg] = self.base.get_arg(arg_info["config"], default, indirect=False) + if have_all_args: + self.log(f"Starting {component_info['name']} interface") + self.components[component_name] = component_info["class"](*arg_dict.values(), self.base) + self.component_tasks[component_name] = self.base.create_task(self.components[component_name].start()) + if not self.components[component_name].wait_api_started(): + self.log(f"Error: {component_info['name']} API failed to start") + self.record_status(f"Error: {component_info['name']} API failed to start") + raise ValueError(f"{component_info['name']} API failed to start") + + async def stop(self): + 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 + + """ + Pass through events to the appropriate component + """ + + async def select_event(self, entity_id, value): + for component_name, component in self.components.items(): + 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(): + 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(): + 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): + """Check if a component is alive, or check if all are alive""" + return all(self.is_alive(name) for name in self.components.keys()) + + 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 + 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] + return active_components + + def get_component(self, name): + return self.components.get(name, None) + + def get_all(self): + all_components = [name for name in self.components.keys()] + return all_components diff --git a/apps/predbat/db_manager.py b/apps/predbat/db_manager.py index b0f0abe03..a059be3ed 100644 --- a/apps/predbat/db_manager.py +++ b/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): """ diff --git a/apps/predbat/fetch.py b/apps/predbat/fetch.py index 3c7e47ebd..66024ae47 100644 --- a/apps/predbat/fetch.py +++ b/apps/predbat/fetch.py @@ -1000,11 +1000,12 @@ def fetch_sensor_data(self): # Log current values self.log("Current data so far today: load {} kWh import {} kWh export {} kWh pv {} kWh".format(dp2(self.load_minutes_now), dp2(self.import_today_now), dp2(self.export_today_now), dp2(self.pv_today_now))) + octopus_api_direct = self.components.get_component("octopus") if "rates_import_octopus_url" in self.args: # Fixed URL for rate import self.log("Downloading import rates directly from URL {}".format(self.get_arg("rates_import_octopus_url", indirect=False))) self.rate_import = self.download_octopus_rates(self.get_arg("rates_import_octopus_url", indirect=False)) - elif self.octopus_api_direct: + elif octopus_api_direct: self.log("Downloading import rates directly from Octopus API") self.rate_import = self.get_octopus_direct("import") if not self.rate_import: @@ -1032,7 +1033,7 @@ def fetch_sensor_data(self): self.rate_import = self.basic_rates(self.get_arg("rates_import", [], indirect=False), "rates_import") # Gas rates if set - if self.octopus_api_direct: + if octopus_api_direct: self.log("Downloading gas rates directly from Octopus API") self.rate_gas = self.get_octopus_direct("gas") self.rate_gas, self.rate_gas_replicated = self.rate_replicate(self.rate_gas, is_import=False, is_gas=False) @@ -1079,16 +1080,16 @@ def fetch_sensor_data(self): slot_is_configured = "octopus_intelligent_slot" in self.args ohme_automatic = self.get_arg("ohme_automatic", False) - if (self.octopus_api_direct and self.octopus_api_direct.get_intelligent_device()) or (not self.octopus_api_direct and slot_is_configured): + if (octopus_api_direct and octopus_api_direct.get_intelligent_device()) or (not octopus_api_direct and slot_is_configured): completed = [] planned = [] vehicle = {} vehicle_pref = {} - if self.octopus_api_direct and not ohme_automatic: - completed = self.octopus_api_direct.get_intelligent_completed_dispatches() - planned = self.octopus_api_direct.get_intelligent_planned_dispatches() - vehicle = self.octopus_api_direct.get_intelligent_vehicle() + if octopus_api_direct and not ohme_automatic: + completed = octopus_api_direct.get_intelligent_completed_dispatches() + planned = octopus_api_direct.get_intelligent_planned_dispatches() + vehicle = octopus_api_direct.get_intelligent_vehicle() vehicle_pref = vehicle self.log("Octopus API planned and completed slots: {} - {}".format(completed, planned)) else: @@ -1124,7 +1125,7 @@ def fetch_sensor_data(self): if self.num_cars >= 1: # Extract vehicle data if we can get it - if vehicle or (self.octopus_api_direct and not ohme_automatic): + if vehicle or (octopus_api_direct and not ohme_automatic): self.car_charging_battery_size[0] = float(vehicle.get("vehicleBatterySizeInKwh", self.car_charging_battery_size[0])) self.car_charging_rate[0] = float(vehicle.get("chargePointPowerInKw", self.car_charging_rate[0])) else: @@ -1139,7 +1140,7 @@ def fetch_sensor_data(self): self.car_charging_limit[0] = dp3((float(self.get_arg("car_charging_limit", 100.0, index=0)) * self.car_charging_battery_size[0]) / 100.0) # Extract vehicle preference if we can get it - if (vehicle_pref or (self.octopus_api_direct and not ohme_automatic)) and self.octopus_intelligent_charging: + if (vehicle_pref or (octopus_api_direct and not ohme_automatic)) and self.octopus_intelligent_charging: octopus_limit = max(float(vehicle_pref.get("weekdayTargetSoc", 100)), float(vehicle_pref.get("weekendTargetSoc", 100))) octopus_ready_time = vehicle_pref.get("weekdayTargetTime", None) if not octopus_ready_time: @@ -1200,7 +1201,7 @@ def fetch_sensor_data(self): # Fixed URL for rate export self.log("Downloading export rates directly from URL {}".format(self.get_arg("rates_export_octopus_url", indirect=False))) self.rate_export = self.download_octopus_rates(self.get_arg("rates_export_octopus_url", indirect=False)) - elif self.octopus_api_direct: + elif octopus_api_direct: self.log("Downloading export rates directly from Octopus API") self.rate_export = self.get_octopus_direct("export") elif "metric_octopus_export" in self.args: @@ -1225,7 +1226,7 @@ def fetch_sensor_data(self): self.octopus_free_slots, self.octopus_saving_slots = self.fetch_octopus_sessions() # Standing charge - if self.octopus_api_direct: + if octopus_api_direct: self.metric_standing_charge = self.get_standing_charge_direct() self.log("Octopus Import standing charge is set to {} p".format(self.metric_standing_charge)) else: diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index b6fb0ecb2..35828b639 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -190,14 +190,14 @@ class GECloudDirect: - def __init__(self, base): + def __init__(self, api_key, automatic, base): """ Setup client """ self.base = base self.log = base.log - self.api_key = self.base.args.get("ge_cloud_key", None) - self.automatic = self.base.args.get("ge_cloud_automatic", False) + self.api_key = api_key + self.automatic = automatic self.register_list = {} self.settings = {} self.status = {} @@ -231,6 +231,12 @@ def wait_api_started(self): return False return True + def is_alive(self): + """ + Check if the API is alive + """ + return self.api_started + async def switch_event(self, entity_id, service): """ Switch event diff --git a/apps/predbat/ha.py b/apps/predbat/ha.py index 9674f9522..fbcb20c12 100644 --- a/apps/predbat/ha.py +++ b/apps/predbat/ha.py @@ -17,10 +17,9 @@ 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 - class RunThread(threading.Thread): def __init__(self, coro): @@ -51,29 +50,43 @@ class HAInterface: Direct interface to Home Assistant """ - def is_running(self): + def is_active(self): + return self.is_alive() + + 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 @@ -81,6 +94,7 @@ def __init__(self, base): self.websocket_task = None self.api_errors = 0 self.stop_thread = False + self.api_started = False self.base = base self.log = base.log @@ -93,9 +107,8 @@ 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: @@ -103,33 +116,29 @@ def __init__(self, base): 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): """ diff --git a/apps/predbat/octopus.py b/apps/predbat/octopus.py index 1ccc967e3..f51cdbb22 100644 --- a/apps/predbat/octopus.py +++ b/apps/predbat/octopus.py @@ -261,6 +261,7 @@ def __init__(self, api_key, log, timeout_in_seconds=20): raise Exception("Octopus API KEY is not set") self.api_key = api_key + self.log = log self.base_url = "https://api.octopus.energy" self.default_headers = {"user-agent": f"{user_agent_value}/1.0"} @@ -310,6 +311,15 @@ def __init__(self, api_key, account_id, base): os.makedirs(self.cache_path) self.cache_file = self.cache_path + "/octopus.yaml" + async def select_event(self, entity_id, value): + pass + + async def number_event(self, entity_id, value): + pass + + async def switch_event(self, entity_id, service): + pass + def wait_api_started(self): """ Wait for the API to start @@ -324,6 +334,9 @@ def wait_api_started(self): return False return True + def is_alive(self): + return self.api_started and self.account_data + async def start(self): """ Main run loop @@ -403,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): @@ -880,6 +893,8 @@ async def async_get_intelligent_device(self, account_id, device_id, completed): chargePointVariants = device_result.get("chargePointVariants", []) electricVehicles = device_result.get("electricVehicles", []) devices = device_result.get("devices", []) + if not devices: + return result for device in device_result["devices"]: deviceType = device.get("deviceType", None) status = device.get("status", {}).get("current", None) @@ -1144,8 +1159,9 @@ def download_octopus_rates(self, url): return pdata # Retry up to 3 minutes + octopus_api_direct = self.components.get_component("octopus") for retry in range(3): - pdata = self.download_octopus_rates_func(url, api=self.octopus_api_direct) + pdata = self.download_octopus_rates_func(url, api=octopus_api_direct) if pdata: break @@ -1177,8 +1193,9 @@ def get_octopus_direct(self, tariff_type, standingCharge=False): """ Get the direct import rates from Octopus """ - if self.octopus_api_direct: - tariff = self.octopus_api_direct.get_tariff(tariff_type) + octopus_api_direct = self.components.get_component("octopus") + if octopus_api_direct: + tariff = octopus_api_direct.get_tariff(tariff_type) if tariff and ("data" in tariff): if standingCharge: tariff_data = tariff["standing"] @@ -1642,7 +1659,8 @@ def fetch_octopus_sessions(self): # Octopus saving session octopus_saving_slots = [] - if self.octopus_api_direct or ("octopus_saving_session" in self.args): + octopus_api_direct = self.components.get_component("octopus") + if octopus_api_direct or ("octopus_saving_session" in self.args): saving_rate = 200 # Default rate if not reported octopoints_per_penny = self.get_arg("octopus_saving_session_octopoints_per_penny", 8) # Default 8 octopoints per found @@ -1650,8 +1668,8 @@ def fetch_octopus_sessions(self): available_events = [] state = False - if self.octopus_api_direct: - available_events, joined_events = self.octopus_api_direct.get_saving_session_data() + if octopus_api_direct: + available_events, joined_events = octopus_api_direct.get_saving_session_data() else: entity_id = self.get_arg("octopus_saving_session", indirect=False) if entity_id: @@ -1673,8 +1691,8 @@ def fetch_octopus_sessions(self): saving_rate = event.get("octopoints_per_kwh", saving_rate * octopoints_per_penny) / octopoints_per_penny # Octopoints per pence if code: # Join the new Octopus saving event and send an alert self.log("Joining Octopus saving event code {} {}-{} at rate {} p/kWh".format(code, start_time.strftime("%a %d/%m %H:%M"), end_time.strftime("%H:%M"), saving_rate)) - if self.octopus_api_direct: - self.octopus_api_direct.join_saving_session_event(code) + if octopus_api_direct: + octopus_api_direct.join_saving_session_event(code) else: self.call_service_wrapper("octopus_energy/join_octoplus_saving_session_event", event_code=code, entity_id=entity_id) self.call_notify("Predbat: Joined Octopus saving event {}-{}, {} p/kWh".format(start_time.strftime("%a %d/%m %H:%M"), end_time.strftime("%H:%M"), saving_rate)) diff --git a/apps/predbat/ohme.py b/apps/predbat/ohme.py index 81aa3ee1b..65b3db7b8 100644 --- a/apps/predbat/ohme.py +++ b/apps/predbat/ohme.py @@ -165,7 +165,7 @@ class ChargerPower: class OhmeAPI: """Ohme API exception.""" - def __init__(self, base, email, password, ohme_automatic_octopus_intelligent=False): + def __init__(self, email, password, ohme_automatic_octopus_intelligent, base): self.email = email self.base = base self.log = base.log @@ -191,6 +191,12 @@ def wait_api_started(self): return False return True + def is_alive(self): + """ + Check if the API is alive + """ + return self.api_started + async def start(self): """ Main run loop diff --git a/apps/predbat/plan.py b/apps/predbat/plan.py index eab22bc07..156cdc9af 100644 --- a/apps/predbat/plan.py +++ b/apps/predbat/plan.py @@ -1069,8 +1069,8 @@ def calculate_plan(self, recompute=True, debug_mode=False, publish=True): self.publish_html_plan(pv_forecast_minute_step, pv_forecast_minute10_step, load_minutes_step, load_minutes_step10, self.end_record) # Web history - if self.web_interface: - self.web_interface.history_update() + if self.components.get_component("web_interface"): + self.components.get_component("web_interface").history_update() # Destroy pool if self.pool: diff --git a/apps/predbat/predbat.py b/apps/predbat/predbat.py index 54582dc20..7845e9942 100644 --- a/apps/predbat/predbat.py +++ b/apps/predbat/predbat.py @@ -39,7 +39,7 @@ THIS_VERSION = "v8.25.0" # fmt: off -PREDBAT_FILES = ["predbat.py", "config.py", "prediction.py", "gecloud.py","utils.py", "inverter.py", "ha.py", "download.py", "unit_test.py", "web.py", "web_helper.py", "predheat.py", "futurerate.py", "octopus.py", "solcast.py","execute.py", "plan.py", "fetch.py", "output.py", "userinterface.py", "energydataservice.py", "alertfeed.py", "compare.py", "db_manager.py", "db_engine.py", "plugin_system.py", "ohme.py"] +PREDBAT_FILES = ["predbat.py", "config.py", "prediction.py", "gecloud.py","utils.py", "inverter.py", "ha.py", "download.py", "unit_test.py", "web.py", "web_helper.py", "predheat.py", "futurerate.py", "octopus.py", "solcast.py","execute.py", "plan.py", "fetch.py", "output.py", "userinterface.py", "energydataservice.py", "alertfeed.py", "compare.py", "db_manager.py", "db_engine.py", "plugin_system.py", "ohme.py", "components.py"] # fmt: on from download import predbat_update_move, predbat_update_download, check_install @@ -70,14 +70,12 @@ ) from prediction import reset_prediction_globals from utils import minutes_since_yesterday, dp1, dp2, dp3 -from ha import HAInterface -from web import WebInterface from predheat import PredHeat -from octopus import Octopus, OctopusAPI +from octopus import Octopus from energydataservice import Energidataservice from solcast import Solcast -from gecloud import GECloud, GECloudDirect -from ohme import OhmeAPI +from gecloud import GECloud +from components import Components from execute import Execute from plan import Plan from fetch import Fetch @@ -348,12 +346,7 @@ def reset(self): self.arg_errors = {} self.ha_interface = None self.fatal_error = False - self.ge_cloud_direct = None - self.ge_cloud_direct_task = None - self.octopus_api_direct = None - self.octopus_api_direct_task = None - self.ohme_api_direct = None - self.ohme_api_direct_task = None + self.components = None self.CONFIG_ITEMS = copy.deepcopy(CONFIG_ITEMS) self.comparison = None self.predheat = None @@ -608,8 +601,6 @@ def reset(self): self.iboost_running_solar = False self.last_service_hash = {} self.count_inverter_writes = {} - self.web_interface = None - self.web_interface_task = None self.rate_slots = [] self.io_adjusted = {} self.low_rates = [] @@ -1286,27 +1277,9 @@ def is_running(self): if not self.ha_interface: return False - if not self.ha_interface.is_running(): - return False - - if self.octopus_api_direct_task: - # Check if the task is still running - if not self.octopus_api_direct_task.is_alive(): - return False - - if self.ohme_api_direct_task: - # Check if the task is still running - if not self.ohme_api_direct_task.is_alive(): - return False - - if self.ge_cloud_direct_task: - # Check if the task is still running - if not self.ge_cloud_direct_task.is_alive(): - return False - if self.web_interface_task: - # Check if the task is still running - if not self.web_interface_task.is_alive(): + if self.components: + if not self.components.is_all_alive(): return False # Read predbat.status @@ -1342,57 +1315,17 @@ 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 - self.web_interface = None - self.web_interface_task = None - self.log("Starting web interface") - self.web_interface = WebInterface(self) + + # 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() - self.web_interface_task = self.create_task(self.web_interface.start()) - - # Octopus API - if self.get_arg("octopus_api_key", "", indirect=False) and self.get_arg("octopus_api_account", "", indirect=False): - self.log("Starting Octopus API interface") - self.octopus_api_direct = OctopusAPI(self.get_arg("octopus_api_key", "", indirect=False), self.get_arg("octopus_api_account", "", indirect=False), self) - self.octopus_api_direct_task = self.create_task(self.octopus_api_direct.start()) - if not self.octopus_api_direct.wait_api_started(): - self.log("Error: Octopus API failed to start") - self.record_status("Error: Octopus API failed to start") - raise ValueError("Octopus API failed to start") - - # GE Cloud Direct API - if self.get_arg("ge_cloud_direct", False): - self.log("Starting GE cloud direct interface") - self.ge_cloud_direct = GECloudDirect(self) - self.ge_cloud_direct_task = self.create_task(self.ge_cloud_direct.start()) - # Allow GE Cloud API to start - if not self.ge_cloud_direct.wait_api_started(): - self.log("Error: GE Cloud API failed to start") - self.record_status("Error: GE Cloud API failed to start") - raise ValueError("GE Cloud API failed to start") - - # Ohme API - if self.get_arg("ohme_login", "", indirect=False) and self.get_arg("ohme_password", "", indirect=False): - self.log("Starting Ohme API interface") - self.ohme_api_direct = OhmeAPI(self, self.get_arg("ohme_login", "", indirect=False), self.get_arg("ohme_password", "", indirect=False), self.get_arg("ohme_automatic_octopus_intelligent", False)) - self.ohme_api_direct_task = self.create_task(self.ohme_api_direct.start()) - if not self.ohme_api_direct.wait_api_started(): - self.log("Error: Ohme API failed to start") - self.record_status("Error: Ohme API failed to start") - raise ValueError("Ohme API failed to start") - else: - self.log("Ohme EV charger not configured") - # Printable config root self.config_root_p = self.config_root slug = self.ha_interface.get_slug() @@ -1453,16 +1386,8 @@ async def terminate(self): """ self.log("Predbat terminating") self.stop_thread = True - if self.web_interface: - await self.web_interface.stop() - if self.ge_cloud_direct: - await self.ge_cloud_direct.stop() - if self.ohme_api_direct: - await self.ohme_api_direct.stop() - if self.octopus_api_direct: - self.octopus_api_direct.stop() - if self.ha_interface: - self.ha_interface.stop() + if self.components: + await self.components.stop() await asyncio.sleep(0) if hasattr(self, "pool"): diff --git a/apps/predbat/userinterface.py b/apps/predbat/userinterface.py index 90d3048ca..0ebb92ac8 100644 --- a/apps/predbat/userinterface.py +++ b/apps/predbat/userinterface.py @@ -25,8 +25,7 @@ DEBUG_EXCLUDE_LIST = [ "pool", "ha_interface", - "web_interface", - "web_interface_task", + "components", "prediction", "logfile", "predheat", @@ -38,12 +37,6 @@ "CONFIG_ITEMS", "config_index", "comparison", - "octopus_api_direct", - "octopus_api_direct_task", - "ge_cloud_direct", - "ge_cloud_direct_task", - "ohme_api_direct", - "ohme_api_direct_task", "plugin_system", ] @@ -274,12 +267,7 @@ async def select_event(self, event, data, kwargs): entities = [entities] for entity_id in entities: - if "predbat_gecloud_" in entity_id: - if self.ge_cloud_direct: - await self.ge_cloud_direct.select_event(entity_id, value) - if "predbat_ohme" in entity_id: - if self.ohme_api_direct: - await self.ohme_api_direct.select_event(entity_id, value) + await self.components.select_event(entity_id, value) for item in self.CONFIG_ITEMS: if ("entity" in item) and (item["entity"] in entities): @@ -332,12 +320,7 @@ async def number_event(self, event, data, kwargs): entities = [entities] for entity_id in entities: - if "predbat_gecloud_" in entity_id: - if self.ge_cloud_direct: - await self.ge_cloud_direct.number_event(entity_id, value) - if "predbat_ohme" in entity_id: - if self.ohme_api_direct: - await self.ohme_api_direct.number_event(entity_id, value) + await self.components.number_event(entity_id, value) for item in self.CONFIG_ITEMS: if ("entity" in item) and (item["entity"] in entities): @@ -381,12 +364,7 @@ async def switch_event(self, event, data, kwargs): entities = [entities] for entity_id in entities: - if "predbat_gecloud_" in entity_id: - if self.ge_cloud_direct: - await self.ge_cloud_direct.switch_event(entity_id, service) - if "predbat_ohme" in entity_id: - if self.ohme_api_direct: - await self.ohme_api_direct.switch_event(entity_id, service) + await self.components.switch_event(entity_id, service) for item in self.CONFIG_ITEMS: if ("entity" in item) and (item["entity"] in entities): diff --git a/apps/predbat/web.py b/apps/predbat/web.py index 372a44fb4..10af5da09 100644 --- a/apps/predbat/web.py +++ b/apps/predbat/web.py @@ -18,7 +18,8 @@ import json import shutil import html as html_module -from web_helper import get_header_html, get_plan_css, get_editor_js, get_editor_css, get_log_css, get_charts_css, get_apps_css, get_html_config_css, get_apps_js +import time +from web_helper import get_header_html, get_plan_css, get_editor_js, get_editor_css, get_log_css, get_charts_css, get_apps_css, get_html_config_css, get_apps_js, get_components_css from utils import calc_percent_limit, str2time, dp0, dp2 from config import TIME_FORMAT, TIME_FORMAT_DAILY @@ -30,7 +31,7 @@ class WebInterface: - def __init__(self, base) -> None: + def __init__(self, web_port, base) -> None: self.abort = False self.base = base self.log = base.log @@ -42,12 +43,22 @@ def __init__(self, base) -> None: self.cost_yesterday_hist = {} self.cost_yesterday_car_hist = {} self.cost_yesterday_no_car = {} - self.web_port = self.base.get_arg("web_port", 5052) + self.web_port = web_port self.default_log = "warnings" + self.api_started = False # Plugin registration system self.registered_endpoints = [] + async def select_event(self, entity_id, value): + pass + + async def number_event(self, entity_id, value): + pass + + async def switch_event(self, entity_id, service): + pass + def register_endpoint(self, path, handler, method="GET"): """ Register a new endpoint with the web interface @@ -114,6 +125,7 @@ async def start(self): app.router.add_post("/config", self.html_config_post) app.router.add_get("/dash", self.html_dash) app.router.add_post("/dash", self.html_dash_post) + app.router.add_get("/components", self.html_components) app.router.add_get("/debug_yaml", self.html_debug_yaml) app.router.add_get("/debug_log", self.html_debug_log) app.router.add_get("/debug_apps", self.html_debug_apps) @@ -149,9 +161,11 @@ async def start(self): site = web.TCPSite(runner, "0.0.0.0", self.web_port) await site.start() print("Web interface started") + self.api_started = True while not self.abort: await asyncio.sleep(1) await runner.cleanup() + self.api_started = False print("Web interface stopped") async def stop(self): @@ -159,6 +173,23 @@ async def stop(self): self.abort = True await asyncio.sleep(1) + def wait_api_started(self): + """ + Wait for the API to start + """ + self.log("Web: 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: Web: Failed to start") + return False + return True + + def is_alive(self): + return self.api_started + def get_attributes_html(self, entity, from_db=False): """ Return the attributes of an entity as HTML @@ -441,7 +472,7 @@ def get_status_html(self, status, version): if status and (("Warn:" in status) or ("Error:" in status)): text += "Status{}\n".format(status) elif not is_running: - text += "Predbat has errors\n" + text += "{} (with errors)\n".format(status) else: text += "Status{}\n".format(status) text += "Last Updated{}\n".format(last_updated) @@ -2546,6 +2577,118 @@ async def html_plan_override(self, request): self.log(f"ERROR: Failed to process plan override: {str(e)}") return web.json_response({"success": False, "message": str(e)}, status=500) + def count_entities_matching_filter(self, event_filter): + """ + Count the number of entities that match the given event filter pattern + """ + count = 0 + try: + # Get all entities from Home Assistant + all_entities = self.base.get_state_wrapper() + if all_entities: + for entity_id in all_entities.keys(): + if event_filter in entity_id: + count += 1 + except Exception as e: + self.log(f"Error counting entities for filter '{event_filter}': {e}") + return count + + async def html_components(self, request): + """ + Return the Components view as an HTML page showing status of all components + """ + self.default_page = "./components" + text = self.get_header("Predbat Components", refresh=60) + text += "\n" + text += get_components_css() + + text += "

Component Status

\n" + text += "
\n" + + # Get all component information + all_components = self.base.components.get_all() + active_components = self.base.components.get_active() + + for component_name in all_components: + from components import COMPONENT_LIST + + component_info = COMPONENT_LIST.get(component_name, {}) + component = self.base.components.get_component(component_name) + is_alive = self.base.components.is_alive(component_name) + is_active = component_name in active_components + + # Create component card + text += f'
\n' + text += f'
\n' + text += f'

{component_info.get("name", component_name)}

\n' + + # Status indicator + if is_active and is_alive: + text += '●Active\n' + elif is_active and not is_alive: + text += '●Error\n' + else: + text += '●Disabled\n' + + text += f"
\n" + + # Component details + text += f'
\n' + text += f"

Component: {component_name}

\n" + + # Show args and their current values + args_info = component_info.get("args", {}) + if args_info: + text += f'
\n' + text += f"

Configuration:

\n" + text += f'\n' + text += f"\n" + + for arg_name, arg_info in args_info.items(): + required = arg_info.get("required", False) + config_key = arg_info.get("config", "") + default = arg_info.get("default", "") + + # Get current value + current_value = self.base.get_arg(config_key, default, indirect=False) + + # Hide sensitive values + display_value = current_value + if any(sensitive in arg_name.lower() for sensitive in ["password", "key", "secret", "token"]): + if current_value: + display_value = "***configured***" + else: + display_value = "***not set***" + elif current_value is None: + display_value = "Not set" + elif current_value == "": + display_value = "Empty" + + required_text = "Yes" if required else "No" + text += f'\n' + text += f"\n" + text += f"\n" + text += f"\n" + text += f"\n" + + text += f"
SettingRequiredCurrent Value
{config_key}{required_text}{display_value}
\n" + text += f"
\n" + + # Event filter info - count matching entities + event_filter = component_info.get("event_filter", "") + if event_filter: + # Count entities that match the filter + entity_count = self.count_entities_matching_filter(event_filter) + count_class = "entity-count-zero" if entity_count == 0 else "entity-count-positive" + text += f'

Entities: num_entities: {entity_count}

\n' + + text += f"
\n" + text += f"
\n" + + text += "
\n" + text += "\n" + return web.Response(content_type="text/html", text=text) + async def html_restart(self, request): """ Handle restart request by setting fatal_error to trigger restart diff --git a/apps/predbat/web_helper.py b/apps/predbat/web_helper.py index 331dc974b..9cb779889 100644 --- a/apps/predbat/web_helper.py +++ b/apps/predbat/web_helper.py @@ -1451,6 +1451,216 @@ def get_apps_css(): return text +def get_components_css(): + """ + Return CSS for components page + """ + text = """ + +""" + return text + + def get_charts_css(): text = """