edge-latency-aware-cross-ve.../elac_plan/dsl.py

102 lines
2.4 KiB
Python

from __future__ import annotations
"""Minimal ELAC-Plan DSL sketch
This module provides a tiny, vendor-agnostic DSL representation that maps
the canonical ELAC primitives to a lightweight dictionary-based schema
suitable for cross-adapter interoperability and deterministic replay.
"""
from dataclasses import dataclass, asdict
from typing import Any, Dict
from .core import LocalProblem, SharedVariables, PlanDelta, DualVariables
@dataclass
class DSLObject:
# LocalProblem-like object translated into DSL form
id: str
asset: str
venue: str
objective: str
constraints: Dict[str, Any]
price_target: float
tolerance: float
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@dataclass
class DSLMorphisms:
# SharedVariables-like signals translated into DSL form
contract_id: str
version: int
variables: Dict[str, Any]
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@dataclass
class DSLDualVariables:
# DualVariables-like coupling signals translated into DSL form
contract_id: str
version: int
shadow_prices: Dict[str, float]
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@dataclass
class DSLPlanDelta:
# PlanDelta translated into DSL form along with metadata
contract_id: str
delta: Dict[str, Any]
timestamp: str
author: str
privacy_budget: float
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
def to_dsl_object(lp: LocalProblem) -> DSLObject:
return DSLObject(
id=lp.id,
asset=lp.asset,
venue=lp.venue,
objective=lp.objective,
constraints=lp.constraints,
price_target=lp.price_target,
tolerance=lp.tolerance,
)
def to_dsl_morphisms(sw: SharedVariables) -> DSLMorphisms:
return DSLMorphisms(
contract_id=sw.contract_id,
version=sw.version,
variables=dict(sw.variables),
)
def to_dsl_dual(dd: DualVariables) -> DSLDualVariables:
return DSLDualVariables(
contract_id=dd.contract_id,
version=dd.version,
shadow_prices=dict(dd.shadow_prices),
)
def to_dsl_plan(delta: PlanDelta) -> DSLPlanDelta:
return DSLPlanDelta(
contract_id=delta.contract_id,
delta=dict(delta.delta),
timestamp=delta.timestamp,
author=delta.author,
privacy_budget=delta.privacy_budget,
)