# Project export: Dwell

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: UC Berkeley AI Hackathon 2026
- Tagline: AI agent turning scattered government data into a city's early warning system, flagging at risk neighborhoods, why, and what to fix first, before residents feel it.
- Devpost: https://devpost.com/software/dwell-2a79sw
- GitHub: https://github.com/Rik-Banerjee/dwell
- Video: https://www.youtube.com/embed/jGVV2ODN8nI?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 2 GitHub contributor(s) — Rik Banerjee (7 commits), sreyas (1 commits)

## Devpost submission (written by the team)

### Inspiration

By the time a building failure is visible to a resident, it's already been predictable in public data for months — 311 complaints that went unanswered, permits that expired without inspection, seismic risk that nobody acted on. That data is technically public, but it's buried across a dozen disconnected government databases in formats no normal person can read. A renter has no way of knowing that next year they could be dealing with water damage, foundation problems, or major repair disruptions—all because warning signs were buried in millions of rows of government permit, inspection, and infrastructure records that no one ever checks. The neighborhoods that pay the price for this gap are almost always the ones with the fewest resources to recover. Dwell exists to close it.

### What it does

Enter an address and Dwell pulls real data from USGS, NWS/FEMA, USDA SSURGO, and DataSF — building permits, 311 history, soil quality, seismic and flood risk. It reasons across signals: a building with no soft-story retrofit, a cluster of water damage permits, and two open sewer complaints tells a different story than any one of those alone. The output is a plain-language safety report with color-coded risk findings, a prioritized roadmap, and realistic cost ranges. For renters, every finding maps to action — what a landlord is legally required to fix, how to file a 311 complaint that creates a paper trail.

### How we built it

A Python data layer wraps four public APIs into clean, cacheable functions. A Claude-powered ReAct agent decides which tools to call, then passes results to a deterministic scoring rubric kept separate from the LLM so every grade is auditable, not a black box. A lightweight TF-IDF system grounds cost estimates in real public works data. Every API response is cached on first run so the demo is instant and Wi-Fi-resilient.

### Challenges we ran into

We originally planned a trained ML failure predictor, but the available public data is current-state snapshots — not labeled failure events — so we pivoted to an auditable reasoning system instead. That turned out to be the stronger product: residents making real decisions need to trust the output, and a system that shows its work beats one that doesn't. We also hit a JSON truncation bug from an undersized token budget, and built graceful fallback handling after the SF 311 feed dropped mid-test — because a silent failure producing a falsely clean report is worse than an error.

### What we learned

The hard part of public infrastructure data isn't access — it's synthesis across sources that were never designed to talk to each other. A transparent rubric paired with an LLM that explains its reasoning is more trustworthy than a black-box score, especially when the output affects where someone lives. No one should need to be a data engineer to find out if their building is safe.

## README (from the GitHub repository)

# Dwell

AI agent that turns scattered public government data into a city's early warning system, flagging which locations are at risk of infrastructure failure, why, and what to fix first, before residents feel it.

**Demo video:** https://www.youtube.com/watch?v=jGVV2ODN8nI

Dwell takes unstructured, disparate public data, soil reports, seismic records, weather and disaster history, building permits, complaint logs, and turns it into a structured infrastructure risk assessment: which hazards are active, how severe, on what general timeframe, and what to do about it. The goal is to surface infrastructure risk before it becomes failure, not by predicting exact failure dates, but by giving cities the structured, evidence based read they need to act early.

## What it does

Give Dwell an address and it pulls real data from federal and municipal sources (soil and terrain, earthquake history, weather and FEMA disaster declarations, building permits and 311 complaints), scores the location across four hazard categories using a transparent, auditable rubric, grounds its recommendations in curated public works cost data, and produces:

- A plain language welfare summary of overall risk
- Four color coded hazard scores (green, orange, red): seismic, flood and ground, structural, infrastructure decay
- A prioritized, costed roadmap of concrete preventative actions
- Renter specific actions for people who do not own the property
- Honest disclosure of any data gaps encountered during the assessment

## How it works

1. **Gather.** A Claude powered ReAct agent calls four data tools as needed: USDA SSURGO soil data, USGS earthquake history, NWS weather alerts and FEMA disaster declarations, and DataSF building permits and 311 complaints.
2. **Score.** A deterministic, hand built scoring rubric, not a trained model, turns that raw data into the four hazard grades. Every point is traceable to a specific data field, so a red grade always has a concrete, explainable reason.
3. **Ground.** A lightweight retrieval system pulls relevant public works knowledge (typical retrofit costs, pipe replacement ranges) so the roadmap's cost estimates are grounded rather than invented.
4. **Report.** The agent writes the final structured report using the rubric's grades as authoritative and the retrieved knowledge as its cost basis.

We deliberately did not train a predictive ML model. The available public data is current state snapshots, not labeled historical failure events, so a trained model would project false confidence it has not earned. An auditable rubric paired with an LLM that explains it is the more honest design, and it is what is running here.

## Using the app

The frontend takes a **street address**, not a city or neighborhood name. Type a full address, for example `1 Dr Carlton B Goodlett Pl, San Francisco, CA`, and click **Analyze**. The app geocodes the address, pulls the data, and generates the report.

Click **+ new** in the sidebar to start a fresh report for a different address. Previous reports stay listed in the sidebar so you can revisit them without regenerating.

### San Francisco only, for now

Soil, earthquake, and weather and FEMA data are national and will return results almost anywhere in the United States. Building permits and 311 complaints are currently scoped to San Francisco only, since that is the dataset we built and tested against for this hackathon. Addresses outside San Francisco will still generate a report, but the infrastructure decay and structural hazard scores will be based on partial data, and the report will say so explicitly in its coverage notes rather than presenting a falsely confident score. Adding another city means adding its open data sources to the building and infrastructure data module.

## Setup

```bash
pip install -r requirements.txt
npm install
export ANTHROPIC_API_KEY=sk-ant-your-key-here
```

Run the backend and frontend in two terminals:

```bash
# Terminal 1
python3 api_server.py

# Terminal 2
npm run dev
```

Open the local URL Vite prints, typically `http://127.0.0.1:5173`.

### Demo mode

Set `DATAGENT_MODE=demo` before starting `api_server.py` to serve cached responses for addresses you have already run once, instead of hitting live government APIs and the geocoder every time. Use `DATAGENT_MODE=record` once per address beforehand to populate that cache.

```bash
export DATAGENT_MODE=record
python3 api_server.py
# run your demo addresses once each, then:

export DATAGENT_MODE=demo
python3 api_server.py
# now those exact addresses load fast and without network dependency
```

## Built with

Python, Anthropic Claude API, USGS Earthquake Catalog API, National Weather Service NWS Alerts API, FEMA OpenFEMA Disaster Declarations API, USDA SSURGO Soil Data Access API, DataSF Socrata Open Data API, FCC Census Block API, Census Geocoder API, React, Vite


## Detected evidence (automated analysis)

Indexed codebase: 18 recognized source files, 104 KB.
- Anthropic (technology) — detected in the code
- CSS (language) — detected in the code
- HTML (language) — detected in the code
- JavaScript (language) — detected in the code
- Python (language) — detected in the code
- React (technology) — detected in the code

## Codebase structure (from repository index)

### Files (22 of 22)

```
.gitignore
api_server.py
building_infrastructure_api.py
earthquake_api.py
engine/__init__.py
engine/knowledge_base.json
engine/rag.py
engine/react_engine.py
engine/scoring.py
engine/tools.py
index.html
main.py
package.json
README.md
requirements.txt
run.py
soil_api.py
src/main.jsx
src/styles.css
test_offline.py
test_rag.py
weather_disaster_api.py
```

### Dependencies

- package.json: @vitejs/plugin-react@^5.1.2, react@^19.2.3, react-dom@^19.2.3, vite@^7.3.5
- requirements.txt: anthropic@>=0.39.0, requests@>=2.31.0

### Recent commits (newest first)

- Add project README with demo link and usage instructions
- Fix slope_r string bug in rubric, add renter actions, untrack node_modules
- refresh
- frontend updates
- frontend added
- Add ReAct reasoning engine: tool-use loop, risk rubric, RAG grounding
- data gathering done
- earthquake api
- soil data API
- first commit

## Key source files (fetched from GitHub, selected and truncated for size)

### requirements.txt

```
anthropic>=0.39.0
requests>=2.31.0

```

### package.json

```
{
  "scripts": {
    "dev": "vite --host 127.0.0.1",
    "build": "vite build",
    "preview": "vite preview --host 127.0.0.1"
  },
  "dependencies": {
    "@vitejs/plugin-react": "^5.1.2",
    "vite": "^7.3.5",
    "react": "^19.2.3",
    "react-dom": "^19.2.3"
  },
  "devDependencies": {}
}

```

### main.py

```python
from soil_api import get_soil_data_for_lat_long

from earthquake_api import get_earthquake_data_for_lat_long

from weather_disaster_api import get_weather_disaster_data_for_lat_long 

from building_infrastructure_api import get_building_infrastructure_data_for_lat_long

lat = 37.773972
long = -122.431297

soil_data = get_soil_data_for_lat_long(lat, long)

earthquake_data = get_earthquake_data_for_lat_long(lat, long)

weather_data = get_weather_disaster_data_for_lat_long(lat, long)

building_data = get_building_infrastructure_data_for_lat_long(
    lat,
    long,
    radius_meters=100,
)

data = {'soil_and_terrain_data' : soil_data, 'earthquake_history_data' : earthquake_data, 'weather_history_data' : weather_data, 'building_and_infrastructure_data' : building_data}


print(data)
```

### src/main.jsx

```javascript
import React, { useEffect, useMemo, useState } from "react";
import { createRoot } from "react-dom/client";
import "./styles.css";

const API_URL = "http://127.0.0.1:8000/api/analyze";
const REPORT_STORAGE_KEY = "dwell.locationReports";

const LOADING_STEPS = [
  "Finding the building location ...",
  "Looking for soil and land data ...",
  "Looking for regional earthquake history ...",
  "Checking weather hazards and disaster signals ...",
  "Reviewing building and infrastructure records ...",
  "Running infrastructure risk analysis ...",
  "Preparing the final report ...",
];

const DEFAULT_REPORTS = [
  {
    id: "sample-sf",
    title: "San Francisco, CA",
    subtitle: "1 Dr Carlton B Goodlett Pl",
    createdAt: "Recent",
    result: null,
    draft: { address: "1 Dr Carlton B Goodlett Pl, San Francisco, CA" },
  },
];

function App() {
  const [address, setAddress] = useState(
    "1 Dr Carlton B Goodlett Pl, San Francisco, CA",
  );
  const [loading, setLoading] = useState(false);
  const [stepIndex, setStepIndex] = useState(0);
  const [error, setError] = useState("");
  const [activeReportId, setActiveReportId] = useState(null);
  const [currentResult, setCurrentResult] = useState(null);
  const [reports, setReports] = useState(() => loadReports());

  const report = currentResult?.report;
  const riskProfile = currentResult?.risk_profile;
  const loadingText = useMemo(
    () => LOADING_STEPS[Math.min(stepIndex, LOADING_STEPS.length - 1)],
    [stepIndex],
  );

  useEffect(() => {
    saveReports(reports.filter((item) => item.result));
  }, [reports]);

  function startNewReport() {
    setAddress("");
    setError("");
    setCurrentResult(null);
    setActiveReportId(null);
    setStepIndex(0);
  }

  function openReport(item) {
    setActiveReportId(item.id);
    setError("");
    setCurrentResult(item.result);
    setAddress(item.draft?.address || item.title);
  }

  async function analyzeLocation(event) {
    event.preventDefault();
    setError("");
    setCurrentResult(null);

    const trimmedAddress = address.trim();

    if (!trimmedAddress) {
      setError("Enter a building address.");
      return;
    }

    setLoading(true);
    setStepIndex(0);

    const interval = window.setInterval(() => {
      setStepIndex((current) => Math.min(current + 1, LOADING_STEPS.length - 1));
    }, 1800);

    try {
      const response = await fetch(API_URL, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          address: trimmedAddress,
        }),
      });
      const payload = await response.json();

      if (!response.ok || !payload.ok) {
        throw new Error(payload.error || "Analysis failed.");
      }

      const nextReport = {
        id: String(Date.now()),
        title: payload.data.report?.location?.label || trimmedAddress,
        subtitle: `${payload.data.input?.lat?.toFixed(5)}, ${payload.data.input?.lon?.toFixed(5)}`,
        createdAt: new Date().toLocaleString([], {
          month: "short",
          day: "numeric",
          hour: "numeric",
          minute: "2-digit",
        }),
        result: payload.data,
        draft: { address: trimmedAddress },
      };

      setCurrentResult(payload.data);
      setActiveReportId(nextReport.id);
      setReports((items) => [nextReport, ...items.filter((item) => item.result)]);
      setStepIndex(LOADING_STEPS.length - 1);
    } catch (err) {
      setError(err.message);
    } finally {
      window.clearInterval(interval);
      setLoading(false);
    }
  }

  return (
    <main className="dwell-shell">
      <aside className="sidebar">
        <div className="brand-block">
          <h1>Dwell</h1>
          <button className="new-button" onClick={startNewReport} type="button">
            + new
          </button>
        </div>

        <section className="report-history">
          <div className="history-heading">
            <span>Reports</span>
          </div>
          <div className="history-list">
            {reports.map((item) => (
              <button
                className={
                  item.id === activeReportId ? "history-item active" : "history-item"
                }
                key={item.id}
                onClick={() => openReport(item)}
                type="button"
              >
                <span>{item.title}</span>
                <small>{item.subtitle}</small>
              </button>
            ))}
          </div>
        </section>
      </aside>

      <section className="workspace">
        <div className="workspace-scroll">
          {!report && !loading ? (
            <section className="entry-state">
              <div className="entry-copy">
                <span className="eyebrow">Location report</span>
                <h2>Enter Location</h2>
                <p>
                  Dwell will pull environmental, seismic, weather, and civic
                  infrastructure data, then produce a risk report.
                </p>
              </div>
              <LocationForm
                address={address}
                loading={loading}
                onAddressChange={setAddress}
                onSubmit={analyzeLocation}
              />
              {error ? <div className="error-message">{error}</div> : null}
            </section>
          ) : null}

          {loading ? (
            <section className="analysis-state">
              <div className="analysis-card">
                <div className="spinner" />
                <div>
                  <span className="eyebrow">Dwell is working</span>
                  <h2>{loadingText}</h2>
                </div>
              </div>
              <div className="step-stack">
                {LOADING_STEPS.map((step, index) => (
                  <div
                    className={index <= stepIndex ? "step-row active" : "step-row"}
                    key={step}
                  >
                    <span />
                
[truncated — 4478 more characters]
```

### index.html

```html
<!doctype html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Datagent</title>
  </head>
  <body>
    <div id="root"></div>
    <script type="module" src="/src/main.jsx"></script>
  </body>
</html>

```

### test_rag.py

```python
from engine import scoring, rag, react_engine  # import = post-edit syntax check

fixture = {
  "soil_and_terrain_data": {"components":[{"compname":"Urban land","comppct_r":60,
      "drainagecl":"Poorly drained","hydgrp":"D","hydricrating":"yes","slope_r":18}]},
  "earthquake_history_data": {"summary":{"event_count":14,"max_magnitude":4.7,
      "activity_level":"moderate"}},
  "weather_history_data": {"summary":{"has_active_flood_alert":True,
      "historical_declaration_categories":{"flood":2}}},
  "building_and_infrastructure_data": {"datasets_used":[{"id":"sf_311_cases"}],
      "summary":{"building_permit_count":22,"recent_building_permit_count":0,
      "has_soft_story_retrofit":True,"infrastructure_related_311_count":9,
      "open_service_request_count":6,
      "service_request_categories":{"water_or_sewer":4,"street_defect":2}}},
}
profile = scoring.compute_risk_profile(fixture)
grounding = rag.retrieve_for_profile(profile)
print(f"Retrieved {len(grounding)} grounding docs for elevated hazards:\n")
for g in grounding:
    print(f"  [{g['_relevance']:.3f}] {g['id']:20s} ({g['hazard']}) - {g['title']}")
assert len(grounding) > 0, "RAG returned nothing"
# water/sewer should surface for the infra_decay hazard
ids = {g["id"] for g in grounding}
assert "kb_water_main" in ids or "kb_sewer" in ids, "expected pipe KB doc"
print("\nOK: RAG surfaces relevant pipe/flood/soft-story docs")
print("OK: react_engine still imports cleanly after edits")

```

### run.py

```python
#!/usr/bin/env python3
"""
datagent runner.

Usage:
  # 1. Before the demo, on good wifi, cache all the live data:
  DATAGENT_MODE=record python run.py 37.773972 -122.431297 "San Francisco, CA"

  # 2. On stage, run instantly from cache (no network can break it):
  DATAGENT_MODE=demo python run.py 37.773972 -122.431297 "San Francisco, CA"

  # 3. Normal live run (network with cache fallback):
  python run.py 37.773972 -122.431297 "San Francisco, CA"

Writes the structured report to out/<label>.json for the frontend to load,
and prints it to stdout.
"""
from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path

from engine.react_engine import assess_location


def _slug(text: str) -> str:
    return re.sub(r"[^a-z0-9]+", "_", text.lower()).strip("_") or "location"


def main() -> None:
    lat = float(sys.argv[1]) if len(sys.argv) > 1 else 37.773972
    lon = float(sys.argv[2]) if len(sys.argv) > 2 else -122.431297
    label = sys.argv[3] if len(sys.argv) > 3 else "San Francisco, CA"

    print(f"[mode={os.environ.get('DATAGENT_MODE','live')}] "
          f"Assessing {label} ({lat}, {lon}) ...", file=sys.stderr)

    result = assess_location(lat, lon, label)

    out_dir = Path("out")
    out_dir.mkdir(exist_ok=True)
    out_path = out_dir / f"{_slug(label)}.json"
    out_path.write_text(json.dumps(result, indent=2, default=str))

    print(json.dumps(result["report"], indent=2, default=str))
    print(f"\n[written] {out_path}", file=sys.stderr)


if __name__ == "__main__":
    main()

```

### soil_api.py

```python
from typing import Any

import requests


SSURGO_URL = "https://sdmdataaccess.sc.egov.usda.gov/tabular/post.rest"


class SSURGOError(RuntimeError):
    """Raised when the Soil Data Access service returns an unusable response."""


def _table_to_dicts(payload: dict[str, Any]) -> list[dict[str, Any]]:
    table = payload.get("Table") or []
    if not table:
        return []

    columns = table[0]
    rows = table[1:]
    return [dict(zip(columns, row)) for row in rows]


def run_sql(sql: str) -> list[dict[str, Any]]:
    response = requests.post(
        SSURGO_URL,
        data={"query": sql, "format": "JSON+COLUMNNAME"},
        timeout=30,
    )
    response.raise_for_status()

    try:
        return _table_to_dicts(response.json())
    except ValueError as exc:
        raise SSURGOError(f"SSURGO returned non-JSON data: {response.text}") from exc


def get_mukey_for_lat_long(lat: float, lon: float) -> str | None:
    lat = float(lat)
    lon = float(lon)
    rows = run_sql(
        f"""
        SELECT mukey
        FROM SDA_Get_Mukey_from_intersection_with_WktWgs84('point({lon} {lat})')
        """
    )

    if not rows:
        return None

    return rows[0]["mukey"]


def get_soil_data_for_lat_long(lat: float, lon: float) -> dict[str, Any]:
    """Pull SSURGO map unit, component, and horizon data for a WGS84 point."""
    mukey = get_mukey_for_lat_long(lat, lon)
    if mukey is None:
        return {
            "lat": float(lat),
            "lon": float(lon),
            "mukey": None,
            "mapunit": None,
            "components": [],
            "horizons": [],
        }

    mapunit = run_sql(
        f"""
        SELECT
            mukey,
            musym,
            muname,
            mukind,
            mustatus,
            muacres
        FROM mapunit
        WHERE mukey = {mukey}
        """
    )

    components = run_sql(
        f"""
        SELECT
            cokey,
            compname,
            comppct_r,
            majcompflag,
            hydgrp,
            drainagecl,
            hydricrating,
            taxclname,
            localphase,
            slope_r,
            elev_r
        FROM component
        WHERE mukey = {mukey}
        ORDER BY comppct_r DESC, compname
        """
    )

    horizons = run_sql(
        f"""
        SELECT
            c.cokey,
            c.compname,
            c.comppct_r,
            ch.chkey,
            ch.hzname,
            ch.hzdept_r,
            ch.hzdepb_r,
            ch.awc_r,
            ch.ksat_r,
            ch.om_r,
            ch.claytotal_r,
            ch.sandtotal_r,
            ch.silttotal_r,
            ch.ph1to1h2o_r
        FROM component AS c
        INNER JOIN chorizon AS ch
            ON ch.cokey = c.cokey
        WHERE c.mukey = {mukey}
        ORDER BY c.comppct_r DESC, c.compname, ch.hzdept_r
        """
    )

    return {
        "lat": float(lat),
        "lon": float(lon),
        "mukey": mukey,
        "mapunit": mapunit[0] if mapunit else None,
        "components": components,
        "horizons": horizons,
    }

```

### api_server.py

```python
from __future__ import annotations

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse

import requests

from engine.react_engine import assess_location


HOST = "127.0.0.1"
PORT = 8000
CENSUS_GEOCODER_URL = "https://geocoding.geo.census.gov/geocoder/locations/onelineaddress"


class DatagentAPIHandler(BaseHTTPRequestHandler):
    def do_OPTIONS(self) -> None:
        self._send_empty(204)

    def do_POST(self) -> None:
        if urlparse(self.path).path != "/api/analyze":
            self._send_json({"ok": False, "error": "Unknown endpoint"}, 404)
            return

        try:
            payload = self._read_json_body()
            address = (payload.get("address") or "").strip()
            if address:
                geocoded = geocode_address(address)
                lat = geocoded["lat"]
                lon = geocoded["lon"]
                label = geocoded["matched_address"]
            else:
                lat = float(payload["lat"])
                lon = float(payload["lon"])
                label = payload.get("label") or f"{lat},{lon}"

            result = assess_location(lat, lon, label)
            result["input"] = {
                "address": address or None,
                "lat": lat,
                "lon": lon,
                "label": label,
            }
            self._send_json({"ok": True, "data": result})
        except KeyError as exc:
            self._send_json({"ok": False, "error": f"Missing field: {exc}"}, 400)
        except ValueError as exc:
            self._send_json({"ok": False, "error": str(exc)}, 400)
        except Exception as exc:
            self._send_json({"ok": False, "error": str(exc)}, 500)

    def _read_json_body(self) -> dict:
        length = int(self.headers.get("Content-Length", "0"))
        raw_body = self.rfile.read(length).decode("utf-8")
        return json.loads(raw_body or "{}")

    def _send_empty(self, status: int) -> None:
        self.send_response(status)
        self._send_cors_headers()
        self.end_headers()

    def _send_json(self, payload: dict, status: int = 200) -> None:
        body = json.dumps(payload, default=str).encode("utf-8")
        self.send_response(status)
        self._send_cors_headers()
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _send_cors_headers(self) -> None:
        self.send_header("Access-Control-Allow-Origin", "http://127.0.0.1:5173")
        self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")


def run() -> None:
    server = ThreadingHTTPServer((HOST, PORT), DatagentAPIHandler)
    print(f"Datagent API running at http://{HOST}:{PORT}")
    server.serve_forever()


def geocode_address(address: str) -> dict:
    response = requests.get(
        CENSUS_GEOCODER_URL,
        params={
            "address": address,
            "benchmark": "Public_AR_Current",
            "format": "json",
        },
        timeout=30,
    )
    response.raise_for_status()
    payload = response.json()
    matches = (payload.get("result") or {}).get("addressMatches") or []

    if not matches:
        raise ValueError(f"Could not geocode address: {address}")

    match = matches[0]
    coordinates = match.get("coordinates") or {}
    return {
        "lat": float(coordinates["y"]),
        "lon": float(coordinates["x"]),
        "matched_address": match.get("matchedAddress") or address,
    }


if __name__ == "__main__":
    run()

```

### earthquake_api.py

```python
from datetime import UTC, datetime, timedelta
from typing import Any

import requests


USGS_EARTHQUAKE_URL = "https://earthquake.usgs.gov/fdsnws/event/1/query"


class USGSEarthquakeError(RuntimeError):
    """Raised when the USGS earthquake service returns an unusable response."""


def _iso_utc(dt: datetime) -> str:
    return dt.astimezone(UTC).replace(microsecond=0).isoformat()


def _parse_event(feature: dict[str, Any]) -> dict[str, Any]:
    properties = feature.get("properties") or {}
    geometry = feature.get("geometry") or {}
    coordinates = geometry.get("coordinates") or [None, None, None]

    return {
        "id": feature.get("id"),
        "place": properties.get("place"),
        "magnitude": properties.get("mag"),
        "magnitude_type": properties.get("magType"),
        "time": datetime.fromtimestamp(properties["time"] / 1000, UTC).isoformat()
        if properties.get("time") is not None
        else None,
        "updated": datetime.fromtimestamp(properties["updated"] / 1000, UTC).isoformat()
        if properties.get("updated") is not None
        else None,
        "url": properties.get("url"),
        "detail_url": properties.get("detail"),
        "felt_reports": properties.get("felt"),
        "maximum_reported_intensity": properties.get("cdi"),
        "maximum_estimated_intensity": properties.get("mmi"),
        "alert": properties.get("alert"),
        "status": properties.get("status"),
        "significance": properties.get("sig"),
        "event_type": properties.get("type"),
        "longitude": coordinates[0],
        "latitude": coordinates[1],
        "depth_km": coordinates[2],
    }


def _activity_level(max_magnitude: float | None, event_count: int) -> str:
    if max_magnitude is None or event_count == 0:
        return "none_found"
    if max_magnitude >= 6 or event_count >= 25:
        return "high"
    if max_magnitude >= 4.5 or event_count >= 10:
        return "moderate"
    return "low"


def _summarize_events(events: list[dict[str, Any]]) -> dict[str, Any]:
    magnitudes = [
        event["magnitude"]
        for event in events
        if isinstance(event.get("magnitude"), int | float)
    ]
    max_magnitude = max(magnitudes) if magnitudes else None

    return {
        "event_count": len(events),
        "max_magnitude": max_magnitude,
        "activity_level": _activity_level(max_magnitude, len(events)),
    }


def get_earthquake_data_for_lat_long(
    lat: float,
    lon: float,
    radius_km: float = 100,
    days_back: int = 365,
    min_magnitude: float = 2.5,
    limit: int = 100,
) -> dict[str, Any]:
    """Pull nearby earthquake event data from the USGS catalog for a WGS84 point."""
    lat = float(lat)
    lon = float(lon)
    end_time = datetime.now(UTC)
    start_time = end_time - timedelta(days=days_back)

    response = requests.get(
        USGS_EARTHQUAKE_URL,
        params={
            "format": "geojson",
            "latitude": lat,
            "longitude": lon,
            "maxradiuskm": radius_km,
            "starttime": _iso_utc(start_time),
            "endtime": _iso_utc(end_time),
            "minmagnitude": min_magnitude,
            "orderby": "time",
            "eventtype": "earthquake",
            "limit": limit,
        },
        timeout=30,
    )
    response.raise_for_status()

    try:
        payload = response.json()
    except ValueError as exc:
        raise USGSEarthquakeError(
            f"USGS returned non-JSON data: {response.text}"
        ) from exc

    events = [_parse_event(feature) for feature in payload.get("features", [])]

    return {
        "lat": lat,
        "lon": lon,
        "source": "USGS Earthquake Catalog",
        "search": {
            "radius_km": radius_km,
            "days_back": days_back,
            "start_time": _iso_utc(start_time),
            "end_time": _iso_utc(end_time),
            "min_magnitude": min_magnitude,
            "limit": limit,
        },
        "summary": _summarize_events(events),
        "events": events,
    }

```

[8 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]