ha.py

# -----------------------------------------------------------------------------
# Predbat Home Battery System
# Copyright Trefor Southwell 2026 - 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


"""Home Assistant communication interface.

Provides the HAInterface class for direct WebSocket and REST API communication
with Home Assistant. Manages state caching, entity synchronisation, service
calls, history retrieval, and optional SQLite database mirroring. Also includes
HAHistory for automatic entity history tracking and pruning.
"""

import os
from datetime import timedelta, datetime, timezone
import asyncio
from aiohttp import ClientSession, WSMsgType
import array
import bisect
import json
import requests
import traceback
import threading
import time
from utils import str2time
from const import TIME_FORMAT_HA, TIMEOUT, TIME_FORMAT_HA_TZ
from component_base import ComponentBase

# Maximum days of history fetched per request. Long windows are split so only one chunk's
# response body, decoded string and parsed objects are resident at a time.
HISTORY_CHUNK_DAYS = 3


class RunThread(threading.Thread):
    def __init__(self, coro):
        self.coro = coro
        self.result = None
        super().__init__()

    def run(self):
        self.result = asyncio.run(self.coro)


def run_async(coro):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None
    if loop and loop.is_running():
        thread = RunThread(coro)
        thread.start()
        thread.join()
        return thread.result
    else:
        return asyncio.run(coro)


class EntityHistory:
    """
    Compact columnar storage for a single entity's history.

    Stores timestamps as epoch seconds and numeric states in typed arrays (~16 bytes
    per record) instead of keeping the decoded JSON record dicts (~300 bytes per
    record). Non-numeric states and per-record attribute variations are kept in small
    exception maps keyed by epoch time. Records are materialised back into the
    standard list-of-dicts form on demand by to_records(). Records must be appended
    oldest to newest so the epoch array stays sorted.
    """

    __slots__ = ["epochs", "states", "state_exceptions", "shared_attributes", "attribute_exceptions"]

    def __init__(self):
        """
        Create an empty history store
        """
        self.epochs = array.array("d")
        self.states = array.array("d")
        self.state_exceptions = {}
        self.shared_attributes = None
        self.attribute_exceptions = {}

    def __len__(self):
        """
        Return the number of records stored
        """
        return len(self.epochs)

    def __getitem__(self, index):
        """
        Materialise and return the record dict at the given integer index
        """
        epoch = self.epochs[index]
        if self.state_exceptions and epoch in self.state_exceptions:
            state = self.state_exceptions[epoch]
        else:
            state = self.states[index]
        if self.attribute_exceptions and epoch in self.attribute_exceptions:
            attributes = dict(self.attribute_exceptions[epoch])
        else:
            attributes = dict(self.shared_attributes) if self.shared_attributes else {}
        return {"state": state, "last_updated": datetime.fromtimestamp(epoch, timezone.utc).isoformat(), "attributes": attributes}

    def __iter__(self):
        """
        Iterate over the stored records, materialising each as a record dict
        """
        for index in range(len(self.epochs)):
            yield self[index]

    def append_record(self, record, epoch=None):
        """
        Append a single history record, which must be newer than all stored records.

        Returns True if the record was stored, False if its last_updated could not be parsed.
        """
        if epoch is None:
            last_updated = record.get("last_updated", None)
            if not last_updated:
                return False
            try:
                epoch = str2time(last_updated).timestamp()
            except (ValueError, TypeError):
                return False

        state = record.get("state", None)
        try:
            state_value = float(state)
        except (ValueError, TypeError):
            state_value = None
        if state_value is None or state_value != state_value:
            # Keep non-numeric states (including NaN, which would break equality checks) as-is
            self.state_exceptions[epoch] = state
            state_value = 0.0
        self.epochs.append(epoch)
        self.states.append(state_value)

        attributes = record.get("attributes", None) or {}
        if self.shared_attributes is None:
            self.shared_attributes = dict(attributes)
        elif attributes != self.shared_attributes:
            self.attribute_exceptions[epoch] = dict(attributes)
        return True

    @classmethod
    def from_records(cls, records):
        """
        Build a store from a list of history record dicts ordered oldest to newest
        """
        store = cls()
        for record in records:
            store.append_record(record)
        return store

    def to_records(self):
        """
        Materialise the store back into the standard list of history record dicts.

        Each returned record dict, and its "state"/"last_updated" values, are freshly
        created and safe for callers to reassign. The "attributes" dict is NOT: records
        that shared identical attributes when stored share one attributes dict object in
        the output, so mutating it in place (rather than replacing record["attributes"]
        wholesale) will affect every other record sharing it. Callers that need to edit
        attributes in place must dict.copy() first.
        """
        shared_attributes = dict(self.shared_attributes) if self.shared_attributes else {}
        state_exceptions = self.state_exceptions
        attribute_exceptions = self.attribute_exceptions
        states = self.states
        utc = timezone.utc
        records = []
        for index, epoch in enumerate(self.epochs):
            if state_exceptions and epoch in state_exceptions:
                state = state_exceptions[epoch]
            else:
                state = states[index]
            if attribute_exceptions and epoch in attribute_exceptions:
                attributes = dict(attribute_exceptions[epoch])
            else:
                attributes = shared_attributes
            records.append({"state": state, "last_updated": datetime.fromtimestamp(epoch, utc).isoformat(), "attributes": attributes})
        return records

    def prune(self, cutoff_epoch):
        """
        Drop all records older than the given cutoff epoch time
        """
        index = bisect.bisect_left(self.epochs, cutoff_epoch)
        if index > 0:
            self.epochs = self.epochs[index:]
            self.states = self.states[index:]
            if self.state_exceptions:
                self.state_exceptions = {epoch: value for epoch, value in self.state_exceptions.items() if epoch >= cutoff_epoch}
            if self.attribute_exceptions:
                self.attribute_exceptions = {epoch: value for epoch, value in self.attribute_exceptions.items() if epoch >= cutoff_epoch}


class HAHistory(ComponentBase):
    """
    Home Assistant History Data
    """

    def initialize(self):
        self.history_entities = {}
        self.history_data = {}
        self.history_lock = threading.Lock()

    def add_entity(self, entity_id, days):
        """
        Add an entity to the history list
        """
        if entity_id in self.history_entities:
            if days > self.history_entities[entity_id]:
                self.history_entities[entity_id] = days
        else:
            self.history_entities[entity_id] = days

    def get_history(self, entity_id, days=30, tracked=True):
        """
        Get history data for an entity
        """
        result = None

        with self.history_lock:
            store = self.history_data.get(entity_id, None)
            if store and self.history_entities.get(entity_id, 0) >= days:
                # Materialise fresh record dicts from the compact store (replaces the previous
                # deepcopy) - callers can reassign record["state"]/["attributes"] without
                # corrupting the cache, but must not mutate a returned attributes dict in place
                # (see EntityHistory.to_records docstring: it may be shared across records)
                result = [store.to_records()]

        if result is None:
            ha_interface = self.base.components.get_component("ha")
            if not ha_interface:
                self.log("Error: HAHistory: No HAInterface available, cannot fetch history")
                return None
            if tracked:
                self.add_entity(entity_id, days)
                days = self.history_entities[entity_id]
            history_data = ha_interface.get_history(entity_id, datetime.now(self.local_tz), days=days)
            if history_data and len(history_data) > 0:
                history_data = history_data[0]
                if tracked:
                    self.update_entity(entity_id, history_data)
                # The store keeps its own compact copy of the data, so the fetched records
                # can be handed to the caller directly
                result = [history_data]

        return result

    def prune_history(self, now):
        """
        Prune history data older than required
        """
        with self.history_lock:
            for entity_id in list(self.history_data.keys()):
                max_days = self.history_entities.get(entity_id, 30)
                cutoff_epoch = (now - timedelta(days=max_days)).timestamp()
                self.history_data[entity_id].prune(cutoff_epoch)

    def update_entity(self, entity_id, new_history_data):
        """
        Update history data for an entity
        """
        if not new_history_data:
            return

        FILTER_ATTRIBUTES = ["friendly_name", "unit_of_measurement", "icon", "device_class", "results", "state_class", "entity_id"]
        FILTER_ENTRIES = ["last_changed", "entity_id"]

        # Filter useless data from history data
        for entry in new_history_data:
            if entry.get("attributes"):
                for attr in FILTER_ATTRIBUTES:
                    entry["attributes"].pop(attr, None)
            for entry_attr in FILTER_ENTRIES:
                entry.pop(entry_attr, None)
            # Normalise numeric states to float here (mirroring EntityHistory.append_record) so a
            # tracked entity's first (uncached) fetch returns the same state type as later cache hits
            try:
                state_value = float(entry["state"])
            except (ValueError, TypeError, KeyError):
                state_value = None
            if state_value is not None and state_value == state_value:
                entry["state"] = state_value

        with self.history_lock:
            store = self.history_data.get(entity_id, None)
            if store and len(store) > 0:
                # Scan new data (always in order from oldest to newest), appending entries newer
                # than the stored data and collecting any entries older than the stored data
                first_epoch = store.epochs[0]
                last_epoch = store.epochs[-1]
                older_entries = []
                for entry in new_history_data:
                    this_updated = entry.get("last_updated", None)
                    if not this_updated:
                        continue
                    try:
                        epoch = str2time(this_updated).timestamp()
                    except (ValueError, TypeError):
                        continue
                    if epoch > last_epoch:
                        store.append_record(entry, epoch=epoch)
                        last_epoch = epoch
                    elif epoch < first_epoch:
                        older_entries.append(entry)

                if older_entries:
                    # Rebuild the store with the older entries merged in (rare, only happens
                    # when more days of history are requested than previously cached)
                    combined = older_entries + store.to_records()
                    combined.sort(key=lambda x: str2time(x.get("last_updated")))
                    self.history_data[entity_id] = EntityHistory.from_records(combined)
            else:
                self.history_data[entity_id] = EntityHistory.from_records(new_history_data)

        # Update last success timestamp
        self.update_success_timestamp()

    async def run(self, seconds, first):
        if first:
            self.log("Info: Starting HAHistory")

        ha_interface = self.base.components.get_component("ha")
        if not ha_interface:
            self.log("Error: HAHistory: No HAInterface available, cannot start history updates")
            return False

        if first or (seconds % (2 * 60) == 0):
            # Update history data every 2 minutes
            now = datetime.now(self.local_tz)
            for entity_id in list(self.history_entities.keys()):
                # self.log("HAHistory: Updating history for {}".format(entity_id))
                store = self.history_data.get(entity_id, None)
                last_updated_time = datetime.fromtimestamp(store.epochs[-1], timezone.utc) if store and len(store) > 0 else None
                if last_updated_time:
                    history_data = ha_interface.get_history(entity_id, now, days=1, from_time=last_updated_time)
                    if history_data and len(history_data) > 0:
                        history_data = history_data[0]
                        self.update_entity(entity_id, history_data)
                else:
                    history_data = ha_interface.get_history(entity_id, now, days=self.history_entities[entity_id])
                    if history_data and len(history_data) > 0:
                        history_data = history_data[0]
                        self.update_entity(entity_id, history_data)

        if first or (seconds % (60 * 60) == 0):
            # Prune history data every hour
            self.log("Info: HAHistory: Pruning history data started.")
            self.prune_history(datetime.now(self.local_tz))
            self.log("Info: HAHistory: Pruning history data completed.")
        return True


class HAInterface(ComponentBase):
    """
    Direct interface to Home Assistant
    """

    def is_alive(self):
        if not self.api_started:
            return False
        if self.ha_key and not self.websocket_active:
            return False
        return True

    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 initialize(self, ha_url, ha_key, db_enable, db_mirror_ha, db_primary):
        """
        Initialise the interface to Home Assistant.
        """
        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_mirror_list = {}
        self.db = None
        self.db_cursor = None
        self.websocket_active = False
        self.api_errors = 0
        self.last_success_timestamp = None

        self.state_data = {}
        self.slug = None

        # Websocket command queue infrastructure
        self.ws_command_queue = []
        self.ws_pending_requests = {}
        self.ws_pending_lock = threading.Lock()
        self.ws_sync_event = threading.Event()
        self.ws_async_event = None  # Created in async context
        self.ws_event_loop = None
        self.ws_bridge_thread = None

        if not self.ha_key:
            if not (self.db_enable and self.db_primary):
                self.log("Error: ha_key or SUPERVISOR_TOKEN not found, you must set ha_url/ha_key in apps.yaml")
                raise ValueError
            else:
                self.log("Info: Using SQL Lite database as primary data source, no HA interface available")

        if self.ha_key:
            # Get the current app info, but suppress warning message if the API call fails as non-HAOS installs won't have supervisor running
            #
            # HA changed terminology from 'addons' to 'apps' in HA 2026.2 but retained the old service calls for transition
            #
            # At present have not changed Predbat API call in order to not break installations that are still using an older HA supervisor
            # Propose in Feb 2027 that Predbat be changed to use the new service call
            res = self.api_call("/addons/self/info", core=False, silent=True)
            if res:
                # get app slug name which is the actual directory name under /app_configs that /config is mounted to
                self.slug = res["data"]["slug"]
                self.log("Info: App slug is {}".format(self.slug))

            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

        if self.db_enable:
            # Open the SQL Lite database called predbat.db using the DatabaseManager
            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

    async def start(self):
        if self.ha_key:
            self.log("Info: Starting HA interface")
            self.websocket_active = True

            # Create async event and start bridge thread
            self.ws_async_event = asyncio.Event()
            self.ws_event_loop = asyncio.get_event_loop()
            self.ws_bridge_thread = threading.Thread(target=self.bridge_event, args=(self.ws_event_loop,), daemon=True)
            self.ws_bridge_thread.start()

            await self.socketLoop()
        else:
            self.log("Info: Starting Dummy HA interface")
            seconds = 0
            self.api_started = True
            while not self.api_stop:
                if seconds % 60 == 0:
                    self.update_success_timestamp()
                await asyncio.sleep(5)
                seconds += 5
        self.api_started = False
        self.log("Info: HA interface stopped")

    def get_slug(self):
        """
        Get the app slug.
        """
        return self.slug

    def bridge_event(self, loop):
        """
        Bridge sync event to async event loop (runs in separate thread)
        """
        while not self.api_stop:
            self.ws_sync_event.wait(timeout=1.0)
            self.ws_sync_event.clear()
            if self.ws_event_loop and self.ws_async_event and not loop.is_closed():
                loop.call_soon_threadsafe(self.ws_async_event.set)

    def call_service_websocket_command(self, domain, service, data):
        """
        Call a service via the web socket interface
        """
        return run_async(self.async_call_service_websocket_command(domain, service, data))

    async def async_call_service_websocket_command(self, domain, service, service_data):
        """
        Call a service via the web socket interface (queued via socketLoop)
        """
        return_response = service_data.get("return_response", False)
        if "return_response" in service_data:
            del service_data["return_response"]

        # Create event and result holder for this request
        event = threading.Event()
        result_holder = {"response": None, "success": None, "error": None, "ha_error": None}

        # Add to command queue
        with self.ws_pending_lock:
            self.ws_command_queue.append((domain, service, service_data, return_response, event, result_holder))

        # Signal bridge thread to wake socketLoop
        self.ws_sync_event.set()

        # Wait for response with 2 minute timeout
        event.wait(timeout=2 * 60)

        # Extract result
        if result_holder.get("error"):
            self.log("Warn: Service call {}/{} failed: {}".format(domain, service, result_holder["error"]))
            return None

        # Check for timeout (neither success nor error was set). This is indistinguishable here from
        # a call that actually succeeded but whose response arrived (or was processed) just after the
        # 2 minute deadline - callers that treat a falsy return as "try a different service name"
        # (e.g. octopus.py's join fallback) can in principle be tricked into a harmless-but-redundant
        # duplicate call by this specific edge case. Not fully resolved - the timeout window is long
        # enough that this should be rare in practice, and the alternative (a real explicit failure)
        # is by far the more common falsy case.
        if result_holder.get("success") is None and not result_holder.get("error"):
            self.log("Warn: Service call {}/{} failed or timed out: result {}".format(domain, service, result_holder))
            return None

        success = result_holder.get("success", False)
        if not success:
            self.log("Warn: Service call {}/{} data {} failed: {}".format(domain, service, service_data, result_holder.get("ha_error")))
            return None

        # Return response data if requested
        if return_response:
            return result_holder.get("response")

        return True

    async def socketLoop(self):
        """
        Web socket loop for HA interface
        """
        error_count = 0

        while True:
            if self.api_stop or self.fatal_error:
                self.log("Info: Web socket stopping")
                break
            if error_count >= 10:
                self.log("Error: Web socket failed 10 times, stopping")
                self.fatal_error_occurred()
                break

            url = "{}/api/websocket".format(self.ha_url)
            self.log("Info: Start socket for url {}".format(url))
            async with ClientSession() as session:
                try:
                    async with session.ws_connect(url) as websocket:
                        await websocket.send_json({"type": "auth", "access_token": self.ha_key})
                        sid = 1

                        # Connected okay, reset error count
                        error_count = 0

                        # Subscribe to all state changes
                        await websocket.send_json({"id": sid, "type": "subscribe_events", "event_type": "state_changed"})
                        sid += 1

                        # Listen for services
                        await websocket.send_json({"id": sid, "type": "subscribe_events", "event_type": "call_service"})
                        sid += 1

                        # Fire events to say we have registered services, this will be repeated later after startup
                        # To make sure we don't miss any events due to them not being active yet.
                        for item in self.base.SERVICE_REGISTER_LIST:
                            await websocket.send_json({"id": sid, "type": "fire_event", "event_type": "service_registered", "event_data": {"service": item["service"], "domain": item["domain"]}})
                            sid += 1

                        self.log("Info: Web Socket active")
                        self.base.update_pending = True  # Force an update when web-socket reconnects
                        self.api_started = True

                        # Timeout tracking
                        last_timeout_check = time.time()

                        while not self.api_stop and not self.fatal_error:
                            # Check for incoming messages with timeout so we can process commands even if no messages arrive
                            try:
                                message = await asyncio.wait_for(websocket.receive(), timeout=0.1)
                            except asyncio.TimeoutError:
                                # No message received, but we can still process commands
                                message = None

                            if message and message.type == WSMsgType.TEXT:
                                try:
                                    data = json.loads(message.data)
                                    if data:
                                        message_type = data.get("type", "")
                                        if message_type == "event":
                                            event_info = data.get("event", {})
                                            event_type = event_info.get("event_type", "")
                                            if event_type == "state_changed":
                                                event_data = event_info.get("data", {})
                                                old_state = event_data.get("old_state", {})
                                                new_state = event_data.get("new_state", {})
                                                if new_state:
                                                    entity_id = new_state.get("entity_id", None)
                                                    if entity_id:
                                                        self.update_state_item(new_state, entity_id)
                                                        # Only trigger on value change or you get too many updates
                                                        if not old_state or (new_state.get("state", None) != old_state.get("state", None)):
                                                            await self.base.trigger_watch_list(entity_id, event_data.get("attribute", None), event_data.get("old_state", None), new_state)
                                                        error_count = 0  # Reset error count on successful result
                                                        self.update_success_timestamp()
                                                    else:
                                                        self.log("Warn: Web Socket state_changed event has no entity_id {}".format(new_state))
                                                        error_count += 1
                                            elif event_type == "call_service":
                                                service_data = event_info.get("data", {})
                                                await self.base.trigger_callback(service_data)
                                                error_count = 0  # Reset error count on successful result
                                                self.update_success_timestamp()
                                            else:
                                                self.log("Info: Web Socket unknown message {}".format(data))
                                        elif message_type == "result":
                                            # Route result to waiting thread
                                            result_id = data.get("id")
                                            if result_id:
                                                with self.ws_pending_lock:
                                                    if result_id in self.ws_pending_requests:
                                                        request_info = self.ws_pending_requests.pop(result_id)
                                                        result_holder = request_info["result_holder"]
                                                        # #3460: .get()'s default only covers a *missing* key - some
                                                        # services (observed for notify.notify) return a "result"
                                                        # message with "success" explicitly present but null, which
                                                        # .get("success", False) passes through as None rather than
                                                        # False. That left success/error both None, indistinguishable
                                                        # from a genuine 2-minute timeout to the caller and producing
                                                        # a misleading "failed or timed out" warning immediately.
                                                        result_holder["success"] = bool(data.get("success"))
                                                        result_holder["response"] = data.get("result", {}).get("response", None)
                                                        result_holder["error"] = None
                                                        # HA's own reported reason when success is False (e.g. {"code": "not_found", "message": "..."})
                                                        # - distinct from "error" above, which is reserved for a local send/transport failure.
                                                        result_holder["ha_error"] = data.get("error")
                                                        request_info["event"].set()

                                            success = data.get("success", False)
                                            if not success:
                                                self.log("Warn: Web Socket result failed {}".format(data))
                                                error_count += 1
                                            else:
                                                self.update_success_timestamp()
                                                error_count = 0  # Reset error count on successful result
                                        elif message_type == "auth_required":
                                            pass
                                        elif message_type == "auth_ok":
                                            pass
                                        elif message_type == "auth_invalid":
                                            self.log("Warn: Web Socket auth failed, check your ha_key setting")
                                            self.websocket_active = False
                                            error_count += 1
                                            raise Exception("Web Socket auth failed")
                                        else:
                                            self.log("Info: Web Socket unknown message {}".format(data))

                                except Exception as e:
                                    self.log("Error: Web Socket exception in update loop: {}".format(e))
                                    self.log("Error: " + traceback.format_exc())
                                    error_count += 1
                                    break

                            elif message and message.type == WSMsgType.CLOSED:
                                error_count += 1
                                break
                            elif message and message.type == WSMsgType.ERROR:
                                error_count += 1
                                break

                            # Process queued commands (runs even if no message received)
                            while True:
                                command = None
                                with self.ws_pending_lock:
                                    if self.ws_command_queue:
                                        command = self.ws_command_queue.pop(0)

                                if not command:
                                    break

                                domain, service, service_data, return_response, event, result_holder = command

                                # Send command with current sid
                                try:
                                    if domain == "fire_event":
                                        await websocket.send_json({"id": sid, "type": domain, "event_type": service, "event_data": {"service": service_data["event_service"], "domain": service_data["event_domain"]}})
                                    else:
                                        # HA's call_service command expects 'target' (entity_id/device_id/area_id
                                        # addressing) as a sibling of service_data, not nested inside it - a
                                        # nested target is rejected with invalid_format: extra keys not allowed
                                        # @ data['target'] (#4662). Callers commonly configure services with a
                                        # target: entity_id: ... block (the standard HA action syntax), which
                                        # lands as a "target" key inside service_data, so it must be pulled out
                                        # here. Pop from a copy, not service_data itself - the original dict is
                                        # the same object async_call_service_websocket_command() logs on failure
                                        # ("Warn: Service call ... data ... failed"), so mutating it in place
                                        # would silently drop target from that diagnostic.
                                        outgoing_data = dict(service_data) if isinstance(service_data, dict) else service_data
                                        target = outgoing_data.pop("target", None) if isinstance(outgoing_data, dict) else None
                                        call_frame = {"id": sid, "type": "call_service", "domain": domain, "service": service, "service_data": outgoing_data, "return_response": return_response}
                                        if target:
                                            call_frame["target"] = target
                                        await websocket.send_json(call_frame)

                                    # Track pending request (only if send succeeded)
                                    with self.ws_pending_lock:
                                        self.ws_pending_requests[sid] = {"event": event, "result_holder": result_holder, "timestamp": time.time()}

                                    sid += 1
                                    self.update_success_timestamp()
                                    error_count = 0  # Reset error count on successful result
                                except Exception as e:
                                    # Failed to send - notify caller immediately
                                    self.log("Warn: Failed to send service call {}/{}: exception: {}".format(domain, service, e))
                                    error_count += 1
                                    result_holder["error"] = "send_failed: {}".format(e)
                                    result_holder["success"] = False
                                    event.set()

                            # Check for command queue updates via async event (non-blocking)
                            if self.ws_async_event and self.ws_async_event.is_set():
                                self.ws_async_event.clear()

                            # Periodic timeout cleanup (every 10 seconds), check for requests older than 2 minutes
                            current_time = time.time()
                            if current_time - last_timeout_check > 10.0:
                                last_timeout_check = current_time
                                with self.ws_pending_lock:
                                    timed_out = []
                                    for req_id, req_info in list(self.ws_pending_requests.items()):
                                        if current_time - req_info["timestamp"] > 2 * 60.0:
                                            timed_out.append(req_id)

                                    for req_id in timed_out:
                                        req_info = self.ws_pending_requests.pop(req_id)
                                        req_info["result_holder"]["error"] = "timeout"
                                        req_info["result_holder"]["success"] = False
                                        req_info["event"].set()
                                        self.log("Warn: Service call timeout for request id {}".format(req_id))
                                        error_count += 1

                except Exception as e:
                    self.log("Error: Web Socket exception in startup: {}".format(e))
                    self.log("Error: " + traceback.format_exc())
                    error_count += 1

                # Fail all pending requests on connection drop
                with self.ws_pending_lock:
                    for _req_id, req_info in list(self.ws_pending_requests.items()):
                        req_info["result_holder"]["error"] = "connection_lost"
                        req_info["result_holder"]["success"] = False
                        req_info["event"].set()
                    self.ws_pending_requests.clear()
                    if self.ws_command_queue:
                        self.log("Warn: {} queued commands dropped due to connection loss".format(len(self.ws_command_queue)))
                        for dropped_cmd in self.ws_command_queue:
                            dropped_domain, dropped_service, dropped_data, dropped_rr, dropped_event, dropped_result = dropped_cmd
                            dropped_result["error"] = "connection_lost"
                            dropped_result["success"] = False
                            dropped_event.set()
                        self.ws_command_queue.clear()

            if not self.api_stop:
                self.log("Warn: Web Socket closed, will try to reconnect in 5 seconds - error count {}".format(error_count))
                await asyncio.sleep(5)

        self.api_started = False
        if not self.api_stop:
            self.log("Error: Web Socket failed to reconnect, stopping....")
            self.websocket_active = False
            self.fatal_error_occurred()

        self.log("Info: Web Socket stopped")

    def get_state(self, entity_id=None, default=None, attribute=None, refresh=False, raw=False):
        """
        Get state from cached HA data
        """
        if entity_id:
            if isinstance(entity_id, list):
                self.log("Error: get_state called with list entity_id: {}, this should be a single entity string".format(entity_id))
                return default
            self.db_mirror_list[entity_id.lower()] = True

        if not entity_id:
            return self.state_data
        elif entity_id.lower() in self.state_data:
            if refresh and not self.ha_key:
                # Only refresh from DB/HA if we are not the primary data source
                self.update_state(entity_id)
            state_info = self.state_data[entity_id.lower()]
            if raw:
                return state_info
            elif attribute:
                if attribute in state_info["attributes"]:
                    return state_info["attributes"][attribute]
                else:
                    return default
            else:
                return state_info["state"]
        else:
            return default

    def update_state_db(self, entity_id):
        """
        Update state for entity_id from the SQLLite database
        """
        self.db_mirror_list[entity_id.lower()] = True
        if not self.db_enable:
            return
        item = self.db_manager.get_state_db(entity_id)
        if item:
            self.update_state_item(item, entity_id, nodb=True)

    def update_state(self, entity_id):
        """
        Update state for entity_id
        """
        self.db_mirror_list[entity_id.lower()] = True

        if self.db_primary and self.db_enable and not self.ha_key:
            self.update_state_db(entity_id)
            return

        if not self.ha_key:
            return

        item = self.api_call("/api/states/{}".format(entity_id))
        if item:
            self.update_state_item(item, entity_id)

    def update_state_item(self, item, entity_id, nodb=False):
        """
        Update state table for item
        """
        entity_id = entity_id.lower()
        attributes = item.get("attributes", {})
        last_changed = item.get("last_changed", item.get("last_updated", None))
        if "state" in item:
            state = item["state"]
            # Preserve observation freshness separately from a state transition.
            # An unchanged SOC may have a fresh device timestamp/attribute update.
            self.state_data[entity_id] = {"state": state, "attributes": attributes, "last_changed": last_changed, "last_updated": item.get("last_updated", last_changed)}
            if not nodb and self.db_enable and ((self.db_mirror_ha and (entity_id in self.db_mirror_list)) or self.db_primary):
                # Instead of appending to a local mirror_updates list, call the database manager to schedule the update
                if last_changed:
                    try:
                        last_changed = datetime.strptime(last_changed, TIME_FORMAT_HA_TZ)
                    except (ValueError, TypeError):
                        # Try fallback format without microseconds
                        try:
                            last_changed = datetime.strptime(last_changed, "%Y-%m-%dT%H:%M:%S%z")
                        except (ValueError, TypeError) as e:
                            self.log("Warn: Failed to parse last_changed time {} for entity {} : {}".format(last_changed, entity_id, e))
                            last_changed = datetime.now()
                else:
                    last_changed = datetime.now()

                self.db_manager.set_state_db(entity_id, state, attributes, timestamp=last_changed)

    def update_states_db(self):
        """
        Update the state data from the SQL Lite database
        """
        if not self.db_enable:
            return

        entities = self.db_manager.get_all_entities_db()
        if entities:
            for entity_name in entities:
                self.update_state_db(entity_name)
        else:
            self.log("Warn: Failed to update state data from DB")

    def update_states(self):
        """
        Update the state data from Home Assistant.
        """

        if self.db_primary and self.db_enable and not self.ha_key:
            self.update_states_db()
            return

        if not self.ha_key:
            return

        res = self.api_call("/api/states")
        if res:
            self.state_data = {}
            for item in res:
                entity_id = item.get("entity_id", None)
                if entity_id:
                    self.update_state_item(item, entity_id)
                else:
                    self.log("Warn: Failed to update state data from HA, item has no entity_id {}".format(item))
        else:
            self.log("Warn: Failed to update state data from HA")

    def get_history_window(self, sensor, start, end):
        """
        Fetch a single window of history for a sensor.

        :param sensor: The sensor to get the history for.
        :param start: Start of the window.
        :param end: End of the window.
        :return: The raw API response, or None.
        """
        res = self.api_call("/api/history/period/{}".format(start.strftime(TIME_FORMAT_HA)), {"filter_entity_id": sensor, "end_time": end.strftime(TIME_FORMAT_HA)})
        if isinstance(res, list) and len(res) > 0:
            return res
        return None

    def get_history(self, sensor, now, days=30, from_time=None, force_db=False, chunk_days=HISTORY_CHUNK_DAYS):
        """
        Get the history for a sensor from Home Assistant.

        Long windows are fetched in chunks so only one chunk's response body, decoded string
        and parsed objects are resident at a time rather than the whole window's. A 21 day
        window of a power sensor is tens of megabytes of JSON, and holding all three
        representations of it at once dominated the peak memory of a plan cycle.

        :param sensor: The sensor to get the history for.
        :param now: Current time, the end of the window.
        :param days: How many days of history to fetch.
        :param from_time: Explicit window start, overriding days.
        :param force_db: Read from the database rather than Home Assistant.
        :param chunk_days: Maximum days per request; 0 or None fetches the window in one request.
        :return: The history for the sensor, oldest first, or None.
        """
        if not sensor:
            return None

        self.db_mirror_list[sensor.lower()] = True

        if (self.db_primary or force_db) and self.db_enable:
            return self.db_manager.get_history_db(sensor, now, days=days)

        if from_time:
            start = from_time
        else:
            start = now - timedelta(days=days)
        end = now

        if not chunk_days or (end - start) <= timedelta(days=chunk_days):
            return self.get_history_window(sensor, start, end)

        history = []
        cursor = start
        while cursor < end:
            window_end = min(cursor + timedelta(days=chunk_days), end)
            res = self.get_history_window(sensor, cursor, window_end)
            if res:
                for item in res[0]:
                    # Home Assistant opens every window with the state in effect at start_time.
                    # For chunks after the first that instant is already covered by the previous
                    # chunk, and the synthesised record can land inside a gap in the recording
                    # where it would add a data point a single request never returns, changing
                    # how minute_data interpolates across that gap.
                    if cursor > start and item.get("last_updated") and str2time(item["last_updated"]) <= cursor:
                        continue
                    history.append(item)
            cursor = window_end

        return [history] if history else None

    async def set_state_external(self, entity_id, state, attributes=None):
        """
        Used for external changes to Predbat state data
        """
        if attributes is None:
            attributes = {}
        new_value = state
        new_state = {"entity_id": entity_id, "state": state, "attributes": attributes}
        old_value = self.get_state(entity_id)
        old_state = self.state_data.get(entity_id.lower(), None)

        # See if the modified entity is in the CONFIG_ITEMS list
        for item in self.base.CONFIG_ITEMS:
            if item.get("entity") == entity_id:
                old_value = item.get("value", "")
                step = item.get("step", 1)
                itemtype = item.get("type", "")
                if old_value is None:
                    old_value = item.get("default", "")
                if itemtype == "input_number" and step == 1:
                    old_value = int(old_value)
                if old_value != new_value:
                    service_data = {}
                    service_data["domain"] = itemtype
                    if itemtype == "switch":
                        service_data["service"] = "turn_on" if new_value else "turn_off"
                        service_data["service_data"] = {"entity_id": entity_id}
                    elif itemtype == "input_number":
                        service_data["service"] = "set_value"
                        service_data["service_data"] = {"entity_id": entity_id, "value": new_value}
                    elif itemtype == "select":
                        service_data["service"] = "select_option"
                        service_data["service_data"] = {"entity_id": entity_id, "option": new_value}
                    else:
                        continue

                    await self.base.trigger_callback(service_data)
                    return

        # Call back to Predbat handled entities
        domain = entity_id.split(".")[0] if "." in entity_id else ""
        service_data = {"domain": domain}
        if domain in ["switch", "input_boolean"]:
            # For toggle switches, check if value is 'on' or 'off'
            service_data["service"] = "turn_on" if new_value else "turn_off"
            service_data["service_data"] = {"entity_id": entity_id}
        elif domain in ["number", "input_number"]:
            service_data["service"] = "set_value"
            service_data["service_data"] = {"entity_id": entity_id, "value": new_value}
        elif domain in ["select", "input_select"]:
            service_data["service"] = "select_option"
            service_data["service_data"] = {"entity_id": entity_id, "option": new_value}
        elif domain == "input_datetime":
            value_str = str(new_value)
            current_attributes = old_state.get("attributes", {}) if old_state else {}
            has_date = current_attributes.get("has_date", None)
            has_time = current_attributes.get("has_time", None)
            service_data["service"] = "set_datetime"
            if has_date is True and has_time is False:
                service_data["service_data"] = {"entity_id": entity_id, "date": value_str}
            elif has_date is False and has_time is True:
                service_data["service_data"] = {"entity_id": entity_id, "time": value_str}
            elif has_date is True and has_time is True:
                service_data["service_data"] = {"entity_id": entity_id, "datetime": value_str}
            else:
                if ("T" in value_str) or (" " in value_str and ":" in value_str):
                    service_data["service_data"] = {"entity_id": entity_id, "datetime": value_str}
                elif ":" in value_str:
                    service_data["service_data"] = {"entity_id": entity_id, "time": value_str}
                else:
                    service_data["service_data"] = {"entity_id": entity_id, "date": value_str}
        elif domain == "input_text":
            value_str = str(new_value)
            service_data["service"] = "set_value"
            service_data["service_data"] = {"entity_id": entity_id, "value": value_str}
        else:
            service_data = None
        if service_data:
            await self.base.trigger_callback(service_data)
            return

        # If its not found then its a sensor and we can set the state directly
        self.set_state(entity_id, state, attributes)

        # Only trigger on value change or you get too many updates
        if (old_value is None) or (new_value != old_value):
            await self.base.trigger_watch_list(entity_id, attributes, old_state, new_state)

    def set_state(self, entity_id, state, attributes=None):
        """
        Set the state of an entity in Home Assistant.
        """
        if attributes is None:
            attributes = {}
        self.db_mirror_list[entity_id] = True

        if self.db_enable and (self.db_mirror_ha or self.db_primary):
            item = self.db_manager.set_state_db(entity_id, state, attributes)
            # Locally cache state until DB update happens
            self.update_state_item(item, entity_id, nodb=True)

        if self.ha_key:
            data = {"state": state}
            if attributes:
                data["attributes"] = attributes
                data["unrecorded_attributes"] = ["results"]
            self.api_call("/api/states/{}".format(entity_id), data, post=True)
            self.update_state(entity_id)

    def call_service(self, service, **kwargs):
        """
        Call a service in Home Assistant via Websocket
        or loopback to Predbat in standalone mode
        """
        data = {}
        for key in kwargs:
            data[key] = kwargs[key]
        if "/" in service:
            domain, service = service.split("/")
        elif "." in service:
            domain, service = service.split(".")
        else:
            domain = ""

        if self.websocket_active:
            return self.call_service_websocket_command(domain, service, data)
        else:
            # Loopback with no home assistant
            data_frame = {"domain": domain, "service": service, "service_data": data}
            return run_async(self.base.trigger_callback(data_frame))

    def api_call(self, endpoint, data_in=None, post=False, core=True, silent=False):
        """
        Make an API call to Home Assistant.

        :param endpoint: The API endpoint to call.
        :param data_in: The data to send in the body of the request.
        :param post: True if this is a POST request, False for GET.
        :param core: True is this is a call to HA Core, False if it is a Supervisor call
        :param silent: True if warning message from the API call is to be suppressed
        :return: The response from the API.
        """
        if core:
            url = self.ha_url + endpoint
            key = self.ha_key
        else:
            # make an API call to the Supervisor using the supervisor token
            url = "http://supervisor" + endpoint
            key = os.environ.get("SUPERVISOR_TOKEN", None)

        if key is None:
            return None

        headers = {
            "Authorization": "Bearer " + key,
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        try:
            if post:
                if data_in:
                    response = requests.post(url, headers=headers, json=data_in, timeout=TIMEOUT)
                else:
                    response = requests.post(url, headers=headers, timeout=TIMEOUT)
            else:
                if data_in:
                    response = requests.get(url, headers=headers, params=data_in, timeout=TIMEOUT)
                else:
                    response = requests.get(url, headers=headers, timeout=TIMEOUT)
            data = response.json()
            self.api_errors = 0
        except requests.exceptions.JSONDecodeError:
            if not silent:  # suppress warning message for call to get slug id from supervisor because in docker installs this will always error (no supervisor)
                self.log("Warn: Failed to decode response {} from {}".format(response, url))
                self.api_errors += 1

            data = None
        except (requests.Timeout, requests.exceptions.ReadTimeout):
            self.log("Warn: Timeout from {}".format(url))
            self.api_errors += 1
            data = None
        except requests.exceptions.ConnectionError as e:
            if not silent:
                self.log("Warn: Connection error from {}: {}".format(url, e))
            self.api_errors += 1
            data = None

        if self.api_errors >= 10:
            self.log("Error: Too many API errors, stopping")
            self.fatal_error_occurred()

        return data