diff --git a/src/idea15_edgemind_verifiable_onboard/energi_bridge.py b/src/idea15_edgemind_verifiable_onboard/energi_bridge.py index 6835cdc..02b5ca3 100644 --- a/src/idea15_edgemind_verifiable_onboard/energi_bridge.py +++ b/src/idea15_edgemind_verifiable_onboard/energi_bridge.py @@ -64,6 +64,18 @@ class EnergiBridge: "safety_tags": list(delta.safety_tags), } + def map_shared_signals(self, signals: SharedSignals) -> Dict[str, Any]: + """Map SharedSignals to a canonical representation. + + This mirrors map_local_problem / map_plan_delta and provides a + symmetric entry point for adapter-to-canonical data exchange. + """ + return { + "canonical": "SharedSignals", + "version": signals.version, + "data": dict(signals.data), + } + def register_adapter(self, adapter: AdapterContract) -> None: self.adapters[adapter.name] = adapter diff --git a/tests/test_energi_bridge.py b/tests/test_energi_bridge.py new file mode 100644 index 0000000..3c59e55 --- /dev/null +++ b/tests/test_energi_bridge.py @@ -0,0 +1,44 @@ +import sys +import os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))) + +from idea15_edgemind_verifiable_onboard.energi_bridge import ( + EnergiBridge, + LocalProblem, + SharedSignals, + PlanDelta, + AdapterContract, +) + + +def test_map_local_problem_basic(): + lp = LocalProblem(asset_id="robot-01", tasks=["navigate"], budgets={"energy": 10.0}) + eb = EnergiBridge() + canon = eb.map_local_problem(lp) + assert canon["canonical"] == "LocalProblem" + assert canon["asset_id"] == "robot-01" + assert canon["tasks"] == ["navigate"] + assert canon["budgets"]["energy"] == 10.0 + + +def test_map_shared_signals_and_adapter_registration(): + eb = EnergiBridge() + ad = AdapterContract(name="gateway", version="0.1.0") + eb.register_adapter(ad) + assert eb.get_adapter("gateway").version == "0.1.0" + + signals = SharedSignals(version=1, data={"latency": 12}) + canon = eb.map_shared_signals(signals) + assert canon["canonical"] == "SharedSignals" + assert canon["version"] == 1 + assert canon["data"]["latency"] == 12 + + +def test_map_plan_delta_basic(): + delta = PlanDelta(timestamp=123.456, delta_actions=["move_forward"], safety_tags=["ok"]) + eb = EnergiBridge() + canon = eb.map_plan_delta(delta) + assert canon["canonical"] == "PlanDelta" + assert canon["timestamp"] == 123.456 + assert canon["delta_actions"] == ["move_forward"] + assert canon["safety_tags"] == ["ok"]