hass.py

"""Standalone mode wrapper for running PredBat outside AppDaemon.

Provides the Hass class that emulates the AppDaemon interface for standalone
execution, including YAML configuration loading, secret management, log
rotation, scheduled callback execution, and file change detection for
development hot-reload.
"""

import io
import yaml
import sys
import asyncio
import os
import subprocess


def write_git_version_marker():
    """
    Best-effort: when running from a git checkout - directly, or via the symlinked
    .py files that coverage/standalone_ha sets up for a live-HA dev run - record the
    commit as git_version.txt next to this file (predbat.py resolves its own __file__
    to the same directory) so predbat.py can show it instead of just the release tag.
    Runs before predbat is imported, since that's when predbat.py reads the marker.
    Silently does nothing if git isn't available or this isn't a checkout.
    """
    this_dir = os.path.dirname(os.path.abspath(__file__))
    repo_dir = os.path.dirname(os.path.realpath(__file__))
    try:
        commit = subprocess.check_output(["git", "-C", repo_dir, "rev-parse", "--short", "HEAD"], stderr=subprocess.DEVNULL).decode().strip()
        dirty = bool(subprocess.check_output(["git", "-C", repo_dir, "status", "--porcelain"], stderr=subprocess.DEVNULL).decode().strip())
        with open(os.path.join(this_dir, "git_version.txt"), "w") as f:
            f.write(commit + ("-dirty" if dirty else ""))
    except Exception:
        pass


write_git_version_marker()

import predbat
import time
from datetime import datetime, timedelta
from multiprocessing import set_start_method
import concurrent.futures
import threading
import traceback


def check_modified(py_files, start_time):
    """
    Check if .py file was changed since we started
    """

    # Check the last modified timestamp of each .py file
    for file_path in py_files:
        if os.path.exists(file_path):
            last_modified = os.path.getmtime(file_path)
            last_modified_timestamp = datetime.fromtimestamp(last_modified)
            if last_modified_timestamp > start_time:
                print("File {} was modified".format(file_path))
                return True
        else:
            print("File {} does not exist".format(file_path))
            return True
    return False


def resolve_apps_yaml_path():
    """
    Resolve the one real apps.yaml path this instance is actually configured to use, using the
    same PREDBAT_APPS_FILE resolution applied when the config itself is loaded.
    """
    return os.path.abspath(os.getenv("PREDBAT_APPS_FILE", "apps.yaml"))


def collect_watch_files(roots, apps_file_path):
    """
    Build the list of files to watch for changes: every .py file under the given root
    directories, plus the one real apps.yaml file this instance is actually configured to use.

    Deliberately does NOT match a file just because it happens to be named "apps.yaml" -
    tools that write their own scratch apps.yaml-shaped files elsewhere under the same tree
    (e.g. the Annual prediction tool's isolated headless work directory, see annual.py's
    write_minimal_apps_yaml()) would otherwise be indistinguishable from the real config file,
    forcing an unwanted restart of the live instance whenever they run (#4397, #4396).
    """
    apps_file_path = os.path.abspath(apps_file_path)
    py_files = []
    seen_files = set()
    for root_dir in roots:
        for root, dirs, files in os.walk(root_dir):
            for file in files:
                if file.startswith("."):
                    continue
                full_path = os.path.abspath(os.path.join(root, file))
                if full_path in seen_files:
                    continue
                if file.endswith(".py") or full_path == apps_file_path:
                    py_files.append(full_path)
                    seen_files.add(full_path)

    # The real configured apps.yaml might not live under any of the walked roots at all
    # (e.g. PREDBAT_APPS_FILE pointing somewhere the roots don't reach) - always include it
    # explicitly if it exists, rather than silently dropping config-change watching entirely
    # just because the walk never happened to encounter it (Copilot review on #4401).
    if apps_file_path not in seen_files and os.path.exists(apps_file_path):
        py_files.append(apps_file_path)
        seen_files.add(apps_file_path)

    return py_files


async def main():
    print("**** Starting Standalone Predbat ****")
    start_time = datetime.now()

    try:
        p_han = predbat.PredBat()
    except Exception as e:
        print("Error: Failed to construct predbat {}".format(e))
        print(traceback.format_exc())
        return

    try:
        p_han.initialize()
    except Exception as e:
        print("Error: Failed to initialise predbat {}".format(e))
        print(traceback.format_exc())
        await p_han.stop_all()
        return

    #
    #   Adding additional root to monitor
    #

    # List of root directories to search
    roots = [".", "/addon"]

    # Find all .py files in the directory hierarchy, plus the one real apps.yaml this
    # instance is actually configured to use (not any file that merely happens to share
    # that name elsewhere in the tree - see #4397/#4396)
    py_files = collect_watch_files(roots, resolve_apps_yaml_path())

    print("Watching {} for changes".format(py_files))

    # Runtime loop
    count = 0
    while True:
        time.sleep(1)
        await p_han.timer_tick()
        if (count % 5 == 0) and check_modified(py_files, start_time):
            print("Stopping Predbat due to file changes....")
            await p_han.stop_all()
            break
        if p_han.fatal_error:
            print("Stopping Predbat due to fatal error....")
            await p_han.stop_all()
            break
        count += 1


if __name__ == "__main__":
    set_start_method("fork")
    asyncio.run(main())
    sys.exit(0)


class Hass:
    """Standalone mode wrapper emulating the AppDaemon interface.

    Enables PredBat to run outside Home Assistant/AppDaemon with YAML
    config loading, secret management, log rotation, scheduled callbacks,
    and file change detection for development hot-reload.
    """

    def log(self, msg, quiet=True):
        """
        Log a message to the logfile
        """
        message = "{}: {}\n".format(datetime.now(), msg)
        self.logfile.write(message)
        self.logfile.flush()
        msg_lower = msg.lower()
        if not quiet or msg_lower.startswith("error") or msg_lower.startswith("warn") or msg_lower.startswith("info"):
            print(message, end="")

        # maximum number of historic logfiles to retain
        max_logs = 9

        log_size = self.logfile.tell()
        if log_size > 10000000 and threading.current_thread() is threading.main_thread():
            # Only rotate from the main thread to avoid race conditions with
            # component threads that also call log().
            for num_logs in range(max_logs - 1, 0, -1):
                filename = "predbat." + format(num_logs) + ".log"
                if os.path.isfile(filename):
                    newfile = "predbat." + format(num_logs + 1) + ".log"
                    os.rename(filename, newfile)

            self.logfile.close()
            os.rename("predbat.log", "predbat.1.log")
            self.logfile = open("predbat.log", "w")

    async def run_in_executor(self, callback, *args):
        """
        Run a function in the executor
        """
        with concurrent.futures.ThreadPoolExecutor() as pool:
            future = pool.submit(callback, *args)
            return future

    async def task_waiter_async(self, task):
        """
        Waits for a task to complete async
        """
        await task

    def task_waiter(self, task):
        """
        Waits for a task to complete
        """
        asyncio.run(self.task_waiter_async(task))

    def create_task(self, task, name="TaskCreate"):
        """
        Creates a new thread to run the task in
        """
        self.log("Creating task: {}".format(task), quiet=False)
        t1 = threading.Thread(name=name, target=self.task_waiter, args=[task])
        t1.start()
        self.threads.append(t1)
        return t1

    async def stop_all(self):
        """
        Stop Predbat
        """
        self.log("Stopping Predbat", quiet=False)
        await self.terminate()

        for t in self.threads:
            t.join(5 * 60)
        self.logfile.close()

    def load_secrets(self):
        """
        Load secrets from secrets.yaml file
        Priority: PREDBAT_SECRETS_FILE env var, ./secrets.yaml, /config/secrets.yaml
        """
        secrets = {}
        secrets_file = None

        # Try loading from different locations in priority order
        possible_locations = [
            os.getenv("PREDBAT_SECRETS_FILE"),
            "secrets.yaml",
            "/homeassistant/secrets.yaml",
            "/conf/secrets.yaml",
            "/config/secrets.yaml",
        ]

        for location in possible_locations:
            if location and os.path.isfile(location):
                secrets_file = location
                break

        if secrets_file:
            self.log(f"Loading secrets from {secrets_file}", quiet=False)
            try:
                with io.open(secrets_file, "r") as stream:
                    secrets = yaml.safe_load(stream) or {}
                    # Check for debug logging option
                    if secrets.get("logger") == "debug":
                        self.log(f"Info: Secrets loaded from {secrets_file}", quiet=False)
            except yaml.YAMLError as exc:
                self.log(f"Error: Failed to load secrets from {secrets_file}: {exc}", quiet=False)
            except Exception as exc:
                self.log(f"Error: Failed to open secrets file {secrets_file}: {exc}", quiet=False)
        else:
            self.log("Info: No secrets.yaml file found", quiet=False)

        return secrets

    def secret_constructor(self, loader, node):
        """
        YAML constructor for !secret tag
        """
        secret_key = loader.construct_scalar(node)
        if secret_key in self.secrets:
            return self.secrets[secret_key]
        else:
            self.log(f"Warn: Secret '{secret_key}' not found in secrets.yaml")
            return None

    def __init__(self):
        """
        Start Predbat
        """
        self.args = {}
        self.run_list = []
        self.threads = []
        self.fatal_error = False
        self.hass_api_version = 2

        self.logfile = open("predbat.log", "a")

        # Load secrets first
        self.secrets = self.load_secrets()

        # Register custom YAML constructor for !secret tag
        yaml.add_constructor("!secret", self.secret_constructor, Loader=yaml.SafeLoader)

        # Open YAML file apps.yaml and read it
        apps_file = os.getenv("PREDBAT_APPS_FILE", "apps.yaml")
        self.log(f"Loading {apps_file}", quiet=False)
        with io.open(apps_file, "r") as stream:
            try:
                config = yaml.safe_load(stream)
                self.args = config["pred_bat"]
            except yaml.YAMLError as exc:
                print(exc)
                sys.exit(1)

    def run_every(self, callback, next_time, run_every, **kwargs):
        """
        Run a function every x seconds
        """
        self.run_list.append({"callback": callback, "next_time": next_time, "run_every": run_every, "kwargs": kwargs})
        return True

    async def timer_tick(self):
        """
        Timer tick function, executes tasks at the correct time
        """
        now = datetime.now()
        for item in self.run_list:
            if now > item["next_time"]:
                try:
                    item["callback"](None)
                except Exception as e:
                    self.log("Error: timer_tick caught exception: {}".format(e), quiet=False)
                    print(traceback.format_exc())
                while now > item["next_time"]:
                    run_every = timedelta(seconds=item["run_every"])
                    item["next_time"] += run_every