perf: batch the prediction fan-out into one kernel call (35% faster planning)

.cspell/custom-dictionary-workspace.txt

@@ -25,6 +25,7 @@ armhf
 armv
 ASHP
 asyncio
+atfork
 authoriser
 autodocstring
 autoflake
@@ -51,6 +52,7 @@ byok
 cadata
 calib
 Cantarell
+CDLL
 cdn
 Cerbo
 cexxxx
@@ -77,6 +79,7 @@ corruptplans
 cprofile
 creds
 crosscharge
+ctypes
 customauthorizer
 customisation
 Customise
@@ -378,6 +381,7 @@ preseeded
 prevs
 protobuf
 psum
+pthread
 pvbat
 pvenergytotal
 pvlib
@@ -447,6 +451,7 @@ slugifies
 slugify
 slugifying
 snakeviz
+snprintf
 socb
 socketloop
 socs
@@ -472,6 +477,7 @@ stdlib
 stepline
 Stromligning
 strptime
+strtod
 Strømligning
 subclassing
 submessage
@@ -489,6 +495,7 @@ temperation
 Teslemetry
 testname
 thirdparty
+threadpool
 timea
 timeb
 timec
@@ -520,6 +527,8 @@ unconfigured
 undiscounted
 unnormalised
 unparseable
+unpickled
+unpushed
 unsmoothed
 unstaged
 urlsafe
@@ -530,10 +539,14 @@ venv
 Victron
 visibilitychange
 waf
+waitpid
 Wallbox
 weblink
 welink
 Werror
+WEXITSTATUS
+WIFEXITED
+WNOHANG
 workmode
 writeonly
 wrongsha

.github/copilot-instructions.md

@@ -88,44 +88,37 @@ The core prediction/planning happens in `Plan` mixin (`apps/predbat/plan.py`) wh
 - `optimise_detailed_pass()` - Fine-tune charge/export windows
 - `run_prediction()` - Execute battery simulation via `Prediction` class
 
-### Thread Pool System
+### Batched Predictions
 
-**Multiprocessing for predictions** (in `apps/predbat/plan.py`):
+There is no process pool and no Python thread pool. Every optimiser fan-out in `plan.py` is
+launch-all-then-collect-all, so the launches queue instead of running and the whole fan-out is
+handed to the C++ kernel in one call, which spreads it across its own threads with the GIL
+released. Python itself is strictly serial.
 
-```python
-from multiprocessing import Pool, cpu_count
-
-# Pool created on demand
-if not self.pool:
-    threads = self.get_arg("threads", "auto")
-    if threads == "auto":
-        self.pool = Pool(processes=cpu_count())
-    else:
-        self.pool = Pool(processes=int(threads))
-```
+**How it works** (`apps/predbat/prediction_batch.py`, mixed into `Prediction`):
 
-**How it works**:
-
-- `self.pool` is a `multiprocessing.Pool` instance (NOT thread pool - uses processes)
-- Pool runs prediction scenarios in parallel using `pool.apply_async()`
-- Wrapped functions in `prediction.py`: `wrapped_run_prediction_single()`, `wrapped_run_prediction_charge()`, etc.
-- Global state shared via `PRED_GLOBAL` dict (copied to child processes)
-- Pool configured via `threads` config option: `"auto"` for CPU count or specific number
-- Returns `DummyThread` objects when pool disabled (for single-threaded testing)
+- `Plan.launch_run_prediction_*()` calls `Prediction.queue_run_prediction_*()`, which builds the
+  trial inputs, appends a `BatchJob` to `self.pending_batch` and returns a `BatchHandle`
+- Nothing has run at that point. The first `BatchHandle.get()` calls `flush_batch()`, which resolves
+  prediction-cache hits and intra-batch duplicates in Python and sends the remainder through one
+  `run_prediction_kernel_batch()` / `pk_run_batch` call; every handle in the fan-out is then resolved
+- A job the kernel will not take (no kernel, stale binary, debug run) falls back to
+  `Prediction.run_prediction()`, which is also the direct path `thread_run_prediction_*()` uses
+- `threads` config option sets `Prediction.batch_threads`, the kernel's thread count. `"auto"` uses
+  the CPU count. Results are identical at any thread count - it only trades CPU for planning time
 
 **Key pattern**:
 
 ```python
-# Async call to run prediction in pool
-if self.pool and self.pool._state == "RUN":
-    han = self.pool.apply_async(wrapped_run_prediction_charge, (args...))
-    return han
-else:
-    # Fallback to synchronous execution
-    result = wrapped_run_prediction_charge(args...)
-    return DummyThread(result)
+# Queue the whole fan-out, then read the handles - the first get() runs all of it
+handles = [self.launch_run_prediction_charge(soc, window_n, ...) for soc in try_socs]
+results = [handle.get() for handle in handles]
 ```
 
+**Invariant**: a caller must not mutate any list or window dict it has passed to
+`launch_run_prediction_*` until it has read the matching handle, because those inputs are not read
+until the batch is flushed.
+
 ### Component System (Critical Pattern)
 
 All major features inherit from `ComponentBase` (see `apps/predbat/component_base.py`):
@@ -281,8 +274,7 @@ self.log("Error: Failed with: {}".format(e))
 Core prediction happens in `apps/predbat/prediction.py`:
 
 - `Prediction` class runs battery simulation in 5-min steps
-- Multiprocessing via `wrapped_run_prediction_*` functions
-- Global state in `PRED_GLOBAL` dict for process sharing
+- Batched predictions via `launch_run_prediction_*()` queuing jobs and `BatchHandle.get()` flushing through the C++ kernel
 - Optimisation in `apps/predbat/plan.py` via `optimise_all_windows()`
 
 ### State Management

apps/predbat/annual.py

@@ -997,14 +997,14 @@ def prepare_sample(predbat, config, weather, tariff, load_source, day, midnight_
     """
     reset_sample_state(predbat)
 
-    # calculate_plan() spins up a multiprocessing Pool sized from args["threads"] (plan.py)
-    # unless it is exactly 0. The annual tool plans hundreds of individual days per run, each a
-    # small, fast calculation, so per-day pool creation is both wasted overhead and, on a
-    # spawn-based multiprocessing start method, unsafe: pool workers read the module-level
-    # PRED_GLOBAL dict in prediction.py, which is only populated in the parent process. This
-    # matches create_headless_predbat()'s own choice for a fully offline run; it is forced here
-    # too so run_day() behaves the same way against a caller-supplied PredBat instance (such as
-    # the standard unit test fixture) that has not gone through that bootstrap.
+    # calculate_plan() sizes the C++ kernel's batch thread count from args["threads"] (plan.py); it
+    # no longer spins up a Python multiprocessing Pool. pk_run_batch creates and joins its own
+    # std::thread pool fresh on every call, so leaving threads at its 'auto' default would pay that
+    # creation cost on each of the hundreds of small, fast per-day calculations the annual tool runs
+    # - overhead a plan this size cannot recoup. This matches create_headless_predbat()'s own choice
+    # for a fully offline run; it is forced here too so run_day() behaves the same way against a
+    # caller-supplied PredBat instance (such as the standard unit test fixture) that has not gone
+    # through that bootstrap.
     predbat.args["threads"] = 0
 
     predbat.midnight_utc = midnight_utc

apps/predbat/hass.py

@@ -164,9 +164,7 @@ def log(self, msg, quiet=True):
         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() — a Threading.Lock cannot
-            # be used here because the logfile object must survive pickle for
-            # the multiprocessing pool.
+            # 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):

apps/predbat/load_ml_component.py

@@ -736,8 +736,10 @@ async def run(self, seconds, first):
             self.log("ML Component: No training needed, model age is {} hours".format(dp2(retrain_age_seconds / 3600.0)))
 
         if should_train or should_fetch:
-            # Set load_ml_calculating across all NumPy-heavy work (training + predict + save)
-            # so that the plan's multiprocessing pool is never fork()ed while Numpy threads are active.
+            # Set load_ml_calculating across all NumPy-heavy work (training + predict + save). This used
+            # to gate the plan's multiprocessing pool off while NumPy threads were active (fork() is
+            # unsafe with live threads); the pool is gone, so this is now purely a status flag surfaced
+            # via is_calculating() and logged by calculate_plan().
             if self.base.prediction_started:
                 self.log("ML Component: Waiting for current prediction cycle to complete before running ML work")
                 while self.base.prediction_started:

apps/predbat/marginal.py

@@ -86,8 +86,7 @@ def calculate_marginal_costs(self):
                     if minute in modified_load:
                         modified_load[minute] += extra_per_step
 
-                # Create a fresh Prediction with the modified load; this updates PRED_GLOBAL
-                # which is safe since we run synchronously (pool is idle at this point)
+                # Create a fresh Prediction with the modified load.
                 # No need to include 10% extra load as we only run normal simulations.
                 pred = Prediction(self, self.pv_forecast_minute_step, self.pv_forecast_minute_step, modified_load, modified_load)