build(agent): new-agents#a6e6ec iteration

src/cosmosmesh_privacy_preserving_federated/dsl_seed.py

@@ -0,0 +1,107 @@
+"""Minimal DSL seeds for interoperability bootstrap.
+
+This module provides lightweight seed data classes that mirror the
+CosmosMesh MVP primitives and offer a convenient helper to produce a
+CatOpt-like IR using the existing bridge utilities.
+
+The goal is to give testers and downstream adapters a stable, small surface
+to bootstrap interoperability without pulling in extra dependencies or
+writing bespoke test fixtures.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict
+from typing import List, Any, Dict
+
+# Reuse core primitives from the existing bridge for compatibility
+from .catopt_bridge import LocalProblem as CP_LocalProblem
+from .catopt_bridge import SharedVariables as CP_SharedVariables
+from .catopt_bridge import PlanDelta as CP_PlanDelta
+from .catopt_bridge import DualVariables as CP_DualVariables
+
+
+@dataclass
+class LocalProblemSeed:
+    id: str
+    domain: str
+    assets: List[str]
+    objective: Any
+    constraints: Any
+    version: int | None = None
+
+    def to_catopt(self) -> CP_LocalProblem:
+        # Build a CP LocalProblem using seed values
+        return CP_LocalProblem(
+            id=self.id,
+            domain=self.domain,
+            assets=self.assets,
+            objective=self.objective,
+            constraints=self.constraints,
+            version=self.version,
+        )
+
+
+@dataclass
+class SharedVariablesSeed:
+    forecasts: Dict[str, Any] | None = None
+    priors: Dict[str, Any] | None = None
+    version: int = 1
+
+    def to_catopt(self) -> CP_SharedVariables:
+        return CP_SharedVariables(
+            forecasts=self.forecasts or {},
+            priors=self.priors or {},
+            version=self.version,
+        )
+
+
+@dataclass
+class DualVariablesSeed:
+    values: Dict[str, Any] | None = None
+    version: int = 1
+
+    def to_catopt(self) -> CP_DualVariables:
+        return CP_DualVariables(values=self.values or {}, version=self.version)
+
+
+@dataclass
+class PlanDeltaSeed:
+    delta: Dict[str, Any]
+    timestamp: str
+    author: str
+    contract_id: str
+    signature: str | None = None
+
+    def to_catopt(self) -> CP_PlanDelta:
+        return CP_PlanDelta(delta=self.delta, timestamp=self.timestamp, author=self.author, contract_id=self.contract_id, signature=self.signature)
+
+
+def seed_end_to_catopt(
+    lp_seed: LocalProblemSeed,
+    sv_seed: SharedVariablesSeed,
+    dv_seed: DualVariablesSeed,
+    delta_seed: PlanDeltaSeed,
+) -> Dict[str, Any]:
+    """Convenience: produce a CatOpt-like IR from seeds.
+
+    This function intentionally reuses the existing bridge data models so
+    downstream adapters can treat seeds as first-class inputs without
+    duplicating logic.
+    """
+    lp = lp_seed.to_catopt()
+    sv = sv_seed.to_catopt()
+    dv = dv_seed.to_catopt()
+    delta = delta_seed.to_catopt()
+    # Reuse the canonical bridge to convert to a CatOpt-like IR
+    from .catopt_bridge import to_catopt as _to_catopt  # local import to avoid cycles
+
+    return _to_catopt(lp, sv, delta)  # type: ignore[arg-type]
+
+
+__all__ = [
+    "LocalProblemSeed",
+    "SharedVariablesSeed",
+    "DualVariablesSeed",
+    "PlanDeltaSeed",
+    "seed_end_to_catopt",
+]