Ambulance data

backend/api/routes/get_routes/get_legend.py

@@ -4,6 +4,7 @@
 
 from query import (
     get_soil_suit_legend,
+    get_ambulance_legend,
 )
 
 logger = logging.getLogger(__name__)
@@ -15,3 +16,9 @@
 async def wastewater_soil_suit_legend():
     data = get_soil_suit_legend()
     return Response(content=data, media_type="application/json")
+
+
+@router.get("/load/mapping/ambulance/ambulance_legend")
+async def ambulance_legend():
+    data = get_ambulance_legend()
+    return Response(content=data, media_type="application/json")

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,30 @@
+from fastapi import APIRouter, Response
+
+from api.core_functions import request_to_source, spec_to_source
+from api.models import FilterRequest, FilterSpec
+from query import (
+    get_ambulance_geojson,
+    get_ambulance_legend,
+)
+
+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")
+
+
+@router.post("/load/mapping/ambulance/service_area_new")
+async def ambulance_info_geojson_new(specs: list[FilterSpec]):
+    sources = [spec_to_source(spec, "default") for spec in specs]
+    data = get_ambulance_geojson(sources)
+    return Response(content=data, media_type="application/json")
+
+
+@router.get("/load/mapping/ambulance/ambulance_legend")
+async def ambulance_legend():
+    data = get_ambulance_legend()
+    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
+
+_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("""--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/cdc.py

@@ -69,7 +69,7 @@ def build_places(name: str, path: Path, indicators: str) -> None:
     df = CON.execute(sql).df()
     # Needed to get rid of df variable assignment to pass ruff linting check
     if name == "county":
-        build_PCA_table(df)
+        pca_df = build_PCA_table(df)
     pct_df = add_national_percentile(df)  # df here is still national
     df = df[df["StateAbbr"] == "VT"].copy()
     df, edge_df = bin_measures(df, variable_col="Measure", value_col="Data_Value")
@@ -81,18 +81,18 @@ def build_places(name: str, path: Path, indicators: str) -> None:
     CON.execute(f"""--sql
             CREATE OR REPLACE TABLE {name}_places
             AS SELECT * 
-            FROM df            
+            FROM {df}            
         """)
     CON.execute(f"""--sql
         CREATE OR REPLACE TABLE {name}_edges
         AS SELECT * 
-        FROM edge_df            
+        FROM {edge_df}            
     """)
     if name == "county":
         CON.execute(f"""--sql
                 CREATE OR REPLACE TABLE {name}PcaData
                 AS SELECT * 
-                FROM pca_df            
+                FROM {pca_df}            
             """)
 
 

backend/build/consolidate.py

@@ -16,7 +16,6 @@
     "Data/_Processed/all_data-copy.duckdb",
 ]:
     path = Path(__file__).parent.parent / db_path
-    path.touch(exist_ok=True)
     db = duckdb.connect(str(db_path))
     db.execute("INSTALL SPATIAL")
     db.execute("LOAD SPATIAL")

backend/build/main.py

@@ -7,7 +7,7 @@
     Runs all the build scripts one after the other.
 """
 
-from build import FIPS_data, acs5, cdc, wastewater, zoning
+from build import FIPS_data, acs5, ambulance, cdc, wastewater, zoning
 
 
 def main():
@@ -18,6 +18,7 @@ def main():
     FIPS_data.main()
     zoning.main()
     wastewater.main()
+    ambulance.main()
 
 
 if __name__ == "__main__":

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,42 @@
+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()