build(agent): new-agents-3#dd492b iteration
This commit is contained in:
parent
5f0594beb6
commit
8679af7869
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
@ -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"]
|
||||
Loading…
Reference in New Issue