fix(fetch): publish rate_import/rate_export atomically to remove a read race

update_pred rebuilds rate_import/rate_export on an AppDaemon worker thread
while async components (e.g. the Teslemetry tariff builder) read them, with no
lock. The build reset them to {}, reassigned per step, and mutated them in
place (load_saving_slot/load_free_slot/load_axle_slot), so a concurrent reader
could see an empty or half-merged dict.

Build each side into a local (import_rates/export_rates) through the whole
fetch + replicate + slot pipeline and assign self.rate_import/self.rate_export
exactly once at the end, so a reader always sees either the previous complete
dict or the new complete one - never an intermediate. The three in-place slot
mutators now take the working dict as a parameter instead of touching
self.rate_*.

Behaviour-preserving: derived attrs (rate_import_base/_replicated/_no_io,
rate_export_base/_replicated) are unchanged, and the full --quick suite
(octopus rates, saving/free/axle slots, overrides, manual rates, IO) passes.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: springfall2008

apps/predbat/axle.py

@@ -647,9 +647,9 @@ def fetch_axle_sessions(base):
     return axle_events_deduplicated
 
 
-def load_axle_slot(base, axle_sessions, export, rate_replicate=None):
+def load_axle_slot(base, axle_sessions, rate_dict, export, rate_replicate=None):
     """
-    Load Axle VPP session slot
+    Load Axle VPP session slot into rate_dict (in place)
     """
     if rate_replicate is None:
         rate_replicate = {}
@@ -679,11 +679,11 @@ def load_axle_slot(base, axle_sessions, export, rate_replicate=None):
                 base.log("Setting Axle VPP session in range {} - {} export {} pence_per_kwh {}".format(base.time_abs_str(start_minutes), base.time_abs_str(end_minutes), export, pence_per_kwh))
                 for minute in range(start_minutes, end_minutes):
                     if export:
-                        base.rate_export[minute] = base.rate_export.get(minute, 0) + pence_per_kwh
+                        rate_dict[minute] = rate_dict.get(minute, 0) + pence_per_kwh
                         base.load_scaling_dynamic[minute] = base.load_scaling_saving
                         rate_replicate[minute] = "saving"
                     else:
-                        base.rate_import[minute] = base.rate_import.get(minute, 0) - pence_per_kwh
+                        rate_dict[minute] = rate_dict.get(minute, 0) - pence_per_kwh
                         base.load_scaling_dynamic[minute] = base.load_scaling_free
                         rate_replicate[minute] = "saving"
 

apps/predbat/fetch.py

@@ -708,9 +708,9 @@ def fetch_sensor_data(self, save=True):
         prev_octopus_free_slots = self.octopus_free_slots.copy()
         prev_axle_sessions = self.axle_sessions.copy()
 
-        self.rate_import = {}
+        import_rates = {}
         self.rate_import_replicated = {}
-        self.rate_export = {}
+        export_rates = {}
         self.rate_export_replicated = {}
         self.rate_slots = []
         self.io_adjusted = {}
@@ -849,36 +849,36 @@ def fetch_sensor_data(self, save=True):
             # Fixed URL for rate import
             self.log("Downloading import rates directly from URL {}".format(self.get_arg("rates_import_octopus_url", indirect=False)))
             # Need to take a copy, as saving sessions will repeatedly increment cached import rates
-            self.rate_import = copy.deepcopy(self.download_octopus_rates(self.get_arg("rates_import_octopus_url", indirect=False)))
+            import_rates = copy.deepcopy(self.download_octopus_rates(self.get_arg("rates_import_octopus_url", indirect=False)))
         elif "metric_octopus_import" in self.args:
             # Octopus import rates
             entity_id = self.get_arg("metric_octopus_import", None, indirect=False)
-            self.rate_import = self.fetch_octopus_rates(entity_id, adjust_key="is_intelligent_adjusted")
-            if not self.rate_import:
+            import_rates = self.fetch_octopus_rates(entity_id, adjust_key="is_intelligent_adjusted")
+            if not import_rates:
                 self.log("Error: metric_octopus_import is not set correctly in apps.yaml, or no energy rates can be read")
                 self.record_status(message="Error: metric_octopus_import not set correctly in apps.yaml, or no energy rates can be read", had_errors=True)
         elif "metric_energidataservice_import" in self.args:
             # Energi Data Service import rates
             entity_id = self.get_arg("metric_energidataservice_import", None, indirect=False)
-            self.rate_import = self.fetch_energidataservice_rates(entity_id, adjust_key="is_intelligent_adjusted")
-            if not self.rate_import:
+            import_rates = self.fetch_energidataservice_rates(entity_id, adjust_key="is_intelligent_adjusted")
+            if not import_rates:
                 self.log("Error: metric_energidataservice_import is not set correctly in apps.yaml, or no energy rates can be read")
                 self.record_status(message="Error: metric_energidataservice_import not set correctly in apps.yaml, or no energy rates can be read", had_errors=True)
         elif "metric_stromligning_import_today" in self.args or "metric_stromligning_import_tomorrow" in self.args:
             # Strømligning import rates
             entity_id_today = self.get_arg("metric_stromligning_import_today", None, indirect=False)
             entity_id_tomorrow = self.get_arg("metric_stromligning_import_tomorrow", None, indirect=False)
-            self.rate_import = self.fetch_stromligning_rates(entity_id_today, entity_id_tomorrow, adjust_key="is_intelligent_adjusted")
-            if not self.rate_import:
+            import_rates = self.fetch_stromligning_rates(entity_id_today, entity_id_tomorrow, adjust_key="is_intelligent_adjusted")
+            if not import_rates:
                 self.log("Error: metric_stromligning_import sensors are not set correctly or no energy rates can be read")
                 self.record_status(message="Error: metric_stromligning_import sensors not set correctly or no energy rates can be read", had_errors=True)
 
         # Fallback if no other rate types are set
-        if not self.rate_import:
+        if not import_rates:
             # Basic rates defined by user over time
             rate_import_dict = self.get_arg("rates_import", [], indirect=False)
             if rate_import_dict:
-                self.rate_import = self.basic_rates(rate_import_dict, "rates_import")
+                import_rates = self.basic_rates(rate_import_dict, "rates_import")
 
         # Gas rates if set
         if "metric_octopus_gas" in self.args:
@@ -925,36 +925,36 @@ def fetch_sensor_data(self, save=True):
             # Fixed URL for rate export
             self.log("Downloading export rates directly from URL {}".format(self.get_arg("rates_export_octopus_url", indirect=False)))
             # Need to take a copy, as saving sessions will repeatedly increment cached export rates
-            self.rate_export = copy.deepcopy(self.download_octopus_rates(self.get_arg("rates_export_octopus_url", indirect=False)))
+            export_rates = copy.deepcopy(self.download_octopus_rates(self.get_arg("rates_export_octopus_url", indirect=False)))
         elif "metric_octopus_export" in self.args:
             # Octopus export rates
             entity_id = self.get_arg("metric_octopus_export", None, indirect=False)
-            self.rate_export = self.fetch_octopus_rates(entity_id)
-            if not self.rate_export:
+            export_rates = self.fetch_octopus_rates(entity_id)
+            if not export_rates:
                 self.log("Warning: metric_octopus_export is not set correctly in apps.yaml, or no energy rates can be read")
                 self.record_status(message="Error: metric_octopus_export not set correctly in apps.yaml, or no energy rates can be read", had_errors=True)
         elif "metric_energidataservice_export" in self.args:
             # Energi Data Service export rates
             entity_id = self.get_arg("metric_energidataservice_export", None, indirect=False)
-            self.rate_export = self.fetch_energidataservice_rates(entity_id)
-            if not self.rate_export:
+            export_rates = self.fetch_energidataservice_rates(entity_id)
+            if not export_rates:
                 self.log("Warning: metric_energidataservice_export is not set correctly in apps.yaml, or no energy rates can be read")
                 self.record_status(message="Error: metric_energidataservice_export not set correctly in apps.yaml, or no energy rates can be read", had_errors=True)
         elif "metric_stromligning_export_today" in self.args or "metric_stromligning_export_tomorrow" in self.args:
             # Strømligning export rates
             entity_id_today = self.get_arg("metric_stromligning_export_today", None, indirect=False)
             entity_id_tomorrow = self.get_arg("metric_stromligning_export_tomorrow", None, indirect=False)
-            self.rate_export = self.fetch_stromligning_rates(entity_id_today, entity_id_tomorrow)
-            if not self.rate_export:
+            export_rates = self.fetch_stromligning_rates(entity_id_today, entity_id_tomorrow)
+            if not export_rates:
                 self.log("Warning: metric_stromligning_export sensors are not set correctly or no energy rates can be read")
                 self.record_status(message="Error: metric_stromligning_export sensors not set correctly or no energy rates can be read", had_errors=True)
 
         # Fallback if no other rate types are set
-        if not self.rate_export:
+        if not export_rates:
             # Basic rates defined by user over time
             rate_export_dict = self.get_arg("rates_export", [], indirect=False)
             # Allow all zero export rates, as some users have a feed-in tariff that is zero
-            self.rate_export = self.basic_rates(rate_export_dict, "rates_export")
+            export_rates = self.basic_rates(rate_export_dict, "rates_export")
 
         # Fetch Axle sessions first so Octopus auto-join can skip saving sessions that overlap an Axle VPP session
         self.axle_sessions = fetch_axle_sessions(self)
@@ -967,44 +967,47 @@ def fetch_sensor_data(self, save=True):
 
         # futurerate data
         futurerate = FutureRate(self)
-        self.future_energy_rates_import, self.future_energy_rates_export = futurerate.futurerate_analysis(self.rate_import, self.rate_export)
+        self.future_energy_rates_import, self.future_energy_rates_export = futurerate.futurerate_analysis(import_rates, export_rates)
 
         # Replicate and scan import rates
-        if self.rate_import:
-            self.rate_scan(self.rate_import, print=False)
+        if import_rates:
+            self.rate_scan(import_rates, print=False)
             self.rate_max_base = self.rate_max  # True peak rate before saving sessions / overrides inflate it
             self.rate_min_base = self.rate_min  # True off-peak rate before free sessions / overrides deflate it
-            self.rate_import_base, _ = self.rate_replicate(self.rate_import.copy(), {}, is_import=True)  # True import rates, gap-filled but without IO/saving/override distortion
-            self.rate_import, self.rate_import_replicated = self.rate_replicate(self.rate_import, self.io_adjusted, is_import=True)
-            self.rate_import_no_io = self.rate_import.copy()
+            self.rate_import_base, _ = self.rate_replicate(import_rates.copy(), {}, is_import=True)  # True import rates, gap-filled but without IO/saving/override distortion
+            import_rates, self.rate_import_replicated = self.rate_replicate(import_rates, self.io_adjusted, is_import=True)
+            self.rate_import_no_io = import_rates.copy()
             for car_n in range(self.num_cars):
-                self.rate_import = self.rate_add_io_slots(car_n, self.rate_import, self.octopus_slots[car_n])
-            self.load_saving_slot(self.octopus_saving_slots, export=False, rate_replicate=self.rate_import_replicated)
-            self.load_free_slot(self.octopus_free_slots, export=False, rate_replicate=self.rate_import_replicated)
-            load_axle_slot(self, self.axle_sessions, export=False, rate_replicate=self.rate_import_replicated)
-            self.rate_import = self.basic_rates(self.get_arg("rates_import_override", [], indirect=False), "rates_import_override", self.rate_import, self.rate_import_replicated)
-            self.rate_import = self.apply_manual_rates(self.rate_import, self.manual_import_rates, is_import=True, rate_replicate=self.rate_import_replicated)
-            self.rate_scan(self.rate_import, print=True)
+                import_rates = self.rate_add_io_slots(car_n, import_rates, self.octopus_slots[car_n])
+            self.load_saving_slot(self.octopus_saving_slots, import_rates, export=False, rate_replicate=self.rate_import_replicated)
+            self.load_free_slot(self.octopus_free_slots, import_rates, export=False, rate_replicate=self.rate_import_replicated)
+            load_axle_slot(self, self.axle_sessions, import_rates, export=False, rate_replicate=self.rate_import_replicated)
+            import_rates = self.basic_rates(self.get_arg("rates_import_override", [], indirect=False), "rates_import_override", import_rates, self.rate_import_replicated)
+            import_rates = self.apply_manual_rates(import_rates, self.manual_import_rates, is_import=True, rate_replicate=self.rate_import_replicated)
+            self.rate_scan(import_rates, print=True)
         else:
             self.rate_import_no_io = {}
             self.log("Warning: No import rate data provided")
             self.record_status(message="Error: No import rate data provided", had_errors=True)
+        # Atomic publish: readers (e.g. async components) never see a half-built or empty rate_import.
+        self.rate_import = import_rates
 
         # Replicate and scan export rates
-        if self.rate_export:
-            self.rate_scan_export(self.rate_export, print=False)
-            self.rate_export, self.rate_export_replicated = self.rate_replicate(self.rate_export, is_import=False)
-            self.rate_export_base = self.rate_export.copy()
+        if export_rates:
+            self.rate_scan_export(export_rates, print=False)
+            export_rates, self.rate_export_replicated = self.rate_replicate(export_rates, is_import=False)
+            self.rate_export_base = export_rates.copy()
             # For export tariff only load the saving session if enabled
             if self.rate_export_max > 0:
-                self.load_saving_slot(self.octopus_saving_slots, export=True, rate_replicate=self.rate_export_replicated)
-            load_axle_slot(self, self.axle_sessions, export=True, rate_replicate=self.rate_export_replicated)
-            self.rate_export = self.basic_rates(self.get_arg("rates_export_override", [], indirect=False), "rates_export_override", self.rate_export, self.rate_export_replicated)
-            self.rate_export = self.apply_manual_rates(self.rate_export, self.manual_export_rates, is_import=False, rate_replicate=self.rate_export_replicated)
-            self.rate_scan_export(self.rate_export, print=True)
+                self.load_saving_slot(self.octopus_saving_slots, export_rates, export=True, rate_replicate=self.rate_export_replicated)
+            load_axle_slot(self, self.axle_sessions, export_rates, export=True, rate_replicate=self.rate_export_replicated)
+            export_rates = self.basic_rates(self.get_arg("rates_export_override", [], indirect=False), "rates_export_override", export_rates, self.rate_export_replicated)
+            export_rates = self.apply_manual_rates(export_rates, self.manual_export_rates, is_import=False, rate_replicate=self.rate_export_replicated)
+            self.rate_scan_export(export_rates, print=True)
         else:
             self.log("Warning: No export rate data provided")
             self.record_status(message="Error: No export rate data provided", had_errors=True)
+        self.rate_export = export_rates
 
         # Set rate thresholds
         if self.rate_import or self.rate_export:

apps/predbat/octopus.py

@@ -2405,9 +2405,9 @@ def _parse_slot_time(self, value):
         except (ValueError, TypeError):
             return None
 
-    def load_free_slot(self, octopus_free_slots, export=False, rate_replicate=None):
+    def load_free_slot(self, octopus_free_slots, rate_dict, export=False, rate_replicate=None):
         """
-        Load octopus free session slot
+        Load octopus free session slot into rate_dict (in place)
         """
         if rate_replicate is None:
             rate_replicate = {}
@@ -2436,15 +2436,15 @@ def load_free_slot(self, octopus_free_slots, export=False, rate_replicate=None):
                 self.log("Setting Octopus free session in range {} - {} export {} rate {}".format(self.time_abs_str(start_minutes), self.time_abs_str(end_minutes), export, rate))
                 for minute in range(start_minutes, end_minutes):
                     if export:
-                        self.rate_export[minute] = rate
+                        rate_dict[minute] = rate
                     else:
-                        self.rate_import[minute] = min(rate, self.rate_import[minute])
+                        rate_dict[minute] = min(rate, rate_dict[minute])
                         self.load_scaling_dynamic[minute] = self.load_scaling_free
                     rate_replicate[minute] = "saving"
 
-    def load_saving_slot(self, octopus_saving_slots, export=False, rate_replicate=None):
+    def load_saving_slot(self, octopus_saving_slots, rate_dict, export=False, rate_replicate=None):
         """
-        Load octopus saving session slot
+        Load octopus saving session slot into rate_dict (in place)
         """
         if rate_replicate is None:
             rate_replicate = {}
@@ -2476,15 +2476,11 @@ def load_saving_slot(self, octopus_saving_slots, export=False, rate_replicate=No
             if start_minutes < (self.forecast_minutes + self.minutes_now):
                 self.log("Octopus: Setting Octopus saving session in range {} - {} export {} rate {}".format(self.time_abs_str(start_minutes), self.time_abs_str(end_minutes), export, rate))
                 for minute in range(start_minutes, end_minutes):
-                    if export:
-                        if minute in self.rate_export:
-                            self.rate_export[minute] += rate
-                            rate_replicate[minute] = "saving"
-                    else:
-                        if minute in self.rate_import:
-                            self.rate_import[minute] += rate
+                    if minute in rate_dict:
+                        rate_dict[minute] += rate
+                        rate_replicate[minute] = "saving"
+                        if not export:
                             self.load_scaling_dynamic[minute] = self.load_scaling_saving
-                            rate_replicate[minute] = "saving"
 
     def decode_octopus_slot(self, car_n, slot, raw=False):
         """

apps/predbat/tests/test_axle.py

@@ -1072,7 +1072,7 @@ def get_arg(self, name, indirect=True):
 
     # Load the Axle export slot
     rate_replicate = {}
-    load_axle_slot(base, axle_sessions, export=True, rate_replicate=rate_replicate)
+    load_axle_slot(base, axle_sessions, base.rate_export, export=True, rate_replicate=rate_replicate)
 
     # Verify rates were increased by 100p/kWh during the event period
     for minute in range(start_minutes, end_minutes):
@@ -1155,7 +1155,7 @@ def get_arg(self, name, indirect=True):
     original_rate = base.rate_import[start_minutes]
 
     rate_replicate = {}
-    load_axle_slot(base, axle_sessions, export=False, rate_replicate=rate_replicate)
+    load_axle_slot(base, axle_sessions, base.rate_import, export=False, rate_replicate=rate_replicate)
 
     # Verify import rates were decreased by pence_per_kwh during the event period
     for minute in range(start_minutes, end_minutes):