Compliance Checklists: NIMS ICS-209, FEMA BPAS & OGC API Features

import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Mapping

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger(__name__)


class Severity(str, Enum):
    REQUIRED = "required"   # absence is a hard fail
    ADVISORY = "advisory"   # absence degrades but does not fail the record


@dataclass(frozen=True)
class Requirement:
    """One checkable clause of a standard."""
    key: str                                   # stable id, e.g. "ics209.incident_number"
    label: str                                 # human description for the audit log
    severity: Severity
    predicate: Callable[[Mapping[str, Any]], bool]


@dataclass
class Finding:
    key: str
    label: str
    severity: Severity
    passed: bool
    detail: str = ""


@dataclass
class Checklist:
    standard: str                              # e.g. "NIMS ICS-209"
    version: str                               # e.g. "form-edition-2022.1"
    requirements: List[Requirement] = field(default_factory=list)

    def evaluate(self, payload: Mapping[str, Any]) -> List[Finding]:
        """Evaluate every requirement; never fail fast so operators see all gaps at once."""
        findings: List[Finding] = []
        for req in self.requirements:
            try:
                passed = bool(req.predicate(payload))
                detail = "" if passed else "predicate returned False"
            except Exception as exc:
                # A predicate that raises is treated as a failed check, not a crash,
                # so one malformed field cannot abort the whole conformance run.
                logger.warning("Requirement %s raised during evaluation: %s", req.key, exc)
                passed, detail = False, f"predicate error: {exc}"
            findings.append(
                Finding(req.key, req.label, req.severity, passed, detail)
            )
        return findings

import logging
from datetime import datetime, timezone
from typing import Any, List, Mapping

logger = logging.getLogger(__name__)


def _present(field_name: str) -> "Callable[[Mapping[str, Any]], bool]":
    return lambda p: p.get(field_name) not in (None, "")


def _pct_in_range(field_name: str) -> "Callable[[Mapping[str, Any]], bool]":
    def check(p: Mapping[str, Any]) -> bool:
        value = p.get(field_name)
        return isinstance(value, (int, float)) and 0 <= value <= 100
    return check


def _utc_timestamp(field_name: str) -> "Callable[[Mapping[str, Any]], bool]":
    def check(p: Mapping[str, Any]) -> bool:
        raw = p.get(field_name)
        if not isinstance(raw, str):
            return False
        try:
            dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
        except ValueError:
            return False
        return dt.tzinfo is not None and dt.utcoffset() == timezone.utc.utcoffset(None)
    return check


def _lat_lon_valid(p: Mapping[str, Any]) -> bool:
    lat, lon = p.get("origin_latitude"), p.get("origin_longitude")
    return (
        isinstance(lat, (int, float)) and -90 <= lat <= 90
        and isinstance(lon, (int, float)) and -180 <= lon <= 180
    )


def build_ics209_checklist() -> "Checklist":
    reqs = [
        Requirement("ics209.incident_number", "Incident number present",
                    Severity.REQUIRED, _present("incident_number")),
        Requirement("ics209.incident_name", "Incident name present",
                    Severity.REQUIRED, _present("incident_name")),
        Requirement("ics209.incident_kind", "Incident kind present",
                    Severity.REQUIRED, _present("incident_kind")),
        Requirement("ics209.report_period_utc", "Reporting period end is UTC ISO 8601",
                    Severity.REQUIRED, _utc_timestamp("report_period_end")),
        Requirement("ics209.incident_size", "Incident size (acres) is a positive number",
                    Severity.REQUIRED, lambda p: isinstance(p.get("size_acres"), (int, float)) and p["size_acres"] >= 0),
        Requirement("ics209.percent_contained", "Percent contained within 0-100",
                    Severity.REQUIRED, _pct_in_range("percent_contained")),
        Requirement("ics209.point_of_origin", "Point of origin latitude/longitude valid",
                    Severity.REQUIRED, _lat_lon_valid),
        Requirement("ics209.incident_commander", "Incident commander named",
                    Severity.ADVISORY, _present("incident_commander")),
    ]
    return Checklist(standard="NIMS ICS-209", version="form-edition-2022.1", requirements=reqs)


def run_ics209(report: Mapping[str, Any]) -> List["Finding"]:
    findings = build_ics209_checklist().evaluate(report)
    failed = [f.key for f in findings if not f.passed and f.severity is Severity.REQUIRED]
    if failed:
        logger.error("ICS-209 report failed %d required field(s): %s", len(failed), failed)
    return findings

import logging
from typing import Any, List, Mapping, Sequence

import httpx

logger = logging.getLogger(__name__)

# Canonical OGC API - Features conformance class URIs.
OGC_CORE = "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/core"
OGC_GEOJSON = "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/geojson"
OGC_OPENAPI = "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/oas30"

REQUIRED_BPAS_ELEMENTS: Sequence[str] = (
    "submission_id", "business_process", "authoritative_source",
    "effective_date", "geometry", "record_status",
)


class ConformanceUnavailable(RuntimeError):
    """Raised when a live service cannot be interrogated for its conformance declaration."""


def build_bpas_checklist() -> "Checklist":
    reqs = [
        Requirement(f"bpas.element.{name}", f"Required BPAS element '{name}' present",
                    Severity.REQUIRED, _present(name))
        for name in REQUIRED_BPAS_ELEMENTS
    ]
    reqs.append(
        Requirement("bpas.effective_date_utc", "Effective date is UTC ISO 8601",
                    Severity.REQUIRED, _utc_timestamp("effective_date"))
    )
    return Checklist(standard="FEMA BPAS", version="bpas-2024.03", requirements=reqs)


def fetch_ogc_conformance(base_url: str, timeout_s: float = 10.0) -> List[str]:
    """Return the conformance class URIs a service advertises at /conformance."""
    url = base_url.rstrip("/") + "/conformance"
    try:
        resp = httpx.get(url, headers={"Accept": "application/json"}, timeout=timeout_s)
        resp.raise_for_status()
    except httpx.HTTPError as exc:
        logger.error("Could not reach OGC conformance endpoint %s: %s", url, exc)
        raise ConformanceUnavailable(f"conformance fetch failed for {base_url}") from exc
    classes = resp.json().get("conformsTo", [])
    logger.info("Service %s advertises %d conformance class(es)", base_url, len(classes))
    return list(classes)


def build_ogc_checklist(advertised: Sequence[str]) -> "Checklist":
    advertised_set = set(advertised)
    required = {
        "ogc.core": OGC_CORE,
        "ogc.geojson": OGC_GEOJSON,
    }
    advisory = {"ogc.openapi": OGC_OPENAPI}
    reqs: List[Requirement] = []
    for key, uri in required.items():
        reqs.append(Requirement(key, f"Conformance class advertised: {uri}",
                                Severity.REQUIRED, (lambda u: lambda _p: u in advertised_set)(uri)))
    for key, uri in advisory.items():
        reqs.append(Requirement(key, f"Conformance class advertised: {uri}",
                                Severity.ADVISORY, (lambda u: lambda _p: u in advertised_set)(uri)))
    return Checklist(standard="OGC API - Features", version="part1-1.0", requirements=reqs)

import json
import logging
from dataclasses import asdict
from datetime import datetime, timezone
from typing import Any, Dict, List, Mapping, Tuple

logger = logging.getLogger(__name__)


def build_conformance_record(
    artifact_id: str,
    checklist_results: List[Tuple["Checklist", List["Finding"]]],
) -> Dict[str, Any]:
    """Aggregate multi-standard findings into one immutable, auditable record."""
    gates: List[Dict[str, Any]] = []
    overall_conformant = True

    for checklist, findings in checklist_results:
        required_failures = [
            f for f in findings if not f.passed and f.severity is Severity.REQUIRED
        ]
        gate_pass = not required_failures
        overall_conformant &= gate_pass
        gates.append({
            "standard": checklist.standard,
            "checklist_version": checklist.version,
            "conformant": gate_pass,
            "findings": [asdict(f) | {"severity": f.severity.value} for f in findings],
        })

    record = {
        "artifact_id": artifact_id,
        "evaluated_utc": datetime.now(timezone.utc).isoformat(),
        "conformant": overall_conformant,
        "gates": gates,
    }
    level = logging.INFO if overall_conformant else logging.ERROR
    logger.log(level, "Conformance for %s: %s", artifact_id,
               "PASS" if overall_conformant else "FAIL")
    return record


def emit_conformance_record(record: Mapping[str, Any], sink_path: str) -> None:
    """Append the record to a durable, append-only audit log (never overwrite)."""
    try:
        with open(sink_path, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(record, sort_keys=True) + "\n")
    except OSError as exc:
        # Failing to persist an audit record must be loud: a silent audit gap is a
        # compliance failure in its own right.
        logger.critical("Failed to persist conformance record for %s: %s",
                        record.get("artifact_id"), exc)
        raise

import logging

logger = logging.getLogger(__name__)


def smoke_test() -> None:
    good_report = {
        "incident_number": "CA-ABC-012345",
        "incident_name": "Ridge Complex",
        "incident_kind": "Wildfire",
        "report_period_end": "2026-07-13T06:00:00Z",
        "size_acres": 14200,
        "percent_contained": 35,
        "origin_latitude": 39.51,
        "origin_longitude": -121.44,
        # incident_commander (advisory) intentionally omitted
    }

    # 1. A complete report passes ICS-209 with only an advisory gap.
    findings = run_ics209(good_report)
    required_failed = [f for f in findings if not f.passed and f.severity is Severity.REQUIRED]
    assert not required_failed, "complete report must satisfy every required field"

    # 2. Evaluation does not fail fast: all requirements are reported.
    assert len(findings) == len(build_ics209_checklist().requirements)

    # 3. A missing required field fails the record but not the run.
    bad = {k: v for k, v in good_report.items() if k != "percent_contained"}
    bad_findings = run_ics209(bad)
    assert any(f.key == "ics209.percent_contained" and not f.passed for f in bad_findings)

    # 4. Advisory-only failure still yields an overall conformant record.
    record = build_conformance_record("INC-smoke", [(build_ics209_checklist(), findings)])
    assert record["conformant"] is True, "advisory gap must not veto conformance"

    logger.info("conformance smoke test passed")


smoke_test()

python -c "import httpx, pydantic; print('conformance stack ok')"
python -m emergency_conformance.smoke   # exits non-zero on any failed assertion