diff --git a/interplanetary_edge_orchestrator_privacy/adapters/__init__.py b/interplanetary_edge_orchestrator_privacy/adapters/__init__.py index 69a54f2..bf692ad 100644 --- a/interplanetary_edge_orchestrator_privacy/adapters/__init__.py +++ b/interplanetary_edge_orchestrator_privacy/adapters/__init__.py @@ -1,5 +1,8 @@ """Adapters package for canonical/interoperable surface. This module is intentionally lightweight and serves as a namespace holder -for toy adapters used in MVP experiments. +for adapters used in MVP experiments. """ + +from .canonical import CanonicalAdapter # noqa: F401 +from .catopt_bridge import CatOptBridge # noqa: F401 diff --git a/interplanetary_edge_orchestrator_privacy/adapters/catopt_bridge.py b/interplanetary_edge_orchestrator_privacy/adapters/catopt_bridge.py new file mode 100644 index 0000000..482ac9d --- /dev/null +++ b/interplanetary_edge_orchestrator_privacy/adapters/catopt_bridge.py @@ -0,0 +1,101 @@ +"""CatOpt Bridge + +Minimal, production-friendly scaffold for a CatOpt-style interoperability bridge. + +This module provides a tiny, well-typed surface that demonstrates how a real +adapter could map its internal representations to the DSL primitives defined in +``dsl.LocalProblem``, ``dsl.SharedVariables``, and ``dsl.PlanDelta``. + +The goal is to give collaborators a concrete starting point for building +interoperable adapters across domains (energy, robotics, space) without +breaking existing behavior. +""" + +from __future__ import annotations + +from typing import Any, Dict, Optional, Tuple + +from ..dsl import LocalProblem, SharedVariables, PlanDelta + + +class CatOptBridge: + """Lightweight bridge for mapping between a canonical contract and DSL types. + + This class does not perform any privacy-preserving math; it merely shows how + adapters can marshal/unmarshal between the vendor-agnostic contract surface + and the in-repo DSL data models. + """ + + CONTRACT_VERSION: str = "0.1" + + def map_local_problem(self, problem: LocalProblem) -> Dict[str, Any]: + """Serialize a LocalProblem into a contract-friendly dict.""" + return { + "contract_version": self.CONTRACT_VERSION, + "type": "LocalProblem", + "problem_id": problem.problem_id, + "features": problem.features, + "objective": problem.objective, + "metadata": problem.metadata, + } + + def map_shared_variables(self, shared: SharedVariables) -> Dict[str, Any]: + """Serialize SharedVariables into the contract surface.""" + return { + "contract_version": self.CONTRACT_VERSION, + "type": "SharedVariables", + "version": shared.version, + "data": shared.data, + "timestamp": shared.timestamp, + } + + def map_plan_delta(self, delta: PlanDelta) -> Dict[str, Any]: + """Serialize PlanDelta into the contract surface.""" + return { + "contract_version": self.CONTRACT_VERSION, + "type": "PlanDelta", + "version": delta.version, + "delta": delta.delta, + "insight": delta.insight, + } + + def to_contract(self, problem: LocalProblem, shared: SharedVariables, delta: Optional[PlanDelta] = None) -> Dict[str, Any]: + """Combine DSL primitives into a single contract payload.""" + payload: Dict[str, Any] = { + "contract_version": self.CONTRACT_VERSION, + "LocalProblem": self.map_local_problem(problem), + "SharedVariables": self.map_shared_variables(shared), + } + if delta is not None: + payload["PlanDelta"] = self.map_plan_delta(delta) + return payload + + def from_contract(self, payload: Dict[str, Any]) -> Tuple[LocalProblem, SharedVariables, Optional[PlanDelta]]: + """Deserialize a contract payload back into DSL primitives. + + This is intentionally simple; a real implementation would include validation + and version checks. + """ + lp = LocalProblem( + problem_id=payload.get("LocalProblem", {}).get("problem_id", ""), + features=payload.get("LocalProblem", {}).get("features", []), + objective=payload.get("LocalProblem", {}).get("objective"), + metadata=payload.get("LocalProblem", {}).get("metadata", {}), + ) + + sv = SharedVariables( + version=payload.get("SharedVariables", {}).get("version", 0), + data=payload.get("SharedVariables", {}).get("data", {}), + timestamp=payload.get("SharedVariables", {}).get("timestamp"), + ) + + delta_payload = payload.get("PlanDelta") + delta = None + if delta_payload is not None: + delta = PlanDelta( + version=delta_payload.get("version", 0), + delta=delta_payload.get("delta", {}), + insight=delta_payload.get("insight"), + ) + + return lp, sv, delta diff --git a/interplanetary_edge_orchestrator_privacy/adapters/toy_adapter.py b/interplanetary_edge_orchestrator_privacy/adapters/toy_adapter.py index 3bc3ddd..26e99a9 100644 --- a/interplanetary_edge_orchestrator_privacy/adapters/toy_adapter.py +++ b/interplanetary_edge_orchestrator_privacy/adapters/toy_adapter.py @@ -26,3 +26,26 @@ class RoverAdapter: class HabitatAdapter(RoverAdapter): # Reuse the same mapping for demonstration; in a real system this would vary. pass + + +class RoverHabitatMVPBridge: + """Minimal end-to-end MVP bridge wiring two adapters through CatOptBridge. + + This is a simple, test-oriented helper to demonstrate how two adapters can + participate in a canonical contract payload without requiring a full runtime + orchestration layer. It uses the existing CanonicalAdapter under the hood. + """ + + def __init__(self): + self._canon = CanonicalAdapter() + # Reuse the CatOptBridge scaffold for end-to-end mapping + from .catopt_bridge import CatOptBridge # local import to avoid cycle in tests + self._bridge = CatOptBridge() + + def rover_to_contract(self, problem_id: str, features: List[float]) -> dict: + lp = self._canon.map_local_problem(problem_id, features) + sv = SharedVariables(version=1, data={"seed": 42}) + return self._bridge.to_contract(lp, sv) + + def habitat_from_contract(self, payload: dict) -> tuple[LocalProblem, SharedVariables, PlanDelta | None]: + return self._bridge.from_contract(payload)