feat(log): make the number of retained log files configurable

apps/predbat/config.py

@@ -2523,6 +2523,7 @@
     "db_primary": {"type": "boolean"},
     "threads": {"type": "string|integer", "allowed": ["auto", 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]},
     "prediction_kernel_enable": {"type": "boolean"},
+    "log_count": {"type": "integer", "min": 2, "max": 100},
     "ha_url": {"type": "string", "empty": False},
     "ha_key": {"type": "string", "empty": False},
     "load_filter_threshold": {"type": "integer"},

apps/predbat/config/apps.yaml

@@ -24,6 +24,10 @@ pred_bat:
   # Can be auto for automatic, 0 for off or values 1-N for a fixed number
   threads: auto
 
+  # Number of log files to keep, including the live predbat.log (2-100, default 10)
+  # Rotated logs are predbat.01.log .. predbat.99.log
+  #log_count: 10
+
   # If you are using Predbat outside of HA then set the HA URL and Key (long lived access token here)
   #ha_url: 'http://homeassistant.local:8123'
   #ha_key: !secret ha_key

apps/predbat/hass.py

@@ -36,7 +36,7 @@ def write_git_version_marker():
 write_git_version_marker()
 
 import predbat
-from utils import load_apps_yaml
+from utils import load_apps_yaml, predbat_log_count, predbat_log_name, rotate_predbat_logs
 import time
 from datetime import datetime, timedelta
 from multiprocessing import set_start_method
@@ -188,21 +188,17 @@ def log(self, msg, quiet=True):
         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
+        # Total logfiles to keep including the live one, so max_logs rotated copies.
+        max_logs = predbat_log_count(self.args) - 1
 
         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)
+            rotate_predbat_logs(max_logs)
 
             self.logfile.close()
-            os.rename("predbat.log", "predbat.1.log")
+            os.rename("predbat.log", predbat_log_name(1))
             self.logfile = open("predbat.log", "w")
 
     async def run_in_executor(self, callback, *args):

apps/predbat/predbat.py

@@ -1532,6 +1532,12 @@ def validate_config(self):
                             matches = True
                             if required_entries is not None and len(value) > required_entries:
                                 value = value[:required_entries]
+                            # min/max are declared on some APPS_SCHEMA entries (log_count, #5076)
+                            # but were never actually enforced here - the schema promised a range
+                            # check that validate_config() silently skipped (Copilot review on
+                            # #5076).
+                            schema_min = spec.get("min", None)
+                            schema_max = spec.get("max", None)
                             for item in value:
                                 if not self.validate_is_int(item):
                                     self.log("Warn: Validation of apps.yaml found configuration item '{}' element {} is not an integer".format(name, item))
@@ -1543,6 +1549,11 @@ def validate_config(self):
                                     self.arg_errors[name] = "Invalid value, expected non-zero integer item {}".format(item)
                                     errors += 1
                                     break
+                                if (schema_min is not None and int(item) < schema_min) or (schema_max is not None and int(item) > schema_max):
+                                    self.log("Warn: Validation of apps.yaml found configuration item '{}' value {} is outside the allowed range [{}, {}]".format(name, item, schema_min, schema_max))
+                                    self.arg_errors[name] = "Invalid value, expected between {} and {}".format(schema_min, schema_max)
+                                    errors += 1
+                                    break
                     elif expected_type == "float" or expected_type == "float_list":
                         if expected_type == "float" and (isinstance(value, float) or isinstance(value, int)):
                             value = [value]

apps/predbat/tests/test_log_rotation.py

@@ -0,0 +1,216 @@
+# -----------------------------------------------------------------------------
+# 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
+
+"""
+Unit tests for configurable log rotation (#5076) - the log_count apps.yaml setting, the
+two-digit predbat.NN.log naming, and the migration path that reads the single-digit names an
+older Predbat wrote.
+"""
+
+import os
+import tempfile
+
+from utils import (
+    PREDBAT_LOG_COUNT_DEFAULT,
+    PREDBAT_LOG_COUNT_MAX,
+    PREDBAT_LOG_COUNT_MIN,
+    predbat_log_count,
+    predbat_log_file_prev,
+    predbat_log_name,
+    predbat_log_name_legacy,
+    read_predbat_log,
+    rotate_predbat_logs,
+)
+
+
+def _touch(name, content=""):
+    """Create a file with known content, for asserting which file ended up where."""
+    with open(name, "w", encoding="utf-8") as handle:
+        handle.write(content)
+
+
+def _rotate(args):
+    """
+    Run the rotation exactly as Hass.log() does, against the current working directory.
+
+    Calls the same rotate_predbat_logs() Hass.log() calls, rather than a hand-copy of its loop -
+    a copy can diverge from the code it is meant to be testing and still pass, which is exactly
+    what let the off-by-one Copilot found on #5076 ship with a green test suite (the copy had the
+    same bug, so it agreed with itself). Only the parts of Hass.log() this file does not otherwise
+    exercise - closing/reopening the live predbat.log - are still done directly here.
+    """
+    max_logs = predbat_log_count(args) - 1
+    rotate_predbat_logs(max_logs)
+    os.rename("predbat.log", predbat_log_name(1))
+    _touch("predbat.log", "live")
+
+
+def run_log_rotation_tests(my_predbat):
+    """
+    Test configurable log rotation, two-digit naming and single-digit migration (#5076).
+    """
+    failed = False
+    print("**** Running log_rotation tests ****")
+
+    saved_cwd = os.getcwd()
+    try:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            os.chdir(tmpdir)
+
+            print("Test 1: log_count defaults and clamps")
+            for args, expected, why in (
+                ({}, PREDBAT_LOG_COUNT_DEFAULT, "absent"),
+                ({"log_count": 25}, 25, "in range"),
+                ({"log_count": 1}, PREDBAT_LOG_COUNT_MIN, "below minimum"),
+                ({"log_count": 500}, PREDBAT_LOG_COUNT_MAX, "above maximum"),
+                ({"log_count": "30"}, 30, "numeric string"),
+                ({"log_count": "banana"}, PREDBAT_LOG_COUNT_DEFAULT, "non-numeric"),
+                ({"log_count": None}, PREDBAT_LOG_COUNT_DEFAULT, "null"),
+                # YAML accepts non-finite numeric scalars (.inf/-.inf), which parse as a float
+                # int() cannot convert - OverflowError, not caught by the (TypeError, ValueError)
+                # this already guarded against (Copilot review on #5076).
+                ({"log_count": float("inf")}, PREDBAT_LOG_COUNT_DEFAULT, "positive infinity"),
+                ({"log_count": float("-inf")}, PREDBAT_LOG_COUNT_DEFAULT, "negative infinity"),
+            ):
+                actual = predbat_log_count(args)
+                if actual != expected:
+                    print("  ERROR: log_count {} ({}) should give {}, got {}".format(args, why, expected, actual))
+                    failed = True
+
+            print("Test 2: names are zero-padded to two digits")
+            if predbat_log_name(1) != "predbat.01.log" or predbat_log_name(99) != "predbat.99.log":
+                print("  ERROR: unexpected padded names: {!r} {!r}".format(predbat_log_name(1), predbat_log_name(99)))
+                failed = True
+            if predbat_log_name_legacy(1) != "predbat.1.log":
+                print("  ERROR: unexpected legacy name: {!r}".format(predbat_log_name_legacy(1)))
+                failed = True
+
+            print("Test 3: a rotation shifts every log up one slot")
+            _touch("predbat.log", "current")
+            _touch(predbat_log_name(1), "one")
+            _touch(predbat_log_name(2), "two")
+            _rotate({"log_count": 10})
+            for name, expected in ((predbat_log_name(1), "current"), (predbat_log_name(2), "one"), (predbat_log_name(3), "two")):
+                if not os.path.isfile(name):
+                    print("  ERROR: expected {} to exist after rotation".format(name))
+                    failed = True
+                elif open(name, encoding="utf-8").read() != expected:
+                    print("  ERROR: {} should hold {!r}, got {!r}".format(name, expected, open(name, encoding="utf-8").read()))
+                    failed = True
+
+            print("Test 4: single-digit logs from an older Predbat migrate to two-digit")
+            for name in os.listdir("."):
+                os.remove(name)
+            _touch("predbat.log", "current")
+            _touch(predbat_log_name_legacy(1), "old-one")
+            _touch(predbat_log_name_legacy(2), "old-two")
+            _rotate({"log_count": 10})
+            if os.path.isfile(predbat_log_name_legacy(2)) or os.path.isfile(predbat_log_name_legacy(3)):
+                print("  ERROR: single-digit files should have been renamed, found {}".format(sorted(os.listdir("."))))
+                failed = True
+            for name, expected in ((predbat_log_name(2), "old-one"), (predbat_log_name(3), "old-two")):
+                if not os.path.isfile(name) or open(name, encoding="utf-8").read() != expected:
+                    print("  ERROR: expected {} to hold {!r}, listing is {}".format(name, expected, sorted(os.listdir("."))))
+                    failed = True
+
+            print("Test 4b: a legacy single-digit log in the OLDEST kept slot is migrated, not stranded")
+            # #5076 Copilot review: the shift previously only walked slots 1..max_logs-1, so a
+            # legacy predbat.9.log sitting in the oldest kept slot (max_logs=9 for the default
+            # log_count=10) was never a rename source - it does not collide with anything landing
+            # on slot 9 from below, since that only writes the two-digit name. Left on disk under
+            # the old name forever, invisible to predbat_log_file_prev()'s two-digit-first lookup.
+            for name in os.listdir("."):
+                os.remove(name)
+            _touch("predbat.log", "current")
+            _touch(predbat_log_name_legacy(9), "oldest-legacy")
+            _rotate({"log_count": 10})  # max_logs = 9
+            if os.path.isfile(predbat_log_name_legacy(9)):
+                print("  ERROR: the legacy file in the oldest kept slot should not survive under its old name, found {}".format(sorted(os.listdir("."))))
+                failed = True
+            if os.path.isfile(predbat_log_name(10)) or os.path.isfile(predbat_log_name_legacy(10)):
+                print("  ERROR: log_count=10 keeps only slots 1-9 plus the live log, found {}".format(sorted(os.listdir("."))))
+                failed = True
+
+            print("Test 5: logs beyond the configured count are removed")
+            for name in os.listdir("."):
+                os.remove(name)
+            _touch("predbat.log", "current")
+            for num in range(1, 6):
+                _touch(predbat_log_name(num), "log{}".format(num))
+            _rotate({"log_count": 3})
+            # Keeping 3 files total means the live log plus slots 01 and 02.
+            if os.path.isfile(predbat_log_name(3)):
+                print("  ERROR: slot 03 should have been removed with log_count 3, listing is {}".format(sorted(os.listdir("."))))
+                failed = True
+            remaining = sorted(name for name in os.listdir(".") if name != "predbat.log")
+            if remaining != [predbat_log_name(1), predbat_log_name(2)]:
+                print("  ERROR: expected only slots 01 and 02 to remain, got {}".format(remaining))
+                failed = True
+
+            print("Test 6: lowering log_count clears files stranded above the new count")
+            for name in os.listdir("."):
+                os.remove(name)
+            _touch("predbat.log", "current")
+            for num in range(1, 21):
+                _touch(predbat_log_name(num), "log{}".format(num))
+            _rotate({"log_count": 4})
+            stranded = [name for name in os.listdir(".") if name not in ("predbat.log", predbat_log_name(1), predbat_log_name(2), predbat_log_name(3))]
+            if stranded:
+                print("  ERROR: files above the new count should be removed, found {}".format(sorted(stranded)))
+                failed = True
+
+            print("Test 7: the viewer prefers the two-digit previous log")
+            for name in os.listdir("."):
+                os.remove(name)
+            _touch("predbat.log", "live-line")
+            _touch(predbat_log_name(1), "padded-prev")
+            if predbat_log_file_prev() != predbat_log_name(1):
+                print("  ERROR: expected the padded name to win, got {!r}".format(predbat_log_file_prev()))
+                failed = True
+            data = read_predbat_log()
+            if "padded-prev" not in data or "live-line" not in data:
+                print("  ERROR: the viewer should show both logs, got {!r}".format(data))
+                failed = True
+            if data.index("padded-prev") > data.index("live-line"):
+                print("  ERROR: the previous log should be prefixed, not appended, got {!r}".format(data))
+                failed = True
+
+            print("Test 8: the viewer falls back to a single-digit previous log after an upgrade")
+            for name in os.listdir("."):
+                os.remove(name)
+            _touch("predbat.log", "live-line")
+            _touch(predbat_log_name_legacy(1), "legacy-prev")
+            if predbat_log_file_prev() != predbat_log_name_legacy(1):
+                print("  ERROR: expected the legacy name as fallback, got {!r}".format(predbat_log_file_prev()))
+                failed = True
+            data = read_predbat_log()
+            if "legacy-prev" not in data:
+                print("  ERROR: an upgrade must not lose the previous log, got {!r}".format(data))
+                failed = True
+
+            print("Test 9: no previous log at all is not an error")
+            for name in os.listdir("."):
+                os.remove(name)
+            _touch("predbat.log", "only-line")
+            if predbat_log_file_prev() is not None:
+                print("  ERROR: expected no previous log, got {!r}".format(predbat_log_file_prev()))
+                failed = True
+            if read_predbat_log().strip() != "only-line":
+                print("  ERROR: expected just the live log, got {!r}".format(read_predbat_log()))
+                failed = True
+    finally:
+        os.chdir(saved_cwd)
+
+    if failed:
+        print("\n**** log_rotation tests: FAILED ****")
+    else:
+        print("\n**** log_rotation tests: PASSED ****")
+
+    return failed

apps/predbat/tests/test_validate_config.py

@@ -128,6 +128,23 @@ def test_validate_config(my_predbat):
     print("  [integer zero:False] zero fails")
     _run(my_predbat, {"gateway_mqtt_port": 0}, expect_errors=["gateway_mqtt_port"])
 
+    # min/max constraint (log_count: {"type": "integer", "min": 2, "max": 100}) - declared in the
+    # schema but not actually enforced by validate_config()'s integer branch, so an out-of-range
+    # value silently passed validation despite the schema promising a range check
+    # (Copilot review on #5076).
+    print("  [integer min/max] in-range value passes")
+    _run(my_predbat, {"log_count": 25}, expect_clean=["log_count"])
+
+    print("  [integer min/max] below minimum fails")
+    _run(my_predbat, {"log_count": 1}, expect_errors=["log_count"])
+
+    print("  [integer min/max] above maximum fails")
+    _run(my_predbat, {"log_count": 101}, expect_errors=["log_count"])
+
+    print("  [integer min/max] exactly at the boundary passes")
+    _run(my_predbat, {"log_count": 2}, expect_clean=["log_count"])
+    _run(my_predbat, {"log_count": 100}, expect_clean=["log_count"])
+
     # ==========================================================================
     # INTEGER_LIST type  (days_previous: {"type": "integer_list"})
     # ==========================================================================

apps/predbat/unit_test.py

@@ -229,6 +229,7 @@
 from tests.test_rate_add_io_slots import run_rate_add_io_slots_tests
 from tests.test_iog_charge_skew import run_iog_charge_skew_tests
 from tests.test_dispatch_timeline import run_dispatch_timeline_tests
+from tests.test_log_rotation import run_log_rotation_tests
 from tests.test_battery_curve_keys import run_battery_curve_keys_tests
 from tests.test_balance_inverters import run_balance_inverters_tests
 from tests.test_octopus_download_rates import test_octopus_download_rates_wrapper
@@ -547,6 +548,7 @@ def main():
         ("rate_add_io_slots", run_rate_add_io_slots_tests, "Rate add IO slots tests", False),
         ("iog_charge_skew", run_iog_charge_skew_tests, "IOG earlier-charge skew characterisation tests", False),
         ("dispatch_timeline", run_dispatch_timeline_tests, "Dispatch timeline diagnostic tests (#4516 Stage 1)", False),
+        ("log_rotation", run_log_rotation_tests, "Configurable log rotation and two-digit naming (#5076)", False),
         ("rate_replicate", test_rate_replicate, "Rate replicate comprehensive tests (missing slots, IO, offsets, gas)", False),
         ("find_charge_window", test_find_charge_window, "Find charge window gap handling tests", False),
         ("find_charge_rate", test_find_charge_rate, "Find charge rate tests", False),