Implementation Guide: Analyze 311 / Incident Data & Multi-Source Feeds — Resource Gaps, Equity Issues & Surge Requirements

# Ingests and anonymizes 311, CAD, and public health data for AI analysis

# data_ingestion_pipeline.py
# Ingests and anonymizes 311, CAD, and public health data for AI analysis

import pandas as pd
import sqlalchemy
import requests
import os
from datetime import datetime, timedelta
import hashlib

# Azure SQL (Government) connection
AZURE_SQL_CONN = os.environ["AZURE_SQL_CONNECTION_STRING"]
engine = sqlalchemy.create_engine(AZURE_SQL_CONN)

# ── 311 Data (via Socrata / Tyler API) ──────────────────────────────────────

def fetch_311_data(socrata_domain: str, dataset_id: str, days_back: int = 30) -> pd.DataFrame:
    """Fetch 311 service requests from Socrata API."""
    cutoff = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%dT00:00:00")

    url = f"https://{socrata_domain}/resource/{dataset_id}.json"
    params = {
        "$where": f"created_date > '{cutoff}'",
        "$limit": 50000,
        "$select": "service_request_id,created_date,closed_date,status,service_name,"
                   "agency_responsible,latitude,longitude,zip_code,census_tract,ward"
        # Note: street address deliberately excluded for privacy
    }

    resp = requests.get(url, params=params,
                       headers={"X-App-Token": os.environ["SOCRATA_APP_TOKEN"]})
    resp.raise_for_status()
    return pd.DataFrame(resp.json())


def anonymize_311_data(df: pd.DataFrame) -> pd.DataFrame:
    """Apply anonymization and derived fields to 311 data."""
    # Remove any accidentally included PII columns
    pii_columns = ["address", "street_address", "cross_street", "requestor_name",
                   "requestor_email", "requestor_phone", "account_id"]
    df = df.drop(columns=[c for c in pii_columns if c in df.columns])

    # Convert timestamps
    df["created_date"] = pd.to_datetime(df["created_date"])
    df["closed_date"] = pd.to_datetime(df["closed_date"], errors="coerce")

    # Derived fields for analysis
    df["response_days"] = (df["closed_date"] - df["created_date"]).dt.days
    df["created_month"] = df["created_date"].dt.to_period("M").astype(str)
    df["created_dow"] = df["created_date"].dt.day_name()
    df["created_hour"] = df["created_date"].dt.hour

    # Aggregate to census tract level (remove precise coordinates)
    # Keep census_tract for geographic analysis, drop lat/lon
    df = df.drop(columns=["latitude", "longitude"], errors="ignore")

    return df


# ── CAD / Public Safety Data ─────────────────────────────────────────────────

def fetch_cad_aggregate_data(cad_api_url: str, start_date: str, end_date: str) -> pd.DataFrame:
    """
    Fetch PRE-AGGREGATED incident data from CAD system.
    IMPORTANT: Only fetch aggregated counts, not individual incident records.
    Individual CAD records contain sensitive PII and operational details.
    """
    headers = {
        "Authorization": f"Bearer {os.environ['CAD_API_KEY']}",
        "Content-Type": "application/json"
    }

    # Request aggregate data only (by zone, shift, incident type)
    payload = {
        "start_date": start_date,
        "end_date": end_date,
        "group_by": ["dispatch_zone", "shift", "incident_type", "priority"],
        "metrics": ["incident_count", "avg_response_time_min", "median_response_time_min",
                    "p90_response_time_min", "unit_count"]
        # Do NOT request individual incident IDs, addresses, or person details
    }

    resp = requests.post(f"{cad_api_url}/api/aggregate", headers=headers, json=payload)
    resp.raise_for_status()
    return pd.DataFrame(resp.json().get("data", []))


# ── Public Health Data ────────────────────────────────────────────────────────

def fetch_hospital_capacity(fhir_base_url: str) -> pd.DataFrame:
    """
    Fetch aggregate hospital capacity data via HL7 FHIR API.
    Requires data sharing agreement and HIPAA-compliant data access.
    Returns facility-level aggregate metrics only — no patient-level data.
    """
    headers = {
        "Authorization": f"Bearer {os.environ['FHIR_ACCESS_TOKEN']}",
        "Accept": "application/fhir+json"
    }

    # MeasureReport resource provides aggregate capacity metrics
    resp = requests.get(
        f"{fhir_base_url}/MeasureReport",
        headers=headers,
        params={
            "measure": "http://www.cdc.gov/nhsn/fhirportal/dqm/ig/Measure/NHSNHospitalCapacityAndUtilizationMeasure",
            "_count": 100
        }
    )
    resp.raise_for_status()

    capacity_records = []
    for entry in resp.json().get("entry", []):
        resource = entry.get("resource", {})
        facility_id = resource.get("subject", {}).get("reference", "Unknown")

        for group in resource.get("group", []):
            for pop in group.get("population", []):
                capacity_records.append({
                    "facility_id": hashlib.sha256(facility_id.encode()).hexdigest()[:8],  # Pseudonymize
                    "metric": group.get("code", {}).get("text", ""),
                    "value": pop.get("count", 0),
                    "date": resource.get("date", ""),
                    "period_start": resource.get("period", {}).get("start", ""),
                    "period_end": resource.get("period", {}).get("end", "")
                })

    return pd.DataFrame(capacity_records)


def load_to_analytical_db(df: pd.DataFrame, table_name: str, if_exists: str = "append"):
    """Load processed data to Azure SQL analytical database."""
    df["load_timestamp"] = datetime.utcnow()
    df.to_sql(table_name, engine, if_exists=if_exists, index=False, chunksize=1000)
    print(f"Loaded {len(df)} rows to {table_name}")

# defines generate_311_analysis_report() for 311 service data analysis and
# generate_surge_forecast() for public health surge capacity forecasting
# using Azure OpenAI

# analytical_report_generator.py

from openai import AzureOpenAI
import pandas as pd
import os, json

client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_key=os.environ["AZURE_OPENAI_KEY"],
    api_version="2024-08-01-preview"
)

def generate_311_analysis_report(df_summary: pd.DataFrame, period: str, jurisdiction: str) -> str:
    """Generate AI-assisted analysis of 311 service request data."""

    # Compute summary statistics for AI input
    stats = {
        "period": period,
        "jurisdiction": jurisdiction,
        "total_requests": int(df_summary["incident_count"].sum()),
        "by_service_type": df_summary.groupby("service_name")["incident_count"].sum().nlargest(10).to_dict(),
        "by_zone": df_summary.groupby("census_tract")["incident_count"].sum().to_dict(),
        "avg_response_days_by_type": df_summary.groupby("service_name")["avg_response_days"].mean().to_dict(),
        "open_over_30_days": int(df_summary.get("open_over_30_days", pd.Series([0])).sum()),
        "prior_period_comparison": "Provide if available"
    }

    analysis_prompt = f"""Analyze the following 311 service request data summary for {jurisdiction}.

DATA SUMMARY:
{json.dumps(stats, indent=2, default=str)}

Generate a structured analysis report covering:

## EXECUTIVE SUMMARY
- 3 key findings leadership should know immediately
- Overall service delivery health assessment (Good/Fair/Poor) with rationale

## SERVICE REQUEST VOLUME AND TRENDS
- Volume compared to prior period (if data available)
- Top 5 service categories and their trends
- Day-of-week and time-of-day patterns

## RESPONSE TIME ANALYSIS
- Average response time by service category
- Categories with response times exceeding benchmarks
- Geographic zones with consistently longer response times

## RESOURCE ALLOCATION ANALYSIS
- Which service categories or geographic areas appear under-resourced based on volume and response times?
- Seasonal or event-driven spikes requiring additional capacity?
- Staff or equipment resource gaps implied by the data

## EQUITY ANALYSIS
- Are response times or service levels consistent across geographic areas?
- Flag any census tracts or wards with consistently lower service levels
- Note: demographic data would be needed to make definitive equity conclusions — flag as requiring additional analysis

## RECOMMENDATIONS
- 3-5 specific, actionable resource allocation recommendations
- Each recommendation: what to do, why, expected impact, data confidence level (High/Medium/Low)

## DATA QUALITY NOTES
- Any data quality issues observed (missing values, anomalous entries)
- Limitations of this analysis

[DRAFT — AI GENERATED — REQUIRES DEPARTMENT DIRECTOR REVIEW BEFORE DISTRIBUTION]"""

    response = client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
        messages=[
            {"role": "system", "content": "You are a government performance analytics specialist. Generate evidence-based, actionable analysis reports for local government operations managers."},
            {"role": "user", "content": analysis_prompt}
        ],
        temperature=0.2,
        max_tokens=3000
    )

    return response.choices[0].message.content


def generate_surge_forecast(capacity_df: pd.DataFrame, epi_df: pd.DataFrame,
                            weather_data: dict, jurisdiction: str) -> str:
    """Generate public health surge forecast from multi-source data."""

    capacity_summary = capacity_df.groupby("metric")["value"].agg(["mean", "max", "min"]).to_dict()

    forecast_prompt = f"""Analyze the following public health data and generate a
surge capacity forecast for {jurisdiction}.

HOSPITAL CAPACITY METRICS (aggregate, anonymized):
{json.dumps(capacity_summary, indent=2, default=str)}

EPIDEMIOLOGICAL INDICATORS:
{epi_df.to_json(orient='records')[:2000] if not epi_df.empty else 'No epi data provided'}

WEATHER AND ENVIRONMENTAL FACTORS:
{json.dumps(weather_data, indent=2)}

Generate a Surge Capacity Forecast Report:

## CURRENT CAPACITY STATUS
- Overall system capacity utilization (%)
- Facilities approaching critical thresholds (>85% capacity)
- Specialty capacity concerns (ICU, ED, pediatric, behavioral health)

## 7-DAY SURGE FORECAST
- Projected demand change (%, with confidence interval)
- Driving factors (seasonal illness, weather events, community events)
- Probability of exceeding 90% system capacity: [%]

## RESOURCE GAP ANALYSIS
- Projected staff shortfalls by category (RN, MD, support)
- Equipment and supply concerns
- Transfer capacity if local surge exceeds capacity

## RECOMMENDED ACTIONS
Tiered response recommendations:
- Preparatory actions (take now regardless of surge)
- Trigger-based actions (take if X metric reaches Y threshold)
- Escalation actions (if surge exceeds forecast)

## MUTUAL AID CONSIDERATIONS
- Which neighboring jurisdictions may experience concurrent surge?
- EMAC resource request lead times for key resources

[DRAFT — REQUIRES PUBLIC HEALTH DIRECTOR REVIEW — DO NOT DISTRIBUTE EXTERNALLY]"""

    response = client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
        messages=[
            {"role": "system", "content": "You are a public health emergency management analyst. Generate evidence-based, actionable surge capacity forecasts. Do not overstate certainty — clearly communicate confidence levels and data limitations."},
            {"role": "user", "content": forecast_prompt}
        ],
        temperature=0.1,
        max_tokens=3000
    )

    return response.choices[0].message.content

# display prominently on Page 2 with trend (improving/worsening)

Equity Score = 1 - (CV of response times across census tracts)
where CV = coefficient of variation (std dev / mean)

Score > 0.85  = Equitable
0.70 - 0.85   = Moderate disparity
< 0.70        = Significant disparity

https://api.powerbi.us/v1.0/myorg/reports/{reportId}/ExportToFile

SYSTEM PROMPT:
You are a government equity analyst preparing a plain-language summary of
service delivery equity data for a city council presentation.

Translate the following equity metrics into a clear, factual narrative suitable
for a public audience. The narrative should:
- Describe what the data shows in plain language (no statistical jargon)
- Acknowledge both positive findings and areas of concern
- Be factual and data-grounded — do not editorialize or assign blame
- Note data limitations clearly
- Suggest specific, constructive next steps
- Be suitable for public release (no internal deliberative content)
- Maximum 400 words

EQUITY METRIC DATA:
{equity_metrics}

GEOGRAPHIC CONTEXT:
{geographic_context}