diff --git a/src/cosmosmesh_privacy_preserving_federated/dsl_seed.py b/src/cosmosmesh_privacy_preserving_federated/dsl_seed.py new file mode 100644 index 0000000..bcb06ee --- /dev/null +++ b/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", +]