92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
"""Policy-to-constraint DSL sketch for CityGrid.
|
|
|
|
This module provides a very small, pragmatic DSL surface to translate
|
|
city-level policy goals into global optimization constraints that can be
|
|
consumed by the MVP ADMM-like solver.
|
|
|
|
Design goal:
|
|
- Keep the surface minimal and implementation-friendly for an MVP.
|
|
- Provide a deterministic mapping from a policy dictionary to a constraint
|
|
dictionary that the LocalProblem/SharedVariables layer can understand.
|
|
- Do not attempt to be a full DSL; instead offer a tiny, well-typed bridge
|
|
that can be extended over time.
|
|
"""
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
def _clip(value: float, min_value: float, max_value: float) -> float:
|
|
return max(min_value, min(value, max_value))
|
|
|
|
|
|
def policy_to_constraints(policy: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Translate a policy dict into a global constraint spec.
|
|
|
|
Expected policy keys (all optional):
|
|
- reliability: dict with per-domain service targets, e.g. {"electric": 0.99}
|
|
- equity: dict with fairness targets, e.g. {"per_capita_energy": 0.8}
|
|
- climate: dict with targets like {"co2_reduction": 0.5}
|
|
- peaks: dict with peak load caps, e.g. {"hour_of_day": {"22": 0.9}}
|
|
- privacy: dict with privacy budgets per-signal, e.g. {"signal_a": 0.1}
|
|
|
|
The function returns a structure suitable for consumption by the MVP solver,
|
|
i.e. a mapping to global constraints that can be merged with local constraints.
|
|
"""
|
|
|
|
if not isinstance(policy, dict):
|
|
return {"global_constraints": {"note": "empty-policy"}}
|
|
|
|
global_constraints: Dict[str, Any] = {}
|
|
|
|
# Reliability constraints translate into bounded acceptable ranges for
|
|
# critical signals like voltages or reserve margins. We keep a simple
|
|
# numeric cap if provided.
|
|
reliability = policy.get("reliability")
|
|
if isinstance(reliability, dict):
|
|
# Example: {"electricity": {"target": 0.995, "min_buffer": 0.01}}
|
|
for domain, specs in reliability.items():
|
|
if not isinstance(specs, dict):
|
|
continue
|
|
target = specs.get("target")
|
|
if target is not None:
|
|
global_constraints.setdefault("reliability", {})[domain] = float(target)
|
|
# Optional min buffer applied as a constraint delta
|
|
min_buf = specs.get("min_buffer")
|
|
if min_buf is not None:
|
|
global_constraints.setdefault("reliability", {})[f"{domain}_buffer"] = float(min_buf)
|
|
|
|
# Equity constraints are represented as per-asset or per-domain bounds
|
|
equity = policy.get("equity")
|
|
if isinstance(equity, dict):
|
|
global_constraints["equity"] = equity
|
|
|
|
# Climate targets are mapped to a global CO2 reduction or energy mix target
|
|
climate = policy.get("climate")
|
|
if isinstance(climate, dict):
|
|
global_constraints["climate"] = climate
|
|
|
|
# Peak load caps
|
|
peaks = policy.get("peaks")
|
|
if isinstance(peaks, dict):
|
|
global_constraints["peaks"] = peaks
|
|
|
|
# Privacy budgets (per-signal budgets are recorded for auditability)
|
|
privacy = policy.get("privacy")
|
|
if isinstance(privacy, dict):
|
|
global_constraints["privacy_budgets"] = privacy
|
|
|
|
return {"global_constraints": global_constraints}
|
|
|
|
|
|
class PolicyDSL:
|
|
"""Lightweight namespace to keep policy utilities discoverable."""
|
|
|
|
@staticmethod
|
|
def translate(policy: Dict[str, Any]) -> Dict[str, Any]:
|
|
return policy_to_constraints(policy)
|
|
|
|
|
|
__all__ = ["policy_to_constraints", "PolicyDSL"]
|