From a0d8fee50dea49c48cebf47b3e47169e34b63ecc Mon Sep 17 00:00:00 2001 From: agent-23e5c897f40fd19e Date: Wed, 15 Apr 2026 21:39:25 +0200 Subject: [PATCH] build(agent): molt-y#23e5c8 iteration --- README.md | 9 ++- .../adapters/habitat_life_support.py | 36 +++++++++ .../adapters/rover_planner.py | 51 ++++++------ .../dsl_sketch.py | 77 +++++++++++-------- 4 files changed, 117 insertions(+), 56 deletions(-) create mode 100644 src/cosmosmesh_privacy_preserving_federated/adapters/habitat_life_support.py diff --git a/README.md b/README.md index ff64aa7..a4f4bab 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,14 @@ Core MVP outline - Lightweight transport (TLS-based) and a tiny ADMM-lite solver per asset - Graph-of-Contracts registry for schemas and per-message metadata to support audits and replay protection -What’s included in this patch +-What’s included in this patch +- Added a minimal CatOpt bridge module: src/cosmosmesh_privacy_preserving_federated/catopt_bridge.py (already present, extended with examples) +- Added DSL sketch: src/cosmosmesh_privacy_preserving_federated/dsl_sketch.py +- Added toy adapters: src/cosmosmesh_privacy_preserving_federated/adapters/rover_planner.py and src/cosmosmesh_privacy_preserving_federated/adapters/habitat_life_support.py +- Package initializer: src/cosmosmesh_privacy_preserving_federated/__init__.py +- Lightweight unit tests: tests/test_catopt_bridge.py (unchanged) +- Simple usage example via a RoundTrip encoding path (as in tests) +- Basic README describing MVP approach and how to extend - Added a minimal CatOpt bridge module: src/cosmosmesh_privacy_preserving_federated/catopt_bridge.py - Package initializer: src/cosmosmesh_privacy_preserving_federated/__init__.py - Lightweight unit test: tests/test_catopt_bridge.py diff --git a/src/cosmosmesh_privacy_preserving_federated/adapters/habitat_life_support.py b/src/cosmosmesh_privacy_preserving_federated/adapters/habitat_life_support.py new file mode 100644 index 0000000..a9da33f --- /dev/null +++ b/src/cosmosmesh_privacy_preserving_federated/adapters/habitat_life_support.py @@ -0,0 +1,36 @@ +"""Toy habitat life-support adapter for CosmosMesh CatOpt bridge. + +Demonstrates exposing a LocalProblem and handling commands in a simple +offline-first setting. +""" +from __future__ import annotations + +from typing import Any, Dict + +from cosmosmesh_privacy_preserving_federated.catopt_bridge import LocalProblem, PlanDelta + + +class HabitatLifeSupportAdapter: + def __init__(self, habitat_id: str = "habitat-1") -> None: + self.habitat_id = habitat_id + self._state: Dict[str, Any] = {"life_support_on": True, "battery": 100.0} + + def readState(self) -> Dict[str, Any]: + return {"habitat_id": self.habitat_id, "state": self._state} + + def exposeLocalProblemData(self) -> LocalProblem: + lp = LocalProblem( + problem_id=f"habitat-{self.habitat_id}-lp", + version=1, + variables={"cooling": 22.0, "humidity": 40.0}, + objective=1.0, + constraints=None, + ) + return lp + + def applyCommand(self, delta: PlanDelta) -> None: + # Simple: interpret delta changes into internal state flags if present + changes = delta.changes if isinstance(delta, PlanDelta) else {} + if isinstance(changes, dict): + for k, v in changes.items(): + self._state[k] = v diff --git a/src/cosmosmesh_privacy_preserving_federated/adapters/rover_planner.py b/src/cosmosmesh_privacy_preserving_federated/adapters/rover_planner.py index 014e592..defa798 100644 --- a/src/cosmosmesh_privacy_preserving_federated/adapters/rover_planner.py +++ b/src/cosmosmesh_privacy_preserving_federated/adapters/rover_planner.py @@ -1,34 +1,39 @@ -from __future__ import annotations -from typing import Any, Dict +"""Toy rover planner adapter for CosmosMesh CatOpt bridge. -from cosmosmesh_privacy_preserving_federated.catopt_bridge import LocalProblem, SharedVariables, DualVariables +This adapter implements a minimal interface compatible with the CatOpt bridge +and demonstrates how a device-specific model could be translated into a LocalProblem +and how to expose local data as SharedVariables. +""" +from __future__ import annotations + +from typing import Any, Dict, List + +from cosmosmesh_privacy_preserving_federated.catopt_bridge import LocalProblem, SharedVariable, PlanDelta, CatOptBridge class RoverPlannerAdapter: - """Minimal rover planner adapter scaffold. + def __init__(self, planner_id: str = "rover-1") -> None: + self.planner_id = planner_id + # Use a loosely-typed state dict to handle diverse state shapes across adapters + self._state: Dict[str, Any] = { + "tasks": ["survey", "sample"] + } - Exposes a tiny interface used by the MVP to demonstrate problem translation - and command application. This is intentionally lightweight. - """ + def readState(self) -> Dict[str, Any]: + # Return a small snapshot of rover state as a dict + return {"planner_id": self.planner_id, "state": self._state} - def __init__(self) -> None: - pass - - def read_state(self) -> Dict[str, Any]: - # Placeholder rover state; in a real system this would query sensors. - return {"battery": 78, "position": [0.0, 0.0], "status": "idle"} - - def expose_local_problem_data(self) -> LocalProblem: - # Return a toy LocalProblem instance for the MVP demo. + def exposeLocalProblemData(self) -> LocalProblem: + # Expose a simple LocalProblem for the rover to solve lp = LocalProblem( - id="lp_rover_1", - objective="minimize_energy", - variables={"task": "survey"}, - constraints=["battery>=20"], + problem_id=f"rover-{self.planner_id}-lp", version=1, + variables={"task_weight": 1.0, "battery": 90.0}, + objective=0.0, + constraints=[{"battery_min": 20.0}], ) return lp - def apply_command(self, command: Dict[str, Any]) -> Dict[str, Any]: - # Accept a command and return an acknowledgment. This is a stub. - return {"ack": True, "command": command} + def applyCommand(self, command: Dict[str, Any]) -> None: + # Apply a command to the rover planner (no-op in this toy example) + self._state["last_command"] = command diff --git a/src/cosmosmesh_privacy_preserving_federated/dsl_sketch.py b/src/cosmosmesh_privacy_preserving_federated/dsl_sketch.py index a1280f2..2406a8c 100644 --- a/src/cosmosmesh_privacy_preserving_federated/dsl_sketch.py +++ b/src/cosmosmesh_privacy_preserving_federated/dsl_sketch.py @@ -1,49 +1,62 @@ -"""DSL sketch for CosmosMesh MVP primitives. +"""Minimal DSL sketch for LocalProblem/SharedVariables/PlanDelta. -The goal here is to provide lightweight dataclass-like shells that describe the -contract between agents in a way that's friendly to MVP wiring and testing. These -types are intentionally simple and purely for internal prototyping. +This module provides a lightweight, Pythonic DSL to describe a per-agent +LocalProblem and associated signals (SharedVariables / PlanDelta). It is +intended for prototyping, tests, and documentation, not for production use. """ + from __future__ import annotations -from dataclasses import dataclass, asdict from typing import Any, Dict, List, Optional +from .catopt_bridge import LocalProblem, SharedVariable, PlanDelta, ContractRegistry, CatOptBridge -@dataclass class LocalProblemDSL: - agent_id: str - objective: str - variables: List[str] - constraints: List[str] + """Fluent DSL for building a LocalProblem dataclass.""" - def dict(self) -> Dict[str, Any]: - return asdict(self) + def __init__(self, problem_id: str, version: int = 1) -> None: + self.problem_id = problem_id + self.version = version + self.variables: Dict[str, Any] = {} + self.objective: Optional[Any] = None + self.constraints: List[Dict[str, Any]] = [] + + def var(self, name: str, value: Any) -> 'LocalProblemDSL': + self.variables[name] = value + return self + + def set_objective(self, obj: Any) -> 'LocalProblemDSL': + self.objective = obj + return self + + def add_constraint(self, constraint: Dict[str, Any]) -> 'LocalProblemDSL': + self.constraints.append(constraint) + return self + + def build(self) -> LocalProblem: + return LocalProblem( + problem_id=self.problem_id, + version=self.version, + variables=self.variables, + objective=self.objective if self.objective is not None else 0, + constraints=self.constraints or None, + ) -@dataclass class SharedVariablesDSL: - signals: Dict[str, float] - version: str = "0.1" + """DSL helper to create SharedVariable instances.""" - def dict(self) -> Dict[str, Any]: - return asdict(self) + @staticmethod + def sv(name: str, value: Any, version: int = 1) -> SharedVariable: + return SharedVariable(name, value, version) -@dataclass -class DualVariablesDSL: - lambdas: Dict[str, float] - version: str = "0.1" - - def dict(self) -> Dict[str, Any]: - return asdict(self) - - -@dataclass class PlanDeltaDSL: - delta_id: str - changes: Dict[str, Any] - timestamp: Optional[str] = None + """DSL helper to create PlanDelta instances.""" - def dict(self) -> Dict[str, Any]: - return asdict(self) + @staticmethod + def delta(delta_id: str, changes: Dict[str, Any], timestamp: Optional[float] = None) -> PlanDelta: + return PlanDelta(delta_id=delta_id, changes=changes, timestamp=timestamp) + + +__all__ = ["LocalProblemDSL", "SharedVariablesDSL", "PlanDeltaDSL"]