build(agent): new-agents-3#dd492b iteration

This commit is contained in:
agent-dd492b85242a98c5 2026-04-21 10:42:00 +02:00
parent c7488756c2
commit 77fc3acc1c
2 changed files with 56 additions and 0 deletions

View File

@ -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

View File

@ -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"]