From ab02a502f34f5e7a7687540b513009e65f1bd1cd Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 6 Aug 2026 17:46:51 +0100 Subject: [PATCH] feat(lattice): wire durable auto config runtime --- apps/predbat/gateway.py | 110 +++- apps/predbat/gecloud.py | 84 ++- apps/predbat/lattice_autoconfig.py | 35 +- apps/predbat/lattice_autoconfig_runtime.py | 368 ++++++++++++ apps/predbat/lattice_compiled_publication.py | 17 +- apps/predbat/lattice_durable_storage.py | 562 ++++++++++++++++++ apps/predbat/lattice_fragment_adapters.py | 10 +- apps/predbat/lattice_gateway_fragment.py | 73 ++- apps/predbat/lattice_ge_cloud_fragment.py | 77 ++- apps/predbat/lattice_generated_overlay.py | 109 ++++ apps/predbat/lattice_provider_inputs.py | 335 +++++++++++ apps/predbat/predbat.py | 31 + apps/predbat/tests/test_gateway.py | 243 +++++++- .../tests/test_gecloud_lattice_wiring.py | 184 ++++++ apps/predbat/tests/test_lattice_autoconfig.py | 33 + .../tests/test_lattice_autoconfig_runtime.py | 314 ++++++++++ .../test_lattice_compiled_publication.py | 21 +- .../tests/test_lattice_durable_storage.py | 468 +++++++++++++++ .../tests/test_lattice_gateway_fragment.py | 22 + .../test_lattice_gateway_ge_composition.py | 77 +++ .../tests/test_lattice_ge_cloud_fragment.py | 38 ++ .../tests/test_lattice_generated_overlay.py | 179 ++++++ .../tests/test_lattice_provider_inputs.py | 215 +++++++ apps/predbat/userinterface.py | 41 +- 24 files changed, 3628 insertions(+), 18 deletions(-) create mode 100644 apps/predbat/lattice_autoconfig_runtime.py create mode 100644 apps/predbat/lattice_durable_storage.py create mode 100644 apps/predbat/lattice_generated_overlay.py create mode 100644 apps/predbat/lattice_provider_inputs.py create mode 100644 apps/predbat/tests/test_gecloud_lattice_wiring.py create mode 100644 apps/predbat/tests/test_lattice_autoconfig_runtime.py create mode 100644 apps/predbat/tests/test_lattice_durable_storage.py create mode 100644 apps/predbat/tests/test_lattice_generated_overlay.py create mode 100644 apps/predbat/tests/test_lattice_provider_inputs.py diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index eca02b549..85b54a09e 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -1,3 +1,4 @@ +# cspell:ignore autoconfig """ESP32 Gateway MQTT component. Provides full inverter telemetry and control via the ESP32 gateway's @@ -253,6 +254,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt self._loop = None self._gateway_online = False self._last_telemetry_time = 0 + self._lattice_gateway_stale = False self._last_plan_data = None self._last_plan_publish_time = 0 # Entries and timezone of the last built plan, kept so the periodic re-publish @@ -576,6 +578,8 @@ async def run(self, seconds, first): self.log("Info: GatewayMQTT: Restarting MQTT listener task") self._mqtt_task = asyncio.ensure_future(self._mqtt_loop()) + await self._invalidate_stale_lattice_gateway() + # Publish any queued plan from on_plan_executed hook if self._pending_plan: plan_entries, tz = self._pending_plan @@ -687,7 +691,14 @@ async def _handle_message(self, message): try: if topic == self.topic_status: - self._process_telemetry(message.payload) + lattice_runtime = self._active_lattice_runtime() + if lattice_runtime is None: + self._process_telemetry(message.payload) + else: + await self._process_lattice_telemetry( + message.payload, + lattice_runtime, + ) elif topic == self.topic_online: payload = message.payload.decode("utf-8", errors="replace").strip() was_online = self._gateway_online @@ -700,11 +711,108 @@ async def _handle_message(self, message): attributes=GATEWAY_ATTRIBUTE_TABLE.get("gateway_online", {}), app="gateway", ) + lattice_runtime = self._active_lattice_runtime() + if lattice_runtime is not None: + await self._set_lattice_gateway_liveness( + lattice_runtime, + self._gateway_online, + ) except Exception as e: self._error_count += 1 self.log(f"Warn: GatewayMQTT: Error handling message on {topic}: {e}") self.log(f"Warn: {traceback.format_exc()}") + def _active_lattice_runtime(self): + """Return the enabled live Lattice runtime, if one is installed.""" + runtime = getattr(self.base, "lattice_autoconfig_runtime", None) + if runtime is None or getattr(runtime, "enabled", False) is not True: + return None + if runtime.provider_active("predbat-gateway") is not True: + return None + return runtime + + async def stop(self): + """Invalidate the long-lived provider before a component restart.""" + runtime = self._active_lattice_runtime() + if runtime is not None: + await self._set_lattice_gateway_liveness(runtime, False) + await super().stop() + + async def _set_lattice_gateway_liveness(self, runtime, online): + """Publish Gateway liveness without blocking the MQTT event loop.""" + try: + await asyncio.to_thread(runtime.set_gateway_liveness, online) + return True + except Exception as e: + self._error_count += 1 + self.log(f"Warn: GatewayMQTT: Lattice liveness update failed: {e}") + return False + + async def _invalidate_stale_lattice_gateway(self): + """Invalidate one broker-connected provider after telemetry goes stale.""" + runtime = self._active_lattice_runtime() + stale = self._last_telemetry_time and (time.time() - self._last_telemetry_time) >= _TELEMETRY_STALE_THRESHOLD + if runtime is None or not stale or self._lattice_gateway_stale: + return False + invalidated = await self._set_lattice_gateway_liveness(runtime, False) + if invalidated: + self._lattice_gateway_stale = True + return invalidated + + def _synchronize_lattice_gateway_plan(self, plan): + """Install the pure plan's stable serial selection into Gateway state.""" + selected_serials = tuple(getattr(plan, "selected_serials", ())) + discovered_serials = tuple(getattr(plan, "discovered_serials", ())) + if not selected_serials: + raise ValueError("Lattice Gateway plan selected no inverters") + suffix_to_serial = {_serial_suffix(serial): serial for serial in selected_serials} + if len(suffix_to_serial) != len(selected_serials): + raise ValueError("Lattice Gateway plan contains colliding serial suffixes") + self._suffix_to_serial = suffix_to_serial + self._configured_inverter_serials = frozenset(discovered_serials) + self._configured_ev_chargers = frozenset() + self._auto_configured = True + + async def _process_lattice_telemetry(self, data, runtime): + """Decode, publish, and activate one Lattice-managed Gateway status.""" + try: + status = pb.GatewayStatus() + status.ParseFromString(data) + except Exception as e: + self._error_count += 1 + self.log(f"Warn: GatewayMQTT: Failed to decode telemetry: {e}") + return + + self._debug_dump("RX telemetry", status, raw=data) + if len(status.inverters) == 0: + return + + self._last_status = status + self._last_telemetry_time = time.time() + self.update_success_timestamp() + + try: + plan, result = await asyncio.to_thread( + runtime.ingest_gateway_status, + status, + prefix=self.prefix, + serial_filter=tuple(self.gateway_inverter_serial), + ) + if plan is None or not getattr(result, "accepted", False): + raise ValueError("Lattice runtime did not accept the Gateway status") + self._synchronize_lattice_gateway_plan(plan) + self._inject_entities(status) + self._lattice_gateway_stale = False + except Exception as e: + self._auto_configured = False + self.log(f"Warn: GatewayMQTT: Lattice auto-config rejected Gateway telemetry: {e}") + await self._set_lattice_gateway_liveness(runtime, False) + return + + if not self.api_started: + self.api_started = True + self.log("Info: GatewayMQTT: First telemetry published to Lattice and auto-config complete, API started") + def _debug_dump(self, label, message=None, raw=None, message_type=None): """Log a protobuf message as readable text when debug logging is enabled. diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index 8a7beab26..e29b1fc3c 100644 --- a/apps/predbat/gecloud.py +++ b/apps/predbat/gecloud.py @@ -1,3 +1,4 @@ +# cspell:ignore autoconfig # ----------------------------------------------------------------------------- # Predbat Home Battery System # Copyright Trefor Southwell 2026 - All Rights Reserved @@ -821,6 +822,65 @@ async def publish_registers(self, device, registers, select_key=None): self.dashboard_item(entity_id, state="on" if state else "off", attributes=attributes, app="gecloud") self.register_entity_map[entity_id] = {"device": device, "key": key} + def _lattice_autoconfig_runtime(self): + """Return the enabled runtime only for GE automatic configuration.""" + if not self.automatic: + return None + runtime = getattr(self.base, "lattice_autoconfig_runtime", None) + if runtime is None or getattr(runtime, "enabled", False) is not True: + return None + if runtime.provider_active("ge-cloud") is not True: + return None + return runtime + + async def stop(self): + """Invalidate the long-lived provider before a component restart.""" + runtime = self._lattice_autoconfig_runtime() + if runtime is not None: + await self._set_lattice_liveness(False) + await super().stop() + + async def _refresh_automatic_config(self): + """Select exactly one legacy or Lattice automatic-config writer.""" + if not self.automatic: + return False + runtime = self._lattice_autoconfig_runtime() + if runtime is None: + await self.async_automatic_config(self.devices_dict) + return True + await asyncio.to_thread( + runtime.ingest_gecloud_state, + self.devices_dict, + self.settings, + self.info, + prefix=self.prefix, + load_today_ignore=self.get_arg( + "ge_cloud_load_today_ignore", + default=False, + ), + split_pv=self.get_arg( + "ge_cloud_automatic_split_pv", + default=False, + ), + split_ct=self.get_arg( + "ge_cloud_automatic_split_ct", + default=False, + ), + shared_ct=self.get_arg( + "ge_cloud_automatic_shared_ct", + default=False, + ), + ) + return True + + async def _set_lattice_liveness(self, health): + """Publish one bounded cloud-health transition when Lattice is active.""" + runtime = self._lattice_autoconfig_runtime() + if runtime is None: + return False + await asyncio.to_thread(runtime.set_gecloud_liveness, health) + return True + async def async_automatic_config(self, devices): """ Automatically configure predbat using GE Cloud auto-detected devices. @@ -1109,7 +1169,12 @@ async def run(self, seconds, first): else: self.log("GECloud: No valid settings found in storage cache, will poll") + lattice_runtime = self._lattice_autoconfig_runtime() + lattice_refresh_due = False + lattice_health = "unchanged" + if first or (seconds % 120 == 0): + lattice_success_before = self.last_success_timestamp if lattice_runtime is not None else None inverter_auth_denied = False for device in self.device_list: self.status[device] = await self.async_get_inverter_status(device, self.status.get(device, {})) @@ -1123,6 +1188,15 @@ async def run(self, seconds, first): self.info[device] = await self.async_get_device_info(device, self.info.get(device, {})) await self.publish_info(device, self.info[device]) + if lattice_runtime is not None: + if inverter_auth_denied: + lattice_health = False + elif self.last_success_timestamp != lattice_success_before: + lattice_health = True + lattice_refresh_due = True + else: + lattice_health = None + # Surface a clear, correct status when the GivEnergy cloud API denied access to the core # inverter data, rather than letting stale data be misdiagnosed downstream (e.g. as # inverter clock skew). Scoped to inverter — not EVC — auth failures, and reported only @@ -1163,7 +1237,9 @@ async def run(self, seconds, first): # One shot tasks if first: if self.automatic: - await self.async_automatic_config(self.devices_dict) + if lattice_runtime is None or lattice_health is True: + await self._refresh_automatic_config() + lattice_refresh_due = False now_utc = self.now_utc_exact options_due = self.default_options_stamp is None or (now_utc - self.default_options_stamp) >= timedelta(hours=24) @@ -1172,6 +1248,12 @@ async def run(self, seconds, first): for device in self.device_list: await self.enable_default_options(device, self.settings[device]) + if lattice_runtime is not None: + if lattice_refresh_due and lattice_health is not False: + await self._refresh_automatic_config() + if lattice_health != "unchanged" and lattice_health is not True: + await self._set_lattice_liveness(lattice_health) + # Clear pending writes for device in self.device_list: if device in self.pending_writes: diff --git a/apps/predbat/lattice_autoconfig.py b/apps/predbat/lattice_autoconfig.py index f89f9a928..095b8cfa1 100644 --- a/apps/predbat/lattice_autoconfig.py +++ b/apps/predbat/lattice_autoconfig.py @@ -1449,8 +1449,14 @@ def _field_provenance( return tuple(sorted(provenance, key=lambda item: (item.field_path, item.provider_id, item.generation, item.source_path))) -def compile_auto_config(snapshots, user_overrides=()): +def compile_auto_config( + snapshots, + user_overrides=(), + atomic_materializer=False, +): """Compile usable provider snapshots into one deterministic immutable plan.""" + if not isinstance(atomic_materializer, bool): + raise ValueError("atomic_materializer must be a boolean") snapshots = tuple(sorted(snapshots, key=lambda item: item.provider_id)) if not snapshots: raise AutoConfigCompileError("no usable provider snapshots") @@ -1516,9 +1522,11 @@ def compile_auto_config(snapshots, user_overrides=()): user_overrides, ) if config_arguments: - blockers = tuple(blocker for blocker in readiness.blockers if blocker != "config_projection_bindings_missing") + ("atomic_materializer_missing",) + blockers = tuple(blocker for blocker in readiness.blockers if blocker != "config_projection_bindings_missing") + if not atomic_materializer: + blockers += ("atomic_materializer_missing",) readiness = MaterializationReadiness( - ready=False, + ready=not blockers, blockers=blockers, ) @@ -1666,10 +1674,22 @@ def compile_auto_config(snapshots, user_overrides=()): class LatticeAutoConfigCompiler: """Thread-safe invalidation, compilation, and last-known-good coordinator.""" - def __init__(self, readers=None, materializer=None): + def __init__( + self, + readers=None, + materializer=None, + atomic_materializer=False, + allow_provider_failover=False, + ): """Create an idle compiler over provider snapshot readers.""" + if not isinstance(atomic_materializer, bool): + raise ValueError("atomic_materializer must be a boolean") + if not isinstance(allow_provider_failover, bool): + raise ValueError("allow_provider_failover must be a boolean") self._readers = {} self._materializer = materializer + self._atomic_materializer = atomic_materializer + self._allow_provider_failover = allow_provider_failover self._lock = threading.RLock() self._compiling = False self._pending = False @@ -1787,11 +1807,14 @@ def _compile_attempt(self): active_providers = set(dict(self._active_plan.provider_generations)) if self._active_plan is not None else set() usable_providers = {snapshot.provider_id for snapshot in snapshots} unavailable_active = sorted(active_providers - usable_providers) - if unavailable_active: + if unavailable_active and not self._allow_provider_failover: detail = "previously active provider(s) unavailable: {}".format(", ".join(unavailable_active)) return None, issues + (CompileIssue("active_provider_unavailable", detail), CompileIssue("compile_failed", detail)) try: - plan = compile_auto_config(snapshots) + plan = compile_auto_config( + snapshots, + atomic_materializer=self._atomic_materializer, + ) except (AutoConfigCompileError, TopologyValidationError, TypeError, ValueError) as exc: return None, issues + (CompileIssue("compile_failed", str(exc)),) return plan, issues diff --git a/apps/predbat/lattice_autoconfig_runtime.py b/apps/predbat/lattice_autoconfig_runtime.py new file mode 100644 index 000000000..7a77205c6 --- /dev/null +++ b/apps/predbat/lattice_autoconfig_runtime.py @@ -0,0 +1,368 @@ +# cspell:ignore XIAO +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - live Lattice automatic configuration runtime +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Default-off runtime coordinator for durable generated configuration.""" + +# cspell:ignore autoconfig + +import threading +from dataclasses import dataclass + +from gateway_autoconfig import compile_gateway_auto_config +from lattice_autoconfig import CompileStatus, ProviderHealth, _plain +from lattice_durable_storage import ( + PredBatCompiledLatticeStateStore, + PredBatFragmentAdapterStateStore, +) +from lattice_fragment_adapters import ( + FragmentAdapterConflict, + FragmentAdapterReadError, + FragmentAdapterRegistry, +) +from lattice_gateway_fragment import GatewayRetainedTopologyFragmentPublisher +from lattice_ge_cloud_fragment import ( + GECloudDeviceSnapshot, + GECloudFragmentPublisher, +) +from lattice_generated_overlay import GeneratedConfigOverlay +from lattice_provider_inputs import ( + gateway_status_to_lattice_input, + gecloud_state_to_lattice_input, +) + + +@dataclass(frozen=True) +class LatticeRuntimeResult: + """Observable result of one provider-triggered reconciliation.""" + + accepted: bool + staged: bool + digest: object + status: object + issues: tuple + + +class LatticeAutoConfigRuntime: + """Own durable publishers, compilation, and one generated config overlay.""" + + GATEWAY_PROVIDER_ID = "predbat-gateway" + GECLOUD_PROVIDER_ID = "ge-cloud" + + def __init__(self, base, storage, enabled=False): + """Create an unwired runtime; disabled construction has no durable writes.""" + if not isinstance(enabled, bool): + raise ValueError("enabled must be a boolean") + self.base = base + self.log = base.log + self.enabled = enabled + self.overlay = GeneratedConfigOverlay( + authoritative_arguments=("inverter_hybrid",), + ) + self.base.lattice_generated_overlay = self.overlay + self._lock = threading.RLock() + self._registry = None + self._compiler = None + self._gateway = None + self._gecloud = None + self._compiled_store = None + self._pending = None + self._pending_digest = None + self._pending_disable = False + self._applied_digest = None + self._provider_ids = frozenset() + self._fresh_providers = set() + if not enabled: + return + if storage is None: + raise ValueError("enabled Lattice auto-config requires storage") + + self._gateway = GatewayRetainedTopologyFragmentPublisher( + self.GATEWAY_PROVIDER_ID, + PredBatFragmentAdapterStateStore(storage, self.GATEWAY_PROVIDER_ID), + enabled=True, + ) + self._gecloud = GECloudFragmentPublisher( + self.GECLOUD_PROVIDER_ID, + PredBatFragmentAdapterStateStore(storage, self.GECLOUD_PROVIDER_ID), + enabled=True, + ) + self._compiled_store = PredBatCompiledLatticeStateStore(storage) + self._seed_or_revalidate_publishers() + + @property + def gateway_publisher(self): + """Return the long-lived Gateway publisher shared across restarts.""" + return self._gateway if self.enabled else None + + @property + def gecloud_publisher(self): + """Return the long-lived GE Cloud publisher shared across restarts.""" + return self._gecloud if self.enabled else None + + @property + def compiler(self): + """Return the bound compiler for diagnostics.""" + return self._compiler + + def _seed_or_revalidate_publishers(self): + """Ensure registry-visible offline state without trusting restart health.""" + empty_gateway = { + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": 0, + "producer": { + "name": "PredBat Gateway", + "provider": self.GATEWAY_PROVIDER_ID, + "authority": 10, + }, + "nodes": [], + } + try: + gateway_state = self._gateway.read_state() + except FragmentAdapterReadError: + self._gateway.ingest_retained_topology(empty_gateway, online=False) + else: + if gateway_state.snapshot.health is not ProviderHealth.OFFLINE: + self._gateway.set_liveness(False) + + try: + ge_state = self._gecloud.read_state() + except FragmentAdapterReadError: + self._gecloud.ingest_discovery( + 0, + ( + GECloudDeviceSnapshot( + serial="LATTICE-PENDING", + kind="inverter", + online=False, + ), + ), + health=False, + ) + else: + if ge_state.snapshot.health is not ProviderHealth.OFFLINE: + self._gecloud.set_liveness(False) + + def bind(self, gateway_enabled=False, gecloud_enabled=False): + """Freeze the active provider set before phase-one components start.""" + if not self.enabled: + return () + if not isinstance(gateway_enabled, bool) or not isinstance( + gecloud_enabled, + bool, + ): + raise ValueError("provider activation flags must be booleans") + with self._lock: + if self._compiler is not None: + raise RuntimeError("Lattice auto-config runtime is already bound") + registry = FragmentAdapterRegistry(enabled=True) + provider_ids = [] + if gateway_enabled: + registry.register(self._gateway) + provider_ids.append(self.GATEWAY_PROVIDER_ID) + if gecloud_enabled: + registry.register(self._gecloud) + provider_ids.append(self.GECLOUD_PROVIDER_ID) + if not provider_ids: + self._registry = registry + return () + self._compiler = registry.create_compiler( + self._compiled_store, + atomic_materializer=True, + allow_provider_failover=True, + ) + self._registry = registry + self._provider_ids = frozenset(provider_ids) + return tuple(provider_ids) + + def provider_active(self, provider_id): + """Return whether one provider belongs to this frozen runtime binding.""" + return self.enabled and provider_id in self._provider_ids + + def _invalidate_pending(self): + """Cancel stale queued config and stage removal of an active overlay.""" + changed = self._pending is not None + self._pending = None + self._pending_digest = None + if self.overlay.snapshot.enabled: + changed = changed or not self._pending_disable + self._pending_disable = True + else: + self._pending_disable = False + if changed: + self.base.update_pending = True + self.base.plan_valid = False + return changed + + def _stage_publication(self, publication): + """Queue one complete immutable config replacement by digest.""" + if publication is None: + return False + plan = publication.plan + if not plan.materialization_readiness.ready: + return False + if plan.digest == self._applied_digest: + self._pending = None + self._pending_digest = None + self._pending_disable = False + return False + if plan.digest == self._pending_digest and not self._pending_disable: + return False + self._pending = _plain(plan.projected_config) + self._pending_digest = plan.digest + self._pending_disable = False + self.base.update_pending = True + self.base.plan_valid = False + return True + + def reconcile(self): + """Drain invalidations and queue only a fresh durable publication.""" + if not self.enabled or self._compiler is None: + return LatticeRuntimeResult(False, False, None, None, ()) + with self._lock: + run = self._compiler.drain() + publication = run.publication + accepted = publication is not None and run.status in (CompileStatus.FRESH, CompileStatus.DEGRADED) and publication.plan.materialization_readiness.ready + staged = self._stage_publication(publication) if accepted else self._invalidate_pending() + return LatticeRuntimeResult( + accepted, + staged, + publication.digest if publication is not None else None, + run.status, + tuple(run.issues), + ) + + def apply_pending(self): + """Atomically apply one queued full replacement at a config boundary.""" + if not self.enabled: + return False + with self._lock: + if self._pending_disable: + self.overlay.disable() + self._pending_disable = False + self._pending = None + self._pending_digest = None + self._applied_digest = None + self.base.inverters = [] + self.base.plan_valid = False + return True + if self._pending is None: + return False + pending = self._pending + digest = self._pending_digest + self.overlay.replace(pending) + self._pending = None + self._pending_digest = None + self._pending_disable = False + self._applied_digest = digest + self.base.inverters = [] + self.base.plan_valid = False + return True + + def disable(self): + """Remove all generated values without mutating explicit arguments.""" + with self._lock: + changed = self.overlay.snapshot.enabled or self._pending is not None or self._pending_disable + self._pending = None + self._pending_digest = None + self._pending_disable = False + self._applied_digest = None + self.overlay.disable() + if changed: + self.base.inverters = [] + self.base.update_pending = True + self.base.plan_valid = False + return changed + + def ingest_gateway_status(self, status, prefix="predbat", serial_filter=()): + """Publish one complete XIAO/Gateway telemetry-derived fragment.""" + if not self.enabled or self._compiler is None: + return None, LatticeRuntimeResult(False, False, None, None, ()) + live = gateway_status_to_lattice_input( + status, + provider_id=self.GATEWAY_PROVIDER_ID, + ) + current = self._gateway.read_state() + previous = _plain(current.snapshot.topology_fragment) + previous_version = int(previous.get("docVersion", 0)) + candidate = dict(live.topology_fragment) + previous_without_version = dict(previous) + previous_without_version.pop("docVersion", None) + document_version = previous_version if candidate == previous_without_version else previous_version + 1 + candidate["docVersion"] = document_version + plan = compile_gateway_auto_config( + live.inverters, + prefix=prefix, + serial_filter=serial_filter, + ) + self._gateway.ingest_complete(candidate, plan, online=True) + self._fresh_providers.add(self.GATEWAY_PROVIDER_ID) + return plan, self.reconcile() + + def set_gateway_liveness(self, online): + """Publish Gateway liveness and reconcile local/cloud selection.""" + if not self.enabled or self._compiler is None: + return LatticeRuntimeResult(False, False, None, None, ()) + if not online: + with self._lock: + self._fresh_providers.discard(self.GATEWAY_PROVIDER_ID) + # Cancel the in-memory candidate before the durable write. If + # persistence fails, a rejected telemetry batch still cannot be + # activated at the next configuration boundary. + self._invalidate_pending() + if online and self.GATEWAY_PROVIDER_ID not in self._fresh_providers: + return self.reconcile() + self._gateway.set_liveness(online) + return self.reconcile() + + def ingest_gecloud_state( + self, + devices, + settings, + info, + prefix="predbat", + **flags, + ): + """Publish one complete GE discovery/config state and reconcile.""" + if not self.enabled or self._compiler is None: + return None, LatticeRuntimeResult(False, False, None, None, ()) + live = gecloud_state_to_lattice_input( + devices, + settings, + info, + prefix=prefix, + **flags, + ) + version = self._gecloud.discovery_version + try: + self._gecloud.ingest_complete( + version, + live.devices, + live.auto_config, + health=True, + ) + except FragmentAdapterConflict: + self._gecloud.ingest_complete( + version + 1, + live.devices, + live.auto_config, + health=True, + ) + self._fresh_providers.add(self.GECLOUD_PROVIDER_ID) + return live, self.reconcile() + + def set_gecloud_liveness(self, health): + """Publish GE Cloud health and reconcile local/cloud selection.""" + if not self.enabled or self._compiler is None: + return LatticeRuntimeResult(False, False, None, None, ()) + if health is False: + with self._lock: + self._fresh_providers.discard(self.GECLOUD_PROVIDER_ID) + self._invalidate_pending() + if health is not False and self.GECLOUD_PROVIDER_ID not in self._fresh_providers: + return self.reconcile() + self._gecloud.set_liveness(health) + return self.reconcile() diff --git a/apps/predbat/lattice_compiled_publication.py b/apps/predbat/lattice_compiled_publication.py index 004ff0c27..6409c4c57 100644 --- a/apps/predbat/lattice_compiled_publication.py +++ b/apps/predbat/lattice_compiled_publication.py @@ -353,6 +353,8 @@ def __init__( readers=None, state_store=None, override_reader=None, + atomic_materializer=False, + allow_provider_failover=False, ): """Create a publisher over registered fragment and override readers.""" if state_store is None: @@ -362,7 +364,12 @@ def __init__( if override_reader is not None and not callable(override_reader): raise ValueError("override_reader must be callable") - super().__init__(readers=readers, materializer=None) + super().__init__( + readers=readers, + materializer=None, + atomic_materializer=atomic_materializer, + allow_provider_failover=allow_provider_failover, + ) self._state_store = state_store self._override_reader = override_reader self._override_requested_generation = -1 @@ -540,7 +547,7 @@ def _compile_attempt(self): active_providers = set(dict(self._active_plan.provider_generations)) if self._active_plan is not None else set() usable_providers = {snapshot.provider_id for snapshot in snapshots} unavailable_active = sorted(active_providers - usable_providers) - if unavailable_active: + if unavailable_active and not self._allow_provider_failover: detail = "previously active provider(s) unavailable: {}".format(", ".join(unavailable_active)) self._candidate_issues = issues + ( CompileIssue("active_provider_unavailable", detail), @@ -549,7 +556,11 @@ def _compile_attempt(self): return None, self._candidate_issues try: - plan = compile_auto_config(snapshots, overrides) + plan = compile_auto_config( + snapshots, + overrides, + atomic_materializer=self._atomic_materializer, + ) except ( AutoConfigCompileError, TopologyValidationError, diff --git a/apps/predbat/lattice_durable_storage.py b/apps/predbat/lattice_durable_storage.py new file mode 100644 index 000000000..7c0725ef7 --- /dev/null +++ b/apps/predbat/lattice_durable_storage.py @@ -0,0 +1,562 @@ +# cspell:ignore journalled +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - durable Lattice state storage +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Crash-tolerant synchronous stores for durable Lattice state. + +The compiler and fragment publishers intentionally expose synchronous store +contracts. PredBat's shared storage component is asynchronous, so callers must +run these stores on a worker thread rather than an active asyncio event loop. + +Each logical value is journalled across two non-expiring JSON slots. A write +replaces only the inactive slot and is accepted only after an exact read-back, +leaving the previous valid slot available after an interrupted storage write. +""" + +# cspell:ignore autoconfig checksummed fsync + +import asyncio +import hashlib +import json +import math +import os +import threading +from dataclasses import fields, is_dataclass +from enum import Enum +from types import MappingProxyType +from typing import Mapping + +from lattice_autoconfig import ( + AliasBinding, + AliasRole, + AutoConfigField, + AutoConfigPlan, + CompileIssue, + CompileStatus, + FieldProvenance, + IdentityBinding, + IndexedRoleTarget, + MaterializationReadiness, + ProjectedConfigArgument, + ProjectionCandidate, + ProjectionCardinality, + ProjectionRouting, + ProjectionValueKind, + ProviderAlias, + ProviderConfigProjection, + ProviderHealth, + ProviderIdentityAlias, + ProviderProjectionValue, + ProviderRoleAssignment, + ProviderSnapshot, + RoleAssignmentBinding, + UserConfigOverride, +) +from lattice_compiled_publication import ( + CompilationInvalidation, + CompiledLatticeDiagnostics, + CompiledLatticePublication, + CompiledLatticeStateStore, +) +from lattice_fragment_adapters import FragmentAdapterState, FragmentAdapterStateStore + + +_STORAGE_MODULE = "lattice_autoconfig" +_SCHEMA = "predbat-lattice-state/v1" +_LOCKS_GUARD = threading.Lock() +_LOCKS = {} + + +class LatticeDurableStorageError(RuntimeError): + """Base error for durable Lattice persistence.""" + + +class LatticeDurableReadError(LatticeDurableStorageError): + """Persisted Lattice state is unavailable or invalid.""" + + +class LatticeDurableWriteError(LatticeDurableStorageError): + """A durable Lattice journal write could not be verified.""" + + +_DATACLASS_TYPES = { + cls.__name__: cls + for cls in ( + ProviderAlias, + ProviderIdentityAlias, + ProviderRoleAssignment, + ProviderProjectionValue, + ProviderConfigProjection, + UserConfigOverride, + ProviderSnapshot, + AliasBinding, + IdentityBinding, + RoleAssignmentBinding, + IndexedRoleTarget, + MaterializationReadiness, + FieldProvenance, + AutoConfigField, + ProjectionCandidate, + ProjectedConfigArgument, + AutoConfigPlan, + CompileIssue, + CompilationInvalidation, + CompiledLatticeDiagnostics, + CompiledLatticePublication, + FragmentAdapterState, + ) +} +_ENUM_TYPES = { + cls.__name__: cls + for cls in ( + ProviderHealth, + AliasRole, + ProjectionValueKind, + ProjectionRouting, + ProjectionCardinality, + CompileStatus, + ) +} + + +def _canonical_json(value): + """Return deterministic strict JSON for checksums and persisted values.""" + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + + +def _encode_node(value): + """Encode only explicitly supported immutable Lattice value types.""" + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError("non-finite floats cannot be persisted") + return value + if isinstance(value, Enum): + enum_name = type(value).__name__ + if _ENUM_TYPES.get(enum_name) is not type(value): + raise ValueError("unsupported enum {}".format(enum_name)) + return {"type": "enum", "name": enum_name, "value": value.value} + if is_dataclass(value) and not isinstance(value, type): + class_name = type(value).__name__ + if _DATACLASS_TYPES.get(class_name) is not type(value): + raise ValueError("unsupported dataclass {}".format(class_name)) + return { + "type": "dataclass", + "name": class_name, + "fields": {field.name: _encode_node(getattr(value, field.name)) for field in fields(value)}, + } + if isinstance(value, Mapping): + items = [] + for key, item in value.items(): + if not isinstance(key, str): + raise ValueError("persisted mapping keys must be strings") + items.append([key, _encode_node(item)]) + return {"type": "mapping", "items": sorted(items)} + if isinstance(value, tuple): + return {"type": "tuple", "items": [_encode_node(item) for item in value]} + if isinstance(value, frozenset): + encoded = [_encode_node(item) for item in value] + return { + "type": "frozenset", + "items": sorted(encoded, key=_canonical_json), + } + if isinstance(value, list): + return {"type": "list", "items": [_encode_node(item) for item in value]} + raise ValueError("unsupported persisted value {}".format(type(value).__name__)) + + +def _exact_keys(value, expected, description): + """Reject missing and forward-unknown persistence fields.""" + if not isinstance(value, dict) or set(value) != set(expected): + raise ValueError("{} fields are invalid".format(description)) + + +def _mutable_plain(value): + """Thaw decoded JSON for constructors that defensively deepcopy inputs.""" + if isinstance(value, Mapping): + return {key: _mutable_plain(item) for key, item in value.items()} + if isinstance(value, (tuple, list)): + return [_mutable_plain(item) for item in value] + if isinstance(value, frozenset): + return {_mutable_plain(item) for item in value} + return value + + +def _decode_node(value): + """Decode one node through the explicit type whitelist.""" + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError("non-finite floats cannot be restored") + return value + if not isinstance(value, dict): + raise ValueError("persisted nodes must be tagged objects or JSON scalars") + node_type = value.get("type") + if node_type == "enum": + _exact_keys(value, ("type", "name", "value"), "enum") + enum_type = _ENUM_TYPES.get(value["name"]) + if enum_type is None: + raise ValueError("unsupported persisted enum {}".format(value["name"])) + return enum_type(value["value"]) + if node_type == "dataclass": + _exact_keys(value, ("type", "name", "fields"), "dataclass") + class_type = _DATACLASS_TYPES.get(value["name"]) + if class_type is None: + raise ValueError("unsupported persisted dataclass {}".format(value["name"])) + encoded_fields = value["fields"] + expected_fields = tuple(field.name for field in fields(class_type)) + _exact_keys(encoded_fields, expected_fields, value["name"]) + arguments = {name: _decode_node(encoded_fields[name]) for name in expected_fields} + if class_type is ProviderSnapshot: + arguments["topology_fragment"] = _mutable_plain(arguments["topology_fragment"]) + elif class_type is UserConfigOverride: + arguments["value"] = _mutable_plain(arguments["value"]) + return class_type(**arguments) + if node_type in ("tuple", "list", "frozenset"): + _exact_keys(value, ("type", "items"), node_type) + if not isinstance(value["items"], list): + raise ValueError("{} items must be a list".format(node_type)) + items = tuple(_decode_node(item) for item in value["items"]) + if node_type == "tuple": + return items + if node_type == "frozenset": + return frozenset(items) + return list(items) + if node_type == "mapping": + _exact_keys(value, ("type", "items"), "mapping") + if not isinstance(value["items"], list): + raise ValueError("mapping items must be a list") + decoded = {} + for pair in value["items"]: + if not isinstance(pair, list) or len(pair) != 2 or not isinstance(pair[0], str) or pair[0] in decoded: + raise ValueError("persisted mapping items are invalid") + decoded[pair[0]] = _decode_node(pair[1]) + return MappingProxyType(decoded) + raise ValueError("unsupported persisted node type {!r}".format(node_type)) + + +def encode_fragment_adapter_state(state): + """Encode a fragment state into strict plain JSON data.""" + if state is not None and not isinstance(state, FragmentAdapterState): + raise ValueError("fragment state must be FragmentAdapterState or None") + return _encode_node(state) + + +def decode_fragment_adapter_state(payload): + """Decode and validate a fragment state from plain JSON data.""" + state = _decode_node(payload) + if state is not None and not isinstance(state, FragmentAdapterState): + raise ValueError("persisted fragment payload has the wrong root type") + return state + + +def encode_compiled_lattice_publication(publication): + """Encode one compiled publication into strict plain JSON data.""" + if publication is not None and not isinstance( + publication, + CompiledLatticePublication, + ): + raise ValueError("compiled publication must be CompiledLatticePublication or None") + return _encode_node(publication) + + +def decode_compiled_lattice_publication(payload): + """Decode and validate one compiled publication from plain JSON data.""" + publication = _decode_node(payload) + if publication is not None and not isinstance( + publication, + CompiledLatticePublication, + ): + raise ValueError("persisted compiled payload has the wrong root type") + return publication + + +def _journal_lock(storage, key): + """Return the shared in-process lock for one durable storage key.""" + identity = (id(storage), _STORAGE_MODULE, key) + with _LOCKS_GUARD: + return _LOCKS.setdefault(identity, threading.RLock()) + + +class _TwoSlotJournal: + """Synchronous two-slot journal over PredBat's asynchronous storage API.""" + + def __init__(self, storage, key, kind, encoder, decoder): + """Bind one logical value and verify its persisted state immediately.""" + for method_name in ("load", "save", "age"): + if not callable(getattr(storage, method_name, None)): + raise ValueError("storage must provide async {}".format(method_name)) + if kind not in ("fragment", "compiled"): + raise ValueError("unsupported journal kind") + self._storage = storage + self._key = key + self._kind = kind + self._encoder = encoder + self._decoder = decoder + self._lock = _journal_lock(storage, key) + self.load() + + def _call(self, method_name, *args, **kwargs): + """Run one storage coroutine only from a synchronous worker thread.""" + try: + asyncio.get_running_loop() + except RuntimeError: + pass + else: + raise RuntimeError("durable Lattice storage must run outside an active asyncio event loop") + method = getattr(self._storage, method_name) + return asyncio.run(method(*args, **kwargs)) + + def _slot_name(self, slot): + """Return one deterministic physical journal key.""" + return "{}_{}".format(self._key, slot) + + def _local_slot_exists(self, slot_name): + """Detect an unreadable local slot without weakening storage abstraction.""" + backend = getattr(self._storage, "backend", self._storage) + meta_path = getattr(backend, "_meta_path", None) + data_path = getattr(backend, "_data_path", None) + if not callable(meta_path) or not callable(data_path): + return None + return os.path.exists(meta_path(_STORAGE_MODULE, slot_name)) or os.path.exists(data_path(_STORAGE_MODULE, slot_name, "json")) + + def _checksum(self, envelope): + """Hash all safety-relevant envelope fields except the checksum.""" + content = { + "schema": envelope["schema"], + "kind": envelope["kind"], + "sequence": envelope["sequence"], + "payload": envelope["payload"], + } + return hashlib.sha256(_canonical_json(content).encode("utf-8")).hexdigest() + + def _envelope(self, sequence, value): + """Create one checksummed plain-data envelope.""" + envelope = { + "schema": _SCHEMA, + "kind": self._kind, + "sequence": sequence, + "payload": self._encoder(value), + } + envelope["checksum"] = self._checksum(envelope) + return envelope + + def _decode_envelope(self, raw, slot_name): + """Strictly validate and decode one persisted envelope.""" + _exact_keys( + raw, + ("schema", "kind", "sequence", "payload", "checksum"), + "journal envelope", + ) + if raw["schema"] != _SCHEMA: + raise LatticeDurableReadError("unsupported durable Lattice schema in {}".format(slot_name)) + if raw["kind"] != self._kind: + raise LatticeDurableReadError("durable Lattice kind mismatch in {}".format(slot_name)) + sequence = raw["sequence"] + if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence < 1: + raise LatticeDurableReadError("durable Lattice sequence is invalid in {}".format(slot_name)) + if not isinstance(raw["checksum"], str) or raw["checksum"] != self._checksum(raw): + raise LatticeDurableReadError("durable Lattice checksum mismatch in {}".format(slot_name)) + try: + value = self._decoder(raw["payload"]) + except (TypeError, ValueError) as exc: + raise LatticeDurableReadError( + "durable Lattice payload is invalid in {}: {}".format( + slot_name, + exc, + ) + ) from exc + return sequence, value, raw + + def _read_slot(self, slot): + """Read one slot, distinguishing a missing key from unreadable data.""" + slot_name = self._slot_name(slot) + try: + raw = self._call("load", _STORAGE_MODULE, slot_name) + except Exception as exc: + raise LatticeDurableReadError( + "durable Lattice read failed for {}: {}: {}".format( + slot_name, + type(exc).__name__, + exc, + ) + ) from exc + if raw is None: + try: + age = self._call("age", _STORAGE_MODULE, slot_name) + except Exception as exc: + raise LatticeDurableReadError( + "durable Lattice age check failed for {}: {}: {}".format( + slot_name, + type(exc).__name__, + exc, + ) + ) from exc + if age is None: + exists = self._local_slot_exists(slot_name) + if exists: + raise LatticeDurableReadError( + "durable Lattice slot {} exists but is unreadable".format( + slot_name, + ) + ) + return None + raise LatticeDurableReadError("durable Lattice slot {} exists but is unreadable".format(slot_name)) + if not isinstance(raw, dict): + raise LatticeDurableReadError("durable Lattice envelope in {} must be a mapping".format(slot_name)) + try: + return self._decode_envelope(raw, slot_name) + except LatticeDurableReadError: + raise + except (KeyError, TypeError, ValueError) as exc: + raise LatticeDurableReadError( + "durable Lattice envelope is invalid in {}: {}".format( + slot_name, + exc, + ) + ) from exc + + def _scan(self): + """Return the newest valid slot, tolerating one interrupted inactive slot.""" + valid = [] + errors = [] + for slot in (0, 1): + try: + restored = self._read_slot(slot) + except LatticeDurableReadError as exc: + errors.append(exc) + continue + if restored is not None: + valid.append((restored[0], slot, restored[1], restored[2])) + if not valid: + if errors: + raise errors[0] + return 0, None, None, None + if len(valid) == 2 and valid[0][0] == valid[1][0] and valid[0][3] != valid[1][3]: + raise LatticeDurableReadError("durable Lattice sequence {} was reused with different content".format(valid[0][0])) + return max(valid, key=lambda item: item[0]) + + def load(self): + """Fresh-read the newest verified durable value.""" + with self._lock: + _sequence, _slot, value, _raw = self._scan() + return value + + def compare_and_store(self, expected, replacement): + """Atomically compare, write the inactive slot, and verify the result.""" + with self._lock: + sequence, active_slot, current, _raw = self._scan() + if current != expected: + return False + next_sequence = sequence + 1 + next_slot = 1 if active_slot in (None, 0) else 0 + candidate = self._envelope(next_sequence, replacement) + slot_name = self._slot_name(next_slot) + write_error = None + try: + saved = self._call( + "save", + _STORAGE_MODULE, + slot_name, + candidate, + format="json", + expiry=None, + ) + if saved is not True: + write_error = "storage.save returned {!r}".format(saved) + except Exception as exc: + write_error = "{}: {}".format(type(exc).__name__, exc) + + try: + winner = self._read_slot(next_slot) + except LatticeDurableReadError as exc: + winner = None + if write_error is None: + write_error = str(exc) + if winner is not None and winner[0] == next_sequence and winner[2] == candidate: + return True + raise LatticeDurableWriteError( + "durable Lattice write could not be verified for {}: {}".format( + slot_name, + write_error or "read-back differed from candidate", + ) + ) + + +class PredBatFragmentAdapterStateStore(FragmentAdapterStateStore): + """Production fragment state store backed by PredBat storage.""" + + def __init__(self, storage, provider_id): + """Bind one provider to its collision-resistant durable journal.""" + if not isinstance(provider_id, str) or not provider_id.strip(): + raise ValueError("provider_id must be a non-empty string") + self.provider_id = provider_id.strip() + provider_key = hashlib.sha256(self.provider_id.encode("utf-8")).hexdigest() + self._journal = _TwoSlotJournal( + storage, + "fragment_{}".format(provider_key), + "fragment", + encode_fragment_adapter_state, + decode_fragment_adapter_state, + ) + + def load(self): + """Fresh-read this provider's current durable fragment state.""" + state = self._journal.load() + if state is not None and state.provider_id != self.provider_id: + raise LatticeDurableReadError("durable fragment belongs to provider {}".format(state.provider_id)) + return state + + def compare_and_store(self, expected, replacement): + """Atomically install a complete fragment state or tombstone.""" + for value, description in ( + (expected, "expected"), + (replacement, "replacement"), + ): + if value is not None and not isinstance(value, FragmentAdapterState): + raise ValueError("{} must be FragmentAdapterState or None".format(description)) + if value is not None and value.provider_id != self.provider_id: + raise ValueError("{} belongs to provider {}".format(description, value.provider_id)) + return self._journal.compare_and_store(expected, replacement) + + +class PredBatCompiledLatticeStateStore(CompiledLatticeStateStore): + """Production compiled-publication store backed by PredBat storage.""" + + def __init__(self, storage): + """Restore the sole compiled-Lattice publication journal.""" + self._journal = _TwoSlotJournal( + storage, + "compiled", + "compiled", + encode_compiled_lattice_publication, + decode_compiled_lattice_publication, + ) + + def load(self): + """Fresh-read the current compiled publication.""" + return self._journal.load() + + def compare_and_publish(self, expected_version, publication): + """Atomically publish exactly the next compiled Lattice version.""" + if not isinstance(expected_version, int) or isinstance(expected_version, bool) or expected_version < 0: + raise ValueError("expected_version must be a non-negative integer") + if not isinstance(publication, CompiledLatticePublication): + raise ValueError("publication must be a CompiledLatticePublication") + if publication.lattice_version != expected_version + 1: + raise ValueError("publication version must be exactly expected_version + 1") + current = self.load() + current_version = current.lattice_version if current is not None else 0 + if current_version != expected_version: + return False + return self._journal.compare_and_store(current, publication) diff --git a/apps/predbat/lattice_fragment_adapters.py b/apps/predbat/lattice_fragment_adapters.py index 22f21e1d5..c62dbb0ad 100644 --- a/apps/predbat/lattice_fragment_adapters.py +++ b/apps/predbat/lattice_fragment_adapters.py @@ -511,7 +511,13 @@ def unregister(self, provider_id): del self._adapters[provider_id] return True - def create_compiler(self, state_store, override_reader=None): + def create_compiler( + self, + state_store, + override_reader=None, + atomic_materializer=False, + allow_provider_failover=False, + ): """Freeze membership and create the sole compiled-Lattice coordinator.""" if not self._enabled: raise RuntimeError("fragment adapter registry is disabled") @@ -533,6 +539,8 @@ def create_compiler(self, state_store, override_reader=None): readers, state_store=state_store, override_reader=override_reader, + atomic_materializer=atomic_materializer, + allow_provider_failover=allow_provider_failover, ) unsubscribers = [] diff --git a/apps/predbat/lattice_gateway_fragment.py b/apps/predbat/lattice_gateway_fragment.py index 5796a2f9c..b6abdd9b9 100644 --- a/apps/predbat/lattice_gateway_fragment.py +++ b/apps/predbat/lattice_gateway_fragment.py @@ -95,6 +95,7 @@ "ge_cloud_serial", "ge_cloud_data", "ge_cloud_direct", + "inverter_hybrid", "ems_total_soc", "ems_total_charge", "ems_total_discharge", @@ -180,7 +181,10 @@ def _materializable_snapshot(document, plan, provider_id, generation, health): assignments.add((AliasRole.CONTROL, "inverters", index, node_id)) projections = [] - for argument, raw_value in sorted(plan.arguments.items()): + arguments = dict(plan.arguments) + if plan.hybrid is not None: + arguments["inverter_hybrid"] = plan.hybrid + for argument, raw_value in sorted(arguments.items()): if not isinstance(raw_value, tuple) and argument not in _SCALAR_ARGUMENTS: raw_value = tuple(raw_value for _serial in plan.selected_serials) values = raw_value if isinstance(raw_value, tuple) else (raw_value,) @@ -651,6 +655,73 @@ def ingest_auto_config(self, plan): "gateway auto-config plan changed", ) + def ingest_complete(self, payload, plan, online=_LIVENESS_UNCHANGED): + """Atomically publish retained topology and its complete mapper plan.""" + if not self._enabled: + return False + document = decode_topology(payload) + _validate_fragment(document, self.provider_id) + + with self._lock: + current = self._current_state() + if online is _LIVENESS_UNCHANGED and current is not None: + health = current.snapshot.health + elif online is _LIVENESS_UNCHANGED: + health = ProviderHealth.DEGRADED + else: + health = _health_from_liveness(online) + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}".format( + self.provider_id, + current.generation, + ) + ) + if current is not None: + previous = _plain(current.snapshot.topology_fragment) + previous_explicit = "docVersion" in previous + incoming_explicit = "docVersion" in document + previous_version = _document_version(previous) + incoming_version = _document_version(document) + if previous_explicit and not incoming_explicit: + return False + if previous_explicit and incoming_explicit: + if incoming_version < previous_version: + return False + if incoming_version == previous_version and document != previous: + raise TopologyValidationError( + "provider {} reused docVersion {} for different content".format( + self.provider_id, + incoming_version, + ) + ) + candidate = _materializable_snapshot( + document, + plan, + self.provider_id, + current.generation, + health, + ) + if _fingerprint_snapshot(candidate) == _fingerprint_snapshot(current.snapshot): + return False + generation = current.generation + 1 + else: + generation = 1 + snapshot = _materializable_snapshot( + document, + plan, + self.provider_id, + generation, + health, + ) + published = self._adapter.publish( + snapshot, + "gateway topology and auto-config changed", + ) + if published: + self._seeded = True + return published + def remove(self): """Publish a durable provider-removal tombstone and invalidate.""" if not self._enabled: diff --git a/apps/predbat/lattice_ge_cloud_fragment.py b/apps/predbat/lattice_ge_cloud_fragment.py index 8187d9294..7629a1af6 100644 --- a/apps/predbat/lattice_ge_cloud_fragment.py +++ b/apps/predbat/lattice_ge_cloud_fragment.py @@ -85,6 +85,7 @@ "ge_cloud_serial", "ge_cloud_data", "ge_cloud_direct", + "inverter_hybrid", ) ) @@ -219,7 +220,8 @@ def _materializable_snapshot( ) projections = [] - for argument, raw_value in plan.arguments: + arguments = tuple(plan.arguments) + (("inverter_hybrid", not plan.ac_coupled),) + for argument, raw_value in arguments: if not isinstance(raw_value, tuple) and argument not in _SCALAR_ARGUMENTS: raw_value = tuple(raw_value for _serial in plan.primary_targets) values = raw_value if isinstance(raw_value, tuple) else (raw_value,) @@ -742,6 +744,79 @@ def ingest_auto_config(self, config): "GE Cloud auto-config plan changed", ) + def ingest_complete( + self, + discovery_version, + devices, + config, + health=_HEALTH_UNCHANGED, + ): + """Atomically publish complete discovery and mapper projections.""" + if not self._enabled: + return False + discovery_version = _discovery_version(discovery_version) + devices = _normalize_devices(devices) + document = _topology_document( + self.provider_id, + discovery_version, + devices, + ) + plan = compile_gecloud_auto_config(config) + + with self._lock: + current = self._current_state() + if current is not None and current.removed: + raise FragmentAdapterRemoved( + "provider {} was removed at generation {}".format( + self.provider_id, + current.generation, + ) + ) + if current is None: + next_health = ProviderHealth.DEGRADED if health is _HEALTH_UNCHANGED else _provider_health(health) + generation = 1 + else: + previous = _plain(current.snapshot.topology_fragment) + previous_version = previous["docVersion"] + if discovery_version < previous_version: + return False + if discovery_version == previous_version and document != previous: + raise FragmentAdapterConflict( + "provider {} reused GE Cloud discovery version {} for different content".format( + self.provider_id, + discovery_version, + ) + ) + next_health = current.snapshot.health if health is _HEALTH_UNCHANGED else _provider_health(health) + candidate = _materializable_snapshot( + document, + devices, + config, + plan, + self.provider_id, + current.generation, + next_health, + ) + if _fingerprint_snapshot(candidate) == _fingerprint_snapshot(current.snapshot): + return False + generation = current.generation + 1 + snapshot = _materializable_snapshot( + document, + devices, + config, + plan, + self.provider_id, + generation, + next_health, + ) + published = self._adapter.publish( + snapshot, + "GE Cloud discovery and auto-config changed", + ) + if published: + self._seeded = True + return published + def remove(self): """Publish an irreversible provider-removal tombstone.""" if not self._enabled: diff --git a/apps/predbat/lattice_generated_overlay.py b/apps/predbat/lattice_generated_overlay.py new file mode 100644 index 000000000..301911111 --- /dev/null +++ b/apps/predbat/lattice_generated_overlay.py @@ -0,0 +1,109 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - generated Lattice configuration overlay +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Atomic, immutable storage for generated Lattice configuration arguments.""" + +import copy +import threading +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType + + +def _freeze(value): + """Detach and recursively freeze a generated configuration value.""" + if isinstance(value, Mapping): + return MappingProxyType({key: _freeze(item) for key, item in value.items()}) + if isinstance(value, (list, tuple)): + return tuple(_freeze(item) for item in value) + if isinstance(value, (set, frozenset)): + return frozenset(_freeze(item) for item in value) + return copy.deepcopy(value) + + +def _thaw(value): + """Return a detached mutable value suitable for Predbat's argument readers.""" + if isinstance(value, Mapping): + return {key: _thaw(item) for key, item in value.items()} + if isinstance(value, tuple): + return [_thaw(item) for item in value] + if isinstance(value, frozenset): + return set(copy.deepcopy(value)) + return copy.deepcopy(value) + + +@dataclass(frozen=True) +class GeneratedConfigSnapshot: + """One immutable full-replacement generated configuration snapshot.""" + + enabled: bool + version: int + arguments: Mapping + + +class GeneratedConfigOverlay: + """Publish and read complete generated argument snapshots atomically. + + Readers capture one immutable snapshot reference without locking. Writers are + serialized and replace that reference only after the complete next snapshot + has been detached and frozen. Values are thawed on every successful read so + existing argument-resolution code cannot mutate the published snapshot. + """ + + def __init__(self, authoritative_arguments=()): + """Create a disabled overlay with no generated arguments.""" + if any(not isinstance(argument, str) or not argument.strip() for argument in authoritative_arguments): + raise ValueError("authoritative argument names must be non-empty strings") + self._write_lock = threading.Lock() + self._authoritative_arguments = frozenset(authoritative_arguments) + self._snapshot = GeneratedConfigSnapshot( + enabled=False, + version=0, + arguments=MappingProxyType({}), + ) + + @property + def snapshot(self): + """Return the current immutable snapshot for diagnostics.""" + return self._snapshot + + def replace(self, arguments): + """Atomically enable and replace the complete generated argument set.""" + if not isinstance(arguments, Mapping): + raise TypeError("generated arguments must be a mapping") + if any(not isinstance(argument, str) or not argument.strip() for argument in arguments): + raise ValueError("generated argument names must be non-empty strings") + frozen = _freeze(dict(arguments)) + with self._write_lock: + current = self._snapshot + self._snapshot = GeneratedConfigSnapshot( + enabled=True, + version=current.version + 1, + arguments=frozen, + ) + return self._snapshot + + def disable(self): + """Atomically disable and remove every generated argument.""" + with self._write_lock: + current = self._snapshot + self._snapshot = GeneratedConfigSnapshot( + enabled=False, + version=current.version + 1, + arguments=MappingProxyType({}), + ) + return self._snapshot + + def read(self, argument): + """Return ``(present, detached_value)`` from one captured snapshot.""" + snapshot = self._snapshot + if not snapshot.enabled or argument not in snapshot.arguments: + return False, None + return True, _thaw(snapshot.arguments[argument]) + + def authoritative(self, argument): + """Return whether a published value is a topology-owned hardware fact.""" + snapshot = self._snapshot + return snapshot.enabled and argument in snapshot.arguments and argument in self._authoritative_arguments diff --git a/apps/predbat/lattice_provider_inputs.py b/apps/predbat/lattice_provider_inputs.py new file mode 100644 index 000000000..40218ddfa --- /dev/null +++ b/apps/predbat/lattice_provider_inputs.py @@ -0,0 +1,335 @@ +# cspell:ignore autoconfig XIAO +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - pure live provider input adapters +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Detach live Gateway and GE Cloud state into Lattice mapper inputs. + +This module deliberately has no component lifecycle, persistence, publication, +or configuration side effects. Runtime wiring can call these helpers after a +successful provider read and hand their results to the existing pure fragment +publishers. +""" + +from dataclasses import dataclass + +from gecloud_autoconfig import GECloudAutoConfigInput, normalize_register_name +from lattice_ge_cloud_fragment import GECloudDeviceSnapshot + + +class LiveProviderInputError(ValueError): + """Live provider state cannot be represented safely by the current slice.""" + + +# Stable values from gateway_status.proto. Keeping the adapter duck-typed +# avoids importing generated protobuf code into this pure conversion seam. +_GIVENERGY_EMS_TYPE = 7 +_GIVENERGY_GATEWAY_TYPE = 8 + + +@dataclass(frozen=True) +class GatewayInverterInput: + """Detached fields consumed by ``compile_gateway_auto_config``.""" + + serial: str + kind: str + primary: bool + battery_present: bool + battery_capacity_wh: int + ems_num_inverters: int + model: str + + +@dataclass(frozen=True) +class GatewayLiveInput: + """One detached mapper input and matching synthetic topology fragment.""" + + inverters: tuple + topology_fragment: object + + +@dataclass(frozen=True) +class GECloudLiveInput: + """Detached GE auto-config policy input and provider discovery devices.""" + + auto_config: GECloudAutoConfigInput + devices: tuple + + +def _field(value, name, default=None): + """Read one field from a mapping or proto-like object.""" + if isinstance(value, dict): + return value.get(name, default) + return getattr(value, name, default) + + +def _iterable_field(value, name): + """Detach one repeated field and report malformed provider state clearly.""" + items = _field(value, name, ()) + try: + return tuple(items) + except TypeError as exc: + raise LiveProviderInputError("{} must be iterable".format(name)) from exc + + +def _required_text(value, name): + """Normalize a required provider identity.""" + if not isinstance(value, str) or not value.strip(): + raise LiveProviderInputError("{} must be a non-empty string".format(name)) + return value.strip() + + +def _optional_text(value): + """Normalize provider text while preserving absence.""" + if value is None: + return None + value = str(value).strip() + return value or None + + +def _non_negative_int(value, name): + """Normalize a non-negative integer without accepting booleans.""" + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise LiveProviderInputError("{} must be a non-negative integer".format(name)) + return value + + +def _gateway_kind(inverter_type): + """Map a numeric or named Gateway enum value to the pure mapper kind.""" + normalized = str(inverter_type).strip().upper() + if inverter_type == _GIVENERGY_EMS_TYPE or normalized in ( + "EMS", + "INVERTER_TYPE_GIVENERGY_EMS", + ): + return "ems" + if inverter_type == _GIVENERGY_GATEWAY_TYPE or normalized in ( + "GATEWAY", + "INVERTER_TYPE_GIVENERGY_GATEWAY", + ): + return "gateway" + return "inverter" + + +def _message_present(message): + """Match protobuf ``ByteSize`` presence while supporting test doubles.""" + if message is None: + return False + byte_size = getattr(message, "ByteSize", None) + if callable(byte_size): + return byte_size() > 0 + if isinstance(message, dict): + return bool(message) + return bool(message) + + +def _gateway_inverter(inverter): + """Detach one proto-like inverter entry.""" + serial = _required_text(_field(inverter, "serial"), "Gateway inverter serial") + battery = _field(inverter, "battery") + ems = _field(inverter, "ems") + capacity = _field(battery, "capacity_wh", 0) if battery is not None else 0 + ems_count = _field(ems, "num_inverters", 0) if ems is not None else 0 + return GatewayInverterInput( + serial=serial, + kind=_gateway_kind(_field(inverter, "type", "inverter")), + primary=bool(_field(inverter, "primary", False)), + battery_present=_message_present(battery), + battery_capacity_wh=_non_negative_int(capacity or 0, "battery capacity_wh"), + ems_num_inverters=_non_negative_int(ems_count or 0, "EMS num_inverters"), + model=str(_field(inverter, "model", "") or ""), + ) + + +def _gateway_node(inverter, provider_id): + """Build one stable provider-owned v0.3 topology node.""" + serial = inverter.serial.upper() + topology_kind = "gateway" if inverter.kind == "gateway" else "inverter" + return { + "id": "gateway:{}".format(serial), + "kind": topology_kind, + "attributes": {"serial": serial}, + "accessPaths": [ + { + "id": "gateway-mqtt", + "provider": provider_id, + "preference": 10, + } + ], + "capabilities": [], + } + + +def gateway_status_to_lattice_input( + status, + provider_id="predbat-gateway", + document_version=None, +): + """Convert one non-empty Gateway status into mapper and topology inputs. + + Plant EMS and EV discovery fail closed because their legacy side effects are + outside the current XIAO/Gateway auto-config slice. + """ + provider_id = _required_text(provider_id, "provider_id") + inverters = tuple(_gateway_inverter(item) for item in _iterable_field(status, "inverters")) + if not inverters: + raise LiveProviderInputError("Gateway status contains no inverters") + if _iterable_field(status, "ev_chargers"): + raise LiveProviderInputError("Gateway EV discovery is outside the Lattice auto-config slice") + if any(item.kind == "ems" for item in inverters): + raise LiveProviderInputError("Gateway EMS discovery requires the multi-battery coordinator model") + + serials = [item.serial.upper() for item in inverters] + if len(serials) != len(set(serials)): + raise LiveProviderInputError("Gateway status contains duplicate inverter serials") + inverters = tuple(sorted(inverters, key=lambda item: item.serial.upper())) + topology = { + "topologyVersion": "0.3.0", + "scope": "fragment", + "producer": { + "name": "PredBat Gateway", + "provider": provider_id, + "authority": 10, + }, + "nodes": [_gateway_node(item, provider_id) for item in inverters], + } + if document_version is not None: + topology["docVersion"] = _non_negative_int( + document_version, + "document_version", + ) + return GatewayLiveInput(inverters, topology) + + +def _ge_device_kinds(devices): + """Return one collision-checked serial-to-kind lookup.""" + result = {} + groups = ( + ("battery", "battery-inverter"), + ("pv", "pv-inverter"), + ("gateway", "gateway"), + ("ems", "ems"), + ) + for key, kind in groups: + raw = devices.get(key, ()) + values = (raw,) if key in ("gateway", "ems") and raw else tuple(raw or ()) + for value in values: + serial = _required_text(value, "GE Cloud {} serial".format(key)) + normalized = serial.upper() + previous = result.get(normalized) + if previous is not None and previous[1] != kind: + raise LiveProviderInputError("GE Cloud serial {} has conflicting device kinds".format(serial)) + result[normalized] = (serial, kind) + return result + + +def _ge_info(info, serial): + """Read one GE inverter info record case-insensitively.""" + record = info.get(serial) + if record is None: + record = info.get(serial.lower()) + if record is None: + record = info.get(serial.upper()) + return record if isinstance(record, dict) else {} + + +def _ge_model(info, serial): + """Mirror the legacy ``self.info[serial]['info']['model']`` lookup.""" + record = _ge_info(info, serial) + details = record.get("info") or {} + return str(details.get("model", "") or "") if isinstance(details, dict) else "" + + +def _ge_register_names(settings, serial): + """Detach the HA-normalized register-name set for one device.""" + registers = settings.get(serial) + if registers is None: + registers = settings.get(serial.lower()) + if registers is None: + registers = settings.get(serial.upper(), {}) + if not isinstance(registers, dict): + raise LiveProviderInputError("GE Cloud settings for {} must be a mapping".format(serial)) + names = [] + for setting in registers.values(): + if not isinstance(setting, dict): + raise LiveProviderInputError("GE Cloud setting for {} must be a mapping".format(serial)) + name = setting.get("name", "") + if name: + names.append(normalize_register_name(str(name))) + return tuple(sorted(set(names))) + + +def _bool_flag(value, name): + """Require explicit boolean runtime flags.""" + if not isinstance(value, bool): + raise LiveProviderInputError("{} must be a boolean".format(name)) + return value + + +def gecloud_state_to_lattice_input( + devices, + settings, + info, + prefix="predbat", + load_today_ignore=False, + split_pv=False, + split_ct=False, + shared_ct=False, +): + """Convert detached GE discovery/settings/info into existing pure inputs.""" + if not isinstance(devices, dict): + raise LiveProviderInputError("GE Cloud devices must be a mapping") + if not isinstance(settings, dict) or not isinstance(info, dict): + raise LiveProviderInputError("GE Cloud settings and info must be mappings") + prefix = _required_text(prefix, "prefix") + batteries = tuple(devices.get("battery") or ()) + if not batteries: + raise LiveProviderInputError("GE Cloud discovery contains no batteries") + batteries = tuple(_required_text(item, "GE Cloud battery serial") for item in batteries) + if len({item.upper() for item in batteries}) != len(batteries): + raise LiveProviderInputError("GE Cloud discovery contains duplicate battery serials") + pv = tuple(_required_text(item, "GE Cloud PV serial") for item in (devices.get("pv") or ())) + ems = _optional_text(devices.get("ems")) + gateway = _optional_text(devices.get("gateway")) + kinds = _ge_device_kinds(devices) + + referenced = set(batteries + pv) + if ems: + referenced.add(ems) + if gateway: + referenced.add(gateway) + register_names = tuple((serial, _ge_register_names(settings, serial)) for serial in sorted(referenced, key=str.upper)) + + meters = devices.get("battery_meters") or {} + if not isinstance(meters, dict): + raise LiveProviderInputError("GE Cloud battery_meters must be a mapping") + battery_meters = tuple((serial, tuple(meters.get(serial, meters.get(serial.lower(), ())))) for serial in batteries) + models = tuple((serial, _ge_model(info, serial)) for serial in batteries) + config = GECloudAutoConfigInput( + batteries=batteries, + ems=ems, + gateway=gateway, + pv=pv, + battery_meters=battery_meters, + register_names=register_names, + models=models, + prefix=prefix, + load_today_ignore=_bool_flag(load_today_ignore, "load_today_ignore"), + split_pv=_bool_flag(split_pv, "split_pv"), + split_ct=_bool_flag(split_ct, "split_ct"), + shared_ct=_bool_flag(shared_ct, "shared_ct"), + ) + + snapshots = [] + for normalized, (serial, kind) in sorted(kinds.items()): + record = _ge_info(info, serial) + snapshots.append( + GECloudDeviceSnapshot( + serial=normalized, + kind=kind, + model=_ge_model(info, serial) or None, + site_id=record.get("site_id"), + uuid=_optional_text(record.get("uuid")), + ) + ) + return GECloudLiveInput(config, tuple(snapshots)) diff --git a/apps/predbat/predbat.py b/apps/predbat/predbat.py index 76a977639..d15829c1d 100644 --- a/apps/predbat/predbat.py +++ b/apps/predbat/predbat.py @@ -1,3 +1,4 @@ +# cspell:ignore autoconfig # ----------------------------------------------------------------------------- # Predbat Home Battery System # Copyright Trefor Southwell 2025-2026 - All Rights Reserved @@ -85,6 +86,7 @@ from plugin_system import PluginSystem from github import GitHub from ha import run_async +from lattice_autoconfig_runtime import LatticeAutoConfigRuntime class PredBat(hass.Hass, Octopus, Energidataservice, Stromligning, Fetch, Plan, Marginal, Execute, Output, UserInterface, GitHub): @@ -305,6 +307,7 @@ def reset(self): self.num_cars = 0 self.fatal_error = False self.components = None + self.lattice_autoconfig_runtime = None self.CONFIG_ITEMS = copy.deepcopy(CONFIG_ITEMS) self.comparison = None self.predheat = None @@ -1644,11 +1647,35 @@ def initialize(self): self.load_user_config(quiet=False, register=False, load_config=True) self.comparison = Compare(self) + lattice_enabled = self.get_arg( + "lattice_autoconfig_enable", + False, + indirect=False, + ) + self.lattice_autoconfig_runtime = LatticeAutoConfigRuntime( + self, + self.components.get_component("storage"), + enabled=lattice_enabled, + ) + self.components.initialize(phase=1) + gecloud = self.components.get_component("gecloud") + providers = self.lattice_autoconfig_runtime.bind( + gateway_enabled=self.components.get_component("gateway") is not None, + gecloud_enabled=gecloud is not None and gecloud.automatic, + ) + if providers: + self.log( + "Lattice auto-config enabled for {}".format( + ", ".join(providers), + ) + ) if not self.components.start(phase=1): self.log("Error: Some components failed to start (phase 1)") self.record_status("Error: Some components failed to start (phase 1)", had_errors=True) + self.lattice_autoconfig_runtime.apply_pending() + self.components.initialize(phase=2) if not self.components.start(phase=2): self.log("Error: Some components failed to start (phase 2)") @@ -1660,6 +1687,10 @@ def initialize(self): # Restore the last saved plan so it is immediately active before the first calculation self.load_plan() + if self.lattice_autoconfig_runtime.enabled: + # Saved plans predate generated-config digests. Keep their windows for + # diagnostics, but never activate one against a newly compiled overlay. + self.plan_valid = False except Exception as e: self.log("Error: Exception raised {}".format(e)) diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index 02054a364..4692f9feb 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -1,3 +1,4 @@ +# cspell:ignore autoconfig """ Tests for GatewayMQTT component. """ @@ -11,7 +12,7 @@ import pytz import gateway_status_pb2 as pb -from gateway import _STARTUP_WAIT_TICKS +from gateway import _STARTUP_WAIT_TICKS, _TELEMETRY_STALE_THRESHOLD import importlib.util @@ -4486,6 +4487,245 @@ async def publish(self, topic, payload, qos=0, retain=False): assert observed.get("thread_ident") == owner_thread.ident, "client.publish() ran on the wrong thread/loop" +class TestGatewayLatticeRuntimeWiring(unittest.TestCase): + """Default-off live Gateway telemetry routing into Lattice.""" + + class Runtime: + """Synchronous runtime double invoked through ``asyncio.to_thread``.""" + + enabled = True + + def __init__(self, reject=None, selected_serials=("SER123",), discovered_serials=("SER123",)): + self.reject = reject + self.selected_serials = selected_serials + self.discovered_serials = discovered_serials + self.ingests = [] + self.liveness = [] + + def ingest_gateway_status(self, status, prefix="predbat", serial_filter=()): + import threading + from types import SimpleNamespace + + self.ingests.append((status, prefix, serial_filter, threading.get_ident())) + if self.reject: + raise ValueError(self.reject) + plan = SimpleNamespace( + selected_serials=self.selected_serials, + discovered_serials=self.discovered_serials, + ) + return plan, SimpleNamespace(accepted=True) + + def set_gateway_liveness(self, online): + import threading + + self.liveness.append((online, threading.get_ident())) + + def provider_active(self, provider_id): + return provider_id == "predbat-gateway" + + class Message: + """Minimal aiomqtt message-shaped value.""" + + def __init__(self, topic, payload): + self.topic = topic + self.payload = payload + + def _make_gateway(self, runtime=None): + from gateway import GatewayMQTT + from unittest.mock import MagicMock + + gw = GatewayMQTT.__new__(GatewayMQTT) + gw.base = MagicMock() + gw.base.args = {} + if runtime is not None: + gw.base.lattice_autoconfig_runtime = runtime + else: + del gw.base.lattice_autoconfig_runtime + gw.args = gw.base.args + gw.log = MagicMock() + gw.initialize( + gateway_device_id="pbgw_test", + mqtt_host="mqtt.example.com", + mqtt_token="token", + gateway_inverter_serial=["SER123"], + ) + gw.local_tz = pytz.timezone("Europe/London") + gw.api_started = False + gw.dashboard_item = MagicMock() + gw.update_success_timestamp = MagicMock() + return gw + + def _status(self, include_ev=False, inverter_type=pb.INVERTER_TYPE_GIVENERGY): + status = pb.GatewayStatus() + status.device_id = "pbgw_test" + status.firmware = "1.0.0" + status.timestamp = 1741789200 + inverter = status.inverters.add() + inverter.type = inverter_type + inverter.serial = "SER123" + inverter.primary = True + inverter.battery.soc_percent = 50 + inverter.battery.capacity_wh = 9500 + inverter.battery.rate_max_w = 5000 + if include_ev: + status.ev_chargers.add().charge_point_id = "ev-1" + return status + + def test_enabled_runtime_bypasses_legacy_and_starts_after_ingest(self): + import asyncio + import threading + from unittest.mock import MagicMock + + runtime = self.Runtime(discovered_serials=("SER123", "SER999")) + gw = self._make_gateway(runtime) + gw.automatic_config = MagicMock(side_effect=AssertionError("legacy auto-config called")) + message = self.Message(gw.topic_status, self._status().SerializeToString()) + caller_thread = threading.get_ident() + + asyncio.run(gw._handle_message(message)) + + gw.automatic_config.assert_not_called() + assert runtime.ingests[0][1:3] == ("predbat", ("SER123",)) + assert runtime.ingests[0][3] != caller_thread + assert gw._suffix_to_serial == {"ser123": "SER123"} + assert gw._configured_inverter_serials == frozenset(("SER123", "SER999")) + assert gw._auto_configured is True + assert gw.api_started is True + assert gw.base.args == {} + assert any(call.args[0] == "sensor.predbat_gateway_ser123_soc" for call in gw.dashboard_item.call_args_list) + + def test_duplicate_telemetry_remains_accepted(self): + import asyncio + from unittest.mock import MagicMock + + runtime = self.Runtime() + gw = self._make_gateway(runtime) + gw.automatic_config = MagicMock(side_effect=AssertionError("legacy auto-config called")) + message = self.Message(gw.topic_status, self._status().SerializeToString()) + + asyncio.run(gw._handle_message(message)) + asyncio.run(gw._handle_message(message)) + + assert len(runtime.ingests) == 2 + assert gw._auto_configured is True + assert gw.api_started is True + gw.automatic_config.assert_not_called() + + def test_disabled_runtime_keeps_legacy_telemetry_path(self): + import asyncio + from unittest.mock import MagicMock + + runtime = self.Runtime() + runtime.enabled = False + gw = self._make_gateway(runtime) + + def legacy_config(): + gw._auto_configured = True + + gw.automatic_config = MagicMock(side_effect=legacy_config) + message = self.Message(gw.topic_status, self._status().SerializeToString()) + + asyncio.run(gw._handle_message(message)) + + gw.automatic_config.assert_called_once_with() + assert runtime.ingests == [] + assert gw.api_started is True + + def test_adapter_rejection_fails_closed_and_marks_liveness_offline(self): + import asyncio + from unittest.mock import MagicMock + + runtime = self.Runtime(reject="Gateway EV discovery is outside the Lattice auto-config slice") + gw = self._make_gateway(runtime) + gw.automatic_config = MagicMock(side_effect=AssertionError("legacy auto-config called")) + message = self.Message( + gw.topic_status, + self._status(include_ev=True).SerializeToString(), + ) + + asyncio.run(gw._handle_message(message)) + + gw.automatic_config.assert_not_called() + assert gw._auto_configured is False + assert gw.api_started is False + assert gw._suffix_to_serial == {} + assert [online for online, _thread in runtime.liveness] == [False] + assert any("Lattice auto-config rejected" in str(call) for call in gw.log.call_args_list) + + def test_entity_injection_failure_invalidates_accepted_publication(self): + import asyncio + from unittest.mock import MagicMock + + runtime = self.Runtime() + gw = self._make_gateway(runtime) + gw._inject_entities = MagicMock(side_effect=RuntimeError("injection failed")) + message = self.Message(gw.topic_status, self._status().SerializeToString()) + + asyncio.run(gw._handle_message(message)) + + assert len(runtime.ingests) == 1 + assert [online for online, _thread in runtime.liveness] == [False] + assert gw._auto_configured is False + assert gw.api_started is False + + def test_ems_adapter_rejection_fails_closed(self): + import asyncio + from unittest.mock import MagicMock + + runtime = self.Runtime(reject="Gateway EMS discovery requires the multi-battery coordinator model") + gw = self._make_gateway(runtime) + gw.automatic_config = MagicMock(side_effect=AssertionError("legacy auto-config called")) + message = self.Message( + gw.topic_status, + self._status(inverter_type=pb.INVERTER_TYPE_GIVENERGY_EMS).SerializeToString(), + ) + + asyncio.run(gw._handle_message(message)) + + gw.automatic_config.assert_not_called() + assert gw._auto_configured is False + assert gw.api_started is False + assert gw._suffix_to_serial == {} + assert [online for online, _thread in runtime.liveness] == [False] + + def test_lwt_changes_publish_lattice_liveness_off_event_loop(self): + import asyncio + import threading + + runtime = self.Runtime() + gw = self._make_gateway(runtime) + caller_thread = threading.get_ident() + + asyncio.run(gw._handle_message(self.Message(gw.topic_online, b"1"))) + asyncio.run(gw._handle_message(self.Message(gw.topic_online, b"0"))) + + assert [online for online, _thread in runtime.liveness] == [True, False] + assert all(thread != caller_thread for _online, thread in runtime.liveness) + + def test_stale_telemetry_invalidates_lattice_provider_once(self): + import asyncio + import time + + runtime = self.Runtime() + gw = self._make_gateway(runtime) + gw._last_telemetry_time = time.time() - _TELEMETRY_STALE_THRESHOLD - 1 + + self.assertTrue(asyncio.run(gw._invalidate_stale_lattice_gateway())) + self.assertFalse(asyncio.run(gw._invalidate_stale_lattice_gateway())) + + assert [online for online, _thread in runtime.liveness] == [False] + + def test_stop_invalidates_long_lived_gateway_provider(self): + import asyncio + + runtime = self.Runtime() + gw = self._make_gateway(runtime) + + asyncio.run(gw.stop()) + + assert [online for online, _thread in runtime.liveness] == [False] + + def run_gateway_tests(my_predbat=None): """Run all GatewayMQTT tests. Returns True on failure, False on success.""" from tests.test_gateway_token_refresh import TestIsAuthFailure, TestApplyRefreshResponse, TestMaybeRefreshOnAuthError @@ -4523,6 +4763,7 @@ def run_gateway_tests(my_predbat=None): TestMaybeRefreshOnAuthError, TestRateAnchors, TestPublishRawLoopSafety, + TestGatewayLatticeRuntimeWiring, ] for cls in test_classes: instance = cls() diff --git a/apps/predbat/tests/test_gecloud_lattice_wiring.py b/apps/predbat/tests/test_gecloud_lattice_wiring.py new file mode 100644 index 000000000..ed3d4a18e --- /dev/null +++ b/apps/predbat/tests/test_gecloud_lattice_wiring.py @@ -0,0 +1,184 @@ +# cspell:ignore autoconfig gecloud +"""Focused tests for default-off GE Cloud Lattice live wiring.""" + +import asyncio +import os +import sys +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from gecloud import GECloudDirect # noqa: E402 + + +class RecordingRuntime: + """Record synchronous runtime calls made through ``asyncio.to_thread``.""" + + def __init__(self, enabled=True, active=True): + self.enabled = enabled + self.active = active + self.ingests = [] + self.health = [] + + def ingest_gecloud_state(self, devices, settings, info, **kwargs): + """Record one complete provider state.""" + self.ingests.append((devices, settings, info, kwargs)) + + def set_gecloud_liveness(self, health): + """Record one provider health transition.""" + self.health.append(health) + + def provider_active(self, provider_id): + return self.active and provider_id == "ge-cloud" + + +def component(runtime, automatic=True): + """Build a minimal GE component for the isolated wiring helpers.""" + ge = object.__new__(GECloudDirect) + ge.automatic = automatic + ge.base = SimpleNamespace(lattice_autoconfig_runtime=runtime) + ge.prefix = "site" + ge.devices_dict = { + "battery": ["bat1"], + "pv": [], + "gateway": None, + "ems": None, + "battery_meters": {"bat1": [42]}, + } + ge.settings = {"bat1": {1: {"name": "Battery Charge Power"}}} + ge.info = {"bat1": {"info": {"model": "Hybrid"}}} + flags = { + "ge_cloud_load_today_ignore": True, + "ge_cloud_automatic_split_pv": False, + "ge_cloud_automatic_split_ct": True, + "ge_cloud_automatic_shared_ct": False, + } + ge.get_arg = lambda name, default=None, **_kwargs: flags.get(name, default) + ge.async_automatic_config = AsyncMock() + return ge + + +def steady_state_component(runtime): + """Extend the helper with the state needed by one two-minute poll.""" + ge = component(runtime) + ge.device_list = ["bat1"] + ge.evc_device_list = [] + ge.status = {} + ge.meter = {} + ge.pending_writes = {"bat1": []} + ge.api_auth_failed = False + ge.auth_denied_reported = False + ge.last_success_timestamp = object() + ge.async_get_inverter_meter = AsyncMock(return_value={}) + ge.async_get_device_info = AsyncMock( + return_value={"info": {"model": "Hybrid"}}, + ) + ge.publish_status = AsyncMock() + ge.publish_meter = AsyncMock() + ge.publish_info = AsyncMock() + return ge + + +class TestGECloudLatticeWriterSelection(unittest.TestCase): + """One and only one automatic-configuration writer is selected.""" + + def test_enabled_runtime_replaces_legacy_writer_with_complete_state(self): + runtime = RecordingRuntime() + ge = component(runtime) + + self.assertTrue(asyncio.run(ge._refresh_automatic_config())) + + ge.async_automatic_config.assert_not_awaited() + self.assertEqual(len(runtime.ingests), 1) + devices, settings, info, flags = runtime.ingests[0] + self.assertIs(devices, ge.devices_dict) + self.assertIs(settings, ge.settings) + self.assertIs(info, ge.info) + self.assertEqual( + flags, + { + "prefix": "site", + "load_today_ignore": True, + "split_pv": False, + "split_ct": True, + "shared_ct": False, + }, + ) + + def test_absent_disabled_or_nonautomatic_runtime_preserves_safe_path(self): + for runtime in ( + None, + RecordingRuntime(enabled=False), + RecordingRuntime(active=False), + ): + with self.subTest(runtime=runtime): + ge = component(runtime) + self.assertTrue(asyncio.run(ge._refresh_automatic_config())) + ge.async_automatic_config.assert_awaited_once_with(ge.devices_dict) + if runtime is not None: + self.assertEqual(runtime.ingests, []) + + runtime = RecordingRuntime() + ge = component(runtime, automatic=False) + self.assertFalse(asyncio.run(ge._refresh_automatic_config())) + ge.async_automatic_config.assert_not_awaited() + self.assertEqual(runtime.ingests, []) + + def test_liveness_is_only_published_for_enabled_automatic_runtime(self): + runtime = RecordingRuntime() + ge = component(runtime) + + self.assertTrue(asyncio.run(ge._set_lattice_liveness(None))) + self.assertTrue(asyncio.run(ge._set_lattice_liveness(False))) + self.assertEqual(runtime.health, [None, False]) + + disabled = component(RecordingRuntime(), automatic=False) + self.assertFalse(asyncio.run(disabled._set_lattice_liveness(False))) + self.assertEqual(disabled.base.lattice_autoconfig_runtime.health, []) + + def test_successful_two_minute_poll_republishes_complete_state_once(self): + runtime = RecordingRuntime() + ge = steady_state_component(runtime) + + async def successful_status(_device, _previous): + ge.last_success_timestamp = object() + return {"status": "NORMAL"} + + ge.async_get_inverter_status = successful_status + + self.assertTrue(asyncio.run(ge.run(seconds=120, first=False))) + + self.assertEqual(len(runtime.ingests), 1) + self.assertEqual(runtime.health, []) + ge.async_automatic_config.assert_not_awaited() + + def test_denied_core_poll_marks_offline_without_rewriting_config(self): + runtime = RecordingRuntime() + ge = steady_state_component(runtime) + ge.base.record_status = lambda *_args, **_kwargs: None + + async def denied_status(_device, _previous): + ge.api_auth_failed = True + return {} + + ge.async_get_inverter_status = denied_status + + self.assertTrue(asyncio.run(ge.run(seconds=120, first=False))) + + self.assertEqual(runtime.ingests, []) + self.assertEqual(runtime.health, [False]) + ge.async_automatic_config.assert_not_awaited() + + def test_stop_invalidates_long_lived_cloud_provider(self): + runtime = RecordingRuntime() + ge = component(runtime) + + asyncio.run(ge.stop()) + + self.assertEqual(runtime.health, [False]) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_autoconfig.py b/apps/predbat/tests/test_lattice_autoconfig.py index 523f58b6b..d4cb2436d 100644 --- a/apps/predbat/tests/test_lattice_autoconfig.py +++ b/apps/predbat/tests/test_lattice_autoconfig.py @@ -636,6 +636,39 @@ def test_disjoint_providers_compose_partial_indexed_slots(self): ("atomic_materializer_missing",), ) + def test_atomic_materializer_capability_makes_projected_plan_ready(self): + """Only an explicitly installed atomic materializer clears its blocker.""" + provider = projection_snapshot( + "gateway", + ("GW1",), + indexed_roles(("GW1",)), + ( + config_projection( + "battery_power", + ( + projection_value( + "GW1", + ProjectionValueKind.ENTITY, + "sensor.gateway_battery_power", + ), + ), + ), + ), + ) + + plan = compile_auto_config( + (provider,), + atomic_materializer=True, + ) + + self.assertTrue(plan.materialization_readiness.ready) + self.assertEqual(plan.materialization_readiness.blockers, ()) + with self.assertRaisesRegex(ValueError, "must be a boolean"): + compile_auto_config( + (provider,), + atomic_materializer="yes", + ) + def test_provider_owned_slot_disambiguates_repeated_canonical_node(self): """Explicit roles place correlated providers in their owned slots.""" identity = "SER-SHARED" diff --git a/apps/predbat/tests/test_lattice_autoconfig_runtime.py b/apps/predbat/tests/test_lattice_autoconfig_runtime.py new file mode 100644 index 000000000..a10626ee5 --- /dev/null +++ b/apps/predbat/tests/test_lattice_autoconfig_runtime.py @@ -0,0 +1,314 @@ +# cspell:ignore autoconfig +"""Focused tests for the default-off live Lattice auto-config coordinator.""" + +import os +import sys +import threading +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_autoconfig import ProviderHealth # noqa: E402 +from lattice_autoconfig_runtime import LatticeAutoConfigRuntime # noqa: E402 +from lattice_fragment_adapters import FragmentAdapterConflict # noqa: E402 +from tests.test_lattice_provider_inputs import inverter, status # noqa: E402 + + +class MemoryAsyncStorage: + """Small thread-safe async storage double for durable journal tests.""" + + def __init__(self): + self.values = {} + self._lock = threading.RLock() + self.fail_save = False + + async def load(self, module, filename): + """Return a detached-enough stored JSON value.""" + with self._lock: + return self.values.get((module, filename)) + + async def save(self, module, filename, value, **_kwargs): + """Store one complete journal envelope.""" + if self.fail_save: + raise OSError("injected durable write failure") + with self._lock: + self.values[(module, filename)] = value + return True + + async def age(self, module, filename): + """Distinguish absent keys from unreadable values.""" + with self._lock: + return 0 if (module, filename) in self.values else None + + +class RuntimeBase: + """Minimal PredBat surface mutated by the runtime.""" + + def __init__(self): + self.args = {} + self.prefix = "predbat" + self.update_pending = False + self.plan_valid = True + self.inverters = [object()] + self.messages = [] + + def log(self, message): + """Capture diagnostics without printing test output.""" + self.messages.append(message) + + +def ge_devices(serial="ser123"): + """Return one complete GE discovery input.""" + return { + "battery": [serial], + "pv": [], + "gateway": None, + "ems": None, + "battery_meters": {serial: ["meter1"]}, + } + + +def ge_settings(serial="ser123"): + """Return minimum direct-control register discovery.""" + return { + serial: { + 1: {"name": "Battery Charge Power"}, + 2: {"name": "Battery Discharge Power"}, + 3: {"name": "Battery Reserve Percent Limit"}, + } + } + + +class TestLatticeAutoConfigRuntime(unittest.TestCase): + """Durable publication is separate from atomic config activation.""" + + def test_disabled_runtime_has_no_storage_requirement_or_generated_values(self): + base = RuntimeBase() + + runtime = LatticeAutoConfigRuntime(base, None, enabled=False) + + self.assertFalse(runtime.enabled) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + self.assertEqual(runtime.bind(gateway_enabled=True), ()) + + def test_gateway_publication_is_queued_then_applied_as_one_overlay(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime( + base, + MemoryAsyncStorage(), + enabled=True, + ) + self.assertEqual( + runtime.bind(gateway_enabled=True), + ("predbat-gateway",), + ) + + plan, result = runtime.ingest_gateway_status( + status(inverter("SER123")), + ) + + self.assertTrue(result.accepted) + self.assertTrue(result.staged) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + self.assertTrue(base.update_pending) + self.assertTrue(runtime.apply_pending()) + self.assertEqual(base.inverters, []) + self.assertEqual( + base.lattice_generated_overlay.read("battery_power"), + (True, ["sensor.predbat_gateway_ser123_battery_power"]), + ) + self.assertEqual(plan.selected_serials, ("SER123",)) + + def test_running_gateway_to_cloud_failover_replaces_the_overlay(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime( + base, + MemoryAsyncStorage(), + enabled=True, + ) + runtime.bind(gateway_enabled=True, gecloud_enabled=True) + runtime.ingest_gecloud_state( + ge_devices(), + ge_settings(), + {"ser123": {"info": {"model": "AC 3ph"}}}, + ) + runtime.ingest_gateway_status(status(inverter("SER123"))) + self.assertTrue(runtime.apply_pending()) + self.assertEqual( + base.lattice_generated_overlay.read("battery_power")[1], + ["sensor.predbat_gateway_ser123_battery_power"], + ) + + result = runtime.set_gateway_liveness(False) + + self.assertTrue(result.accepted) + self.assertTrue(result.staged) + self.assertTrue(runtime.apply_pending()) + self.assertEqual( + base.lattice_generated_overlay.read("battery_power")[1], + ["sensor.predbat_gecloud_ser123_battery_power"], + ) + + def test_pending_gateway_config_is_cancelled_if_provider_fails_before_apply(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime(base, MemoryAsyncStorage(), enabled=True) + runtime.bind(gateway_enabled=True) + runtime.ingest_gateway_status(status(inverter("SER123"))) + + invalidated = runtime.set_gateway_liveness(False) + + self.assertFalse(invalidated.accepted) + self.assertFalse(runtime.apply_pending()) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + + def test_rejection_cancels_pending_even_if_offline_persistence_fails(self): + base = RuntimeBase() + storage = MemoryAsyncStorage() + runtime = LatticeAutoConfigRuntime(base, storage, enabled=True) + runtime.bind(gateway_enabled=True) + runtime.ingest_gateway_status(status(inverter("SER123"))) + storage.fail_save = True + + with self.assertRaises(FragmentAdapterConflict): + runtime.set_gateway_liveness(False) + + self.assertFalse(runtime.apply_pending()) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + + def test_applied_overlay_is_removed_when_sole_provider_goes_offline(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime(base, MemoryAsyncStorage(), enabled=True) + runtime.bind(gateway_enabled=True) + runtime.ingest_gateway_status(status(inverter("SER123"))) + self.assertTrue(runtime.apply_pending()) + + invalidated = runtime.set_gateway_liveness(False) + + self.assertFalse(invalidated.accepted) + self.assertTrue(runtime.apply_pending()) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + + def test_return_to_applied_provider_cancels_unapplied_failover(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime(base, MemoryAsyncStorage(), enabled=True) + runtime.bind(gateway_enabled=True, gecloud_enabled=True) + runtime.ingest_gecloud_state( + ge_devices(), + ge_settings(), + {"ser123": {"info": {"model": "Hybrid"}}}, + ) + runtime.ingest_gateway_status(status(inverter("SER123"))) + self.assertTrue(runtime.apply_pending()) + + runtime.set_gateway_liveness(False) + runtime.ingest_gateway_status(status(inverter("SER123"))) + + self.assertFalse(runtime.apply_pending()) + self.assertEqual( + base.lattice_generated_overlay.read("battery_power")[1], + ["sensor.predbat_gateway_ser123_battery_power"], + ) + + def test_restart_does_not_activate_cached_plan_before_fresh_provider_read(self): + storage = MemoryAsyncStorage() + first_base = RuntimeBase() + first = LatticeAutoConfigRuntime(first_base, storage, enabled=True) + first.bind(gateway_enabled=True) + first.ingest_gateway_status(status(inverter("SER123"))) + self.assertTrue(first.apply_pending()) + + restarted_base = RuntimeBase() + restarted = LatticeAutoConfigRuntime( + restarted_base, + storage, + enabled=True, + ) + self.assertEqual( + restarted.gateway_publisher.read_snapshot().health, + ProviderHealth.OFFLINE, + ) + restarted.bind(gateway_enabled=True) + + retained_online = restarted.set_gateway_liveness(True) + self.assertFalse(retained_online.accepted) + self.assertFalse(restarted.apply_pending()) + + stale = restarted.reconcile() + + self.assertFalse(stale.accepted) + self.assertFalse(restarted.apply_pending()) + self.assertFalse(restarted_base.lattice_generated_overlay.snapshot.enabled) + _plan, fresh = restarted.ingest_gateway_status(status(inverter("SER123"))) + self.assertTrue(fresh.accepted) + self.assertTrue(restarted.apply_pending()) + + def test_restart_cloud_liveness_cannot_promote_cached_config(self): + storage = MemoryAsyncStorage() + first = LatticeAutoConfigRuntime(RuntimeBase(), storage, enabled=True) + first.bind(gecloud_enabled=True) + first.ingest_gecloud_state( + ge_devices(), + ge_settings(), + {"ser123": {"info": {"model": "Hybrid"}}}, + ) + + base = RuntimeBase() + restarted = LatticeAutoConfigRuntime(base, storage, enabled=True) + restarted.bind(gecloud_enabled=True) + + degraded = restarted.set_gecloud_liveness(None) + + self.assertFalse(degraded.accepted) + self.assertFalse(restarted.apply_pending()) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + + def test_component_stop_requires_new_gateway_telemetry_after_retained_online(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime(base, MemoryAsyncStorage(), enabled=True) + runtime.bind(gateway_enabled=True) + runtime.ingest_gateway_status(status(inverter("SER123"))) + self.assertTrue(runtime.apply_pending()) + + runtime.set_gateway_liveness(False) + retained = runtime.set_gateway_liveness(True) + + self.assertFalse(retained.accepted) + self.assertTrue(runtime.apply_pending()) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + + def test_component_stop_requires_new_ge_poll_after_degraded_health(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime(base, MemoryAsyncStorage(), enabled=True) + runtime.bind(gecloud_enabled=True) + runtime.ingest_gecloud_state( + ge_devices(), + ge_settings(), + {"ser123": {"info": {"model": "Hybrid"}}}, + ) + self.assertTrue(runtime.apply_pending()) + + runtime.set_gecloud_liveness(False) + degraded = runtime.set_gecloud_liveness(None) + + self.assertFalse(degraded.accepted) + self.assertTrue(runtime.apply_pending()) + self.assertFalse(base.lattice_generated_overlay.snapshot.enabled) + + def test_ac_coupled_hardware_fact_is_materialized_authoritatively(self): + base = RuntimeBase() + runtime = LatticeAutoConfigRuntime(base, MemoryAsyncStorage(), enabled=True) + runtime.bind(gecloud_enabled=True) + + runtime.ingest_gecloud_state( + ge_devices(), + ge_settings(), + {"ser123": {"info": {"model": "All-in-One"}}}, + ) + + self.assertTrue(runtime.apply_pending()) + self.assertEqual(base.lattice_generated_overlay.read("inverter_hybrid"), (True, False)) + self.assertTrue(base.lattice_generated_overlay.authoritative("inverter_hybrid")) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_compiled_publication.py b/apps/predbat/tests/test_lattice_compiled_publication.py index afdb205ae..1344370d0 100644 --- a/apps/predbat/tests/test_lattice_compiled_publication.py +++ b/apps/predbat/tests/test_lattice_compiled_publication.py @@ -1,6 +1,6 @@ """Tests for immutable durable publication of compiled Lattice state.""" -# cspell:ignore autoconfig +# cspell:ignore autoconfig materializable import os import sys @@ -1012,6 +1012,25 @@ def compare_and_publish(self, expected_version, publication): self.assertEqual(provider.calls, 1) self.assertEqual(override_reader.calls, 1) + def test_runtime_capabilities_are_preserved_in_durable_publication(self): + """The publishing compiler records a runtime-materializable plan.""" + provider = MutableReader(override_projection_snapshot()) + compiler = CompiledLatticeCompiler( + {"gateway": provider}, + state_store=InMemoryCompiledLatticeStateStore(), + atomic_materializer=True, + allow_provider_failover=True, + ) + + run = compiler.drain() + + self.assertTrue(run.published) + self.assertTrue(run.publication.plan.materialization_readiness.ready) + self.assertEqual( + run.publication.plan.materialization_readiness.blockers, + (), + ) + if __name__ == "__main__": unittest.main() diff --git a/apps/predbat/tests/test_lattice_durable_storage.py b/apps/predbat/tests/test_lattice_durable_storage.py new file mode 100644 index 000000000..a86692a41 --- /dev/null +++ b/apps/predbat/tests/test_lattice_durable_storage.py @@ -0,0 +1,468 @@ +"""Focused tests for production durable Lattice state stores.""" + +# cspell:ignore aiofiles autoconfig fsync + +import asyncio +import copy +import hashlib +import json +import os +import shutil +import sys +import tempfile +import threading +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_autoconfig import ( # noqa: E402 + AliasRole, + ProjectionCardinality, + ProjectionRouting, + ProjectionValueKind, + ProviderAlias, + ProviderConfigProjection, + ProviderHealth, + ProviderIdentityAlias, + ProviderProjectionValue, + ProviderRoleAssignment, + ProviderSnapshot, +) +from lattice_compiled_publication import ( # noqa: E402 + CompiledLatticeCompiler, + InMemoryCompiledLatticeStateStore, +) +from lattice_durable_storage import ( # noqa: E402 + LatticeDurableReadError, + LatticeDurableWriteError, + PredBatCompiledLatticeStateStore, + PredBatFragmentAdapterStateStore, + decode_compiled_lattice_publication, + decode_fragment_adapter_state, + encode_compiled_lattice_publication, + encode_fragment_adapter_state, +) +from lattice_fragment_adapters import ( # noqa: E402 + FragmentAdapterState, + _semantic_fingerprint, +) +from tests.test_lattice_autoconfig import snapshot # noqa: E402 + +try: # pragma: no cover - depends on the optional aiofiles test environment + from storage import StorageLocalFiles # noqa: E402 +except ModuleNotFoundError: # pragma: no cover + StorageLocalFiles = None + + +class MemoryStorage: + """Thread-safe async storage double with controllable write outcomes.""" + + def __init__(self): + """Create empty non-expiring storage.""" + self.values = {} + self.lock = threading.RLock() + self.write_mode = "ok" + self.saves = 0 + + async def load(self, module, filename): + """Return a detached stored value.""" + with self.lock: + value = self.values.get((module, filename)) + if value is None: + return None + return json.loads(json.dumps(value, allow_nan=False)) + + async def age(self, module, filename): + """Distinguish a present unreadable value from a missing key.""" + with self.lock: + return 0.0 if (module, filename) in self.values else None + + async def save(self, module, filename, data, format="yaml", expiry=None): + """Save, fail before save, or raise after an ambiguous successful save.""" + del format, expiry + with self.lock: + self.saves += 1 + if self.write_mode == "false_before": + return False + self.values[(module, filename)] = json.loads(json.dumps(data, allow_nan=False)) + if self.write_mode == "raise_after": + raise OSError("ambiguous fsync result") + return True + + +def rich_fragment_state(provider_id="gateway.local", generation=3, removed=False): + """Build one fragment exercising every persisted provider contract type.""" + node_id = "INV-1" + provider_snapshot = ProviderSnapshot( + provider_id=provider_id, + generation=generation, + health=ProviderHealth.DEGRADED, + topology_fragment={ + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": generation, + "producer": { + "name": "Gateway", + "provider": provider_id, + "authority": 10, + }, + "nodes": [{"id": node_id, "kind": "inverter"}], + "relationships": [], + }, + aliases=( + ProviderAlias( + "primary", + node_id, + frozenset((AliasRole.REFERENCE, AliasRole.PRIMARY)), + ), + ), + identity_aliases=(ProviderIdentityAlias("serial", "ABC123", node_id),), + role_assignments=(ProviderRoleAssignment(AliasRole.PRIMARY, "battery", 0, node_id),), + config_projections=( + ProviderConfigProjection( + "battery_power", + AliasRole.PRIMARY, + "battery", + ProjectionRouting.LEAF, + ProjectionCardinality.PER_INDEX, + ( + ProviderProjectionValue( + node_id, + ProjectionValueKind.ENTITY, + "sensor.gateway_battery_power", + capability="battery.power", + identity_kind="serial", + identity_value="ABC123", + access_path_id="gateway-mqtt", + ), + ), + required=True, + transforms=("watts",), + ), + ), + ) + return FragmentAdapterState( + provider_id, + generation, + _semantic_fingerprint(provider_snapshot, removed), + provider_snapshot, + removed, + ) + + +def compiled_publication(): + """Build one real compiled publication through the production compiler.""" + provider_snapshot = snapshot("gateway", generation=1) + compiler = CompiledLatticeCompiler( + {"gateway": lambda: provider_snapshot}, + state_store=InMemoryCompiledLatticeStateStore(), + ) + assert compiler.invalidate("gateway", 1, "initial durable discovery") + run = compiler.drain() + assert run.publication is not None + return run.publication + + +class TestLatticeDurableCodecs(unittest.TestCase): + """Verify strict, lossless codecs for both durable roots.""" + + def test_fragment_round_trip_preserves_immutable_contract(self): + """All provider enums, tuples, sets and mappings survive a round trip.""" + state = rich_fragment_state() + restored = decode_fragment_adapter_state(copy.deepcopy(encode_fragment_adapter_state(state))) + + self.assertEqual(restored, state) + self.assertEqual(restored.semantic_fingerprint, state.semantic_fingerprint) + self.assertIsInstance(restored.snapshot.topology_fragment, type(state.snapshot.topology_fragment)) + self.assertIsInstance(restored.snapshot.aliases[0].roles, frozenset) + + def test_fragment_tombstone_round_trip(self): + """Removal state remains removed with its removal-bound fingerprint.""" + state = rich_fragment_state(removed=True) + restored = decode_fragment_adapter_state(encode_fragment_adapter_state(state)) + + self.assertEqual(restored, state) + self.assertTrue(restored.removed) + + def test_compiled_publication_round_trip(self): + """A complete plan and durable cursor restore exactly.""" + publication = compiled_publication() + restored = decode_compiled_lattice_publication(copy.deepcopy(encode_compiled_lattice_publication(publication))) + + self.assertEqual(restored, publication) + self.assertEqual(restored.digest, publication.plan.digest) + self.assertEqual(restored.provider_fingerprints, publication.provider_fingerprints) + + def test_unknown_root_type_is_rejected(self): + """A whitelisted but incorrect durable root cannot cross store kinds.""" + encoded = encode_fragment_adapter_state(rich_fragment_state()) + with self.assertRaisesRegex(ValueError, "wrong root type"): + decode_compiled_lattice_publication(encoded) + + def test_unknown_dataclass_and_non_finite_values_are_rejected(self): + """The decoder and encoder accept no open-ended object types.""" + encoded = encode_fragment_adapter_state(rich_fragment_state()) + encoded["name"] = "ArbitraryCode" + with self.assertRaisesRegex(ValueError, "unsupported persisted dataclass"): + decode_fragment_adapter_state(encoded) + + state = rich_fragment_state() + object.__setattr__(state.snapshot.config_projections[0].values[0], "value", float("nan")) + with self.assertRaisesRegex(ValueError, "non-finite"): + encode_fragment_adapter_state(state) + + +class TestPredBatFragmentAdapterStateStore(unittest.TestCase): + """Verify restart durability, CAS and journal recovery for fragments.""" + + def test_cas_restart_and_stale_expected(self): + """Only the exact durable cursor can be replaced across restarts.""" + storage = MemoryStorage() + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + first = rich_fragment_state(generation=1) + second = rich_fragment_state(generation=2) + + self.assertIsNone(store.load()) + self.assertTrue(store.compare_and_store(None, first)) + self.assertFalse(store.compare_and_store(None, second)) + self.assertTrue(store.compare_and_store(first, second)) + + restarted = PredBatFragmentAdapterStateStore(storage, "gateway.local") + self.assertEqual(restarted.load(), second) + + def test_provider_ids_use_separate_journals(self): + """Provider names that sanitize alike cannot alias durable state.""" + storage = MemoryStorage() + dotted = PredBatFragmentAdapterStateStore(storage, "gateway.local") + slashed = PredBatFragmentAdapterStateStore(storage, "gateway/local") + dotted_state = rich_fragment_state("gateway.local", generation=1) + slashed_state = rich_fragment_state("gateway/local", generation=1) + + self.assertTrue(dotted.compare_and_store(None, dotted_state)) + self.assertTrue(slashed.compare_and_store(None, slashed_state)) + self.assertEqual(dotted.load(), dotted_state) + self.assertEqual(slashed.load(), slashed_state) + + def test_write_failure_retains_previous_slot(self): + """A failed inactive-slot write never replaces the durable winner.""" + storage = MemoryStorage() + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + first = rich_fragment_state(generation=1) + second = rich_fragment_state(generation=2) + self.assertTrue(store.compare_and_store(None, first)) + + storage.write_mode = "false_before" + with self.assertRaises(LatticeDurableWriteError): + store.compare_and_store(first, second) + storage.write_mode = "ok" + + self.assertEqual( + PredBatFragmentAdapterStateStore(storage, "gateway.local").load(), + first, + ) + + def test_ambiguous_success_is_recovered(self): + """An exception after the exact slot write is treated as committed.""" + storage = MemoryStorage() + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + state = rich_fragment_state(generation=1) + storage.write_mode = "raise_after" + + self.assertTrue(store.compare_and_store(None, state)) + storage.write_mode = "ok" + self.assertEqual( + PredBatFragmentAdapterStateStore(storage, "gateway.local").load(), + state, + ) + + def test_one_corrupt_slot_falls_back_but_two_fail_closed(self): + """The prior slot survives one torn write; no valid journal fails closed.""" + storage = MemoryStorage() + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + first = rich_fragment_state(generation=1) + second = rich_fragment_state(generation=2) + self.assertTrue(store.compare_and_store(None, first)) + self.assertTrue(store.compare_and_store(first, second)) + + keys = list(storage.values) + newest = max(keys, key=lambda key: storage.values[key]["sequence"]) + storage.values[newest]["checksum"] = "corrupt" + self.assertEqual(store.load(), first) + + for key in keys: + storage.values[key]["checksum"] = "corrupt" + with self.assertRaises(LatticeDurableReadError): + store.load() + + def test_duplicate_sequence_with_different_content_fails_closed(self): + """One storage revision cannot ambiguously name two different values.""" + storage = MemoryStorage() + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + first = rich_fragment_state(generation=1) + second = rich_fragment_state(generation=2) + self.assertTrue(store.compare_and_store(None, first)) + self.assertTrue(store.compare_and_store(first, second)) + + values = list(storage.values.values()) + older = min(values, key=lambda item: item["sequence"]) + newer = max(values, key=lambda item: item["sequence"]) + older["sequence"] = newer["sequence"] + content = {key: older[key] for key in ("schema", "kind", "sequence", "payload")} + older["checksum"] = hashlib.sha256( + json.dumps( + content, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + ).hexdigest() + + with self.assertRaisesRegex(LatticeDurableReadError, "reused"): + store.load() + + def test_same_expected_thread_race_has_one_winner(self): + """Two publishing threads cannot both commit the same cursor.""" + storage = MemoryStorage() + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + first = rich_fragment_state(generation=1) + candidates = ( + rich_fragment_state(generation=2), + rich_fragment_state(generation=3), + ) + self.assertTrue(store.compare_and_store(None, first)) + barrier = threading.Barrier(3) + results = [] + + def race(candidate): + """Attempt one CAS after both workers are ready.""" + barrier.wait() + results.append(store.compare_and_store(first, candidate)) + + threads = [threading.Thread(target=race, args=(item,)) for item in candidates] + for thread in threads: + thread.start() + barrier.wait() + for thread in threads: + thread.join() + + self.assertEqual(sorted(results), [False, True]) + self.assertIn(store.load(), candidates) + + def test_two_store_wrappers_share_the_same_cas_lock(self): + """Separate wrappers over one component still serialize one journal key.""" + storage = MemoryStorage() + first_store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + second_store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + first = rich_fragment_state(generation=1) + candidates = ( + rich_fragment_state(generation=2), + rich_fragment_state(generation=3), + ) + self.assertTrue(first_store.compare_and_store(None, first)) + barrier = threading.Barrier(3) + results = [] + + def race(store, candidate): + """Attempt one CAS through an independently constructed wrapper.""" + barrier.wait() + results.append(store.compare_and_store(first, candidate)) + + threads = [ + threading.Thread(target=race, args=(store, candidate)) + for store, candidate in zip( + (first_store, second_store), + candidates, + ) + ] + for thread in threads: + thread.start() + barrier.wait() + for thread in threads: + thread.join() + + self.assertEqual(sorted(results), [False, True]) + self.assertIn(first_store.load(), candidates) + + @unittest.skipIf(StorageLocalFiles is None, "aiofiles is not installed") + def test_real_local_storage_restart_round_trip(self): + """The journal survives actual JSON serialization and a fresh backend.""" + temp_dir = tempfile.mkdtemp() + logs = [] + try: + storage = StorageLocalFiles(temp_dir, logs.append) + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + state = rich_fragment_state(generation=1) + self.assertTrue(store.compare_and_store(None, state)) + + restarted_storage = StorageLocalFiles(temp_dir, logs.append) + restarted = PredBatFragmentAdapterStateStore( + restarted_storage, + "gateway.local", + ) + self.assertEqual(restarted.load(), state) + finally: + shutil.rmtree(temp_dir, ignore_errors=True) + + @unittest.skipIf(StorageLocalFiles is None, "aiofiles is not installed") + def test_real_local_storage_corrupt_metadata_fails_closed(self): + """Malformed sidecar metadata is not mistaken for an empty journal.""" + temp_dir = tempfile.mkdtemp() + logs = [] + try: + storage = StorageLocalFiles(temp_dir, logs.append) + store = PredBatFragmentAdapterStateStore(storage, "gateway.local") + self.assertTrue( + store.compare_and_store( + None, + rich_fragment_state(generation=1), + ) + ) + provider_key = hashlib.sha256(b"gateway.local").hexdigest() + meta_path = storage._meta_path( + "lattice_autoconfig", + "fragment_{}_1".format(provider_key), + ) + with open(meta_path, "w", encoding="utf-8") as metadata: + metadata.write("{broken") + + with self.assertRaisesRegex(LatticeDurableReadError, "unreadable"): + PredBatFragmentAdapterStateStore( + StorageLocalFiles(temp_dir, logs.append), + "gateway.local", + ) + finally: + shutil.rmtree(temp_dir, ignore_errors=True) + + def test_store_rejects_active_event_loop(self): + """Synchronous persistence cannot silently block an asyncio owner loop.""" + storage = MemoryStorage() + + async def construct(): + return PredBatFragmentAdapterStateStore(storage, "gateway.local") + + with self.assertRaisesRegex(RuntimeError, "outside an active asyncio"): + asyncio.run(construct()) + + +class TestPredBatCompiledLatticeStateStore(unittest.TestCase): + """Verify compiled publication version CAS and restart behavior.""" + + def test_publication_round_trip_and_version_guard(self): + """Only version one can replace an empty compiled journal.""" + storage = MemoryStorage() + store = PredBatCompiledLatticeStateStore(storage) + publication = compiled_publication() + + self.assertTrue(store.compare_and_publish(0, publication)) + self.assertFalse(store.compare_and_publish(0, publication)) + self.assertEqual( + PredBatCompiledLatticeStateStore(storage).load(), + publication, + ) + with self.assertRaisesRegex(ValueError, "exactly expected_version"): + store.compare_and_publish(1, publication) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_gateway_fragment.py b/apps/predbat/tests/test_lattice_gateway_fragment.py index 6ef8b98f2..678b2bb11 100644 --- a/apps/predbat/tests/test_lattice_gateway_fragment.py +++ b/apps/predbat/tests/test_lattice_gateway_fragment.py @@ -174,6 +174,7 @@ def test_pure_plan_publishes_materializable_roles_and_config(self): "kind": "inverter", "battery_present": True, "battery_capacity_wh": 9600, + "model": "All-in-One", }, ) ) @@ -191,11 +192,32 @@ def test_pure_plan_publishes_materializable_roles_and_config(self): compiled.projected_config["battery_power"], ("sensor.predbat_gateway_-inv-1_battery_power",), ) + self.assertFalse(compiled.projected_config["inverter_hybrid"]) self.assertEqual( snapshot.identity_aliases[-1].value, "GW-INV-1", ) + def test_complete_ingest_publishes_topology_and_config_in_one_generation(self): + """Live runtime never persists a topology-only intermediate state.""" + adapter, state_store = publisher() + plan = compile_gateway_auto_config( + ( + { + "serial": "GW-INV-1", + "kind": "inverter", + "battery_present": True, + "battery_capacity_wh": 9600, + }, + ) + ) + + self.assertTrue(adapter.ingest_complete(topology(), plan, online=True)) + + self.assertEqual(state_store.writes, 1) + self.assertEqual(adapter.generation, 1) + self.assertTrue(adapter.read_snapshot().config_projections) + def test_projection_metadata_retains_provider_local_capability_coordinates(self): """Projection inspection exposes refs but no materialization authority.""" adapter, _state_store = publisher() diff --git a/apps/predbat/tests/test_lattice_gateway_ge_composition.py b/apps/predbat/tests/test_lattice_gateway_ge_composition.py index d5ed3a284..8c0c76b3f 100644 --- a/apps/predbat/tests/test_lattice_gateway_ge_composition.py +++ b/apps/predbat/tests/test_lattice_gateway_ge_composition.py @@ -160,6 +160,83 @@ def test_gateway_wins_and_cloud_takes_over_when_gateway_offline(self): ("sensor.predbat_gecloud_ser123_battery_power",), ) + def test_running_compiler_fails_over_to_cloud_when_gateway_goes_offline(self): + """The opt-in runtime policy replaces an unavailable local provider.""" + serial = "SER123" + gateway = GatewayRetainedTopologyFragmentPublisher( + "predbat-gateway", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + gateway.ingest_retained_topology( + gateway_topology(serial), + online=True, + ) + gateway.ingest_auto_config( + compile_gateway_auto_config( + ( + { + "serial": serial, + "kind": "inverter", + "battery_present": True, + "battery_capacity_wh": 9600, + "model": "Hybrid", + }, + ) + ) + ) + cloud = GECloudFragmentPublisher( + "ge-cloud", + InMemoryFragmentAdapterStateStore(), + enabled=True, + ) + cloud.ingest_discovery( + 1, + ( + GECloudDeviceSnapshot( + serial=serial.lower(), + kind="battery-inverter", + model="Hybrid", + online=True, + ), + ), + health=True, + ) + cloud.ingest_auto_config(ge_config(serial)) + compiler = LatticeAutoConfigCompiler( + { + "predbat-gateway": gateway.read_snapshot, + "ge-cloud": cloud.read_snapshot, + }, + atomic_materializer=True, + allow_provider_failover=True, + ) + + first = compiler.drain() + self.assertEqual( + first.plan.projected_config["battery_power"], + ("sensor.predbat_gateway_ser123_battery_power",), + ) + + self.assertTrue(gateway.set_liveness(False)) + self.assertTrue( + compiler.invalidate( + "predbat-gateway", + gateway.generation, + "gateway offline", + ) + ) + fallback = compiler.drain() + + self.assertEqual( + fallback.plan.projected_config["battery_power"], + ("sensor.predbat_gecloud_ser123_battery_power",), + ) + self.assertNotIn( + "active_provider_unavailable", + {issue.code for issue in fallback.issues}, + ) + if __name__ == "__main__": unittest.main() diff --git a/apps/predbat/tests/test_lattice_ge_cloud_fragment.py b/apps/predbat/tests/test_lattice_ge_cloud_fragment.py index d52d49d14..f490f3c5f 100644 --- a/apps/predbat/tests/test_lattice_ge_cloud_fragment.py +++ b/apps/predbat/tests/test_lattice_ge_cloud_fragment.py @@ -204,6 +204,44 @@ def test_pure_plan_publishes_materializable_roles_and_config(self): compiled.projected_config["charge_rate"], ("number.predbat_gecloud_bat1_battery_charge_power",), ) + self.assertTrue(compiled.projected_config["inverter_hybrid"]) + + def test_complete_ingest_publishes_discovery_and_config_in_one_generation(self): + """Live runtime never persists a discovery-only intermediate state.""" + adapter, state_store = publisher() + devices = (device(serial="bat1", model="Hybrid"),) + config = GECloudAutoConfigInput( + batteries=("bat1",), + ems=None, + gateway=None, + pv=(), + battery_meters=(("bat1", ("meter1",)),), + register_names=( + ( + "bat1", + ( + "battery_charge_power", + "battery_discharge_power", + "battery_reserve_percent_limit", + ), + ), + ), + models=(("bat1", "Hybrid"),), + prefix="predbat", + ) + + self.assertTrue( + adapter.ingest_complete( + 1, + devices, + config, + health=ProviderHealth.HEALTHY, + ) + ) + + self.assertEqual(state_store.writes, 1) + self.assertEqual(adapter.generation, 1) + self.assertTrue(adapter.read_snapshot().config_projections) def test_device_order_and_serial_case_are_replay_stable(self): """Equivalent provider snapshots do not invent new generations.""" diff --git a/apps/predbat/tests/test_lattice_generated_overlay.py b/apps/predbat/tests/test_lattice_generated_overlay.py new file mode 100644 index 000000000..5feecb1a1 --- /dev/null +++ b/apps/predbat/tests/test_lattice_generated_overlay.py @@ -0,0 +1,179 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - generated Lattice configuration overlay tests +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- + +import importlib.util +from pathlib import Path +import unittest +import sys +import types + +from lattice_generated_overlay import GeneratedConfigOverlay + + +def _load_user_interface(): + """Load the mixin directly without changing the process's real modules.""" + saved_config = sys.modules.get("config") + saved_predbat = sys.modules.get("predbat") + config_stub = types.ModuleType("config") + config_stub.CONFIG_API_OVERRIDE = {"inverter_limit": True} + predbat_stub = types.ModuleType("predbat") + predbat_stub.THIS_VERSION = "test" + try: + sys.modules["config"] = config_stub + sys.modules["predbat"] = predbat_stub + module_path = Path(__file__).resolve().parents[1] / "userinterface.py" + spec = importlib.util.spec_from_file_location("_lattice_overlay_userinterface", module_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.UserInterface + finally: + if saved_config is None: + sys.modules.pop("config", None) + else: + sys.modules["config"] = saved_config + if saved_predbat is None: + sys.modules.pop("predbat", None) + else: + sys.modules["predbat"] = saved_predbat + + +UserInterface = _load_user_interface() + + +class OverlayUserInterface(UserInterface): + """Minimal argument-reader harness with controllable API and HA sources.""" + + def __init__(self, args=None, overlay=None, ha_values=None, api_values=None): + self.args = dict(args or {}) + self.lattice_generated_overlay = overlay + self.ha_values = dict(ha_values or {}) + self.api_values = dict(api_values or {}) + + def get_ha_config(self, name, default): + if name in self.ha_values: + return self.ha_values[name], default + return None, default + + def get_manual_api(self, name): + return list(self.api_values.get(name, ())) + + def get_state_wrapper(self, entity_id=None, default=None, **kwargs): + return default + + def log(self, message): + pass + + def record_status(self, message, **kwargs): + pass + + +class TestGeneratedConfigOverlay(unittest.TestCase): + """Verify immutable publication and effective argument precedence.""" + + def test_overlay_is_default_off_and_absent_overlay_keeps_legacy_default(self): + overlay = GeneratedConfigOverlay() + overlay.replace({"example": "generated"}) + overlay.disable() + + disabled = OverlayUserInterface(overlay=overlay) + absent = OverlayUserInterface(overlay=None) + + self.assertEqual(disabled.get_arg("example", "default", indirect=False), "default") + self.assertEqual(absent.get_arg("example", "default", indirect=False), "default") + + def test_precedence_is_ha_then_explicit_then_generated_then_default(self): + overlay = GeneratedConfigOverlay() + overlay.replace( + { + "ha_wins": "generated", + "explicit_wins": "generated", + "generated_wins": "generated", + } + ) + ui = OverlayUserInterface( + args={"ha_wins": "apps", "explicit_wins": "apps"}, + overlay=overlay, + ha_values={"ha_wins": "ha"}, + ) + + self.assertEqual(ui.get_arg("ha_wins", "default", indirect=False), "ha") + self.assertEqual(ui.get_arg("explicit_wins", "default", indirect=False), "apps") + self.assertEqual(ui.get_arg("generated_wins", "default", indirect=False), "generated") + self.assertEqual(ui.get_arg("missing", "default", indirect=False), "default") + + def test_topology_owned_hardware_fact_precedes_ha_but_allows_explicit_escape_hatch(self): + overlay = GeneratedConfigOverlay(authoritative_arguments=("inverter_hybrid",)) + overlay.replace({"inverter_hybrid": False}) + + generated = OverlayUserInterface( + overlay=overlay, + ha_values={"inverter_hybrid": True}, + ) + explicit = OverlayUserInterface( + args={"inverter_hybrid": True}, + overlay=overlay, + ha_values={"inverter_hybrid": False}, + ) + + self.assertFalse(generated.get_arg("inverter_hybrid", True, indirect=False)) + self.assertTrue(explicit.get_arg("inverter_hybrid", False, indirect=False)) + + def test_explicit_none_suppresses_generated_value(self): + overlay = GeneratedConfigOverlay() + overlay.replace({"example": "generated"}) + ui = OverlayUserInterface(args={"example": None}, overlay=overlay) + + self.assertIsNone(ui.get_arg("example", "default", indirect=False)) + + def test_generated_per_index_value_uses_requested_index(self): + overlay = GeneratedConfigOverlay() + overlay.replace({"battery_rate_max": [3000, 4000]}) + ui = OverlayUserInterface(overlay=overlay) + + self.assertEqual(ui.get_arg("battery_rate_max", 0.0, index=1, indirect=False), 4000.0) + + def test_api_list_override_does_not_mutate_generated_snapshot(self): + overlay = GeneratedConfigOverlay() + overlay.replace({"inverter_limit": [3000, 4000]}) + ui = OverlayUserInterface( + overlay=overlay, + api_values={"inverter_limit": ({"index": 1, "value": "4500"},)}, + ) + + self.assertEqual(ui.get_arg("inverter_limit", [0, 0], indirect=False), [3000, 4500]) + + ui.api_values = {} + self.assertEqual(ui.get_arg("inverter_limit", [0, 0], indirect=False), [3000, 4000]) + self.assertEqual(overlay.read("inverter_limit"), (True, [3000, 4000])) + + def test_domain_arguments_never_fall_back_to_generated_overlay(self): + overlay = GeneratedConfigOverlay() + overlay.replace({"example": "generated"}) + ui = OverlayUserInterface(args={"provider": {"example": "domain"}}, overlay=overlay) + + self.assertEqual(ui.get_arg("example", "default", domain="provider", indirect=False), "domain") + self.assertEqual(ui.get_arg("missing", "default", domain="provider", indirect=False), "default") + + def test_replace_is_full_replacement_and_reads_are_detached(self): + overlay = GeneratedConfigOverlay() + original = {"removed": [1], "retained": {"nested": [2]}} + first = overlay.replace(original) + original["retained"]["nested"].append(99) + + present, value = overlay.read("retained") + self.assertTrue(present) + self.assertEqual(value, {"nested": [2]}) + value["nested"].append(3) + self.assertEqual(overlay.read("retained"), (True, {"nested": [2]})) + + second = overlay.replace({"retained": [4]}) + self.assertEqual(second.version, first.version + 1) + self.assertEqual(overlay.read("removed"), (False, None)) + self.assertEqual(overlay.read("retained"), (True, [4])) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_provider_inputs.py b/apps/predbat/tests/test_lattice_provider_inputs.py new file mode 100644 index 000000000..6320f6397 --- /dev/null +++ b/apps/predbat/tests/test_lattice_provider_inputs.py @@ -0,0 +1,215 @@ +# cspell:ignore autoconfig xiaozhi XIAO BATONE PVONE +"""Focused tests for pure live provider input adapters.""" + +import os +import sys +import unittest +from types import SimpleNamespace + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from gateway_autoconfig import compile_gateway_auto_config # noqa: E402 +from gecloud_autoconfig import compile_gecloud_auto_config # noqa: E402 +from lattice_provider_inputs import ( # noqa: E402 + LiveProviderInputError, + gateway_status_to_lattice_input, + gecloud_state_to_lattice_input, +) +from lattice_topology import decode_topology # noqa: E402 + + +class ProtoMessage: + """Small protobuf-like object with explicit message presence.""" + + def __init__(self, present=True, **values): + self._present = present + for name, value in values.items(): + setattr(self, name, value) + + def ByteSize(self): + """Return protobuf-like presence, independent of field values.""" + return 1 if self._present else 0 + + +def inverter( + serial="CE123456789", + inverter_type=6, + primary=True, + capacity_wh=9500, + model="Hybrid 5.0", +): + """Build one XIAO Gateway inverter telemetry entry.""" + return SimpleNamespace( + serial=serial, + type=inverter_type, + primary=primary, + battery=ProtoMessage(capacity_wh=capacity_wh), + ems=ProtoMessage(present=False, num_inverters=0), + model=model, + ) + + +def status(*inverters, ev_chargers=(), timestamp=0): + """Build one proto-like Gateway status frame.""" + return SimpleNamespace( + inverters=inverters, + ev_chargers=ev_chargers, + timestamp=timestamp, + ) + + +class TestGatewayLiveInput(unittest.TestCase): + """Gateway telemetry is detached without live component side effects.""" + + def test_xiaozhi_single_aio_compiles_and_seeds_valid_topology(self): + live = gateway_status_to_lattice_input(status(inverter()), document_version=4) + plan = compile_gateway_auto_config(live.inverters) + document = decode_topology(live.topology_fragment) + + self.assertEqual(plan.selected_serials, ("CE123456789",)) + self.assertEqual(plan.arguments["inverter_type"], ("GWMQTT",)) + self.assertEqual(plan.arguments["soc_max"], ("sensor.predbat_gateway_456789_battery_capacity",)) + self.assertEqual(document["docVersion"], 4) + self.assertEqual(document["nodes"][0]["id"], "gateway:CE123456789") + self.assertEqual(document["nodes"][0]["attributes"]["serial"], "CE123456789") + + def test_topology_is_order_stable_and_version_is_external_metadata(self): + first = gateway_status_to_lattice_input( + status(inverter("B"), inverter("A"), timestamp=100), + document_version=1, + ) + second = gateway_status_to_lattice_input( + status(inverter("A"), inverter("B"), timestamp=999), + document_version=9, + ) + first_document = dict(first.topology_fragment) + second_document = dict(second.topology_fragment) + first_document.pop("docVersion") + second_document.pop("docVersion") + + self.assertEqual(first_document, second_document) + self.assertEqual( + [node["id"] for node in first_document["nodes"]], + ["gateway:A", "gateway:B"], + ) + + def test_ems_ev_and_duplicate_serials_fail_closed(self): + with self.assertRaisesRegex(LiveProviderInputError, "EMS"): + gateway_status_to_lattice_input(status(inverter(inverter_type=7))) + with self.assertRaisesRegex(LiveProviderInputError, "EV"): + gateway_status_to_lattice_input( + status(inverter(), ev_chargers=(SimpleNamespace(),)), + ) + with self.assertRaisesRegex(LiveProviderInputError, "duplicate"): + gateway_status_to_lattice_input(status(inverter("same"), inverter("SAME"))) + + +class TestGECloudLiveInput(unittest.TestCase): + """GE discovery, register state, and flags map to existing contracts.""" + + def test_registers_models_meters_and_flags_are_detached(self): + live = gecloud_state_to_lattice_input( + devices={ + "battery": ["BatOne"], + "pv": ["PvOne"], + "gateway": None, + "ems": None, + "battery_meters": {"BatOne": [12345, 98765]}, + }, + settings={ + "BatOne": { + 1: {"name": "Battery Charge Power"}, + 2: {"name": "Battery Reserve % Limit"}, + 3: {"name": "Enable-AC-Charge"}, + }, + "PvOne": {}, + }, + info={ + "BatOne": { + "info": {"model": "All-In-One 6.0"}, + "site_id": 321, + "uuid": "battery-uuid", + }, + "PvOne": {"info": {"model": "GIV-PV"}, "site_id": 321}, + }, + prefix="site", + load_today_ignore=True, + split_pv=True, + split_ct=True, + shared_ct=False, + ) + config = live.auto_config + plan = compile_gecloud_auto_config(config) + + self.assertEqual(config.batteries, ("BatOne",)) + self.assertEqual(config.battery_meters, (("BatOne", (12345, 98765)),)) + self.assertEqual(config.models, (("BatOne", "All-In-One 6.0"),)) + self.assertEqual( + dict(config.register_names)["BatOne"], + ( + "battery_charge_power", + "battery_reserve_percent_limit", + "enable_ac_charge", + ), + ) + self.assertTrue(config.load_today_ignore) + self.assertTrue(config.split_pv) + self.assertTrue(config.split_ct) + self.assertFalse(config.shared_ct) + self.assertNotIn("load_today", plan.legacy_arguments()) + self.assertTrue(plan.ac_coupled) + self.assertEqual(plan.pv_sources, ("BatOne", "PvOne")) + self.assertEqual( + [(item.serial, item.kind, item.model, item.site_id) for item in live.devices], + [ + ("BATONE", "battery-inverter", "All-In-One 6.0", "321"), + ("PVONE", "pv-inverter", "GIV-PV", "321"), + ], + ) + + def test_gateway_registers_and_shared_meter_inputs_are_included(self): + live = gecloud_state_to_lattice_input( + devices={ + "battery": ["bat2", "bat1"], + "pv": [], + "gateway": "gw1", + "ems": None, + "battery_meters": { + "bat1": [44], + "bat2": [44], + }, + }, + settings={ + "bat1": {}, + "bat2": {}, + "gw1": {1: {"name": "Charge Power Rate"}}, + }, + info={ + "bat1": {"info": {"model": "Hybrid"}}, + "bat2": {"info": {"model": "Hybrid"}}, + "gw1": {"info": {"model": "Gateway"}}, + }, + shared_ct=True, + ) + config = live.auto_config + plan = compile_gecloud_auto_config(config) + + self.assertEqual(dict(config.register_names)["gw1"], ("charge_power_rate",)) + self.assertEqual(plan.primary_targets, ("gw1",)) + self.assertEqual(plan.arguments[0], ("inverter_type", ("GEC",))) + self.assertTrue(config.shared_ct) + + def test_missing_batteries_and_non_boolean_flags_fail_closed(self): + with self.assertRaisesRegex(LiveProviderInputError, "no batteries"): + gecloud_state_to_lattice_input({}, {}, {}) + with self.assertRaisesRegex(LiveProviderInputError, "split_pv"): + gecloud_state_to_lattice_input( + {"battery": ["bat"], "battery_meters": {}}, + {"bat": {}}, + {}, + split_pv="true", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/userinterface.py b/apps/predbat/userinterface.py index 59eab2f69..4b3681401 100644 --- a/apps/predbat/userinterface.py +++ b/apps/predbat/userinterface.py @@ -1,3 +1,4 @@ +# cspell:ignore autoconfig # ----------------------------------------------------------------------------- # Predbat Home Battery System # Copyright Trefor Southwell 2026 - All Rights Reserved @@ -238,19 +239,51 @@ def get_arg(self, arg, default=None, indirect=True, combine=False, attribute=Non self.log("Note: API Overridden arg {} value {}".format(arg, value)) break + generated_present = False + generated_value = None + generated_authoritative = False + generated_overlay = None + if not domain: + generated_overlay = getattr(self, "lattice_generated_overlay", None) + if generated_overlay is not None: + generated_present, generated_value = generated_overlay.read(arg) + generated_authoritative = generated_overlay.authoritative(arg) + + # Topology-owned hardware facts take precedence over the legacy HA entity, + # but an explicit apps.yaml value remains an escape hatch. API overrides + # have already been resolved above. Ordinary generated arguments remain + # fallbacks and retain the legacy HA precedence. + if value is None and generated_authoritative: + if arg in self.args: + value = self.args[arg] + else: + value = generated_value + # Get From HA config (not for domain specific which are apps.yaml options only) if value is None and not domain: value, default = self.get_ha_config(arg, default) # Resolve locally if no HA config if value is None: - if (arg not in self.args) and (default is not None) and (index is not None): + if not domain and arg not in self.args and generated_overlay is None: + # Generated Lattice configuration is an optional, default-off fallback. + # Its reader returns a detached value so API list overrides and other + # legacy resolution paths cannot mutate the immutable publication. + generated_overlay = getattr(self, "lattice_generated_overlay", None) + if generated_overlay is not None: + generated_present, generated_value = generated_overlay.read(arg) + if (arg not in self.args) and (not generated_present) and (default is not None) and (index is not None): # Allow default to apply to all indices if there is not config item set index = None if domain: value = self.args.get(domain, {}).get(arg, default) + elif arg in self.args: + # Explicit apps.yaml membership always wins, including an explicit None. + value = self.args[arg] + elif generated_present: + value = generated_value else: - value = self.args.get(arg, default) + value = default value = self.resolve_arg(arg, value, default=default, indirect=indirect, combine=combine, attribute=attribute, index=index, required_unit=required_unit) if isinstance(default, float): @@ -933,6 +966,10 @@ def load_user_config(self, quiet=True, register=False, load_config=False): Load config from HA """ + lattice_runtime = getattr(self, "lattice_autoconfig_runtime", None) + if lattice_runtime is not None: + lattice_runtime.apply_pending() + self.config_index = {} self.log("Refreshing Predbat configuration")