Wildfire Incident Response¶

import asyncio
import contextlib
import hashlib
import time
from typing import Literal

from pydantic import BaseModel, Field

from openagentmesh import AgentMesh, AgentSpec, KVEntry, KVKeyExists


class Coords(BaseModel):
    x: float = Field(ge=-5.0, le=5.0)  # km from HQ at the origin
    y: float = Field(ge=-5.0, le=5.0)


class CellState(BaseModel):
    """KV value at wildfire.world.cell.<x_idx>.<y_idx> (200 m grid)."""
    coords: Coords
    temperature: float  # degrees Celsius


class SurveyResult(BaseModel):
    surveyor_instance_id: str
    fire_visible: bool
    persons_detected: int


class DetectionRecord(BaseModel):
    """KV value at wildfire.detection.{detection_id}.

    Lifecycle: pending -> assigned:{instance_id} -> surveyed.
    """
    detection_id: str
    state: str
    coords: Coords
    severity: float  # 0..1, derived from temperature
    survey: SurveyResult | None = None


class TaskRequest(BaseModel):
    operator_id: str
    text: str  # natural language from the operator


class TaskCommand(BaseModel):
    target_fleet: Literal["heli", "ffunit", "medevac"]
    coords: Coords
    priority: Literal["low", "med", "high"]
    persons_estimated: int = 0
    rationale: str  # one line, for the audit log

TEMP_THRESHOLD_C = 100.0


def dedup_id(coords: Coords, now: float) -> str:
    """One detection per 100 m grid bucket per 30 s window."""
    bucket = f"{round(coords.x / 0.1)}:{round(coords.y / 0.1)}:{int(now // 30)}"
    return hashlib.sha1(bucket.encode()).hexdigest()[:16]


def build_uav(mesh: AgentMesh) -> None:
    @mesh.agent(
        AgentSpec(
            name="high-alt.uav",
            description="High-altitude thermal observer; writes pending "
                        "detections from world-cell updates.",
        ),
        sources=[mesh.kv_source("wildfire.world.cell.>", on_init="replay")],
    )
    async def uav(entry: KVEntry[CellState]) -> None:
        if entry.operation == "DELETE":
            return  # cell cooled back to ambient; detections are not retracted

        cell = entry.value
        if cell.temperature <= TEMP_THRESHOLD_C:
            return

        detection_id = dedup_id(cell.coords, time.time())
        record = DetectionRecord(
            detection_id=detection_id,
            state="pending",
            coords=cell.coords,
            severity=min(1.0, (cell.temperature - 100.0) / 700.0),
        )
        # KVKeyExists means this area was already reported in this window:
        # the create is the dedup.
        with contextlib.suppress(KVKeyExists):
            await mesh.kv.create(f"wildfire.detection.{detection_id}", record)

def build_drone(mesh: AgentMesh) -> None:
    @mesh.agent(
        AgentSpec(
            name="low-alt.drone",
            description="Low-altitude survey drone; CAS-elects itself on "
                        "pending detections and writes the survey result.",
        ),
        sources=[mesh.kv_source("wildfire.detection.*", on_init="replay")],
    )
    async def drone(entry: KVEntry[DetectionRecord]) -> None:
        if entry.operation == "DELETE":
            return
        rec = entry.value
        if rec.state != "pending":
            return

        key = f"wildfire.detection.{rec.detection_id}"

        # Election: one CAS attempt. Losing the race is data, not an error.
        async with mesh.kv.try_cas_model(key, DetectionRecord) as claim:
            if claim.value.state != "pending":
                return  # claimed between the event and our read
            claim.value.state = f"assigned:{mesh.instance_id}"
        if not claim.committed:
            return  # another drone won

        await asyncio.sleep(0.1)  # simulate travel + sensor sweep

        async with mesh.kv.try_cas_model(key, DetectionRecord) as done:
            done.value.state = "surveyed"
            done.value.survey = SurveyResult(
                surveyor_instance_id=mesh.instance_id,
                fire_visible=True,
                persons_detected=0,
            )

def build_tasker(mesh: AgentMesh) -> None:
    @mesh.agent(AgentSpec(
        name="tasker",
        description="Translates an operator's natural-language request into "
                    "one typed TaskCommand. Do NOT use it to execute or "
                    "dispatch anything: it only translates.",
    ))
    async def tasker(req: TaskRequest) -> TaskCommand:
        # Ground in live mesh state: target the newest surveyed detection.
        detections = await mesh.kv.list_models(
            "wildfire.detection.>", DetectionRecord,
        )
        surveyed = [e.value for e in detections if e.value.state == "surveyed"]
        target = surveyed[-1].coords if surveyed else Coords(x=0.0, y=0.0)

        # Stand-in for one structured LLM call (see below).
        text = req.text.lower()
        wants_extraction = "person" in text or "extract" in text
        return TaskCommand(
            target_fleet="medevac" if wants_extraction else "heli",
            coords=target,
            priority="high" if ("urgent" in text or "now" in text) else "med",
            rationale=f"Translated from operator {req.operator_id} request.",
        )

async def main(mesh: AgentMesh) -> None:
    build_uav(mesh)
    build_drone(mesh)
    build_tasker(mesh)
    await mesh.catalog()      # flush registrations: KV sources bind, agents go live
    await asyncio.sleep(0.1)  # let the watchers attach

    # Ignite: write one burning cell. Nobody dispatches anything.
    await mesh.kv.put_model(
        "wildfire.world.cell.31.21",
        CellState(coords=Coords(x=1.2, y=-0.8), temperature=620.0),
    )

    # Watch the detection record walk its lifecycle.
    async def until_surveyed() -> DetectionRecord:
        async for value in mesh.kv.watch("wildfire.detection.*"):
            rec = DetectionRecord.model_validate_json(value)
            print(f"detection {rec.detection_id}: {rec.state}")
            if rec.state == "surveyed":
                return rec

    rec = await asyncio.wait_for(until_surveyed(), timeout=10.0)
    assert rec.survey is not None  # a surveyed detection always carries its result
    print(f"survey: fire_visible={rec.survey.fire_visible}")

    # Operator speaks; the mesh answers with a typed, validated command.
    result = await mesh.call(
        "tasker",
        TaskRequest(operator_id="op-1", text="Send the water bomber, urgent"),
    )
    command = TaskCommand.model_validate(result)
    print(f"dispatch: {command.target_fleet} -> "
          f"({command.coords.x}, {command.coords.y}) [{command.priority}]")

import asyncio
from openagentmesh import AgentMesh

async def run():
    async with AgentMesh.local() as mesh:
        await main(mesh)

asyncio.run(run())

uv run python -m demos.wildfire --seed 42