From ea3aa1b1df1a22897c6b2aac1297c7ebfa892e9c Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Sun, 26 Jul 2026 21:17:10 +0100 Subject: [PATCH] feat(lattice): ingest gateway topology with provenance --- apps/predbat/components.py | 1 + apps/predbat/config.py | 6 + apps/predbat/gateway.py | 59 ++- apps/predbat/lattice_topology.py | 410 ++++++++++++++++++++ apps/predbat/tests/test_lattice_topology.py | 189 +++++++++ 5 files changed, 662 insertions(+), 3 deletions(-) create mode 100644 apps/predbat/lattice_topology.py create mode 100644 apps/predbat/tests/test_lattice_topology.py diff --git a/apps/predbat/components.py b/apps/predbat/components.py index 7eb6405ab..c1a49c5f9 100644 --- a/apps/predbat/components.py +++ b/apps/predbat/components.py @@ -511,6 +511,7 @@ "gateway_inverter_serial": {"required": False, "config": "gateway_inverter_serial", "default": None}, "gateway_evc_automatic": {"required": False, "config": "gateway_evc_automatic", "default": False}, "gateway_evc_control": {"required": False, "config": "gateway_evc_control", "default": False}, + "lattice_projection_enable": {"required": False, "config": "lattice_projection_enable", "default": False}, }, "phase": 1, "can_restart": True, diff --git a/apps/predbat/config.py b/apps/predbat/config.py index e8f0e444d..47b80482b 100644 --- a/apps/predbat/config.py +++ b/apps/predbat/config.py @@ -44,6 +44,12 @@ "type": "switch", "default": False, }, + { + "name": "lattice_projection_enable", + "friendly_name": "Lattice Topology Shadow (experimental)", + "type": "switch", + "default": False, + }, { "name": "active", "friendly_name": "Predbat Active", diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 850c6015c..7cfd4c02f 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -19,6 +19,7 @@ import pytz as _pytz from component_base import ComponentBase +from lattice_topology import LatticeTopologyStore, TopologyValidationError try: import gateway_status_pb2 as pb @@ -171,7 +172,18 @@ class GatewayMQTT(ComponentBase): Instance methods handle MQTT lifecycle and ComponentBase integration. """ - def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqtt_token=None, gateway_inverter_serial=None, gateway_evc_automatic=False, gateway_evc_control=False, **kwargs): + def initialize( + self, + gateway_device_id=None, + mqtt_host=None, + mqtt_port=8883, + mqtt_token=None, + gateway_inverter_serial=None, + gateway_evc_automatic=False, + gateway_evc_control=False, + lattice_projection_enable=False, + **kwargs, + ): """Initialize gateway configuration and build MQTT topic strings. Args: @@ -188,11 +200,14 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt the current time falls inside a planned car-charging window and send RemoteStartTransaction plus SetChargingProfile on window entry, or RemoteStopTransaction on window exit. When enabled, car_charging_now is omitted from auto-config to prevent a feedback loop. + lattice_projection_enable: Enable read-only retained topology ingestion and shadow diagnostics. + Off by default. This does not publish Lattice control messages or alter legacy commands. **kwargs: Additional keyword arguments (ignored). """ self.gateway_device_id = gateway_device_id self.gateway_evc_automatic = bool(gateway_evc_automatic) self.gateway_evc_control = bool(gateway_evc_control) + self.lattice_projection_enable = bool(lattice_projection_enable) self.mqtt_host = mqtt_host self.mqtt_port = mqtt_port self.mqtt_token = mqtt_token @@ -223,6 +238,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt self.topic_schedule = f"{self._topic_base}/schedule" self.topic_command = f"{self._topic_base}/command" self.topic_ev_command = f"{self._topic_base}/ev/command" + self.topic_topology = f"{self._topic_base}/topology" # Runtime state self._mqtt_client = None @@ -240,6 +256,7 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt self._plan_version = 0 self._refresh_in_progress = False self._error_count = 0 + self._lattice_topology = LatticeTopologyStore() # Entity naming prefix self.prefix = "predbat" @@ -603,10 +620,15 @@ async def _mqtt_loop(self): backoff = 5 # Reset backoff on successful connection self.log(f"Info: GatewayMQTT: Connected to {self.mqtt_host}:{self.mqtt_port}") - # Subscribe to status and LWT topics + # Subscribe to status and LWT topics. The retained Lattice topology is + # read only and remains completely absent while the shadow flag is off. await client.subscribe(self.topic_status, qos=1) await client.subscribe(self.topic_online, qos=1) - self.log(f"Info: GatewayMQTT: Subscribed to {self.topic_status} and {self.topic_online}") + subscribed_topics = [self.topic_status, self.topic_online] + if self.lattice_projection_enable: + await client.subscribe(self.topic_topology, qos=1) + subscribed_topics.append(self.topic_topology) + self.log("Info: GatewayMQTT: Subscribed to {}".format(", ".join(subscribed_topics))) async for message in client.messages: if self.api_stop: @@ -649,6 +671,8 @@ async def _handle_message(self, message): try: if topic == self.topic_status: self._process_telemetry(message.payload) + elif topic == self.topic_topology and self.lattice_projection_enable: + self._process_lattice_topology(message.payload) elif topic == self.topic_online: payload = message.payload.decode("utf-8", errors="replace").strip() was_online = self._gateway_online @@ -666,6 +690,35 @@ async def _handle_message(self, message): self.log(f"Warn: GatewayMQTT: Error handling message on {topic}: {e}") self.log(f"Warn: {traceback.format_exc()}") + def _process_lattice_topology(self, payload): + """Replace one provider topology and report read-only shadow diagnostics.""" + try: + changed = self._lattice_topology.ingest(payload) + except TopologyValidationError as exc: + self._error_count += 1 + self.log("Warn: GatewayMQTT: Ignoring invalid Lattice topology: {}".format(exc)) + return False + if not changed: + return False + snapshot = self._lattice_topology.snapshot + self.log( + "Info: GatewayMQTT: Lattice shadow topology v{} doc {}: {} node(s), {} source binding(s)".format( + snapshot.site["topologyVersion"], + snapshot.site["docVersion"], + len(snapshot.site.get("nodes", [])), + len(snapshot.provenance), + ) + ) + for warning in snapshot.warnings: + self.log("Warn: GatewayMQTT: Lattice shadow: {}".format(warning)) + return True + + def lattice_source_ref(self, node_id, capability, access_path=None): + """Return provider-local topology coordinates for a future dispatcher.""" + if not self.lattice_projection_enable: + return None + return self._lattice_topology.source_ref(node_id, capability, access_path) + 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/lattice_topology.py b/apps/predbat/lattice_topology.py new file mode 100644 index 000000000..3fca096a5 --- /dev/null +++ b/apps/predbat/lattice_topology.py @@ -0,0 +1,410 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Lattice topology shadow store (read-only) +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Read-only Lattice topology ingestion, merge, and source-reference provenance. + +The merged site document remints capability ``ref`` values. Those refs are not +safe to send to a provider: a control dispatcher must use the ref from the +provider-local source document. This module therefore keeps an out-of-band +sidecar keyed by ``(node id, capability, access path)``. + +There are deliberately no MQTT or Predbat dependencies here. Control +publishing, planner writes, and inverter writes are outside this module. +""" + +import copy +import json +from dataclasses import dataclass +from typing import Optional + + +SUPPORTED_TOPOLOGY_VERSIONS = frozenset(("0.2", "0.3")) + + +class TopologyValidationError(ValueError): + """Raised when an incoming topology document is unsafe to ingest.""" + + +@dataclass(frozen=True) +class SourceCapabilityRef: + """Provider-local coordinates for one winning capability offer.""" + + provider: str + topology_version: str + doc_version: int + cap_ref: int + node_id: str + capability: str + access_path: str + + +@dataclass(frozen=True) +class TopologySnapshot: + """An immutable-by-convention merged site and its provenance sidecar.""" + + site: dict + provenance: dict + warnings: tuple + + def source_ref(self, node_id, capability, access_path=None) -> Optional[SourceCapabilityRef]: + """Return the winning provider-local ref for a future dispatcher. + + When ``access_path`` is omitted, select the highest-preference merged + access path that offers the capability. Exact lookup remains available + so a dispatcher can follow a resolver's chosen fallback path. + """ + node_key = str(node_id) + capability_key = str(capability) + if access_path is not None: + return self.provenance.get((node_key, capability_key, str(access_path))) + + node = next((item for item in self.site.get("nodes", []) if str(item.get("id")) == node_key), None) + if node is None: + return None + preferences = {str(path.get("id")): path.get("preference", 0) or 0 for path in node.get("accessPaths", []) if path.get("id") is not None} + candidates = [] + for key, source in self.provenance.items(): + if key[0] == node_key and key[1] == capability_key: + candidates.append((-(preferences.get(key[2], 0)), key[2], source)) + candidates.sort(key=lambda item: (item[0], item[1])) + return candidates[0][2] if candidates else None + + +def _topology_family(version): + """Return the supported major/minor family for a topology version.""" + parts = str(version).split(".") + return ".".join(parts[:2]) if len(parts) >= 2 else str(version) + + +def _authority(document): + """Return producer authority, excluding bools from Python's integer type.""" + value = (document.get("producer") or {}).get("authority") + return value if isinstance(value, int) and not isinstance(value, bool) else 0 + + +def _doc_version(document): + """Return a validated document version, defaulting legacy 0.2 fragments to zero.""" + value = document.get("docVersion") + if value is None and _topology_family(document.get("topologyVersion")) == "0.2": + return 0 + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise TopologyValidationError("docVersion must be a non-negative integer") + return value + + +def _rank(document, order): + """Build the authority, recency, input-order rank tuple.""" + return (_authority(document), _doc_version(document), -order) + + +def _better(left, right): + """Return whether ``left`` outranks ``right``.""" + return left > right + + +def _winner(contributions): + """Return the highest-ranked contribution, preserving first on an exact tie.""" + best = contributions[0] + for contribution in contributions[1:]: + if _better(contribution[1], best[1]): + best = contribution + return best + + +def _validate_document(document): + """Validate the minimum shape needed for deterministic safe ingestion.""" + if not isinstance(document, dict): + raise TopologyValidationError("topology payload must be a JSON object") + version = document.get("topologyVersion") + if _topology_family(version) not in SUPPORTED_TOPOLOGY_VERSIONS: + raise TopologyValidationError("unsupported topologyVersion {!r}".format(version)) + if document.get("scope") not in ("fragment", "overlay", "site"): + raise TopologyValidationError("scope must be fragment, overlay, or site") + producer = document.get("producer") + if not isinstance(producer, dict) or not isinstance(producer.get("provider"), str) or not producer["provider"]: + raise TopologyValidationError("producer.provider must be a non-empty string") + _doc_version(document) + nodes = document.get("nodes", []) + if not isinstance(nodes, list): + raise TopologyValidationError("nodes must be an array") + for node in nodes: + if not isinstance(node, dict) or node.get("id") is None: + raise TopologyValidationError("every node must be an object with an id") + for field in ("accessPaths", "capabilities"): + if field in node and not isinstance(node[field], list): + raise TopologyValidationError("node {} {} must be an array".format(node["id"], field)) + return document + + +def decode_topology(payload): + """Decode and validate a retained MQTT topology payload.""" + if isinstance(payload, dict): + document = copy.deepcopy(payload) + else: + if isinstance(payload, bytes): + try: + payload = payload.decode("utf-8") + except UnicodeDecodeError as exc: + raise TopologyValidationError("topology payload is not UTF-8") from exc + if not isinstance(payload, str) or not payload.strip(): + raise TopologyValidationError("topology payload is empty") + try: + document = json.loads(payload) + except json.JSONDecodeError as exc: + raise TopologyValidationError("topology payload is malformed JSON") from exc + return _validate_document(document) + + +def _collection_winners(contributions, field, key_function): + """Merge an identity-keyed collection and retain each source document.""" + order = [] + values = {} + for item, rank, document in contributions: + entries = item.get(field, []) + if not isinstance(entries, list): + continue + for entry in entries: + if not isinstance(entry, dict): + continue + key = key_function(entry) + if key is None: + continue + if key not in values: + order.append(key) + current = values.get(key) + if current is None or _better(rank, current[1]): + values[key] = (entry, rank, document) + return [(key, values[key]) for key in order if values[key][0].get("removed") is not True] + + +def _offer_key(offer): + """Return a capability-offer identity.""" + if offer.get("capability") is None: + return None + return (str(offer["capability"]), str(offer.get("accessPath", ""))) + + +def _pick_field(contributions, field, node_id, warnings): + """Choose a scalar field by authority, recency, then input order.""" + setters = [(item[field], rank) for item, rank, _document in contributions if field in item] + if not setters: + return None + value, rank = _winner(setters) + for other_value, other_rank in setters: + if other_rank[:2] == rank[:2] and other_value != value: + warnings.append('node "{}" field "{}" tied; kept first input'.format(node_id, field)) + break + return copy.deepcopy(value) + + +def _merge_bag(contributions, field): + """Merge an attributes/parameters bag per key.""" + values = {} + for item, rank, _document in contributions: + bag = item.get(field) + if not isinstance(bag, dict): + continue + for key, value in bag.items(): + if key not in values or _better(rank, values[key][1]): + values[key] = (copy.deepcopy(value), rank) + return {key: value for key, (value, _rank_value) in values.items()} + + +def _source_ref(document, node_id, offer): + """Build source coordinates for a winning capability offer.""" + source_ref = offer.get("ref") + if not isinstance(source_ref, int) or isinstance(source_ref, bool) or source_ref <= 0: + return None + return SourceCapabilityRef( + provider=document["producer"]["provider"], + topology_version=document["topologyVersion"], + doc_version=_doc_version(document), + cap_ref=source_ref, + node_id=str(node_id), + capability=str(offer["capability"]), + access_path=str(offer.get("accessPath", "")), + ) + + +def _merge_node(contributions, warnings): + """Merge one surviving node and return its source-offer sidecar.""" + winner = _winner([(item, rank) for item, rank, _document in contributions])[0] + node_id = str(winner["id"]) + node = {"id": copy.deepcopy(winner["id"])} + for field in ("kind", "deviceType", "aggregate"): + value = _pick_field(contributions, field, node_id, warnings) + if value is not None: + node[field] = value + for field in ("attributes", "parameters"): + bag = _merge_bag(contributions, field) + if bag: + node[field] = bag + + path_winners = _collection_winners(contributions, "accessPaths", lambda path: str(path["id"]) if path.get("id") is not None else None) + paths = [copy.deepcopy(value[0]) for _key, value in path_winners] + paths.sort(key=lambda path: (-(path.get("preference", 0) or 0), str(path.get("id")))) + if paths: + node["accessPaths"] = paths + + offer_winners = _collection_winners(contributions, "capabilities", _offer_key) + offers = [] + provenance = {} + for key, (offer, _rank_value, document) in offer_winners: + merged_offer = {name: copy.deepcopy(value) for name, value in offer.items() if name not in ("removed", "ref")} + offers.append(merged_offer) + source = _source_ref(document, node_id, offer) + if source is not None: + provenance[(node_id, key[0], key[1])] = source + else: + warnings.append('node "{}" capability "{}" on "{}" has no provider-local ref'.format(node_id, key[0], key[1])) + if offers: + node["capabilities"] = offers + return node, provenance + + +def _digest(document): + """Return a deterministic positive 31-bit digest for merged docVersion.""" + encoded = json.dumps(document, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + value = 0x811C9DC5 + for character in encoded: + value ^= ord(character) + value = (value * 0x01000193) & 0xFFFFFFFF + return (value % 2147483647) + 1 + + +def merge_topologies(documents): + """Authority-merge topology documents while retaining local capability refs.""" + documents = [copy.deepcopy(_validate_document(document)) for document in documents] + if not documents: + raise TopologyValidationError("cannot merge an empty topology set") + + ranked_documents = [(document, _rank(document, order)) for order, document in enumerate(documents)] + top_document = _winner(ranked_documents)[0] + warnings = [] + + node_order = [] + node_contributions = {} + for document, rank in ranked_documents: + for node in document.get("nodes", []): + key = str(node["id"]) + if key not in node_contributions: + node_order.append(key) + node_contributions[key] = [] + node_contributions[key].append((node, rank, document)) + + nodes = [] + provenance = {} + surviving = set() + for node_id in node_order: + contributions = node_contributions[node_id] + if _winner([(item, rank) for item, rank, _document in contributions])[0].get("removed") is True: + continue + tombstones = [rank for item, rank, _document in contributions if item.get("removed") is True] + barrier = max(tombstones) if tombstones else None + live = [(item, rank, document) for item, rank, document in contributions if item.get("removed") is not True and (barrier is None or _better(rank, barrier) or rank == barrier)] + node, node_provenance = _merge_node(live, warnings) + nodes.append(node) + provenance.update(node_provenance) + surviving.add(node_id) + + relationship_contributions = {} + relationship_order = [] + for document, rank in ranked_documents: + for relationship in document.get("relationships", []): + if not isinstance(relationship, dict) or any(relationship.get(field) is None for field in ("from", "to", "type")): + continue + key = (str(relationship["from"]), str(relationship["to"]), str(relationship["type"])) + if key not in relationship_contributions: + relationship_order.append(key) + relationship_contributions[key] = [] + relationship_contributions[key].append((relationship, rank)) + relationships = [] + for key in relationship_order: + relationship = _winner(relationship_contributions[key])[0] + if relationship.get("removed") is True: + continue + if key[0] not in surviving or key[1] not in surviving: + warnings.append("relationship {} dropped because an endpoint is absent".format("|".join(key))) + continue + relationships.append({name: copy.deepcopy(value) for name, value in relationship.items() if name != "removed"}) + + next_ref = 1 + for node in nodes: + refs = {} + for offer in node.get("capabilities", []): + capability = str(offer.get("capability")) + if capability not in refs: + refs[capability] = next_ref + next_ref += 1 + offer["ref"] = refs[capability] + + site = { + "topologyVersion": top_document["topologyVersion"], + "scope": "site", + "producer": { + "name": "predbat-lattice-shadow", + "provider": "predbat", + "inputs": [ + { + "name": document.get("producer", {}).get("name"), + "provider": document["producer"]["provider"], + "authority": _authority(document), + "docVersion": _doc_version(document), + "topologyVersion": document["topologyVersion"], + } + for document, _rank_value in ranked_documents + ], + }, + "nodes": nodes, + } + if relationships: + site["relationships"] = relationships + site["docVersion"] = _digest(site) + return TopologySnapshot(site=site, provenance=provenance, warnings=tuple(warnings)) + + +class LatticeTopologyStore: + """Replace provider documents and expose the current merged shadow snapshot.""" + + def __init__(self): + """Create an empty provider document set.""" + self._documents = {} + self.snapshot = None + + def ingest(self, payload): + """Ingest a provider document, replacing only with a newer version. + + Returns ``True`` when the current snapshot changes. Stale or duplicate + documents are ignored. Invalid documents raise + :class:`TopologyValidationError` and leave the prior snapshot intact. + """ + document = decode_topology(payload) + provider = document["producer"]["provider"] + previous = self._documents.get(provider) + if previous is not None: + incoming_version = _doc_version(document) + previous_version = _doc_version(previous) + if incoming_version < previous_version: + return False + if incoming_version == previous_version: + if document == previous: + return False + # Current gateway 0.2 retained fragments predate docVersion. Arrival + # order is their only replacement signal; explicit versions remain + # protected against reuse with different content. + if "docVersion" in document or "docVersion" in previous: + raise TopologyValidationError("provider {} reused docVersion {} for different content".format(provider, incoming_version)) + candidate_documents = dict(self._documents) + candidate_documents[provider] = document + snapshot = merge_topologies(candidate_documents.values()) + self._documents = candidate_documents + self.snapshot = snapshot + return True + + def source_ref(self, node_id, capability, access_path=None) -> Optional[SourceCapabilityRef]: + """Expose provider-local capability coordinates for a future dispatcher.""" + if self.snapshot is None: + return None + return self.snapshot.source_ref(node_id, capability, access_path) diff --git a/apps/predbat/tests/test_lattice_topology.py b/apps/predbat/tests/test_lattice_topology.py new file mode 100644 index 000000000..577042aaa --- /dev/null +++ b/apps/predbat/tests/test_lattice_topology.py @@ -0,0 +1,189 @@ +"""Tests for read-only Lattice topology ingestion and provenance.""" + +import ast +import json +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_topology import LatticeTopologyStore, TopologyValidationError, merge_topologies + + +def topology(provider, doc_version, authority=0, topology_version="0.3.0", kind="inverter", capability="battery.target_soc", access_path=None, cap_ref=1): + """Build a compact provider-local topology document.""" + access_path = access_path or "{}-local".format(provider) + return { + "topologyVersion": topology_version, + "scope": "fragment", + "docVersion": doc_version, + "producer": {"name": provider, "provider": provider, "authority": authority}, + "nodes": [ + { + "id": "INV1", + "kind": kind, + "accessPaths": [{"id": access_path, "provider": provider, "preference": authority}], + "capabilities": [ + { + "capability": capability, + "accessPath": access_path, + "ref": cap_ref, + "shape": "setpoint", + "control": {"protocol": "mqtt"}, + } + ], + } + ], + } + + +class TestTopologyReplacement(unittest.TestCase): + """Provider documents replace atomically by docVersion.""" + + def test_newer_replaces_and_stale_is_ignored(self): + """A newer retained provider document replaces its predecessor.""" + store = LatticeTopologyStore() + self.assertTrue(store.ingest(topology("gateway", 1, cap_ref=11))) + first_doc_version = store.snapshot.site["docVersion"] + self.assertTrue(store.ingest(topology("gateway", 2, cap_ref=12))) + self.assertEqual(store.source_ref("INV1", "battery.target_soc").cap_ref, 12) + self.assertNotEqual(store.snapshot.site["docVersion"], first_doc_version) + self.assertFalse(store.ingest(topology("gateway", 1, cap_ref=99))) + self.assertEqual(store.source_ref("INV1", "battery.target_soc").cap_ref, 12) + + def test_duplicate_is_ignored_and_reused_version_is_rejected(self): + """A docVersion cannot name two different documents from one provider.""" + store = LatticeTopologyStore() + document = topology("gateway", 4, cap_ref=10) + self.assertTrue(store.ingest(document)) + self.assertFalse(store.ingest(json.dumps(document).encode())) + with self.assertRaises(TopologyValidationError): + store.ingest(topology("gateway", 4, cap_ref=20)) + + def test_malformed_document_preserves_previous_snapshot(self): + """Malformed retained data cannot erase the last known-good topology.""" + store = LatticeTopologyStore() + store.ingest(topology("gateway", 1, cap_ref=11)) + before = store.snapshot + with self.assertRaises(TopologyValidationError): + store.ingest(b"{broken") + self.assertIs(store.snapshot, before) + + def test_gateway_02_without_doc_version_replaces_by_arrival(self): + """Current gateway 0.2 fragments work before firmware adds docVersion.""" + store = LatticeTopologyStore() + first = topology("predbat-gateway", 1, topology_version="0.2.0", cap_ref=11) + second = topology("predbat-gateway", 1, topology_version="0.2.0", cap_ref=12) + del first["docVersion"] + del second["docVersion"] + self.assertTrue(store.ingest(first)) + self.assertTrue(store.ingest(second)) + source = store.source_ref("INV1", "battery.target_soc") + self.assertEqual(source.doc_version, 0) + self.assertEqual(source.cap_ref, 12) + + +class TestAuthorityMerge(unittest.TestCase): + """Authority, recency, and stable input order choose deterministic winners.""" + + def test_authority_then_recency_select_winner(self): + """Higher authority wins before docVersion; recency breaks authority ties.""" + low_new = topology("cloud", 99, authority=1, kind="battery", cap_ref=90) + high_old = topology("gateway", 1, authority=10, kind="inverter", cap_ref=10) + result = merge_topologies([low_new, high_old]) + self.assertEqual(result.site["nodes"][0]["kind"], "inverter") + + newer = topology("installer", 2, authority=10, kind="hybrid-inverter", access_path="installer", cap_ref=20) + result = merge_topologies([high_old, newer]) + self.assertEqual(result.site["nodes"][0]["kind"], "hybrid-inverter") + + def test_exact_tie_keeps_first_and_warns(self): + """Input order is the stable final tiebreak for scalar conflicts.""" + first = topology("a", 1, authority=5, kind="inverter", cap_ref=1) + second = topology("b", 1, authority=5, kind="battery", cap_ref=2) + result = merge_topologies([first, second]) + self.assertEqual(result.site["nodes"][0]["kind"], "inverter") + self.assertTrue(any("tied" in warning for warning in result.warnings)) + + def test_reminted_refs_map_to_provider_local_refs(self): + """Merged refs are independent while the sidecar retains each local ref.""" + gateway = topology("gateway", 7, authority=10, access_path="gw", cap_ref=41) + cloud = topology("cloud", 3, authority=1, access_path="cloud", cap_ref=900) + result = merge_topologies([gateway, cloud]) + offers = result.site["nodes"][0]["capabilities"] + self.assertEqual({offer["ref"] for offer in offers}, {1}) + self.assertEqual(result.source_ref("INV1", "battery.target_soc", "gw").cap_ref, 41) + self.assertEqual(result.source_ref("INV1", "battery.target_soc", "cloud").cap_ref, 900) + self.assertEqual(result.source_ref("INV1", "battery.target_soc").provider, "gateway") + + def test_mixed_02_and_03_providers_retain_source_versions(self): + """Compatible 0.x providers keep their individual topology versions.""" + old = topology("cloud", 1, authority=1, topology_version="0.2.0", access_path="cloud", cap_ref=2) + current = topology("gateway", 2, authority=10, topology_version="0.3.0", access_path="gw", cap_ref=3) + result = merge_topologies([old, current]) + self.assertEqual(result.site["topologyVersion"], "0.3.0") + self.assertEqual(result.source_ref("INV1", "battery.target_soc", "cloud").topology_version, "0.2.0") + self.assertEqual(result.source_ref("INV1", "battery.target_soc", "gw").topology_version, "0.3.0") + + def test_capability_and_node_tombstones_remove_provenance(self): + """Winning tombstones remove merged offers/nodes and their sidecar entries.""" + discovered = topology("gateway", 1, authority=0, access_path="gw", cap_ref=3) + capability_tombstone = { + "topologyVersion": "0.3.0", + "scope": "overlay", + "docVersion": 1, + "producer": {"name": "installer", "provider": "installer", "authority": 50}, + "nodes": [{"id": "INV1", "capabilities": [{"capability": "battery.target_soc", "accessPath": "gw", "removed": True}]}], + } + result = merge_topologies([discovered, capability_tombstone]) + self.assertNotIn("capabilities", result.site["nodes"][0]) + self.assertIsNone(result.source_ref("INV1", "battery.target_soc", "gw")) + + node_tombstone = { + "topologyVersion": "0.3.0", + "scope": "overlay", + "docVersion": 2, + "producer": {"name": "installer", "provider": "installer", "authority": 50}, + "nodes": [{"id": "INV1", "removed": True}], + } + result = merge_topologies([discovered, node_tombstone]) + self.assertEqual(result.site["nodes"], []) + self.assertEqual(result.provenance, {}) + + def test_node_tombstone_exact_tie_respects_stable_input_order(self): + """Equal-rank live/tombstone conflicts keep the first contribution.""" + live = topology("gateway", 1, authority=10, cap_ref=3) + tombstone = { + "topologyVersion": "0.3.0", + "scope": "overlay", + "docVersion": 1, + "producer": {"name": "installer", "provider": "installer", "authority": 10}, + "nodes": [{"id": "INV1", "removed": True}], + } + + live_first = merge_topologies([live, tombstone]) + self.assertEqual([node["id"] for node in live_first.site["nodes"]], ["INV1"]) + self.assertEqual(live_first.source_ref("INV1", "battery.target_soc").cap_ref, 3) + + tombstone_first = merge_topologies([tombstone, live]) + self.assertEqual(tombstone_first.site["nodes"], []) + self.assertEqual(tombstone_first.provenance, {}) + + +class TestGatewayFeatureFlag(unittest.TestCase): + """Gateway topology behavior is absent unless explicitly enabled.""" + + def test_feature_flag_defaults_off(self): + """Default initialization does not expose a dispatcher binding.""" + gateway_path = os.path.join(os.path.dirname(__file__), "..", "gateway.py") + with open(gateway_path, "r", encoding="utf-8") as gateway_file: + module = ast.parse(gateway_file.read()) + gateway_class = next(node for node in module.body if isinstance(node, ast.ClassDef) and node.name == "GatewayMQTT") + initialize = next(node for node in gateway_class.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == "initialize") + defaults = dict(zip([argument.arg for argument in initialize.args.args[-len(initialize.args.defaults) :]], initialize.args.defaults)) + self.assertIs(defaults["lattice_projection_enable"].value, False) + + +if __name__ == "__main__": + unittest.main()