idea143-crisisguard-federat.../crisisguard/toy_scenario/disaster_scenario.py

63 lines
2.0 KiB
Python

"""Toy disaster scenario for CrisisGuard MVP.
This module wires a minimal LocalPlan, applies a PlanDelta through starter
adapters, and records governance events in a GovernanceLog. It is intended
for local experimentation and to demonstrate the end-to-end flow in a
sandbox.
"""
from __future__ import annotations
from crisisguard.core import LocalPlan, PlanDelta, GovernanceLog
from crisisguard.registry import GraphOfContracts
from crisisguard.adapters import supply_chain_optimizer, shelter_planner
from crisisguard.core import GovernanceLog as _GL # for type hints compatibility
def run_scenario() -> dict:
# Base plan with neighborhoods and resources
base = LocalPlan(
plan_id="LP1",
neighborhood="North",
contents={"shelter": {"capacity": 50}, "medical": {"stock": 100}},
)
# Delta updating shelter capacity and shelter plan info
delta = PlanDelta(
delta_id="D1",
base_plan_id=base.plan_id,
updates={"shelter": {"capacity": 60}},
author="partnerA",
)
# Wire adapters (toy path: attempt both adapters to demonstrate usage)
updated_by_supply = supply_chain_optimizer.process(base, delta)
updated_by_shelter = shelter_planner.process(base, delta)
# Governance log entry
log = GovernanceLog()
e = log.append("delta_applied", {"delta_id": delta.delta_id}, "partnerA")
# Simple registry usage (no external persistence in this toy MVP)
g = GraphOfContracts()
g.register_adapter("SupplyChainAdapter", "0.1.0", {"capability": "supply"})
g.register_adapter("ShelterPlannerAdapter", "0.1.0", {"capability": "shelter"})
return {
"base": base.to_dict(),
"updated_by_supply": updated_by_supply.to_dict(),
"updated_by_shelter": updated_by_shelter.to_dict(),
"governance": log.to_dict(),
"registry": {
"adapters": [a for a in g.list_adapters()],
},
"events": e,
}
if __name__ == "__main__":
import json
result = run_scenario()
print(json.dumps(result, indent=2))