My FireWatch AI Wildfire Detection Console

import json
import re
from typing import Dict, List

MAX_CHARS = [[[MAX_CHARS|6000]]]

TEMPLATE_SYSTEM_PROMPT = """You are FireWatch AI, an early wildfire detection analyst.

Task:
- Review the user's wildfire report.
- Combine multiple pieces of evidence when possible.
- Do not rely on one weak signal alone.
- Distinguish evidence from uncertainty.
- Avoid claiming a fire is verified unless evidence is strong.
- Watch for unusual or rapidly changing conditions.

Return plain text with this exact structure:
CLASSIFICATION: one of NORMAL | INVESTIGATE | POSSIBLE FIRE | CONFIRMED/VERIFIED FIRE
CONFIDENCE: Low | Medium | High
EVIDENCE:
- bullet list of observed evidence that supports the assessment
UNCERTAINTIES:
- bullet list of missing, conflicting, or weak signals
UNUSUAL OR RAPID CHANGES:
- bullet list, or '- none noted'
RECOMMENDED NEXT ACTION:
- one primary action
STRUCTURED SUMMARY:
{"classification":"...","confidence":"...","evidence":[...],"uncertainties":[...],"unusual_changes":[...],"recommended_action":"...","location":"...","requires_human_review":true}

Rules:
- If the evidence is incomplete, say so.
- If there is no clear smoke, thermal, or visual evidence, be cautious.
- Use precise operational language.
- Keep the structured summary valid JSON on one line.
"""


def extract_signal_summary(report: str) -> str:
    """Extract a simple signal summary from a wildfire report.

    Args:
        report: Free-form wildfire sensor and observation report text.

    Returns:
        Short bullet-style summary of detected keywords and values.
    """
    text = report[:MAX_CHARS]
    findings: List[str] = []

    patterns = {
        "temperature": r"(?:temperature|temp)\s*[:=]?\s*([-+]?\d+(?:\.\d+)?)",
        "humidity": r"humidity\s*[:=]?\s*(\d+(?:\.\d+)?)\s*%",
        "pm": r"(?:pm2\.5|pm10|particulate|particulates)\s*[:=]?\s*([-+]?\d+(?:\.\d+)?)",
        "thermal": r"(?:thermal|hotspot|heat signature)[^\n]{0,50}?([-+]?\d+(?:\.\d+)?)\s*(?:c|f)?",
        "gps": r"(-?\d{1,3}\.\d+)\s*,\s*(-?\d{1,3}\.\d+)"
    }

    for label, pattern in patterns.items():
        match = re.search(pattern, text, flags=re.IGNORECASE)
        if match:
            if label == "gps":
                findings.append(f"{label}: {match.group(1)}, {match.group(2)}")
            else:
                findings.append(f"{label}: {match.group(1)}")

    keyword_flags = [
        "smoke", "plume", "ash", "burning", "flame", "glow", "hotspot",
        "wind", "gust", "dry brush", "dry vegetation", "rapid increase", "camera"
    ]
    for keyword in keyword_flags:
        if keyword in text.lower():
            findings.append(f"keyword: {keyword}")

    if not findings:
        return "- No distinct signal keywords or values were automatically extracted."
    return "\n".join(f"- {item}" for item in findings[:20])


async def process_user_query(query: str) -> str:
    """Analyze a wildfire monitoring report and return a structured assessment.

    Args:
        query: User-supplied wildfire report plus operating preferences.

    Returns:
        Structured wildfire risk assessment text.
    """
    trimmed = query[:MAX_CHARS]
    signal_summary = extract_signal_summary(trimmed)

    prompt = f"""Analyze this wildfire monitoring report.

Signal summary extracted from the report:
{signal_summary}

Full report:
{trimmed}

Instructions:
- Base the decision on combined evidence.
- Explicitly separate evidence from uncertainty.
- Mention unusual or rapidly changing conditions.
- Recommend the next best action.
- Be careful not to overstate confirmation.
- Follow the exact output structure from the system prompt.
"""

    answer = await agentop_llm.generate(prompt, globals().get("TEMPLATE_SYSTEM_PROMPT", ""))

    if not isinstance(answer, str):
        return str(answer)

    cleaned = answer.strip()
    try:
        json_match = re.search(r"\{.*\}", cleaned, flags=re.DOTALL)
        if json_match:
            parsed = json.loads(json_match.group(0))
            parsed.setdefault("requires_human_review", True)
    except Exception:
        pass
    return cleaned