diff --git a/src/cosmosmesh_privacy_preserving_federated/__init__.py b/src/cosmosmesh_privacy_preserving_federated/__init__.py index ae8c230..39b60b0 100644 --- a/src/cosmosmesh_privacy_preserving_federated/__init__.py +++ b/src/cosmosmesh_privacy_preserving_federated/__init__.py @@ -1,9 +1,16 @@ """CosmosMesh Privacy-Preserving Federated package. Exposes MVP scaffolds for bridging CosmosMesh primitives to a CatOpt-style -representation via the catopt_bridge module. +representation via the catopt_bridge module and a lightweight EnergiBridge +for canonical interoperability with a CatOpt-like IR. """ from .catopt_bridge import CatOptBridge, ContractRegistry, LocalProblem - -__all__ = ["CatOptBridge", "ContractRegistry", "LocalProblem"] +try: + from .energi_bridge import EnergiBridge, LocalProblemEP # type: ignore + __all__ = ["CatOptBridge", "ContractRegistry", "LocalProblem", "EnergiBridge", "LocalProblemEP"] +except Exception: + # EnergiBridge not yet available; avoid import-time failure for MVPs that + # import this package before energibridge module is added in a follow-up + # patch. + __all__ = ["CatOptBridge", "ContractRegistry", "LocalProblem"] diff --git a/src/cosmosmesh_privacy_preserving_federated/energi_bridge.py b/src/cosmosmesh_privacy_preserving_federated/energi_bridge.py new file mode 100644 index 0000000..6a48c10 --- /dev/null +++ b/src/cosmosmesh_privacy_preserving_federated/energi_bridge.py @@ -0,0 +1,112 @@ +"""EnergiBridge: Canonical bridge mapping CosmosMesh → CatOpt-like IR (prototype). + +This module provides a minimal, strongly-typed bridge to translate CosmosMesh +primitives into a vendor-agnostic, CatOpt-inspired intermediate representation +suitable for plug-in adapters across domains (energy, robotics, space). + +The implementation is intentionally compact and dependency-free, focusing on a +clean data model and a convenience encoder to a simple IR dictionary that tests +and adapters can consume. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + + +# Lightweight per-asset local problem representation for EnergiBridge +@dataclass +class LocalProblemEP: + problem_id: str + assets: List[str] + objective: str + constraints: List[str] + data_contracts: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "problem_id": self.problem_id, + "assets": self.assets, + "objective": self.objective, + "constraints": self.constraints, + "data_contracts": self.data_contracts or {}, + } + + +@dataclass +class SharedVariableEP: + channel: str + version: int + payload: Dict[str, Any] + + def to_dict(self) -> Dict[str, Any]: + return { + "channel": self.channel, + "version": self.version, + "payload": self.payload, + } + + +@dataclass +class DualVariableEP: + channel: str + version: int + payload: Dict[str, Any] + + def to_dict(self) -> Dict[str, Any]: + return { + "channel": self.channel, + "version": self.version, + "payload": self.payload, + } + + +@dataclass +class PlanDeltaEP: + delta_id: str + changes: Dict[str, Any] + timestamp: Optional[float] = None + + def to_dict(self) -> Dict[str, Any]: + payload = { + "delta_id": self.delta_id, + "changes": self.changes, + "timestamp": self.timestamp, + } + return payload + + +@dataclass +class EnergiBridge: + """Encoder that builds a canonical, CatOpt-like IR payload from CosmosMesh + primitives. + """ + + def to_ir(self, + lp: LocalProblemEP, + shared: List[SharedVariableEP], + duals: List[DualVariableEP], + plan_deltas: Optional[List[PlanDeltaEP]] = None) -> Dict[str, Any]: + ir: Dict[str, Any] = { + "Version": "0.1", + "Objects": { + "LocalProblem": lp.to_dict(), + }, + } + + morphisms: List[Dict[str, Any]] = [] + for s in shared: + morphisms.append({"Morphisms": {"SharedVariable": s.to_dict()}}) + for d in duals: + morphisms.append({"Morphisms": {"DualVariable": d.to_dict()}}) + if morphisms: + ir["Morphisms"] = morphisms + + if plan_deltas: + ir.setdefault("Objects", {})["PlanDeltas"] = [pd.to_dict() for pd in plan_deltas] + + return ir + + +__all__ = ["LocalProblemEP", "SharedVariableEP", "DualVariableEP", "PlanDeltaEP", "EnergiBridge"] diff --git a/tests/test_energi_bridge.py b/tests/test_energi_bridge.py new file mode 100644 index 0000000..c971c0b --- /dev/null +++ b/tests/test_energi_bridge.py @@ -0,0 +1,36 @@ +import pytest +from cosmosmesh_privacy_preserving_federated.energi_bridge import ( + LocalProblemEP, + SharedVariableEP, + DualVariableEP, + PlanDeltaEP, + EnergiBridge, +) + + +def test_energi_bridge_basic_encoding(): + lp = LocalProblemEP( + problem_id="lp-001", + assets=["rover-1", "drone-A"], + objective="maximize_task_completion", + constraints=["energy<=1000", "time_window=08:00-12:00"], + data_contracts={"Telemetry": {"version": 1}}, + ) + + sv1 = SharedVariableEP(channel="energy_budget", version=1, payload={"value": 42}) + dv1 = DualVariableEP(channel="lag_energy", version=1, payload={"value": 0.5}) + + delta = PlanDeltaEP(delta_id="d-1", changes={"rover-1": {"move_to": "A"}}, timestamp=123456.0) + + bridge = EnergiBridge() + ir = bridge.to_ir(lp, [sv1], [dv1], [delta]) + + # Basic shape checks + assert ir["Version"] == "0.1" + assert "Objects" in ir + assert ir["Objects"]["LocalProblem"]["problem_id"] == "lp-001" + assert "Morphisms" in ir + # We expect at least one SharedVariable and one DualVariable in the Morphisms list + morphs = ir["Morphisms"] + assert any("SharedVariable" in m.get("Morphisms", {}) for m in morphs) or any("DualVariable" in m.get("Morphisms", {}) for m in morphs) + assert "PlanDeltas" in ir["Objects"]