idea168-crisispulse-federat.../crisispolicy.py

113 lines
3.8 KiB
Python

from __future__ import annotations
from typing import Dict, Any, Optional
class CrisisPolicyDSL:
"""Tiny, production-friendly DSL scaffold for crisis response policies.
This is intentionally minimal but designed for extension:
- Policies describe constraints on domain resources (e.g., energy, water, food).
- A policy is expressed as a simple key-operator-value triplet, e.g.:
"allow:energy>0; limit:waste_perishable=5".
The parser returns a normalized dictionary representation for downstream engines
(e.g., a delta allocator or contract registry).
"""
def __init__(self, text: str = "") -> None:
self.text = text
self.parsed: Optional[Dict[str, Any]] = None
@staticmethod
def _parse_token(token: str) -> Optional[Dict[str, Any]]:
token = token.strip()
if not token:
return None
# Very small grammar: key comparator value
# Supported operators: >, >=, <, <=, ==, !=
for op in [">=", "<=", ">", "<", "==", "!="]:
if op in token:
key, val = token.split(op, 1)
key = key.strip()
val = val.strip()
return {"key": key, "op": op, "value": CrisisPolicyDSL._cast(val)}
# If no operator found, treat as a simple flag
return {"flag": token}
@staticmethod
def _cast(v: str) -> Any:
# Best-effort cast to int/float/bool, else string
if v.lower() in {"true", "false"}:
return v.lower() == "true"
try:
if "." in v:
return float(v)
return int(v)
except ValueError:
return v
def parse(self) -> Dict[str, Any]:
if self.parsed is None:
self.parsed = {
"raw": self.text,
"tokens": [],
}
if not self.text:
return self.parsed
# Split by semicolons into tokens
for raw_token in self.text.split(";"):
t = self._parse_token(raw_token)
if t is not None:
self.parsed["tokens"].append(t)
return self.parsed
def validate(self, plan: Dict[str, Any]) -> bool:
"""Very small validator against a plan delta-like structure.
The plan is expected to be a dict with domain keys and numeric allocations.
For example: {"energy": 12, "water": 50}
This function demonstrates the intent and can be extended with a real
rule engine.
"""
if not self.parsed:
self.parse()
if not self.parsed or not self.parsed.get("tokens"):
return True # nothing to validate
# Simple policy semantics: if a policy token constrains a key to be > 0, ensure plan has it > 0
for tok in self.parsed["tokens"]:
if "flag" in tok:
# a bare policy flag does not affect numeric validation in this tiny DSL
continue
key = tok.get("key")
op = tok.get("op")
val = tok.get("value")
if key is None or op is None:
continue
plan_val = plan.get(key)
if plan_val is None:
# missing required resource allocation violates a policy that expects allocation
return False
# Compare using the operator
if not CrisisPolicyDSL._compare(plan_val, op, val):
return False
return True
@staticmethod
def _compare(a: Any, op: str, b: Any) -> bool:
if op == ">":
return a > b
if op == ">=":
return a >= b
if op == "<":
return a < b
if op == "<=":
return a <= b
if op == "==":
return a == b
if op == "!=":
return a != b
return False
__all__ = ["CrisisPolicyDSL"]