Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 49 additions & 13 deletions apps/predbat/hass.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,51 @@ def check_modified(py_files, start_time):
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()
Expand All @@ -65,19 +110,10 @@ async def main():
# List of root directories to search
roots = [".", "/addon"]

# Find all .py files in the directory hierarchy
py_files = []
seen_files = set()

# Directories to walk through
for root_dir in roots:
for root, dirs, files in os.walk(root_dir):
for file in files:
if (file.endswith(".py") or file == "apps.yaml") and not file.startswith("."):
full_path = os.path.abspath(os.path.join(root, file))
if full_path not in seen_files:
py_files.append(full_path)
seen_files.add(full_path)
# 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))

Expand Down
125 changes: 125 additions & 0 deletions apps/predbat/tests/test_hass_watcher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
# -----------------------------------------------------------------------------
# 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

"""
Tests for hass.py's standalone-mode file watcher (collect_watch_files/resolve_apps_yaml_path).

Covers #4397/#4396: the watcher used to match ANY file named "apps.yaml" anywhere in the
watched directory tree, so a tool (e.g. the Annual prediction tool) writing its own
apps.yaml-shaped scratch file elsewhere under the same tree would be indistinguishable from
the real config file, forcing an unwanted restart of the live instance whenever it ran.
"""

import os
import tempfile

from hass import collect_watch_files, resolve_apps_yaml_path


def test_hass_watcher(my_predbat):
"""Tests collect_watch_files() only matches the one real apps.yaml, plus .py files."""
failed = False
print("**** test_hass_watcher ****")

with tempfile.TemporaryDirectory() as root:
# Real apps.yaml at the top level, plus a .py file
real_apps_yaml = os.path.join(root, "apps.yaml")
with open(real_apps_yaml, "w", encoding="utf-8") as f:
f.write("pred_bat:\n timezone: Europe/London\n")

source_py = os.path.join(root, "predbat.py")
with open(source_py, "w", encoding="utf-8") as f:
f.write("# source file\n")

# A scratch subdirectory (mirroring the Annual tool's work_dir) with its own
# unrelated apps.yaml-shaped file, plus a hidden dotfile that should never be watched
scratch_dir = os.path.join(root, "annual_work")
os.makedirs(scratch_dir)
scratch_apps_yaml = os.path.join(scratch_dir, "apps.yaml")
with open(scratch_apps_yaml, "w", encoding="utf-8") as f:
f.write("pred_bat:\n timezone: Europe/London\n")

hidden_py = os.path.join(root, ".hidden.py")
with open(hidden_py, "w", encoding="utf-8") as f:
f.write("# should never be watched\n")

apps_file_path = os.path.abspath(real_apps_yaml)
watched = collect_watch_files([root], apps_file_path)
watched_abs = set(os.path.abspath(p) for p in watched)

print(" [real apps.yaml] the actual configured apps.yaml is watched")
if os.path.abspath(real_apps_yaml) not in watched_abs:
print(f"ERROR: real apps.yaml {real_apps_yaml} should be watched, got {watched_abs}")
failed = True

print(" [source .py file] source files are still watched")
if os.path.abspath(source_py) not in watched_abs:
print(f"ERROR: source file {source_py} should be watched, got {watched_abs}")
failed = True

print(" [scratch apps.yaml] a same-named file elsewhere in the tree is NOT watched (#4397/#4396)")
if os.path.abspath(scratch_apps_yaml) in watched_abs:
print(f"ERROR: scratch apps.yaml {scratch_apps_yaml} should NOT be watched, got {watched_abs}")
failed = True

print(" [hidden dotfile] dotfiles are never watched, even if they end in .py")
if os.path.abspath(hidden_py) in watched_abs:
print(f"ERROR: hidden file {hidden_py} should NOT be watched, got {watched_abs}")
failed = True

print(" [apps.yaml outside the walked roots] still watched explicitly (Copilot review on #4401)")
with tempfile.TemporaryDirectory() as walked_root, tempfile.TemporaryDirectory() as outside_root:
outside_apps_yaml = os.path.join(outside_root, "apps.yaml")
with open(outside_apps_yaml, "w", encoding="utf-8") as f:
f.write("pred_bat:\n timezone: Europe/London\n")

watched = collect_watch_files([walked_root], outside_apps_yaml)
watched_abs = set(os.path.abspath(p) for p in watched)
if os.path.abspath(outside_apps_yaml) not in watched_abs:
print(f"ERROR: apps.yaml outside the walked roots should still be watched, got {watched_abs}")
failed = True

print(" [apps.yaml that does not exist] not added if the configured path is missing entirely")
with tempfile.TemporaryDirectory() as walked_root:
missing_apps_yaml = os.path.join(walked_root, "does_not_exist", "apps.yaml")
watched = collect_watch_files([walked_root], missing_apps_yaml)
watched_abs = set(os.path.abspath(p) for p in watched)
if os.path.abspath(missing_apps_yaml) in watched_abs:
print(f"ERROR: a non-existent apps.yaml path should not be added, got {watched_abs}")
failed = True

print(" [resolve_apps_yaml_path] defaults to 'apps.yaml' in the working directory when PREDBAT_APPS_FILE is unset")
saved_env = os.environ.pop("PREDBAT_APPS_FILE", None)
try:
resolved = resolve_apps_yaml_path()
expected = os.path.abspath("apps.yaml")
if resolved != expected:
print(f"ERROR: expected {expected}, got {resolved}")
failed = True
finally:
if saved_env is not None:
os.environ["PREDBAT_APPS_FILE"] = saved_env

print(" [resolve_apps_yaml_path] honours PREDBAT_APPS_FILE when set (e.g. by the Annual tool's headless instance)")
saved_env = os.environ.get("PREDBAT_APPS_FILE")
try:
os.environ["PREDBAT_APPS_FILE"] = "/tmp/some_other_dir/apps.yaml"
resolved = resolve_apps_yaml_path()
expected = os.path.abspath("/tmp/some_other_dir/apps.yaml")
if resolved != expected:
print(f"ERROR: expected {expected}, got {resolved}")
failed = True
finally:
if saved_env is None:
os.environ.pop("PREDBAT_APPS_FILE", None)
else:
os.environ["PREDBAT_APPS_FILE"] = saved_env

print("**** test_hass_watcher PASSED ****" if not failed else "**** test_hass_watcher FAILED ****")
return failed
2 changes: 2 additions & 0 deletions apps/predbat/unit_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
from tests.test_fetch_config_options import test_fetch_config_options
from tests.test_multi_inverter import run_inverter_multi_tests
from tests.test_window2minutes import test_window2minutes
from tests.test_hass_watcher import test_hass_watcher
from tests.test_history_attribute import test_history_attribute
from tests.test_inverter import run_inverter_tests
from tests.test_basic_rates import test_basic_rates
Expand Down Expand Up @@ -266,6 +267,7 @@ def main():
("rate_min_forward_calc", test_rate_min_forward_calc, "Rate min forward calc tests", False),
("window_sort", run_window_sort_tests, "Window sort tests", False),
("window2minutes", test_window2minutes, "Window to minutes tests", False),
("hass_watcher", test_hass_watcher, "Standalone-mode file watcher tests (#4397/#4396)", False),
("compute_metric", run_compute_metric_tests, "Compute metric tests", False),
("minute_array", test_minute_array, "MinuteArray class tests", False),
("minute_data", test_minute_data, "Minute data tests", False),
Expand Down
Loading