Ambulance data

.gitignore

@@ -11,6 +11,7 @@ venv/
 *_trash*
 *zoning.json*
 *tidy.csv*
+*.ipynb
 
 # Large geodata files — not for version control
 backend/Data/Parcels/

backend/api/routes/post_routes/__init__.py

@@ -5,6 +5,7 @@
 from .post_qcew import router as post_qcew_router
 from .post_zoning import router as post_zoning_router
 from .post_wastewater import router as post_wastewater_router
+from .post_ambulance import router as post_ambulance_router
 
 all_post_routers = [
     post_zoning_router,
@@ -14,4 +15,5 @@
     post_wastewater_router,
     post_export_router,
     post_cdc_router,
+    post_ambulance_router,
 ]

backend/api/routes/post_routes/post_ambulance.py

@@ -0,0 +1,16 @@
+from fastapi import APIRouter, Response
+
+from api.core_functions import request_to_source
+from api.models import FilterRequest
+from query import (
+    get_ambulance_geojson,
+)
+
+router = APIRouter()
+
+
+@router.post("/load/mapping/ambulance/service_area")
+async def ambulance_info_geojson(request: FilterRequest):
+    source = request_to_source(request, "ambulance_ambulance_info", "default")
+    data = get_ambulance_geojson([source])
+    return Response(content=data, media_type="application/json")

backend/api/schema.json

@@ -85,6 +85,14 @@
           "Jurisdiction": "Municipal_Name",
           "Town": "TownName"
       }
+    },
+    "ambulance_ambulance_info": {
+      "join_key":"OBJECTID",
+      "join_type": "inner",
+      "columns": {
+        "Zip Code": "Zip_Code",
+        "City or Town": "City"
+      }
     }
   }
 }

backend/build/ambulance.py

@@ -0,0 +1,117 @@
+"""
+**Author**:
+    Atticus Tarleton
+**Created**:
+    2026-07-20
+**Description**:
+    Build script to convert the ambulance service area files into SQL tables.
+"""
+
+import os
+from pathlib import Path
+
+import duckdb
+import pandas as pd
+
+_project_root = Path.cwd()
+while not (_project_root / "api").exists():
+    _project_root = _project_root.parent
+os.chdir(_project_root)
+print(os.getcwd())
+
+# globals
+con = duckdb.connect()
+proc_dir = Path("Data/_Processed/ambulance")
+data_dir = Path("Data/ambulance/ambulance_service_areas.parquet")
+
+# hardcoded specifics:
+ambulance_info_cols = [
+    "OBJECTID",
+    "Serv_Name",
+    "Cert_Level",
+    "Address",
+    "Street_1",
+    "Street_2",
+    "City",
+    "State",
+    "Zip_Code",
+    "Total_Tran",
+    "Per_No_Tran",
+    "Re_Per_Tran",
+    "Cost_Per",
+    "Cost_Call"
+]
+
+ambulance_geom_cols = [
+    "OBJECTID",
+    "Shape__Area",
+    "Shape__Length",
+    "geometry"
+]
+
+# functions:
+def load_ambulance_data():
+    con.execute("LOAD spatial")
+
+    con.execute(f"""--sql
+      CREATE OR REPLACE VIEW ambulance_service_areas AS
+      SELECT * FROM read_parquet('{data_dir}')
+    """)
+
+
+def build_ambulance_info_table():
+    info_string = ", ".join(ambulance_info_cols)
+
+    con.execute(f"""--sql
+      CREATE OR REPLACE VIEW ambulance_info AS
+      SELECT {info_string}
+      FROM ambulance_service_areas
+    """)
+
+def build_ambulance_geom_table():
+    geom_string = ", ".join(ambulance_geom_cols)
+
+    con.execute(f"""--sql
+      CREATE OR REPLACE VIEW ambulance_geom AS
+      SELECT {geom_string}
+      FROM ambulance_service_areas
+    """)
+
+def build_ambulance_color_table():
+    con.execute(f"""--sql
+    CREATE TABLE ambulance_colors (
+        certification_level   TEXT PRIMARY KEY,
+        hex_color       TEXT NOT NULL,
+        rgba            TEXT NOT NULL  -- '[255,127,14,180]' as JSON-ish text
+    );
+
+    INSERT INTO ambulance_colors VALUES
+        ('Paramedic',         '#2ca02c', '[44, 160, 44, 180]'),
+        ('Advanced EMT',   '#ffcc00', '[255, 204, 0, 180]'),
+        ('Paramedic - Critical Care Endorsement',   '#fd7e14', '[253, 126, 20, 180]')
+    """)
+
+def save_ambulance_tables():
+    load_ambulance_data()
+    build_ambulance_info_table()
+    build_ambulance_geom_table()
+    build_ambulance_color_table()
+
+    for table in [
+        "ambulance_info",
+        "ambulance_geom",
+        "ambulance_colors"
+    ]:
+        con.execute(
+            f"COPY (SELECT * FROM {table}) TO '{proc_dir / f'{table}.parquet'}' "
+        )
+
+## Putting everything together
+def main():
+    proc_dir.mkdir(parents=True, exist_ok=True)
+    con.execute("LOAD spatial")
+    save_ambulance_tables()
+
+
+if __name__ == "__main__":
+    main()

backend/build/consolidate.py

@@ -16,7 +16,7 @@
 db.execute("INSTALL SPATIAL")
 db.execute("LOAD SPATIAL")
 
-for path in Path("Data/_Processed").rglob("*.parquet"):
+for path in Path("../Data/_Processed").rglob("*.parquet"):
     name = f"{path.parent.name}_{path.stem}"
     db.execute(
         f"CREATE OR REPLACE TABLE {name} AS SELECT * FROM read_parquet('{path}')"

backend/build/wastewater.py

@@ -8,7 +8,6 @@
 """
 
 # TODO: move the large SQL code to its own files rather than inside here
-# TODO: also, need to go create the consolidated duckDB database
 
 import os
 from pathlib import Path

backend/data_collection/ambulance.py

@@ -0,0 +1,39 @@
+import pandas as pd
+import requests
+from io import BytesIO
+from pyogrio import read_dataframe
+
+AMBULANCE_SERVICE_AREA = "https://services1.arcgis.com/BkFxaEFNwHqX3tAw/arcgis/rest/services/FS_VCGI_OPENDATA_Emergency_AmbulanceServiceAreas_SP_v1/FeatureServer/0/query?outFields=*&where=1%3D1&f=geojson"
+STORAGE_LOCATION = "../Data/ambulance"
+
+
+# ---------------------------------------------------------------------------
+# Ambulance API fetch
+# ---------------------------------------------------------------------------
+
+# Fetch ambulance data from goverment arcgis website (geojson files)
+
+def fetch_service_areas() -> pd.DataFrame | None:
+    r = requests.get(AMBULANCE_SERVICE_AREA, timeout=30)
+    r.raise_for_status()
+    df = read_dataframe(BytesIO(r.content))
+    return df
+
+
+# ---------------------------------------------------------------------------
+# Main scrape runner
+# ---------------------------------------------------------------------------
+
+
+def run_ambulance_scrape() -> None:
+    """
+    Fetch Wastewater data and save as parquet files.
+    """
+    service_areas = fetch_service_areas()
+
+    if service_areas is not None:
+        service_areas.to_parquet(f"{STORAGE_LOCATION}/ambulance_service_areas.parquet", index=False)
+
+
+if __name__ == "__main__":
+    run_ambulance_scrape()