From 8679af78691c89a3c9f51d5870f678917546afb8 Mon Sep 17 00:00:00 2001 From: agent-dd492b85242a98c5 Date: Mon, 20 Apr 2026 13:51:32 +0200 Subject: [PATCH] build(agent): new-agents-3#dd492b iteration --- README.md | 2 + crisispolicy.py | 112 ++++++++++++++++++ .../crisispolicy.py | 112 ++++++++++++++++++ 3 files changed, 226 insertions(+) create mode 100644 crisispolicy.py create mode 100644 src/idea168_crisispulse_federated_resource/crisispolicy.py diff --git a/README.md b/README.md index 97148ef..27be303 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,8 @@ This repository implements a production-grade MVP for CrisisPulse, a federated resource orchestration platform intended for disaster-relief camp networks. The MVP focuses on core primitives: local ledger with delta-sync, a contract registry (Graph-of-Contracts), adapters for domain-specific devices, a governance ledger, privacy-preserving summaries, and a testbed for cross-domain collaboration. +CrisisPulse also includes a tiny but extensible CrisisPolicy DSL (crisispolicy.py) to bootstrap interoperability across NGOs and humanitarian devices. This DSL is designed to be incrementally enhanced alongside the Graph-of-Contracts and adapters. + Structure overview: - src/idea168_crisispulse_federated_resource/: Python package with core modules - tests/: unit tests for core primitives diff --git a/crisispolicy.py b/crisispolicy.py new file mode 100644 index 0000000..c106cfe --- /dev/null +++ b/crisispolicy.py @@ -0,0 +1,112 @@ +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"] diff --git a/src/idea168_crisispulse_federated_resource/crisispolicy.py b/src/idea168_crisispulse_federated_resource/crisispolicy.py new file mode 100644 index 0000000..c106cfe --- /dev/null +++ b/src/idea168_crisispulse_federated_resource/crisispolicy.py @@ -0,0 +1,112 @@ +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"]