101 lines
3.0 KiB
Python
101 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, asdict
|
|
from typing import Dict, List, Any
|
|
|
|
from .protocol import LocalProblem, SharedVariables, PlanDelta
|
|
|
|
|
|
@dataclass
|
|
class LocalProblemDSL:
|
|
"""Minimal DSL wrapper for a per-shard LocalProblem.
|
|
|
|
This DSL is designed to be mapped into the canonical LocalProblem protocol
|
|
using adapters. It captures the same conceptual fields as LocalProblem, but
|
|
presents a stable, serializable interface for cross-adapter contracts.
|
|
"""
|
|
shard_id: str
|
|
projection: List[str]
|
|
predicates: List[str]
|
|
costs: float
|
|
constraints: Dict[str, Any]
|
|
|
|
def to_protocol(self) -> LocalProblem:
|
|
# Use the canonical constructor; LocalProblem accepts both 'projection' and
|
|
# 'projected_attrs' aliases for compatibility.
|
|
return LocalProblem(
|
|
shard_id=self.shard_id,
|
|
projection=self.projection,
|
|
predicates=self.predicates,
|
|
costs=self.costs,
|
|
constraints=self.constraints,
|
|
)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
@staticmethod
|
|
def from_dict(d: Dict[str, Any]) -> "LocalProblemDSL":
|
|
if d is None:
|
|
d = {}
|
|
return LocalProblemDSL(
|
|
shard_id=d.get("shard_id", ""),
|
|
projection=d.get("projection", d.get("projected_attrs", [])),
|
|
predicates=d.get("predicates", []),
|
|
costs=float(d.get("costs", 0.0)),
|
|
constraints=d.get("constraints", {}),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SharedVariablesDSL:
|
|
version: int
|
|
signals: Dict[str, float]
|
|
priors: Dict[str, float]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
@staticmethod
|
|
def from_dict(d: Dict[str, Any]) -> "SharedVariablesDSL":
|
|
return SharedVariablesDSL(
|
|
version=int(d.get("version", 0)),
|
|
signals=dict(d.get("signals", {})),
|
|
priors=dict(d.get("priors", {})),
|
|
)
|
|
|
|
def to_protocol(self) -> SharedVariables:
|
|
# Convert to the canonical SharedVariables dataclass defined in protocol.py
|
|
return SharedVariables(
|
|
version=self.version,
|
|
signals=self.signals,
|
|
priors=self.priors,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlanDeltaDSL:
|
|
delta_id: str
|
|
timestamp: float
|
|
changes: Dict[str, Any]
|
|
contract_id: str = ""
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_protocol(self) -> PlanDelta:
|
|
return PlanDelta(
|
|
delta_id=self.delta_id,
|
|
timestamp=self.timestamp,
|
|
changes=dict(self.changes),
|
|
contract_id=self.contract_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def from_dict(d: Dict[str, Any]) -> "PlanDeltaDSL":
|
|
return PlanDeltaDSL(
|
|
delta_id=d.get("delta_id", ""),
|
|
timestamp=float(d.get("timestamp", 0.0)),
|
|
changes=dict(d.get("changes", {})),
|
|
contract_id=d.get("contract_id", ""),
|
|
)
|