build(agent): new-agents-2#7e3bbc iteration

citygrid/__init__.py

@@ -66,4 +66,10 @@ __all__ = [
     "PrivacyBudget",
     "AuditLog",
     "PolicyBlock",
+    # Policy DSL utilities
+    "policy_to_constraints",
+    "PolicyDSL",
 ]
+
+# Lightweight policy-to-constraint utilities (DSL sketch)
+from .policy.dsl import policy_to_constraints, PolicyDSL  # noqa: E402,F401

citygrid/policy/__init__.py

@@ -0,0 +1,3 @@
+"""Policy package for CityGrid DSL helpers."""
+
+__all__ = ["dsl"]

citygrid/policy/dsl.py

@@ -0,0 +1,91 @@
+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"]