diff --git a/.gitignore b/.gitignore index 556b5d29..7ca88a27 100644 --- a/.gitignore +++ b/.gitignore @@ -22,9 +22,6 @@ backend/Data/Parcels/ backend/Data/lake backend/Data/lake.files/ -# DuckDB Database with CLEAN tables -backend/Data/warehouse.duckdb - # Local planning/notes files claude_todo.md claude-work-done.md diff --git a/Data/lake b/Data/lake deleted file mode 100644 index a0a6c176..00000000 Binary files a/Data/lake and /dev/null differ diff --git a/README.md b/README.md index 8e304082..73dd315e 100644 --- a/README.md +++ b/README.md @@ -8,13 +8,13 @@ A **React-based Website** for exploring, visualizing, and interpreting Vermont d Install these before you start. Every one of them is used by the standard workflow. -| Tool | Why it's needed | -| --------------------------------------------- | ---------------------------------------------------------------------------------------------------- | -| [git-lfs](https://git-lfs.com/) | The datasets in `Data/` are tracked with Git LFS. | -| [just](https://just.systems/man/en/) | Task runner. Every dev command in this project is a `just` recipe (see [justfile](justfile)). | -| [uv](https://docs.astral.sh/uv/) | Python dependency management and script running for the backend. | -| [Node + npm](https://nodejs.org/) | Frontend dependencies and the Next.js dev server. | -| [podman](https://podman.io/docs/installation) | Builds and runs the containerized stack. | +| Tool | Why it's needed | +| --------------------------------------------- | --------------------------------------------------------------------------------------------- | +| [git-lfs](https://git-lfs.com/) | The datasets in `Data/` are tracked with Git LFS. | +| [just](https://just.systems/man/en/) | Task runner. Every dev command in this project is a `just` recipe (see [justfile](justfile)). | +| [uv](https://docs.astral.sh/uv/) | Python dependency management and script running for the backend. | +| [Node + npm](https://nodejs.org/) | Frontend dependencies and the Next.js dev server. | +| [podman](https://podman.io/docs/installation) | Builds and runs the containerized stack. | > **Note:** podman is required even for the non-containerized workflow, because the `local-*` recipes call `just down` first to make sure a running container isn't already holding the ports. @@ -171,11 +171,8 @@ This project is open-source under the **MIT License**. # VM Deployment - - 1. sudo su - appuser0 - ## Credits - Developed by Ian Sargent and Fitzwilliam Keenan-Koch diff --git a/backend/.sqlfluff b/backend/.sqlfluff index 57f84271..22b13c14 100644 --- a/backend/.sqlfluff +++ b/backend/.sqlfluff @@ -25,9 +25,9 @@ exclude_rules = CP02, CP03, RF04, RF05, ST06, ST07 case_sensitive = True [sqlfluff:templater:jinja:context] -table = acs5_b10_census +table = acs5_demographics_tidy where_string = WHERE "Measure" IN ($1) -cte_filter_block = WITH f0 AS (SELECT DISTINCT OBJECT_ID FROM zoning_info WHERE "County" IN ($1)) +cte_filter_block = WITH f0 AS (SELECT DISTINCT OBJECT_ID FROM VersoZoning_info WHERE "County" IN ($1)) join_filter_block = JOIN f0 USING (OBJECT_ID) info_string = OBJECT_ID, County rule_string = CAST("Residential_Min_Lot" AS VARCHAR) AS residential_min_lot diff --git a/backend/api/models/request_models.py b/backend/api/models/request_models.py index 0d467287..add76358 100644 --- a/backend/api/models/request_models.py +++ b/backend/api/models/request_models.py @@ -1,7 +1,10 @@ +from datetime import datetime from typing import Literal from pydantic import BaseModel, model_validator +MAX_YEAR = datetime.now().year - 2 + class RangeFilter(BaseModel): min: float | None = None @@ -30,7 +33,7 @@ class DPSeriesRequest(BaseModel): variable: str measure: str year_min: int = 2009 - year_max: int = 2024 + year_max: int = MAX_YEAR join_types = Literal["inner", "left", "spatial_intersect"] diff --git a/backend/api/routes/get_routes/get_wholedata.py b/backend/api/routes/get_routes/get_wholedata.py index 18e5cbf6..82ba4dbb 100644 --- a/backend/api/routes/get_routes/get_wholedata.py +++ b/backend/api/routes/get_routes/get_wholedata.py @@ -5,6 +5,9 @@ from app_utils import data_loading from app_utils.flooding import add_flood_color +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) @@ -16,14 +19,33 @@ def read_root(): return {"Default Message": "No endpoint specified"} -# Flood Endpoint (Hardcoded for now) +# Flood Endpoint @router.get("/load/mapping/flood_legal") async def read_flood_data(): - data = data_loading.masterload(name="flood_legal") - # Re-apply zone-based colors at serve time so the static JSON - # does not need to be regenerated when the color scheme changes. - data = add_flood_color(data) - return json.loads(data.to_json()) + result = DB.execute("""--sql + SELECT + *, + ST_AsGeoJSON(geometry)::JSON AS geometry_json + FROM FEMA_floodHazard_geom + """).df() + + result = add_flood_color(result) + + features = [] + for _, row in result.iterrows(): + properties = row.drop(["geometry", "geometry_json"]).to_dict() + features.append( + { + "type": "Feature", + "geometry": json.loads(row["geometry_json"]), + "properties": properties, + } + ) + + return { + "type": "FeatureCollection", + "features": features, + } # Soil Septic Endpoint (Hardcoded for now) diff --git a/backend/api/routes/post_routes/post_acs5_db.py b/backend/api/routes/post_routes/post_acs5_db.py index 6ef9ef58..ed6b15c1 100644 --- a/backend/api/routes/post_routes/post_acs5_db.py +++ b/backend/api/routes/post_routes/post_acs5_db.py @@ -4,15 +4,15 @@ from api.metadata_registry import get_metadata from api.models import DPSeriesRequest, FilterRequest, make_response -from query.processed_db import DB # TODO: Simplify / Refactor this script using the new query folder functions from query.acs5 import ( get_acs5_tidy, - get_median_earnings, - get_snapshot, - get_unemployment_rate_ts, + get_acs5_timeseries, ) +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) router = APIRouter() @@ -21,6 +21,10 @@ # TODO: Percents might need to be weighted averages instead of simple averages for statewide aggregation # TODO: In DB, add an aggregated statewide VT row to each table for easier aggregation requests +# ----------------------------- +# CENSUS TIDY FORMAT TABLES +# ----------------------------- + # Demographics @router.post("/load/acs5-db/tidy/demographics") @@ -36,52 +40,142 @@ async def tidy_education(request: FilterRequest): return make_response(data=rows, metadata=get_metadata("education")) +# Housing @router.post("/load/acs5-db/tidy/housing") async def tidy_housing(request: FilterRequest): rows = get_acs5_tidy(dataset="housing", filters=request.filters) return make_response(data=rows, metadata=get_metadata("housing")) -# Labor Force +# Economics +@router.post("/load/acs5-db/tidy/economics") +async def tidy_economics(request: FilterRequest): + rows = get_acs5_tidy(dataset="economics", filters=request.filters) + return make_response(data=rows, metadata=get_metadata("labor_force")) + + +# Labor Force (FIXME: broken) @router.post("/load/acs5-db/tidy/labor-force") async def tidy_labor_force(request: FilterRequest): rows = get_acs5_tidy(dataset="labor_force", filters=request.filters) return make_response(data=rows, metadata=get_metadata("labor_force")) -# Income +# Income (FIXME: broken) @router.post("/load/acs5-db/tidy/income") async def tidy_income(request: FilterRequest): rows = get_acs5_tidy(dataset="income", filters=request.filters) return make_response(data=rows, metadata=get_metadata("income")) +# ----------------------------- +# CENSUS TIMESERIES TABLES +# ----------------------------- + + +##### DEMOGRAPHICS ##### +# Age Dependency Ratio +@router.post("/load/acs5-db/timeseries/demographics/age-dependency-ratio") +async def get_age_dependency_ratio(request: FilterRequest): + rows = get_acs5_timeseries( + category="demographics", dataset="age_dependency_ratio", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("demographics")) + + # Median Age -@router.post("/load/acs5-db/tidy/demographics/median-age") -async def tidy_median_age(request: FilterRequest): - rows = get_acs5_tidy(dataset="demographics", filters=request.filters) +@router.post("/load/acs5-db/timeseries/demographics/median-age") +async def get_median_age(request: FilterRequest): + rows = get_acs5_timeseries( + category="demographics", dataset="median_age", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("demographics")) + + +# Historic Population +@router.post("/load/acs5-db/timeseries/demographics/historic-population") +async def get_historic_population(request: FilterRequest): + filters = {key: value for key, value in request.filters.items() if key != "year"} + + rows = get_acs5_timeseries( + category="demographics", + dataset="historic_population", + filters=filters, + ) + return make_response(data=rows, metadata=get_metadata("demographics")) -# Unemployment Rate -@router.post("/load/acs5-db/tidy/unemployment-rate") -async def tidy_unemployment_rate(request: FilterRequest): - rows = get_unemployment_rate_ts(filters=request.filters) - return make_response(data=rows, metadata=get_metadata("unemployment_rate")) +##### ECONOMICS ##### +# Heath Insurance Coverage +@router.post("/load/acs5-db/timeseries/economics/health-insurance") +async def get_health_insurance(request: FilterRequest): + rows = get_acs5_timeseries( + category="economics", dataset="health_insurance", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("labor_force")) + + +# Median Household Income +@router.post("/load/acs5-db/timeseries/economics/median-hh-income") +async def get_household_income(request: FilterRequest): + rows = get_acs5_timeseries( + category="economics", dataset="household_income", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("income")) -# Median Earnings -@router.post("/load/acs5-db/tidy/median-earnings") -async def tidy_median_earnings(request: FilterRequest): - rows = get_median_earnings(filters=request.filters) - return make_response(data=rows, metadata=get_metadata("median_earnings")) +# Per Capita Income +@router.post("/load/acs5-db/timeseries/economics/per-capita-income") +async def get_per_capita_income(request: FilterRequest): + rows = get_acs5_timeseries( + category="economics", dataset="per_capita_income", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("income")) + + +# Median Earnings (FIXME: broken) +@router.post("/load/acs5-db/timeseries/economics/median-earnings") +async def get_median_earnings(request: FilterRequest): + rows = get_acs5_timeseries( + category="economics", dataset="median_earnings", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("income")) + + +##### HOUSING ##### +# Total Housing Units +@router.post("/load/acs5-db/timeseries/housing/total-units") +async def get_housing_units(request: FilterRequest): + rows = get_acs5_timeseries( + category="housing", dataset="housing_units", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("housing")) + + +# Median Home Value +@router.post("/load/acs5-db/timeseries/housing/median-home-value") +async def get_median_home_value(request: FilterRequest): + rows = get_acs5_timeseries( + category="housing", dataset="median_home_value", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("housing")) + + +# Vacancy Rates +@router.post("/load/acs5-db/timeseries/housing/vacancy-rates") +async def get_vacancy_rates(request: FilterRequest): + rows = get_acs5_timeseries( + category="housing", dataset="vacancy_rates", filters=request.filters + ) + return make_response(data=rows, metadata=get_metadata("housing")) # Geography Snapshot Variables @router.post("/load/acs5-db/tidy/snapshot") async def tidy_snapshot(request: FilterRequest): - rows = get_snapshot(filters=request.filters) - return make_response(data=rows, metadata=get_metadata("snapshot")) + rows = get_acs5_tidy(dataset="snapshot", filters=request.filters) + return make_response(data=rows, metadata=get_metadata("demographics")) # --------------------------------------------------------------------------- @@ -112,7 +206,7 @@ async def dp_combined_tree(): rows = DB.execute( """--sql SELECT DISTINCT "table", Category, Subcategory, Variable, Measure - FROM acs5_dp_combined + FROM acs5_dp_combined_tidy ORDER BY "table", Category, Subcategory, Variable, Measure """ ).df() @@ -128,7 +222,7 @@ async def dp_combined_series(request: DPSeriesRequest): """--sql SELECT CAST(year AS INTEGER) AS year, CAST(Value AS DOUBLE) AS Value - FROM acs5_dp_combined + FROM acs5_dp_combined_tidy WHERE NAME = ? AND "table" = ? AND Category = ? diff --git a/backend/api/routes/post_routes/post_ambulance.py b/backend/api/routes/post_routes/post_ambulance.py index 5ab349d9..8cb1f08d 100644 --- a/backend/api/routes/post_routes/post_ambulance.py +++ b/backend/api/routes/post_routes/post_ambulance.py @@ -12,7 +12,7 @@ @router.post("/load/mapping/ambulance/service_area") async def ambulance_info_geojson(request: FilterRequest): - source = request_to_source(request, "ambulance_ambulance_info", "default") + source = request_to_source(request, "VCGI_ambulanceService_info", "default") data = get_ambulance_geojson([source]) return Response(content=data, media_type="application/json") diff --git a/backend/api/routes/post_routes/post_cdc.py b/backend/api/routes/post_routes/post_cdc.py index 47b34b81..c019bae3 100644 --- a/backend/api/routes/post_routes/post_cdc.py +++ b/backend/api/routes/post_routes/post_cdc.py @@ -9,7 +9,7 @@ @router.post("/load/mapping/cdc/places/single") async def cdc_single_geojson(request: FilterRequest): - source = request_to_source(request, "cdc_county_places", "default") + source = request_to_source(request, "cdc_places_county", "default") data = single_var_geojson([source]) return data @@ -38,6 +38,7 @@ async def cdc_comparison_tract(specs: list[FilterSpec]) -> APIResponse: return make_response(data=geojson, metadata={"legend": legend}) +# (FIXME) @router.post("/load/mapping/cdc/places/pca_summary") async def cdc_pca(specs: list[FilterSpec]) -> APIResponse: return make_response(data=get_cdc_county_pca(), metadata={}) diff --git a/backend/api/routes/post_routes/post_census.py b/backend/api/routes/post_routes/post_census.py index 8df4109b..04e45fab 100644 --- a/backend/api/routes/post_routes/post_census.py +++ b/backend/api/routes/post_routes/post_census.py @@ -3,12 +3,13 @@ from fastapi import APIRouter, HTTPException from api.models import FilterRequest, make_response -from app_utils import data_loading, timeseries_db +from app_utils import timeseries_db from app_utils.df_filtering import ( filter_from_request, - mass_filter_from_requests, ) -from app_utils.housing import housing_df_metric_dict +from query.production_db import get_db + +DB = get_db() router = APIRouter() @@ -16,81 +17,30 @@ DATADIR = Path(__file__).parent.parent.parent.parent / "Data" CENSUS_DATADIR = DATADIR / "Census" + +CENSUS_DATASETS = ["demographics", "economics", "housing", "social"] + # Maps (category, subcategory) to the timeseries_db view name. # These subcategories are served via DuckDB instead of pandas/CSV. _TIMESERIES_VIEWS: dict[tuple[str, str], str] = { - ("housing", "median_home_value"): "median_home_value", - ("housing", "median_smoc"): "median_smoc", - ("economic", "median_earnings"): "median_earnings", - ("economic", "unemployment_rate"): "unemployment_rate", - ("economic", "commute_habits"): "commute_habits", - ("economic", "commute_time"): "commute_time", - ("demographic", "historic_population"): "historic_population", + ("demographics", "historic_population"): "VCGI_historicPopulation_timeseries", + ( + "demographics", + "age_dependency_ratio", + ): "acs5Demographics_ageDependencyRatio_timeseries", + ("demographics", "median_age"): "acs5Demographics_medianAge_timeseries", + ("economics", "health_insurance"): "acs5Economics_healthInsurance_timeseries", + ("economics", "household_income"): "acs5Economics_medianHouseholdIncome_timeseries", + ("economics", "per_capita_income"): "acs5Economics_perCapitaIncome_timeseries", + ("economics", "unemployment_rate"): "acs5Economics_unemploymentRate_timeseries", + ("housing", "housing_units"): "acs5Housing_housingUnits_timeseries", + ("housing", "median_home_value"): "acs5Housing_medianHomeValue_timeseries", + ("housing", "vacancy_rates"): "acs5Demographics_vacancyRates_timeseries", } -CENSUS_DATASETS = { - "housing": { - "main": CENSUS_DATADIR / "VT_HOUSING_ALL.fgb", - # time-series subcategories handled via _TIMESERIES_VIEWS / DuckDB - }, - "economic": { - "main": CENSUS_DATADIR / "VT_ECONOMIC_ALL.fgb", - # time-series subcategories handled via _TIMESERIES_VIEWS / DuckDB - }, - "demographic": { - "main": CENSUS_DATADIR / "VT_DEMOGRAPHIC_ALL.fgb", - # time-series subcategories handled via _TIMESERIES_VIEWS / DuckDB - }, - "social": {"main": CENSUS_DATADIR / "VT_SOCIAL_ALL.fgb"}, -} - - -# Load the Census "Main" Dataset by Cateogory (housing, economic, demographic, social) -@router.post("/load/census/{category}") -async def read_census_data(category: str, request: FilterRequest): - if category not in CENSUS_DATASETS: - raise HTTPException( - status_code=404, detail=f"Census category '{category}' was not found" - ) - - data = data_loading.load_census_data(CENSUS_DATASETS[category]["main"]) - data = filter_from_request(data, request) - metadata = {} - - return make_response(data, metadata) - - -@router.post("/load/census/housing/snapshot") -async def get_housing_snapshot(request: FilterRequest): - dfs = data_loading.masterload("census_housing") - dfs = mass_filter_from_requests(dfs, request) - metrics, plot_dfs = housing_df_metric_dict(dfs) - - # Convert metrics to JSON-serializable - metrics_json = {k: float(v) if v is not None else None for k, v in metrics.items()} - - # Convert plot dataframes - plot_data = {k: v.to_dict(orient="records") for k, v in plot_dfs.items()} - - response = {"metrics": metrics_json, "plot_data": plot_data} - # Filter response if specific includes requested - if request and request.include: - filtered_response = {} - if "metrics" in request.include: - filtered_response["metrics"] = metrics_json - - # Filter plot_data to only included charts - plot_includes = [i for i in request.include if i in plot_dfs] - if plot_includes: - filtered_response["plot_data"] = {k: plot_data[k] for k in plot_includes} - - return filtered_response - - return response - - -# Load the Census Dataset by `category`(housing, economic, etc.) and `subcategory`(special csv files) +# Load the Census Dataset by `category`(housing, economic, etc.) +# and `subcategory`(special time series tables) @router.post("/load/census/{category}/{subcategory}") async def read_census_data_subcat( category: str, request: FilterRequest, subcategory: str = "main" @@ -105,6 +55,7 @@ async def read_census_data_subcat( if ts_key in _TIMESERIES_VIEWS: view_name = _TIMESERIES_VIEWS[ts_key] filters = request.filters if request else None + data = timeseries_db.query_timeseries(view_name, filters) if data.empty: raise HTTPException( @@ -120,7 +71,8 @@ async def read_census_data_subcat( detail=f"Census subcategory '{subcategory}' was not found in category '{category}'", ) - data = data_loading.load_census_data(CENSUS_DATASETS[category][subcategory]) + # data = data_loading.load_census_data(CENSUS_DATASETS[category][subcategory]) + data = DB.execute(f"SELECT * FROM acs5_{category}_tidy") data = filter_from_request(data, request) metadata = {} diff --git a/backend/api/routes/post_routes/post_qcew.py b/backend/api/routes/post_routes/post_qcew.py index 85b8f048..49901e00 100644 --- a/backend/api/routes/post_routes/post_qcew.py +++ b/backend/api/routes/post_routes/post_qcew.py @@ -3,7 +3,9 @@ from api.metadata_registry import get_metadata from api.models import FilterRequest, make_response -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() router = APIRouter() @@ -36,7 +38,7 @@ def _first(label: str): if not county and is_statewide: query = """ SELECT year, quarter, quarter_label, sector, employment_4qma - FROM qcew_employment + FROM qcew_sectorEmployment_timeseries WHERE sector != 'Total' ORDER BY year, quarter, sector """ @@ -44,7 +46,7 @@ def _first(label: str): elif county: query = """ SELECT year, quarter, quarter_label, sector, employment_4qma - FROM qcew_employment + FROM qcew_sectorEmployment_timeseries WHERE sector != 'Total' AND County = ? ORDER BY year, quarter, sector diff --git a/backend/api/routes/post_routes/post_wastewater.py b/backend/api/routes/post_routes/post_wastewater.py index b13fcc55..77ad17c9 100644 --- a/backend/api/routes/post_routes/post_wastewater.py +++ b/backend/api/routes/post_routes/post_wastewater.py @@ -16,7 +16,7 @@ @router.post("/load/mapping/wastewater/service_area") async def wastewater_service_geojson(request: FilterRequest): - source = request_to_source(request, "service_areas_service_area_info", "default") + source = request_to_source(request, "VersoWastewater_serviceAreas_info", "default") data = get_waste_service_areas_geojson([source]) return Response(content=data, media_type="application/json") @@ -24,17 +24,17 @@ async def wastewater_service_geojson(request: FilterRequest): @router.post("/load/mapping/wastewater/treatment_facility") async def wastewater_facility_geojson(request: FilterRequest): source = request_to_source( - request, "treatment_facilities_treatment_facility_info", "default" + request, "VersoWastewater_treatmentFacilities_info", "default" ) data = get_waste_treatment_facility_geojson([source]) return Response(content=data, media_type="application/json") -@router.post("/load/wastewater/zoning/facility_permits") +@router.post("/load/mapping/wastewater/treatment_facility/permits") async def wastewater_facility_permits(request: FilterRequest): # TODO: the json table might be wrong, check later source = request_to_source( - request, "treatment_facilities_treatment_facility_permit_info", "default" + request, "VersoWastewater_treatmentFacilitiesPermits_info", "default" ) table = get_waste_treatment_facility_permits([source]) return make_response(data=table, metadata=get_metadata("zoning")) @@ -42,7 +42,9 @@ async def wastewater_facility_permits(request: FilterRequest): @router.post("/load/mapping/wastewater/septic_soil_suitability") async def wastewater_soil_suit_geojson(request: FilterRequest): - source = request_to_source(request, "soil_suitability_info_soil_suit", "default") + source = request_to_source( + request, "VersoWastewater_soilSuitability_info", "default" + ) data = get_soil_suit_geojson([source]) return Response(content=data, media_type="application/json") diff --git a/backend/api/routes/post_routes/post_zoning.py b/backend/api/routes/post_routes/post_zoning.py index f5d1855d..68010fa5 100644 --- a/backend/api/routes/post_routes/post_zoning.py +++ b/backend/api/routes/post_routes/post_zoning.py @@ -35,21 +35,21 @@ async def zoning_unzoned(): @router.post("/load/mapping/zoning/standard") async def zoning_geojson_info(request: FilterRequest): - source = request_to_source(request, "zoning_info", "default") + source = request_to_source(request, "VersoZoning_info", "default") data = get_zoning_geojson([source]) return Response(content=data, media_type="application/json") @router.post("/load/data/zoning/aggregated") async def acreage_response(request: FilterRequest): - source = request_to_source(request, "zoning_info", "default") + source = request_to_source(request, "VersoZoning_info", "default") agg, table = get_zoning_aggregated_acres([source]) return make_response(data=agg, metadata=get_metadata("zoning"), tableData=table) @router.post("/load/data/zoning/allowances") async def zoning_allowances(request: FilterRequest): - source = request_to_source(request, "zoning_info", "default") + source = request_to_source(request, "VersoZoning_info", "default") agg, table = get_zoning_allowances([source]) return make_response( data=agg, diff --git a/backend/api/schema.json b/backend/api/schema.json index d4e387ce..f89d304c 100644 --- a/backend/api/schema.json +++ b/backend/api/schema.json @@ -1,6 +1,6 @@ { "default": { - "zoning_info": { + "VersoZoning_info": { "join_key": "OBJECT_ID", "join_type": "inner", "columns": { @@ -10,7 +10,7 @@ "District Name": "District_Name" } }, - "zoning_wide": { + "VersoZoning_wide": { "join_key": "OBJECT_ID", "join_type": "inner", "columns": { @@ -24,7 +24,7 @@ "Planned Unit Development": "PUD_Allowance" } }, - "zoning_rules": { + "VersoZoning_rules": { "join_key": "OBJECT_ID", "join_type": "inner", "value_col": "val", @@ -35,7 +35,7 @@ "Value": "val" } }, - "cdc_county_places": { + "cdc_places_county": { "join_key": "LocationID", "join_type": "inner", "value_col": "Data_Value", @@ -47,8 +47,8 @@ "Prevalence Measure": "Data_Value_Type" } }, - "soil_suitability_info_soil_suit": { - "join_key": "ID", + "VersoWastewater_soilSuitability_info": { + "join_key": "OGC_FID", "join_type": "inner", "columns": { "Regional Planning Commission": "RPC", @@ -56,8 +56,8 @@ "Soil Suitability Level": "Suitability" } }, - "treatment_facilities_treatment_facility_info": { - "join_key": "ID", + "VersoWastewater_treatmentFacilities_info": { + "join_key": "Facility_ID", "join_type": "inner", "columns": { "Regional Planning Commission": "RPC", @@ -66,8 +66,8 @@ "Town": "TownName" } }, - "treatment_facilities_treatment_facility_permit_info": { - "join_key": "ID", + "VersoWastewater_treatmentFacilitiesPermits_info": { + "join_key": "Facility_ID", "join_type": "inner", "columns": { "Regional Planning Commission": "RPC", @@ -76,8 +76,8 @@ "Town": "TownName" } }, - "service_areas_service_area_info": { - "join_key": "ID", + "VersoWastewater_serviceAreas_info": { + "join_key": "Area_ID", "join_type": "inner", "columns": { "Regional Planning Commission": "RPC", @@ -86,7 +86,7 @@ "Town": "TownName" } }, - "ambulance_ambulance_info": { + "VCGI_ambulanceService_info": { "join_key": "OBJECTID", "join_type": "inner", "columns": { diff --git a/backend/app_utils/timeseries_db.py b/backend/app_utils/timeseries_db.py index 7d8df186..99f16629 100644 --- a/backend/app_utils/timeseries_db.py +++ b/backend/app_utils/timeseries_db.py @@ -6,7 +6,9 @@ import logging -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) diff --git a/backend/data_cleaning/clean_acs5.py b/backend/data_cleaning/clean_acs5.py new file mode 100644 index 00000000..bcbcae83 --- /dev/null +++ b/backend/data_cleaning/clean_acs5.py @@ -0,0 +1,110 @@ +import duckdb + +DP_TABLES = { + "DP02": ("acs5_social", "dp_social"), + "DP03": ("acs5_economic", "dp_economic"), + "DP04": ("acs5_housing", "dp_housing"), + "DP05": ("acs5_demographic", "dp_demographic"), +} + +COUNTY_GEOIDS = { + "Addison County, Vermont": 50001, + "Bennington County, Vermont": 50003, + "Caledonia County, Vermont": 50005, + "Chittenden County, Vermont": 50007, + "Essex County, Vermont": 50009, + "Franklin County, Vermont": 50011, + "Grand Isle County, Vermont": 50013, + "Lamoille County, Vermont": 50015, + "Orange County, Vermont": 50017, + "Orleans County, Vermont": 50019, + "Rutland County, Vermont": 50021, + "Washington County, Vermont": 50023, + "Windham County, Vermont": 50025, + "Windsor County, Vermont": 50027, +} + + +def add_dp_tables(con: duckdb.DuckDBPyConnection): + """ + Write each RAW DP table to CLEANED and add the DP identifier. + """ + + for dp, (raw_table, cleaned_table) in DP_TABLES.items(): + con.execute( + f""" + CREATE OR REPLACE TABLE lake.CLEANED.{cleaned_table} AS + SELECT + *, + '{dp}' AS "table" + FROM lake.RAW.{raw_table} + """ + ) + + +def build_dp_combined(con: duckdb.DuckDBPyConnection): + unions = [] + + for dp, (raw_table, _) in DP_TABLES.items(): + unions.append( + f""" + SELECT + NAME, + '{dp}' AS "table", + Category, + Subcategory, + Variable, + Measure, + year, + Value + FROM lake.RAW.{raw_table} + """ + ) + + con.execute( + f""" + CREATE OR REPLACE TABLE + lake.CLEANED.acs5_dp_combined_tidy + AS + {" UNION ALL ".join(unions)} + """ + ) + + +def build_county_geoids(con: duckdb.DuckDBPyConnection): + """ + Create the county GEOID lookup table. + """ + + values = ", ".join(f"('{name}', {geoid})" for name, geoid in COUNTY_GEOIDS.items()) + + con.execute( + f""" + CREATE OR REPLACE TABLE lake.CLEANED.vt_county_geoids AS + SELECT * + FROM ( + VALUES {values} + ) AS t(NAME, GEOID) + """ + ) + + +def clean(con: duckdb.DuckDBPyConnection): + add_dp_tables(con) + build_dp_combined(con) + build_county_geoids(con) + + +def main(con: duckdb.DuckDBPyConnection): + clean(con) + + +if __name__ == "__main__": + from lake_build import get_connection + + con = get_connection() + + try: + main(con) + finally: + con.close() diff --git a/backend/data_cleaning/clean_ambulance.py b/backend/data_cleaning/clean_ambulance.py new file mode 100644 index 00000000..1e3ffefe --- /dev/null +++ b/backend/data_cleaning/clean_ambulance.py @@ -0,0 +1,91 @@ +""" +**Author**: + Atticus Tarleton +**Created**: + 2026-07-20 +**Description**: + Build script to convert the ambulance service area files into SQL tables. +""" + +import duckdb + +# Hardcoded column selections +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", +] + + +def build_ambulance_info_table(con: duckdb.DuckDBPyConnection): + """Create the cleaned info table in DuckLake.""" + info_cols_str = ", ".join(AMBULANCE_INFO_COLS) + + con.execute( + f"""--sql + CREATE OR REPLACE TABLE lake.CLEANED.VCGI_ambulanceService_info AS + SELECT {info_cols_str} + FROM lake.RAW.ambulance + """ + ) + + +def build_ambulance_geom_table(con: duckdb.DuckDBPyConnection): + """Create the cleaned spatial table in DuckLake.""" + geom_cols_str = ", ".join(AMBULANCE_GEOM_COLS) + + con.execute( + f"""--sql + CREATE OR REPLACE TABLE lake.CLEANED.VCGI_ambulanceService_geom AS + SELECT {geom_cols_str} + FROM lake.RAW.ambulance + """ + ) + + +def build_ambulance_color_table(con: duckdb.DuckDBPyConnection): + """Create and populate the certification level colors lookup table.""" + con.execute( + """--sql + CREATE OR REPLACE TABLE lake.CLEANED.VCGI_ambulanceService_colors AS + SELECT * FROM ( + VALUES + ('Paramedic', '#2ca02c', '[44, 160, 44, 180]'), + ('Advanced EMT', '#ffcc00', '[255, 204, 0, 180]'), + ('Paramedic - Critical Care Endorsement', '#fd7e14', '[253, 126, 20, 180]') + ) AS t(certification_level, hex_color, rgba) + """ + ) + + +def clean(con: duckdb.DuckDBPyConnection): + build_ambulance_info_table(con) + build_ambulance_geom_table(con) + build_ambulance_color_table(con) + + +def main(con: duckdb.DuckDBPyConnection): + clean(con) + print("Successfully built ambulance service tables.") + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_cdc.py b/backend/data_cleaning/clean_cdc.py index edde5e3c..d236bd3a 100644 --- a/backend/data_cleaning/clean_cdc.py +++ b/backend/data_cleaning/clean_cdc.py @@ -10,11 +10,11 @@ python -m data_cleaning.clean_cdc """ +import duckdb import pandas as pd from sklearn.decomposition import PCA from build.core_functions import bin_measures -from lake_build import con # Columns we'd like excluded from the cleaned tables, IF they exist on that # particular RAW table. Tract- and county-level releases don't always share @@ -28,7 +28,7 @@ ] -def get_sme_indicators() -> str: +def get_sme_indicators(con: duckdb.DuckDBPyConnection) -> str: """ Get CDC Notes indicators """ @@ -45,32 +45,73 @@ def get_sme_indicators() -> str: def build_PCA_table(us_df: pd.DataFrame) -> pd.DataFrame: """ - Builds a 2-Principal Component DataFrame - for county-level CDC indicators + Builds a 2-component PCA score for Vermont counties. + + PCA is fit using the full national county dataset. Vermont county + observations are then standardized using the national means and + standard deviations before being projected into the fitted PCA space. + + Returns: + DataFrame containing LocationID and the first PCA component score. """ - ## select only shared columns - vt_df = us_df[us_df["stateabbr"] == "VT"].copy() - pv = us_df.pivot(columns="measure", values="data_value", index="locationid").dropna( - axis=0, how="any" + # Build a wide national dataset: + # rows = counties + # columns = CDC measures + pv = us_df.pivot_table( + index="locationid", + columns="measure", + values="data_value", + aggfunc="first", + ).dropna(axis=0, how="any") + + # Build Vermont-wide dataset using the same measures + vt_df = us_df[us_df["stateabbr"].eq("VT")].copy() + + pv_vt = vt_df.pivot_table( + index="locationid", + columns="measure", + values="data_value", + aggfunc="first", ) - pv_vt = vt_df.pivot(columns="measure", values="data_value", index="locationid") - shared = pv.columns.intersection(pv_vt.dropna(axis=1, how="all").columns) - pv = pv[shared] - pv_vt = pv_vt[shared] - - ## standardize US to build column - mean, std = pv.mean(), pv.std() - pv = (pv - mean) / std + + # Keep only measures that exist in both datasets and have + # complete national data. + shared = pv.columns.intersection(pv_vt.columns) + + pv = pv[shared].dropna(axis=1, how="all") + pv_vt = pv_vt[pv.columns] + + # Only retain Vermont counties with complete data for all + # measures used in the PCA. + pv_vt = pv_vt.dropna(axis=0, how="any") + + # Standardize using NATIONAL parameters. + mean = pv.mean() + std = pv.std() + + # Avoid division by zero for constant measures. + valid = std > 0 + pv = pv.loc[:, valid] + pv_vt = pv_vt.loc[:, valid] + mean = mean[valid] + std = std[valid] + + pv_standardized = (pv - mean) / std + vt_standardized = (pv_vt - mean) / std + + # Fit PCA using national observations. pca = PCA(n_components=2) - pca.fit(pv) + pca.fit(pv_standardized) + + # Project Vermont observations into national PCA space. + scores = pca.transform(vt_standardized) - # standardize the VT, transform, add back in, and return - pv_vt = (pv_vt - mean) / std - assert list(pv_vt.columns) == list(pv.columns), "measure columns misaligned" - scores = pca.transform(pv_vt) - pv_vt["pca_score"] = scores[:, 0] - pv_vt = pv_vt.reset_index() - return pv_vt + return pd.DataFrame( + { + "LocationID": pv_vt.index, + "pca_score": scores[:, 0], + } + ) def add_national_percentile(us_df: pd.DataFrame) -> pd.DataFrame: @@ -82,7 +123,7 @@ def add_national_percentile(us_df: pd.DataFrame) -> pd.DataFrame: return df[df["stateabbr"] == "VT"] -def get_columns(table: str) -> list[str]: +def get_columns(table: str, con: duckdb.DuckDBPyConnection) -> list[str]: """ Returns the actual column names for a RAW table. """ @@ -98,7 +139,7 @@ def get_columns(table: str) -> list[str]: def build_places_table( - raw_table: str, geo_filter_col: str, indicators: str + raw_table: str, geo_filter_col: str, indicators: str, con: duckdb.DuckDBPyConnection ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: """ Cleans a single PLACES RAW table (county or tract). @@ -108,7 +149,7 @@ def build_places_table( vt_df: Vermont-only cleaned dataset with measure bins. edge_df: Measure bin edges. """ - existing_cols = set(get_columns(raw_table)) + existing_cols = set(get_columns(raw_table, con)) exclude_cols = [c for c in CANDIDATE_EXCLUDE_COLS if c in existing_cols] exclude_clause = ", ".join(exclude_cols) @@ -143,24 +184,20 @@ def build_places_table( return us_df, vt_df, edge_df -def clean() -> dict[str, pd.DataFrame]: - indicators = get_sme_indicators() +def clean(con: duckdb.DuckDBPyConnection) -> dict[str, pd.DataFrame]: + indicators = get_sme_indicators(con) # County: keep the full national dataset for percentile/PCA calculations county_us, county_places, county_edges = build_places_table( - "cdc_places_county", - "stateabbr", - indicators, + "cdc_places_county", "stateabbr", indicators, con ) # PCA is fit on the national county data and applied to Vermont - # pca_county = build_PCA_table(county_us) + pca_county = build_PCA_table(county_us) # Tract: national data is needed for the national percentile _, tract_places, tract_edges = build_places_table( - "cdc_places_tract", - "stateabbr", - indicators, + "cdc_places_tract", "stateabbr", indicators, con ) return { @@ -168,11 +205,13 @@ def clean() -> dict[str, pd.DataFrame]: "cdc_edges_county": county_edges, "cdc_places_tract": tract_places, "cdc_edges_tract": tract_edges, - # "cdc_pca_county": pca_county, + "cdc_pca_county": pca_county, } -def add_to_lake(tables: dict[str, pd.DataFrame]) -> None: +def add_to_lake( + tables: dict[str, pd.DataFrame], con: duckdb.DuckDBPyConnection +) -> None: for name, df in tables.items(): view_name = f"{name}_df" @@ -186,9 +225,9 @@ def add_to_lake(tables: dict[str, pd.DataFrame]) -> None: con.unregister(view_name) -def main(): - tables = clean() - add_to_lake(tables) +def main(con: duckdb.DuckDBPyConnection): + tables = clean(con) + add_to_lake(tables, con) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_demographics.py b/backend/data_cleaning/clean_demographics.py index d35b9011..82202ace 100644 --- a/backend/data_cleaning/clean_demographics.py +++ b/backend/data_cleaning/clean_demographics.py @@ -10,12 +10,11 @@ python -m data_cleaning.clean_demographics` """ +import duckdb import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT * @@ -26,13 +25,13 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def clean(): - raw_df = read_raw_data() +def clean(con: duckdb.DuckDBPyConnection): + raw_df = read_raw_data(con) # NOTE: Cleaning already included in data fetch --> returning raw dataframe return raw_df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(clean_df: pd.DataFrame, con: duckdb.DuckDBPyConnection): """ Writes the cleaned demographics dataframe to the CLEANED schema in DuckLake. @@ -45,9 +44,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(clean_df, con) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_dependency_ratio.py b/backend/data_cleaning/clean_dependency_ratio.py index e11e858d..0628e9c2 100644 --- a/backend/data_cleaning/clean_dependency_ratio.py +++ b/backend/data_cleaning/clean_dependency_ratio.py @@ -15,13 +15,12 @@ python -m data_cleaning.clean_dependency_ratio """ +import duckdb import numpy as np import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: """ Reading in the raw `demographics` table from lake.RAW schema. Filters to only age-group variables @@ -60,7 +59,7 @@ def replace_unavailable_data(df: pd.DataFrame) -> pd.DataFrame: return df -def calculate_dependency_ratio(df: pd.DataFrame): +def calculate_dependency_ratio(df: pd.DataFrame) -> pd.DataFrame: """ Calculates age-dependency ratio as follows: **Age Dependency Ratio** @@ -98,14 +97,14 @@ def calculate_dependency_ratio(df: pd.DataFrame): return df[["year", "NAME", "Age_Dependency_Ratio", "geo_type"]] -def clean(): - raw_df = read_raw_data() +def clean(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: + raw_df = read_raw_data(con) df = replace_unavailable_data(raw_df) df = calculate_dependency_ratio(df) return df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(clean_df: pd.DataFrame, con: duckdb.DuckDBPyConnection) -> None: """ Writes the cleaned, long-format age_dependency_ratio dataframe to the CLEANED schema in DuckLake. @@ -119,9 +118,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(clean_df, con) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_derived_time_series.py b/backend/data_cleaning/clean_derived_time_series.py index 3818ee43..bcf9fd02 100644 --- a/backend/data_cleaning/clean_derived_time_series.py +++ b/backend/data_cleaning/clean_derived_time_series.py @@ -6,7 +6,7 @@ **Description**: Cleaning script for the single-variable timeseries tables (median age, median household income, median home value, - per capita income, total housing units,vacancy rate, unemployment_rate). + per capita income, total housing units, vacancy rate). Pipeline steps: 1. Read raw data @@ -26,10 +26,11 @@ import sys from dataclasses import dataclass, field +import duckdb import numpy as np import pandas as pd -from lake_build import con +from lake_build import get_connection UNAVAILABLE_SENTINEL = -666666666.0 @@ -101,18 +102,10 @@ class DatasetConfig: output_table="acs5Housing_vacancyRates_timeseries", keep_variable_col=True, ), - "unemployment_rate": DatasetConfig( - source_table="acs5_economic", - variables=["Unemployment Rate"], - value_source_col="Value", - output_value_col="Unemployment_Rate", - output_table="acs5Economics_unemploymentRate_timeseries", - extra_where_statement="Measure = 'Percent'", - ), } -def read_raw_data(cfg: DatasetConfig) -> pd.DataFrame: +def read_raw_data(cfg: DatasetConfig, con: duckdb.DuckDBPyConnection) -> pd.DataFrame: select_parts = ["year", "NAME"] if cfg.keep_variable_col: select_parts.append("Variable") @@ -148,14 +141,16 @@ def replace_unavailable_data(df: pd.DataFrame, value_col: str) -> pd.DataFrame: return df -def clean(cfg: DatasetConfig) -> pd.DataFrame: - df = read_raw_data(cfg) +def clean(con: duckdb.DuckDBPyConnection, cfg: DatasetConfig) -> pd.DataFrame: + df = read_raw_data(cfg, con) df = change_dtype(df, cfg.output_value_col) df = replace_unavailable_data(df, cfg.output_value_col) return df -def add_to_lake(clean_df: pd.DataFrame, output_table: str) -> None: +def add_to_lake( + con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame, output_table: str +) -> None: """ Writes a cleaned, long-format dataframe to the CLEANED schema in DuckLake. """ @@ -168,13 +163,13 @@ def add_to_lake(clean_df: pd.DataFrame, output_table: str) -> None: ) -def run(name: str) -> None: +def run(name: str, con: duckdb.DuckDBPyConnection) -> None: cfg = CONFIGS[name] - clean_df = clean(cfg) - add_to_lake(clean_df, cfg.output_table) + clean_df = clean(con, cfg) + add_to_lake(con, clean_df, cfg.output_table) -def main(names: list[str] | None = None) -> None: +def main(con: duckdb.DuckDBPyConnection, names: list[str] | None = None) -> None: targets = names or list(CONFIGS) unknown = [n for n in targets if n not in CONFIGS] @@ -184,7 +179,7 @@ def main(names: list[str] | None = None) -> None: ) for name in targets: - run(name) + run(name, con) if __name__ == "__main__": @@ -198,5 +193,10 @@ def main(names: list[str] | None = None) -> None: help="Dataset name(s) to clean. Omit to run all.", metavar="DATASET", ) - args = parser.parse_args() - main(args.datasets or None) + con = get_connection() + + try: + args = parser.parse_args() + main(con, args.datasets or None) + finally: + con.close() diff --git a/backend/data_cleaning/clean_economic.py b/backend/data_cleaning/clean_economic.py index 1cc43c84..4af93857 100644 --- a/backend/data_cleaning/clean_economic.py +++ b/backend/data_cleaning/clean_economic.py @@ -10,12 +10,11 @@ python -m data_cleaning.clean_economic """ +import duckdb import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT * @@ -26,13 +25,13 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def clean(): - raw_df = read_raw_data() +def clean(con: duckdb.DuckDBPyConnection): + raw_df = read_raw_data(con) # NOTE: Cleaning already included in data fetch --> returning raw dataframe return raw_df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame): """ Writes the cleaned economic dataframe to the CLEANED schema in DuckLake. @@ -45,9 +44,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_education.py b/backend/data_cleaning/clean_education.py index 65dcf3cc..138ce012 100644 --- a/backend/data_cleaning/clean_education.py +++ b/backend/data_cleaning/clean_education.py @@ -10,12 +10,11 @@ python -m data_cleaning.clean_education """ +import duckdb import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT * @@ -26,13 +25,13 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def clean(): - raw_df = read_raw_data() +def clean(con: duckdb.DuckDBPyConnection): + raw_df = read_raw_data(con) # NOTE: Cleaning already included in data fetch --> returning raw dataframe return raw_df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame): """ Writes the cleaned education dataframe to the CLEANED schema in DuckLake. @@ -45,9 +44,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_fips.py b/backend/data_cleaning/clean_fips.py new file mode 100644 index 00000000..3e327776 --- /dev/null +++ b/backend/data_cleaning/clean_fips.py @@ -0,0 +1,127 @@ +""" +**Author**: + Ian Sargent +**Created**: + 2026-07-16 +**Description**: + Data cleaning script for the raw boundary line tables in the DuckLake. + + Standardizes boundary column names to match the legacy processed boundary + tables used throughout the application. + + County: + CountyFIPS + CountyName + geom + + Town: + FIPS_ID + TOWN_NAME + geometry + + Tract: + LocationID + name + geometry + +Run with: + python -m data_cleaning.clean_boundaries +""" + +import duckdb + + +## BUILD CLEANED VIEWS -------------------- +def build_county_lines(con: duckdb.DuckDBPyConnection) -> None: + """ + Clean VT county boundary lines. + + Standardizes county identifiers and names to the legacy boundary + column conventions: + CNTYGEOID -> CountyFIPS + CNTYNAME -> CountyName + geometry -> geom + """ + con.execute( + """--sql + CREATE OR REPLACE VIEW vt_county_lines AS + SELECT + CNTYGEOID AS CountyFIPS, + CNTYNAME AS CountyName, + geometry + FROM lake.RAW.vt_county_lines + """ + ) + + +def build_town_lines(con: duckdb.DuckDBPyConnection) -> None: + """ + Clean VT town boundary lines. + + Standardizes town identifiers and names to the legacy boundary + column conventions: + GEOID -> FIPS_ID + NAME -> TOWN_NAME + geom -> geometry + """ + con.execute( + """--sql + CREATE OR REPLACE VIEW vt_town_lines AS + SELECT + GEOID AS FIPS_ID, + TRIM(SPLIT_PART("NAME", ',', 1)) AS TOWN_NAME, + geometry + FROM lake.RAW.vt_town_lines + """ + ) + + +def build_tract_lines(con: duckdb.DuckDBPyConnection) -> None: + """ + Clean VT Census tract boundary lines. + + Standardizes the tract identifier to LocationID. + """ + con.execute( + """--sql + CREATE OR REPLACE VIEW vt_tract_lines AS + SELECT + GEOID AS LocationID, + NAMELSAD AS name, + geometry + FROM lake.RAW.vt_tract_lines + """ + ) + + +## WRITE CLEANED TABLES -------------------- +def add_to_lake(con: duckdb.DuckDBPyConnection) -> None: + table_names = [ + "vt_county_lines", + "vt_town_lines", + "vt_tract_lines", + ] + + for name in table_names: + con.execute( + f"""--sql + CREATE OR REPLACE TABLE lake.CLEANED.{name}_geom AS + SELECT * + FROM {name} + """ + ) + + +def clean(con: duckdb.DuckDBPyConnection) -> None: + build_county_lines(con) + build_town_lines(con) + build_tract_lines(con) + + +def main(con: duckdb.DuckDBPyConnection): + clean(con) + add_to_lake(con) + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_flood.py b/backend/data_cleaning/clean_flood.py index 38d057c2..73c5134e 100644 --- a/backend/data_cleaning/clean_flood.py +++ b/backend/data_cleaning/clean_flood.py @@ -6,25 +6,13 @@ **Description**: Data cleaning script for the raw `flood` table in the DuckLake Run with: -python -m ETL.data_cleaning.clean_flood +python -m data_cleaning.clean_flood """ -from lake_build import con +import duckdb -## LOAD SPATIAL EXTENSION FUNCTION -------------------- -def _load_spatial() -> None: - """ - Load the spatial extension, installing it first if necessary. - """ - try: - con.execute("""--sql LOAD spatial""") - except Exception: - con.execute("""--sql INSTALL spatial""") - con.execute("""--sql LOAD spatial""") - - -def build_flood(): +def build_flood(con: duckdb.DuckDBPyConnection) -> None: """ Clean FEMA flood polygons. """ @@ -52,7 +40,7 @@ def build_flood(): ) -def add_to_lake(): +def add_to_lake(con: duckdb.DuckDBPyConnection) -> None: con.execute( """--sql CREATE OR REPLACE TABLE lake.CLEANED.FEMA_floodHazard_geom AS @@ -62,14 +50,13 @@ def add_to_lake(): ) -def clean(): - _load_spatial() - build_flood() +def clean(con: duckdb.DuckDBPyConnection) -> None: + build_flood(con) -def main(): - clean() - add_to_lake() +def main(con: duckdb.DuckDBPyConnection): + clean(con) + add_to_lake(con) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_health_insurance_coverage.py b/backend/data_cleaning/clean_health_insurance_coverage.py index d2829152..75ed1663 100644 --- a/backend/data_cleaning/clean_health_insurance_coverage.py +++ b/backend/data_cleaning/clean_health_insurance_coverage.py @@ -10,13 +10,12 @@ python -m data_cleaning.clean_health_insurance_coverage """ +import duckdb import numpy as np import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT year, NAME, Variable, Value, geo_type @@ -36,27 +35,25 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def change_dtype(df: pd.DataFrame): +def change_dtype(df: pd.DataFrame) -> pd.DataFrame: df["Value"] = pd.to_numeric(df["Value"], errors="coerce") - return df -def replace_unavailable_data(df: pd.DataFrame): +def replace_unavailable_data(df: pd.DataFrame) -> pd.DataFrame: df["Value"] = df["Value"].replace(-666666666.0, np.nan) - return df -def clean(): - raw_df = read_raw_data() +def clean(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: + raw_df = read_raw_data(con) df = change_dtype(raw_df) df = replace_unavailable_data(df) return df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame) -> None: """ Writes the cleaned, long-format health_insurance_coverage dataframe to the CLEANED schema in DuckLake. @@ -69,9 +66,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_historic_population.py b/backend/data_cleaning/clean_historic_population.py index fdc71b79..448484c8 100644 --- a/backend/data_cleaning/clean_historic_population.py +++ b/backend/data_cleaning/clean_historic_population.py @@ -9,12 +9,11 @@ python -m data_cleaning.clean_historic_population """ +import duckdb import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT * @@ -25,7 +24,7 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def clean_column_names(df: pd.DataFrame): +def clean_column_names(df: pd.DataFrame) -> None: """ Cleans historic_population data column names """ @@ -39,7 +38,7 @@ def clean_column_names(df: pd.DataFrame): df["geo_type"] = "town" -def long_format(df: pd.DataFrame): +def long_format(df: pd.DataFrame) -> pd.DataFrame: """ Turns raw historic_population estimates data into long_format """ @@ -49,15 +48,15 @@ def long_format(df: pd.DataFrame): df, id_vars=["geoid", "town", "county", "geo_type"], value_vars=year_cols, - var_name="Year", + var_name="year", value_name="Population", ) - df_long["Year"] = df_long["Year"].astype(int) + df_long["year"] = df_long["year"].astype(int) return df_long -def add_NAME_column(long_df: pd.DataFrame): +def add_NAME_column(long_df: pd.DataFrame) -> pd.DataFrame: import requests json_file = "https://raw.githubusercontent.com/VERSO-UVM/react-vt-data/refs/heads/main/frontend/public/data/municipalites.json" @@ -75,14 +74,14 @@ def add_NAME_column(long_df: pd.DataFrame): return long_df -def add_population_aggregations(df: pd.DataFrame): +def add_population_aggregations(df: pd.DataFrame) -> pd.DataFrame: """ Aggregates town-level population to county and state levels, and appends them as additional rows in the long-format dataframe. """ df["county_geoid"] = df["geoid"].astype(str).str[:5] # County-level aggregation - county_df = df.groupby(["county_geoid", "county", "Year"], as_index=False)[ + county_df = df.groupby(["county_geoid", "county", "year"], as_index=False)[ "Population" ].sum() county_df["NAME"] = county_df["county"] + " County, Vermont" @@ -90,13 +89,13 @@ def add_population_aggregations(df: pd.DataFrame): county_df = county_df.rename(columns={"county_geoid": "geoid"}) # State-level aggregation - state_df = df.groupby("Year", as_index=False)["Population"].sum() + state_df = df.groupby("year", as_index=False)["Population"].sum() state_df["NAME"] = "Vermont" state_df["geoid"] = "50" # Vermont's state FIPS code state_df["geo_type"] = "state" # Align columns before concatenating - cols = ["geoid", "NAME", "Year", "Population", "geo_type"] + cols = ["geoid", "NAME", "year", "Population", "geo_type"] town_df = df[cols] county_df = county_df[cols] @@ -107,17 +106,17 @@ def add_population_aggregations(df: pd.DataFrame): return combined -def clean(): +def clean(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: # Get raw dataframe from DuckLake RAW tables - raw_df = read_raw_data() + raw_df = read_raw_data(con) # Clean column names clean_column_names(raw_df) - # Melt DataFrame into long format (Cols: "geoid", "NAME", "Year", "Population", "geo_type") + # Melt DataFrame into long format (Cols: "geoid", "NAME", "year", "Population", "geo_type") df_long = long_format(raw_df) # Add a census-style "NAME" column for easier filtering df_long_clean = add_NAME_column(df_long) # Reorder columns - column_order = ["geoid", "NAME", "county", "town", "Year", "Population", "geo_type"] + column_order = ["geoid", "NAME", "county", "town", "year", "Population", "geo_type"] df = df_long_clean[column_order] # Append county + state aggregations (total sum) df = add_population_aggregations(df) @@ -125,7 +124,7 @@ def clean(): return df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame) -> None: """ Writes the cleaned, long-format historic population dataframe to the CLEANED schema in DuckLake. @@ -138,9 +137,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_housing.py b/backend/data_cleaning/clean_housing.py index 0063c6b1..22bafb77 100644 --- a/backend/data_cleaning/clean_housing.py +++ b/backend/data_cleaning/clean_housing.py @@ -10,12 +10,11 @@ python -m data_cleaning.clean_housing """ +import duckdb import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT * @@ -26,13 +25,13 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def clean(): - raw_df = read_raw_data() +def clean(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: + raw_df = read_raw_data(con) # NOTE: Cleaning already included in data fetch --> returning raw dataframe return raw_df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame) -> None: """ Writes the cleaned housing dataframe to the CLEANED schema in DuckLake. @@ -45,9 +44,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_median_earnings.py b/backend/data_cleaning/clean_median_earnings.py new file mode 100644 index 00000000..c45cbf46 --- /dev/null +++ b/backend/data_cleaning/clean_median_earnings.py @@ -0,0 +1,79 @@ +""" +**Author**: + Ian Sargent +**Created**: + 2026-08-25 +**Description**: + Data cleaning script for median earnings (Male, Female, All Workers). + Derived from the `RAW.acs5_economic` DuckLake table +**Run with**: +python -m data_cleaning.clean_median_earnings +""" + +from datetime import datetime + +import duckdb +import numpy as np +import pandas as pd + +INFLATION_YEAR = datetime.now().year - 2 + + +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: + raw_df = con.execute( + f"""--sql + SELECT year, NAME, Subcategory AS 'Variable', Value, geo_type + FROM lake.RAW.acs5_economic + WHERE Category LIKE '%INCOME AND BENEFITS ' || chr(40) || 'IN {INFLATION_YEAR}%' + AND Subcategory IN ( + 'Median earnings for male full-time, year-round workers (dollars)', + 'Median earnings for female full-time, year-round workers (dollars)', + 'Median earnings for workers (dollars)' + ) + AND Variable = 'Total' + AND Measure = 'Estimate' + ORDER BY year; + """ + ).df() + + return raw_df + + +def change_dtype(df: pd.DataFrame) -> pd.DataFrame: + df["Value"] = pd.to_numeric(df["Value"], errors="coerce") + return df + + +def replace_unavailable_data(df: pd.DataFrame) -> pd.DataFrame: + df["Value"] = df["Value"].replace(-666666666.0, np.nan) + return df + + +def clean(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: + raw_df = read_raw_data(con) + df = change_dtype(raw_df) + df = replace_unavailable_data(df) + + return df + + +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame) -> None: + """ + Writes the cleaned, long-format health_insurance_coverage dataframe + to the CLEANED schema in DuckLake. + """ + con.execute( + """--sql + CREATE OR REPLACE TABLE lake.CLEANED.acs5Economics_medianEarnings_timeseries AS + SELECT * FROM clean_df + """ + ) + + +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_qcew.py b/backend/data_cleaning/clean_qcew.py index 58bb490f..df27fc3b 100644 --- a/backend/data_cleaning/clean_qcew.py +++ b/backend/data_cleaning/clean_qcew.py @@ -10,12 +10,11 @@ python -m data_cleaning.clean_qcew """ +import duckdb import pandas as pd -from lake_build import con - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT * @@ -26,13 +25,13 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def clean(): - raw_df = read_raw_data() +def clean(con: duckdb.DuckDBPyConnection): + raw_df = read_raw_data(con) # NOTE: Cleaning already included in data fetch --> returning raw dataframe return raw_df -def add_to_lake(clean_df: pd.DataFrame): +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame): """ Writes the cleaned qcew dataframe to the CLEANED schema in DuckLake. @@ -45,9 +44,9 @@ def add_to_lake(clean_df: pd.DataFrame): ) -def main(): - clean_df = clean() - add_to_lake(clean_df) +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_snapshot.py b/backend/data_cleaning/clean_snapshot.py new file mode 100644 index 00000000..1973002c --- /dev/null +++ b/backend/data_cleaning/clean_snapshot.py @@ -0,0 +1,86 @@ +""" +**Author**: + Ian Sargent +**Created**: + 2026-08-21 +**Description**: + Data cleaning script for creating an ACS5 snapshot table by + combining selected indicators from multiple RAW tables in DuckLake. +**Run with**: + python -m data_cleaning.clean_snapshot +""" + +import duckdb +import pandas as pd + + +def read_raw_data( + con: duckdb.DuckDBPyConnection, +) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: + """ + Read the source tables from the RAW schema in DuckLake. + """ + + dem_df = con.execute( + """--sql + SELECT * + FROM lake.RAW.demographics + """ + ).df() + + housing_df = con.execute( + """--sql + SELECT * + FROM lake.RAW.housing + """ + ).df() + + econ_df = con.execute( + """--sql + SELECT * + FROM lake.RAW.economic + """ + ).df() + + return dem_df, housing_df, econ_df + + +def clean(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: + """ + Select the snapshot indicators from the source datasets + and combine them into a single dataframe. + """ + + dem_df, housing_df, econ_df = read_raw_data(con) + dem_vars = dem_df[dem_df["Variable"].isin(["Population (ACS)", "Median Age"])] + housing_vars = housing_df[housing_df["Variable"].isin(["Median Home Value"])] + econ_vars = econ_df[ + econ_df["Variable"].isin( + ["Labor Force Participation Rate (16+)", "Median Household Income"] + ) + ] + + combined = pd.concat([dem_vars, econ_vars, housing_vars], ignore_index=True) + return combined + + +def add_to_lake(con: duckdb.DuckDBPyConnection, clean_df: pd.DataFrame): + """ + Writes the cleaned snapshot dataframe + to the CLEANED schema in DuckLake. + """ + con.execute( + """--sql + CREATE OR REPLACE TABLE lake.CLEANED.acs5_snapshot_indicators_tidy AS + SELECT * FROM clean_df + """ + ) + + +def main(con: duckdb.DuckDBPyConnection): + clean_df = clean(con) + add_to_lake(con, clean_df) + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_wastewater.py b/backend/data_cleaning/clean_wastewater.py index ad39e943..c07b9679 100644 --- a/backend/data_cleaning/clean_wastewater.py +++ b/backend/data_cleaning/clean_wastewater.py @@ -10,27 +10,11 @@ python -m data_cleaning.clean_wastewater """ -from lake_build import con - -# TODO: Path is useful when sql files are created! -# from build import BACKEND -# sql_path = BACKEND / "data_cleaning" / "sql" - - -## LOAD SPATIAL EXTENSION FUNCTION -------------------- -def _load_spatial() -> None: - """ - Load the spatial extension, installing it first if necessary. - """ - try: - con.execute("""--sql LOAD spatial""") - except Exception: - con.execute("""--sql INSTALL spatial""") - con.execute("""--sql LOAD spatial""") +import duckdb ## ADD UNIQUE ID COLUMNS -------------------- -def build_service_area_id() -> None: +def build_service_area_id(con: duckdb.DuckDBPyConnection) -> None: con.execute( """--sql CREATE OR REPLACE VIEW service_areas_with_id AS @@ -42,7 +26,7 @@ def build_service_area_id() -> None: ) -def build_facility_id() -> None: +def build_facility_id(con: duckdb.DuckDBPyConnection) -> None: con.execute( """--sql CREATE OR REPLACE VIEW treatment_facilities_with_id AS @@ -54,7 +38,7 @@ def build_facility_id() -> None: ) -def build_soil_combined() -> None: +def build_soil_combined(con: duckdb.DuckDBPyConnection) -> None: RPCs = [ "ACRPC", "BCRC", @@ -78,7 +62,7 @@ def build_soil_combined() -> None: ) -def build_soil_suitability_id() -> None: +def build_soil_suitability_id(con: duckdb.DuckDBPyConnection) -> None: con.execute( """--sql CREATE OR REPLACE VIEW soil_suitability_with_id AS @@ -91,7 +75,7 @@ def build_soil_suitability_id() -> None: ## SERVICE AREA TABLES -------------------- -def build_service_info() -> None: +def build_service_info(con: duckdb.DuckDBPyConnection) -> None: service_area_info_cols = [ "Area_ID", "TownID", @@ -113,7 +97,7 @@ def build_service_info() -> None: ) -def build_service_geom() -> None: +def build_service_geom(con: duckdb.DuckDBPyConnection) -> None: # service_geom_cols = ["Area_ID", "geometry"] con.execute( """--sql @@ -126,7 +110,7 @@ def build_service_geom() -> None: ) -def build_service_misc() -> None: +def build_service_misc(con: duckdb.DuckDBPyConnection) -> None: service_miscellaneous_info_cols = [ "Area_ID", "GISNotes", @@ -147,7 +131,7 @@ def build_service_misc() -> None: ## TREATMENT FACILITY TABLES -------------------- -def build_facility_info() -> None: +def build_facility_info(con: duckdb.DuckDBPyConnection) -> None: facility_info_cols = [ "Facility_ID", "DesignHydraulicCapacityInMGD", @@ -169,7 +153,7 @@ def build_facility_info() -> None: ) -def build_facility_geom() -> None: +def build_facility_geom(con: duckdb.DuckDBPyConnection) -> None: # facility_geom_cols = ["Facility_ID", "Latitude", "Longitude", "geometry"] con.execute( """--sql @@ -184,7 +168,7 @@ def build_facility_geom() -> None: ) -def build_facility_permits() -> None: +def build_facility_permits(con: duckdb.DuckDBPyConnection) -> None: permit_info_cols = [ "Facility_ID", "PermitID", @@ -203,7 +187,7 @@ def build_facility_permits() -> None: ) -def build_facility_misc() -> None: +def build_facility_misc(con: duckdb.DuckDBPyConnection) -> None: facility_miscellaneous_info_cols = ["Facility_ID", "SourceFile", "GEOIDTXT"] con.execute( @@ -216,7 +200,7 @@ def build_facility_misc() -> None: ## SOIL SUITABILITY TABLES -------------------- -def build_suitability_info() -> None: +def build_suitability_info(con: duckdb.DuckDBPyConnection) -> None: suitability_info_cols = ["OGC_FID", "Suitability", "Jurisdiction", "RPC", "Acres"] con.execute( @@ -228,7 +212,7 @@ def build_suitability_info() -> None: ) -def build_suitability_geom() -> None: +def build_suitability_geom(con: duckdb.DuckDBPyConnection) -> None: # suitability_geom_cols = ["OGC_FID", "geometry"] con.execute( """--sql @@ -241,8 +225,27 @@ def build_suitability_geom() -> None: ) +def build_suitability_colors(con: duckdb.DuckDBPyConnection) -> None: + con.execute( + """--sql + CREATE OR REPLACE TABLE soilSuitability_colors ( + soil_suitability TEXT PRIMARY KEY, + hex_color TEXT NOT NULL, + rgba TEXT NOT NULL + ); + + INSERT INTO soilSuitability_colors VALUES + ('Well Suited', '#2ca02c', '[44, 160, 44, 180]'), + ('Moderately Suited', '#ffcc00', '[255, 204, 0, 180]'), + ('Marginally Suited', '#fd7e14', '[253, 126, 20, 180]'), + ('Not Suited', '#dc3545', '[220, 53, 69, 180]'), + ('Not Rated', '#6c757d', '[108, 117, 125, 180]'); + """ + ) + + ## STORMWATER MANAGEMENT TABLES -------------------- -def build_stormwater_info() -> None: +def build_stormwater_info(con: duckdb.DuckDBPyConnection) -> None: # "Type" labels derived from VERSO WIM GitHub pages con.execute( """--sql @@ -277,7 +280,7 @@ def build_stormwater_info() -> None: ) -def build_stormwater_geom() -> None: +def build_stormwater_geom(con: duckdb.DuckDBPyConnection) -> None: con.execute( """--sql CREATE OR REPLACE VIEW stormwaterManagement_geom AS @@ -290,37 +293,35 @@ def build_stormwater_geom() -> None: ## CLEANING PIPELINE -------------------- -def clean(): - # Load spatial extension in SQL - _load_spatial() - +def clean(con: duckdb.DuckDBPyConnection): # Add unique IDs to each table - build_service_area_id() - build_facility_id() - build_soil_combined() - build_soil_suitability_id() + build_service_area_id(con) + build_facility_id(con) + build_soil_combined(con) + build_soil_suitability_id(con) # Service area tables - build_service_info() - build_service_geom() - build_service_misc() + build_service_info(con) + build_service_geom(con) + build_service_misc(con) # Treatment facility tables - build_facility_info() - build_facility_geom() - build_facility_permits() - build_facility_misc() + build_facility_info(con) + build_facility_geom(con) + build_facility_permits(con) + build_facility_misc(con) # Soil suitability Tables - build_suitability_info() - build_suitability_geom() + build_suitability_info(con) + build_suitability_geom(con) + build_suitability_colors(con) # Stormwater Management Tables - build_stormwater_info() - build_stormwater_geom() + build_stormwater_info(con) + build_stormwater_geom(con) -def add_to_lake(): +def add_to_lake(con: duckdb.DuckDBPyConnection): table_names = [ "serviceAreas_info", "serviceAreas_geom", @@ -332,6 +333,7 @@ def add_to_lake(): "soilSuitability_info", # NOTE: The `soilSuitabilitygeom` dataset below is too large for git storage. Add to .gitignore "soilSuitability_geom", + "soilSuitability_colors", "stormwaterManagement_info", "stormwaterManagement_geom", ] @@ -346,9 +348,9 @@ def add_to_lake(): ) -def main(): - clean() - add_to_lake() +def main(con: duckdb.DuckDBPyConnection): + clean(con) + add_to_lake(con) if __name__ == "__main__": diff --git a/backend/data_cleaning/clean_zoning.py b/backend/data_cleaning/clean_zoning.py index 573396d0..c43d46dd 100644 --- a/backend/data_cleaning/clean_zoning.py +++ b/backend/data_cleaning/clean_zoning.py @@ -11,10 +11,10 @@ from pathlib import Path +import duckdb import pandas as pd from app_utils.sql_render import render_sql -from lake_build import con SQL_PATH = Path(__file__).resolve().parent / "sql" # Town and zoning-district boundaries were digitised separately, so subtracting @@ -59,24 +59,7 @@ } -## LOAD SPATIAL EXTENSION FUNCTION -------------------- -def _load_spatial() -> None: - """ - Load the spatial extension, installing it first if necessary. - """ - try: - con.execute("INSTALL spatial;") - except Exception as e: - print(f"Spatial install note: {e}") - - try: - con.execute("LOAD spatial;") - except Exception as e: - print(f"CRITICAL: Failed to load spatial extension: {e}") - raise e - - -def read_raw_data() -> pd.DataFrame: +def read_raw_data(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: raw_df = con.execute( """--sql SELECT * @@ -89,7 +72,7 @@ def read_raw_data() -> pd.DataFrame: return raw_df -def build_info(): +def build_info(con: duckdb.DuckDBPyConnection) -> None: info_string = ", ".join(info_cols) info_sql = render_sql( @@ -108,10 +91,10 @@ def build_info(): con.register("info", info_df) -def build_geom(): +def build_geom(con: duckdb.DuckDBPyConnection) -> None: con.execute( """--sql - CREATE OR REPLACE VIEW geom AS + CREATE OR REPLACE TEMP VIEW geom AS SELECT OBJECT_ID, ST_GeomFromWKB(geometry) AS geometry @@ -120,7 +103,7 @@ def build_geom(): ) -def get_rule_cols(): +def get_rule_cols(con: duckdb.DuckDBPyConnection) -> list[str]: dropped_cols = ["Shape_Area", "Shape_Length"] all_cols = ( con.execute( @@ -139,7 +122,7 @@ def get_rule_cols(): return rule_cols -def split_col(col: str, use_types: set[str]): +def split_col(col: str, use_types: set[str]) -> tuple[str | None, str | None]: for use_type in use_types: if col.startswith(f"{use_type}_"): rule = col[len(use_type) + 1 :] @@ -148,8 +131,8 @@ def split_col(col: str, use_types: set[str]): return None, None -def build_rules(raw_df: pd.DataFrame): - rule_cols = get_rule_cols() +def build_rules(con: duckdb.DuckDBPyConnection, raw_df: pd.DataFrame) -> None: + rule_cols = get_rule_cols(con) clean_rule_cols = [col.replace("/", "_") for col in rule_cols] cast_df = raw_df[["OBJECT_ID"] + rule_cols].copy() @@ -187,21 +170,21 @@ def build_rules(raw_df: pd.DataFrame): con.register("rules", rules) -def build_full(): +def build_full(con: duckdb.DuckDBPyConnection) -> None: drop_cols = ["geometry", "Shape_Area", "Shape_Length"] exclude = ", ".join(drop_cols) con.execute( f"""--sql - CREATE OR REPLACE VIEW wide AS + CREATE OR REPLACE TEMP VIEW wide AS SELECT * EXCLUDE ({exclude}) FROM zoning_raw """ ) -def build_color(): +def build_color(con: duckdb.DuckDBPyConnection) -> None: con.execute( """--sql - CREATE OR REPLACE VIEW colors AS + CREATE OR REPLACE TEMP VIEW colors AS SELECT * FROM ( VALUES @@ -214,7 +197,7 @@ def build_color(): ) -def build_empty_geom(): +def build_empty_geom(con: duckdb.DuckDBPyConnection) -> None: """ Build the geometry for the polygons *where we don't have zoning information*. @@ -224,7 +207,7 @@ def build_empty_geom(): """ con.execute("""--sql - CREATE OR REPLACE VIEW town_boundaries + CREATE OR REPLACE TEMP VIEW town_boundaries AS SELECT * FROM lake.RAW.vt_town_lines """) @@ -238,20 +221,19 @@ def build_empty_geom(): con.register("empty_geom", empty_geom_df) -def clean(): - _load_spatial() - df = read_raw_data() - build_info() - build_geom() - build_rules(df) - build_empty_geom() - build_color() - build_full() +def clean(con: duckdb.DuckDBPyConnection) -> pd.DataFrame: + df = read_raw_data(con) + build_info(con) + build_geom(con) + build_rules(con, df) + build_empty_geom(con) + build_color(con) + build_full(con) return df -def add_to_lake(): +def add_to_lake(con: duckdb.DuckDBPyConnection) -> None: """ Persists each cleaned zoning table (info, geom, rules, empty_geom, wide, colors) into the CLEANED schema in DuckLake. @@ -266,9 +248,9 @@ def add_to_lake(): ) -def main(): - clean() - add_to_lake() +def main(con: duckdb.DuckDBPyConnection): + clean(con) + add_to_lake(con) if __name__ == "__main__": diff --git a/backend/data_collection/acs5.py b/backend/data_collection/acs5.py index 6509fd1b..60ed069b 100644 --- a/backend/data_collection/acs5.py +++ b/backend/data_collection/acs5.py @@ -1,7 +1,7 @@ """ Fetch ACS 5-Year Data Profile tables (DP02-DP05) for Vermont Geographies: counties + county subdivisions + Vermont statewide + United States -Years: 2009-2024 +Years: 2009 - Latest published data Output: one wide CSV + parquet per table, plus tidy parquet per table Credit: Written largely by Claude, with some fine-tuning and troubleshooting by Fitz Koch @@ -13,6 +13,7 @@ import os import time +from datetime import datetime import pandas as pd import requests @@ -35,7 +36,9 @@ STORAGE_LOCATION = "Data/Census/ACS_5" ID_VARS = ["year", "geo_type", "table", "NAME", "state", "county"] -YEARS = range(2009, 2025) +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) # Default geos list in (label, for_clause, in_clause) format GEOS = [(k, *v) for k, v in ALL_GEOS.items()] @@ -82,89 +85,82 @@ def run_acs5_scrape(years: range = YEARS, geos: list = GEOS, append: bool = Fals for year in years: print(f"\n=== {year} ===") + for geo_label, for_clause, in_clause in geos: for table in TABLES: print(f" {table} / {geo_label}...") - df = fetch_table(year, table, for_clause, in_clause) + + df = fetch_table( + year, + table, + for_clause, + in_clause, + ) + if df is not None: df["geo_type"] = geo_label all_frames[table].append(df) + time.sleep(0.1) - # Save wide + tidy per table results = {} + + # Process each ACS profile table independently for table, frames in all_frames.items(): if not frames: print(f" No frames for {table}, skipping.") continue label = TABLES[table] - title = f"vt_acs5_{label}_data" - combined = pd.concat(frames, ignore_index=True, sort=False) + + combined = pd.concat( + frames, + ignore_index=True, + sort=False, + ) # Key columns to front front = [c for c in ID_VARS if c in combined.columns] rest = [c for c in combined.columns if c not in front] + combined = combined[front + rest] - combined.sort_values(["year", "geo_type", "NAME"], inplace=True) - combined.reset_index(drop=True, inplace=True) - wide_parquet_path = f"{STORAGE_LOCATION}/{title}.parquet" - # wide_csv_path = f"{STORAGE_LOCATION}/{title}.csv" + combined.sort_values( + ["year", "geo_type", "NAME"], + inplace=True, + ) - if append: - new_names = set(combined["year", "geo_type", "NAME"].unique()) - # --- Wide --- - try: - existing_wide = pd.read_parquet(wide_parquet_path) - existing_wide = existing_wide[~existing_wide["NAME"].isin(new_names)] - combined = pd.concat([existing_wide, combined], ignore_index=True) - combined.sort_values(["year", "geo_type", "NAME"], inplace=True) - combined.reset_index(drop=True, inplace=True) - print(f" Wide append: kept {len(existing_wide):,} existing rows.") - except FileNotFoundError: - pass - - # combined.to_csv(wide_csv_path, index=False) - # combined.to_parquet(wide_parquet_path, index=False) - # print(f"Saved wide: {title} ({len(combined):,} rows)") + combined.reset_index(drop=True, inplace=True) # Tidy: run per-year so column labels are year-accurate - # Tidy: run per-year so column labels are year-accurate - tidy_frames = [] + tidy_frames = [] - for year in sorted(combined["year"].unique()): - year_df = combined[combined["year"] == year] + for year in sorted(combined["year"].unique()): + year_df = combined[combined["year"] == year] - if year_df.empty: - continue + if year_df.empty: + continue - try: - tidy_year = tidy_census( - year_df, - year=year, - id_vars=ID_VARS, + try: + tidy_year = tidy_census( + year_df, + year=year, + id_vars=ID_VARS, + ) + + tidy_year["table"] = table + tidy_frames.append(tidy_year) + + except Exception as e: + print(f" SKIP tidy {year} / {table}: {e}") + + if tidy_frames: + tidy = pd.concat( + tidy_frames, + ignore_index=True, ) - tidy_year["table"] = table - tidy_frames.append(tidy_year) - - except Exception as e: - print(f" SKIP tidy {year} / {table}: {e}") - - if tidy_frames: - tidy = pd.concat(tidy_frames, ignore_index=True) - - label = TABLES[table] - results[f"acs5_{label.lower()}"] = tidy - # tidy_parquet_path = f"{STORAGE_LOCATION}/{title}_tidy.parquet" - # tidy_csv_path = f"{STORAGE_LOCATION}/{title}_tidy.csv" - - # No separate append needed for tidy: it's derived from the - # already-merged wide frame, so it naturally contains all geos. - # tidy.to_csv(tidy_csv_path, index=False) - # tidy.to_parquet(tidy_parquet_path, index=False) - # print(f"Saved tidy: {title}_tidy ({len(tidy):,} rows)") + results[f"acs5_{label.lower()}"] = tidy return results @@ -182,7 +178,6 @@ def merge_tidy_tables(): if tidy_frames: combined = pd.concat(tidy_frames, ignore_index=True) # combined.to_parquet(f"{STORAGE_LOCATION}/vt_acs5_combined_TIDY.parquet", index=False) - # print(f"Combined tidy saved: {len(combined):,} rows") return combined return diff --git a/backend/data_collection/ambulance.py b/backend/data_collection/ambulance.py index 4b3a03d2..d7f987b5 100644 --- a/backend/data_collection/ambulance.py +++ b/backend/data_collection/ambulance.py @@ -1,10 +1,10 @@ +from io import BytesIO + 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" # --------------------------------------------------------------------------- @@ -14,7 +14,7 @@ # Fetch ambulance data from goverment arcgis website (geojson files) -def fetch_service_areas() -> pd.DataFrame | None: +def fetch_ambulance_service_areas() -> pd.DataFrame | None: r = requests.get(AMBULANCE_SERVICE_AREA, timeout=30) r.raise_for_status() df = read_dataframe(BytesIO(r.content)) @@ -26,17 +26,13 @@ def fetch_service_areas() -> pd.DataFrame | None: # --------------------------------------------------------------------------- -def run_ambulance_scrape() -> None: +def collect() -> None: """ - Fetch Wastewater data and save as parquet files. + Fetch ambulance 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 - ) + df = fetch_ambulance_service_areas() + return df if __name__ == "__main__": - run_ambulance_scrape() + df = collect() diff --git a/backend/data_collection/base.py b/backend/data_collection/base.py index 6fb229ce..5e854beb 100644 --- a/backend/data_collection/base.py +++ b/backend/data_collection/base.py @@ -24,6 +24,7 @@ import os import time from dataclasses import dataclass +from datetime import datetime import pandas as pd import requests @@ -51,6 +52,8 @@ GEOS = [(k, *v) for k, v in ALL_GEOS.items()] +MAX_YEAR = datetime.now().year - 2 + # --------------------------------------------------------------------------- # Data structures # --------------------------------------------------------------------------- @@ -151,7 +154,7 @@ def run_acs_b_scrape( fetch_specs: dict[str, list[str]], var_groups: list[VarGroup], output_filename: str, - year: int = 2024, + year: int = MAX_YEAR, geos: list = GEOS, append: bool = False, ) -> None: diff --git a/backend/data_collection/demographics.py b/backend/data_collection/demographics.py index 72d63b19..c8fef05d 100644 --- a/backend/data_collection/demographics.py +++ b/backend/data_collection/demographics.py @@ -8,6 +8,8 @@ Output: vt_acs5_b_demographics_tidy.parquet """ +from datetime import datetime + import pandas as pd from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape @@ -27,7 +29,9 @@ ("75 Plus", range(23, 26), range(47, 50)), ] -YEARS = range(2009, 2025) +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) def _b01001_codes(male_r, female_r): @@ -116,7 +120,7 @@ def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: p = argparse.ArgumentParser(description="Scrape ACS B-table demographics data.") p.add_argument("--start-year", type=int, default=2009) - p.add_argument("--end-year", type=int, default=2024) + p.add_argument("--end-year", type=int, default=MAX_YEAR - 1) p.add_argument( "--geos", nargs="+", diff --git a/backend/data_collection/economic.py b/backend/data_collection/economic.py index 0f84a286..0e8cd6cf 100644 --- a/backend/data_collection/economic.py +++ b/backend/data_collection/economic.py @@ -1,6 +1,6 @@ """ Fetch ACS 5-Year economic data for Vermont: - B23025 – Employment Status (labor force participation, 16+) + B23025 – Employment Status (labor force participation, 16+ and unemployment) B23001 – Sex by Age by Employment Status (prime-age 25-54 LFP) B19013 – Median Household Income B19301 – Per Capita Income @@ -20,6 +20,8 @@ Output: vt_acs5_b_economic_tidy.parquet """ +from datetime import datetime + import pandas as pd from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape @@ -37,6 +39,7 @@ "B23001_125E", "B23001_132E", # female 25-54 ] + _PRIME_TOTAL = [ "B23001_024E", "B23001_031E", @@ -55,6 +58,12 @@ ["B23025_002E"], ["B23025_001E"], ), + VarGroup( + "Unemployment Rate", + SL, + ["B23025_005E"], + ["B23025_003E"], + ), VarGroup( "Prime-Age Labor Force Participation Rate (25-54)", SL, @@ -66,13 +75,20 @@ ] fetch_specs = { - "B23025": ["B23025_001E", "B23025_002E"], + "B23025": [ + "B23025_001E", + "B23025_002E", + "B23025_003E", + "B23025_005E", + ], "B23001": _PRIME_TOTAL + _PRIME_IN_LF, "B19013": ["B19013_001E"], "B19301": ["B19301_001E"], } -YEARS = range(2009, 2025) +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: @@ -100,7 +116,7 @@ def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: p = argparse.ArgumentParser(description="Scrape ACS B-table economic data.") p.add_argument("--start-year", type=int, default=2009) - p.add_argument("--end-year", type=int, default=2024) + p.add_argument("--end-year", type=int, default=MAX_YEAR - 1) p.add_argument( "--geos", nargs="+", diff --git a/backend/data_collection/education.py b/backend/data_collection/education.py index 076abf28..4ceb6b18 100644 --- a/backend/data_collection/education.py +++ b/backend/data_collection/education.py @@ -12,6 +12,8 @@ Output: vt_acs5_b_education_tidy.parquet """ +from datetime import datetime + import pandas as pd from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape @@ -39,7 +41,9 @@ "B15003": [TOTAL] + [f"B15003_{str(i).zfill(3)}E" for i in range(2, 26)], } -YEARS = range(2009, 2025) +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: @@ -67,7 +71,7 @@ def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: p = argparse.ArgumentParser(description="Scrape ACS B-table education data.") p.add_argument("--start-year", type=int, default=2009) - p.add_argument("--end-year", type=int, default=2024) + p.add_argument("--end-year", type=int, default=MAX_YEAR - 1) p.add_argument( "--geos", nargs="+", diff --git a/backend/data_collection/housing.py b/backend/data_collection/housing.py index 108fad99..d5784cd6 100644 --- a/backend/data_collection/housing.py +++ b/backend/data_collection/housing.py @@ -15,6 +15,8 @@ Output: vt_acs5_b_housing_tidy.parquet """ +from datetime import datetime + import pandas as pd from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape @@ -43,8 +45,9 @@ "B25077": ["B25077_001E"], } +MAX_YEAR = datetime.now().year - 1 -YEARS = range(2009, 2025) +YEARS = range(2009, MAX_YEAR) def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: @@ -72,7 +75,7 @@ def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: p = argparse.ArgumentParser(description="Scrape ACS B-table housing data.") p.add_argument("--start-year", type=int, default=2009) - p.add_argument("--end-year", type=int, default=2024) + p.add_argument("--end-year", type=int, default=MAX_YEAR - 1) p.add_argument( "--geos", nargs="+", diff --git a/backend/data_collection/qcew.py b/backend/data_collection/qcew.py index d34eb578..ef015ee3 100644 --- a/backend/data_collection/qcew.py +++ b/backend/data_collection/qcew.py @@ -18,6 +18,7 @@ """ import time +from datetime import datetime from io import StringIO from pathlib import Path @@ -88,7 +89,9 @@ BASE_URL = "https://data.bls.gov/cew/data/api/{year}/{q}/area/{fips}.csv" QUARTERS = [1, 2, 3, 4] -YEARS = range(2009, 2025) +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) # --------------------------------------------------------------------------- diff --git a/backend/lake_build.py b/backend/lake_build.py index 86585487..974c1781 100644 --- a/backend/lake_build.py +++ b/backend/lake_build.py @@ -12,37 +12,47 @@ LAKE_PATH = DATA_DIR / "lake" STORAGE_PATH = DATA_DIR / "lake.files" -DATA_DIR.mkdir(parents=True, exist_ok=True) -STORAGE_PATH.mkdir(parents=True, exist_ok=True) - -con = duckdb.connect() - -# Load extension -try: - con.execute("LOAD ducklake") -except duckdb.Error: - con.execute("INSTALL ducklake") - con.execute("LOAD ducklake") - -# Attach DuckLake catalog -con.execute( - f"""--sql - ATTACH '{LAKE_PATH.as_posix()}' - AS lake - ( - TYPE ducklake, - DATA_PATH '{STORAGE_PATH.as_posix()}', - OVERRIDE_DATA_PATH TRUE - ) + +def get_connection() -> duckdb.DuckDBPyConnection: """ -) + Create a connection to the DuckLake. + """ + DATA_DIR.mkdir(parents=True, exist_ok=True) + STORAGE_PATH.mkdir(parents=True, exist_ok=True) + + con = duckdb.connect() -# Create schemas in the lake catalog -con.execute("""CREATE SCHEMA IF NOT EXISTS lake.RAW""") -con.execute("""CREATE SCHEMA IF NOT EXISTS lake.CLEANED""") + for extension in ["ducklake", "spatial"]: + try: + con.execute(f"LOAD {extension}") + except duckdb.Error: + con.execute(f"INSTALL {extension}") + con.execute(f"LOAD {extension}") + con.execute( + f"""--sql + ATTACH '{LAKE_PATH.as_posix()}' + AS lake + ( + TYPE ducklake, + DATA_PATH '{STORAGE_PATH.as_posix()}', + OVERRIDE_DATA_PATH TRUE + ) + """ + ) -def insert_year(name: str, df: pd.DataFrame, years: Union[int, Iterable[int]]): + con.execute("CREATE SCHEMA IF NOT EXISTS lake.RAW") + con.execute("CREATE SCHEMA IF NOT EXISTS lake.CLEANED") + + return con + + +def insert_year( + name: str, + df: pd.DataFrame, + years: Union[int, Iterable[int]], + con: duckdb.DuckDBPyConnection | None = None, +): """ Insert or replace data for specific year(s) in a DuckLake table. """ @@ -58,6 +68,11 @@ def insert_year(name: str, df: pd.DataFrame, years: Union[int, Iterable[int]]): df = df.copy() df["geometry"] = df.geometry.to_wkb() + # If no lake connection, create one + own_connection = con is None + if own_connection: + con = get_connection() + con.register("tmp_df", df) try: @@ -65,73 +80,80 @@ def insert_year(name: str, df: pd.DataFrame, years: Union[int, Iterable[int]]): table_exists = ( con.execute( - """ - SELECT COUNT(*) - FROM information_schema.tables - WHERE table_catalog = 'lake' - AND table_schema = ? - AND table_name = ? - """, + """--sql + SELECT COUNT(*) + FROM information_schema.tables + WHERE table_catalog = 'lake' + AND table_schema = ? + AND table_name = ? + """, [schema, table], ).fetchone()[0] > 0 ) if not table_exists: - con.execute( - f""" - CREATE TABLE lake.{schema}.{table} - AS SELECT * FROM tmp_df - """ - ) + con.execute(f"CREATE TABLE lake.{schema}.{table} AS SELECT * FROM tmp_df") return - # Remove existing data for the target years + # Transaction prevents half-deleted/half-inserted errors + con.execute("BEGIN TRANSACTION") + con.execute( - f""" + f"""--sql DELETE FROM lake.{schema}.{table} WHERE year IN ({",".join("?" for _ in years_list)}) """, years_list, ) - # Insert the replacement data con.execute( - f""" + f"""--sql INSERT INTO lake.{schema}.{table} BY NAME SELECT * FROM tmp_df """ ) + con.execute("COMMIT") + + except Exception: + con.execute("ROLLBACK") + raise finally: con.unregister("tmp_df") + if own_connection: + con.close() -def replace_table(name: str, df: pd.DataFrame): +def replace_table( + name: str, + df: pd.DataFrame, + con: duckdb.DuckDBPyConnection | None = None, +): """ - Replace or create a table in the DuckLake catalog. - - Args: - name : str - Name of the destination table. - df : pd.DataFrame - DataFrame to write. + Replace an entire table in the DuckLake with the updated new one. """ if isinstance(df, gpd.GeoDataFrame): df = df.copy() - # Convert geometry objects to WKB bytes df["geometry"] = df.geometry.to_wkb() - con.register("tmp_df", df) + own_connection = con is None + if own_connection: + con = get_connection() - con.execute( - f"""--sql - CREATE OR REPLACE TABLE lake.{name} - AS - SELECT * - FROM tmp_df - """ - ) + # Parse schema correctly to avoid quoting bugs + schema, table = name.split(".", 1) if "." in name else ("RAW", name) + con.register("tmp_df", df) - con.unregister("tmp_df") + try: + con.execute( + f"""--sql + CREATE OR REPLACE TABLE lake.{schema}.{table} + AS SELECT * FROM tmp_df + """ + ) + finally: + con.unregister("tmp_df") + if own_connection: + con.close() diff --git a/backend/notebooks/zoning/runtime_test.qmd b/backend/notebooks/zoning/runtime_test.qmd index 2371e06a..55faedc6 100644 --- a/backend/notebooks/zoning/runtime_test.qmd +++ b/backend/notebooks/zoning/runtime_test.qmd @@ -152,15 +152,16 @@ def serve_dataframe(filters: dict | None = None): c.hex_color AS hex_color, c.rgba AS rgba, ST_AsGeoJSON(ST_Simplify(g.geom, 0.0001)) AS geometry - FROM zoning_info i - JOIN zoning_geom g USING (OBJECT_ID) - LEFT JOIN zoning_colors c ON c.district_type = i.District_Type + FROM VersoZoning_info i + JOIN VersoZoning_geom g USING (OBJECT_ID) + LEFT JOIN VersoZoning_colors c ON c.district_type = i.District_Type {where} """, ).df() return df -filter_1 = {"County": ["Chittenden"], "Jurisdiction" : ["Hinesburg"]} + +filter_1 = {"County": ["Chittenden"], "Jurisdiction": ["Hinesburg"]} serve_dataframe(filter_1) # print(build_where_query_from_filters(filter_1, FCOLS)) @@ -190,9 +191,9 @@ def serve_geojson(filters: dict | None = None): ) ) ) AS feature - FROM zoning_info i - JOIN zoning_geom g USING (OBJECT_ID) - LEFT JOIN zoning_colors c ON c.district_type = i.District_Type + FROM VersoZoning_info i + JOIN VersoZoning_geom g USING (OBJECT_ID) + LEFT JOIN VersoZoning_colors c ON c.district_type = i.District_Type {where} ) """).fetchone()[0] diff --git a/backend/query/__init__.py b/backend/query/__init__.py index c9822aeb..a47cb172 100644 --- a/backend/query/__init__.py +++ b/backend/query/__init__.py @@ -6,7 +6,7 @@ ) from query.cdc import dual_var_comparison, get_cdc_county_pca, single_var_geojson from query.core_functions import filter_options, filter_tree -from query.processed_db import DB +from query.production_db import get_db from query.wastewater import ( get_soil_suit_geojson, get_soil_suit_legend, diff --git a/backend/query/acs5.py b/backend/query/acs5.py index 252fd17d..ca409cc4 100644 --- a/backend/query/acs5.py +++ b/backend/query/acs5.py @@ -15,7 +15,9 @@ from api.models import FilterSource, RangeFilter from app_utils.sql_render import sql_filter_block from query.core_functions import filter_tree -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) sql_path = Path(__file__).resolve().parent / "sql" / "acs5" @@ -23,19 +25,84 @@ # Per-dataset FIXED filters, expressed as {column: [values]} and folded into the # FilterSource (these replace the old raw-SQL base_conditions) QUERY_CONFIG = { - "demographics": {"table": "acs5_b10_census", "fixed_filters": {}}, - "education": {"table": "acs5_b15003_education", "fixed_filters": {}}, - "housing": {"table": "acs5_b_housing", "fixed_filters": {}}, + "demographics": { + "table": "acs5_demographics_tidy", + "fixed_filters": {}, + "timeseries": { + "age_dependency_ratio": { + "table": "acs5Demographics_ageDependencyRatio_timeseries", + "fixed_filters": {}, + }, + "median_age": { + "table": "acs5Demographics_medianAge_timeseries", + "fixed_filters": {}, + }, + "historic_population": { + "table": "VCGI_historicPopulation_timeseries", + "fixed_filters": {}, + }, + }, + }, + "economics": { + "table": "acs5_economics_tidy", + "fixed_filters": {}, + "timeseries": { + "health_insurance": { + "table": "acs5Economics_healthInsurance_timeseries", + "fixed_filters": {}, + }, + "household_income": { + "table": "acs5Economics_medianHouseholdIncome_timeseries", + "fixed_filters": {}, + }, + "per_capita_income": { + "table": "acs5Economics_perCapitaIncome_timeseries", + "fixed_filters": {}, + }, + "median_earnings": { + "table": "acs5Economics_medianEarnings_timeseries", + "fixed_filters": {}, + }, + }, + }, "labor_force": { - "table": "acs5_b_economic", + "table": "acs5_economics_tidy", "fixed_filters": {"Section": ["Labor Force"]}, + "timeseries": {}, + }, + "income": { + "table": "acs5_economics_tidy", + "fixed_filters": {"Section": ["Income"]}, + "timeseries": {}, }, - "income": {"table": "acs5_b_economic", "fixed_filters": {"Section": ["Income"]}}, - "median_age": { - "table": "acs5_b10_census", - "fixed_filters": {"Variable": ["Median Age"]}, + "housing": { + "table": "acs5_housing_tidy", + "fixed_filters": {}, + "timeseries": { + "housing_units": { + "table": "acs5Housing_housingUnits_timeseries", + "fixed_filters": {}, + }, + "median_home_value": { + "table": "acs5Housing_medianHomeValue_timeseries", + "fixed_filters": {}, + }, + "vacancy_rates": { + "table": "acs5Housing_vacancyRates_timeseries", + "fixed_filters": {}, + }, + }, + }, + "education": { + "table": "acs5_education_tidy", + "fixed_filters": {}, + "timeseries": {}, + }, + "snapshot": { + "table": "acs5_snapshot_indicators_tidy", + "fixed_filters": {}, + "timeseries": {}, }, - "snapshot": {"table": "acs5_snapshot", "fixed_filters": {}}, } # frontend filter label -> database column. Location and the year range both @@ -81,7 +148,7 @@ def get_acs5_tidy(dataset: str, filters: dict | None = None) -> pd.DataFrame: result = DB.execute(sql, params).df() - if result is None: + if result.empty: logger.error( "ACS5 tidy query returned no rows for dataset: %s, filters: %s", dataset, @@ -92,21 +159,46 @@ def get_acs5_tidy(dataset: str, filters: dict | None = None) -> pd.DataFrame: return result -def get_unemployment_rate_ts(filters: dict | None = None) -> pd.DataFrame: - source = _acs5_source(table="acs5_unemployment_rate", filters=filters) +def get_acs5_timeseries( + category: str, + dataset: str, + filters: dict | None = None, +) -> pd.DataFrame: + try: + config = QUERY_CONFIG[category]["timeseries"][dataset] + except KeyError as e: + raise ValueError(f"Unknown ACS5 timeseries: {category}/{dataset}") from e + + source = _acs5_source( + table=config["table"], + filters=filters, + fixed_filters=config.get("fixed_filters"), + ) - sql, params = sql_filter_block(sql_path / "unemployment_rate.sql", [source]) + sql, params = sql_filter_block( + sql_path / "acs5_timeseries.sql", + [source], + ) result = DB.execute(sql, params).df() - if result is None or result.empty: - logger.error("Unemployment rate query returned no rows for filters=%s", filters) - raise ValueError("no results for unemployment_rate query") + if result.empty: + logger.error( + "ACS5 timeseries query returned no rows for category=%s, " + "dataset=%s, filters=%s", + category, + dataset, + filters, + ) + raise ValueError( + f"No results for timeseries: {category}/{dataset}, filters: {filters}" + ) return result -def get_median_earnings(filters: dict | None = None) -> pd.DataFrame: +# FIXME: Link to new database table name (broken for now) +def get_median_earnings_ts(filters: dict | None = None) -> pd.DataFrame: source = _acs5_source(table="acs5_median_earnings", filters=filters) sql, params = sql_filter_block(sql_path / "median_earnings.sql", [source]) @@ -120,19 +212,6 @@ def get_median_earnings(filters: dict | None = None) -> pd.DataFrame: return result -def get_snapshot(filters: dict | None = None) -> pd.DataFrame: - source = _acs5_source(table="snapshot", filters=filters) - - sql, params = sql_filter_block(sql_path / "snapshot.sql", [source]) - - result = DB.execute(sql, params).df() - - if result is None or result.empty: - logger.error("Snapshot query returned no rows for filters=%s", filters) - raise ValueError("no results for snapshot query") - - return result - - +# FIXME: Link to new database table name (broken for now) def get_acs5_filters(): return filter_tree(ACS5_FILTER_COLS, ACS5_TREE_LABELS, "acs5_info") diff --git a/backend/query/ambulance.py b/backend/query/ambulance.py index 7923a2e7..91296862 100644 --- a/backend/query/ambulance.py +++ b/backend/query/ambulance.py @@ -12,7 +12,9 @@ from api.models import FilterSource from app_utils.sql_render import sql_filter_block -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) sql_dir = Path(__file__).resolve().parent / "sql" / "ambulance" @@ -29,7 +31,7 @@ def get_ambulance_geojson(sources: list[FilterSource]): def get_ambulance_legend(): result = DB.execute( - "SELECT json_group_array(to_json(ambulance_ambulance_colors)) FROM ambulance_ambulance_colors;" + "SELECT json_group_array(to_json(VCGI_ambulanceService_colors)) FROM VCGI_ambulanceService_colors;" ).fetchone() if result is None: logger.error("color query returned no rows for the colors dataset") diff --git a/backend/query/cdc.py b/backend/query/cdc.py index d00454dd..d8f779c1 100644 --- a/backend/query/cdc.py +++ b/backend/query/cdc.py @@ -17,7 +17,9 @@ from api.models import FilterSource from app_utils.sql_render import compile_where, sql_filter_block -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) sql_dir = Path(__file__).resolve().parent / "sql" / "cdc" @@ -46,7 +48,7 @@ def single_var_geojson(sources: list[FilterSource]): "geometry": json.loads(r.geometry), "properties": { "rgba_color": RAMP[int(r.bin)], - "tooltip": {"__title__": r.Measure, "value": r.Data_Value}, + "tooltip": {"__title__": r.measure, "value": r.data_value}, }, } ) @@ -54,10 +56,10 @@ def single_var_geojson(sources: list[FilterSource]): def widen_dual_var(df, measures): - cols = ["LocationID", "geometry", "Data_Value", "bin", "natl_pct", "CountyName"] - m1 = df[df.Measure == measures[0]][[c for c in cols if c in df.columns]] - m2 = df[df.Measure == measures[1]][["LocationID", "Data_Value", "bin"]] - wide = m1.merge(m2, on="LocationID", suffixes=("_1", "_2")) + cols = ["locationid", "geometry", "data_value", "bin", "natl_pct", "CountyName"] + m1 = df[df.measure == measures[0]][[c for c in cols if c in df.columns]] + m2 = df[df.measure == measures[1]][["locationid", "data_value", "bin"]] + wide = m1.merge(m2, on="locationid", suffixes=("_1", "_2")) return wide @@ -81,13 +83,13 @@ def _measure_cutpoints(measures: list[str]) -> tuple[list[float], list[float]]: """Bin edges for each measure from the precomputed cdc_edges table.""" params: list = [] where_string = compile_where({"Measure": measures}, params) - sql = f"SELECT * FROM cdc_county_edges {where_string}" + sql = f"SELECT * FROM cdc_edges_county {where_string}" edges = DB.execute(sql, params).df() edges_x = ( - edges[edges["Measure"] == measures[0]].drop(columns="Measure").iloc[0].tolist() + edges[edges["measure"] == measures[0]].drop(columns="measure").iloc[0].tolist() ) edges_y = ( - edges[edges["Measure"] == measures[1]].drop(columns="Measure").iloc[0].tolist() + edges[edges["measure"] == measures[1]].drop(columns="measure").iloc[0].tolist() ) return edges_x, edges_y @@ -106,12 +108,12 @@ def dual_var_comparison( # Both measures ride in one merged FilterSource so the shared places.sql # template serves the single- and dual-variable cases alike. - table = "cdc_county_places" if geoLevel == "county_places" else "cdc_tract_places" + table = "cdc_places_county" if geoLevel == "county_places" else "cdc_places_tract" merged = FilterSource(filter_table=table, filters={"Measure": measures}) sql_path = sql_dir / f"{geoLevel}.sql" + sql, params = sql_filter_block(sql_path, [merged]) df = DB.execute(sql, params).df() - print(df.head()) df = widen_dual_var(df, measures) @@ -122,8 +124,8 @@ def dual_var_comparison( tooltip = { "__title__": "Variable Comparison", # "County": r.CountyName, - f"{measures[0]}": r.Data_Value_1, - f"{measures[1]}": r.Data_Value_2, + f"{measures[0]}": r.data_value_1, + f"{measures[1]}": r.data_value_2, "National Percentage": r.natl_pct, } ## add in County Name if we're in county space. @@ -150,12 +152,19 @@ def dual_var_comparison( def get_cdc_county_pca(): - df = DB.execute("""--sql - SELECT i.LocationID, ROUND(i.pca_score, 2) AS "Health Burden", c.CountyName - FROM cdc_countyPcaData AS i - LEFT JOIN vermont_counties AS c ON i.LocationID = c.CountyFIPS - """).df() + df = DB.execute( + """--sql + SELECT + i.LocationID, + ROUND(i.pca_score, 2) AS "Health Burden", + c.CountyName + FROM cdc_pca_county AS i + LEFT JOIN vt_county_lines_geom AS c + ON i.LocationID = c.CountyFIPS + """ + ).df() + df["CountyName"] = df["CountyName"].str.title() df = df.sort_values(by="CountyName") - ret = df[["CountyName", "Health Burden"]].to_dict(orient="records") - return ret + + return df[["CountyName", "Health Burden"]].to_dict(orient="records") diff --git a/backend/query/core_functions.py b/backend/query/core_functions.py index fa568f96..6d3c3c03 100644 --- a/backend/query/core_functions.py +++ b/backend/query/core_functions.py @@ -13,7 +13,9 @@ import logging from api.models import FilterResponse, RangeDescriptor -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) diff --git a/backend/query/processed_db.py b/backend/query/processed_db.py deleted file mode 100644 index af6c8793..00000000 --- a/backend/query/processed_db.py +++ /dev/null @@ -1,56 +0,0 @@ -""" -**Author**: - Fitz Koch -**Created**: - 2026-06-01 -**Description**: - Pull in all the parquet files into active api. - Uses pre-run ETL from consolidate.py in the build section. -""" - -import logging -import os -from pathlib import Path - -import duckdb - -logger = logging.getLogger(__name__) -proc_dir = Path(__file__).resolve().parent.parent / "Data" / "_Processed" -DATA_DIR = Path(os.environ.get("DATA_DIR", proc_dir)) - - -def _load_spatial(con: duckdb.DuckDBPyConnection) -> None: - """Load the spatial extension, installing it first if necessary.""" - try: - con.execute("LOAD spatial") - except Exception: - con.execute("INSTALL spatial") - con.execute("LOAD spatial") - - -# def _build() -> duckdb.DuckDBPyConnection: -# con = duckdb.connect(":memory:") -# _load_spatial(con) - -# parquets = sorted(proc_dir.rglob("*.parquet")) -# if not parquets: -# logger.warning("No parquet files found under %s", proc_dir) - -# for path in parquets: -# name = f"{path.parent.name}_{path.stem}" -# con.execute(f"""--sql -# CREATE TABLE "{name}" AS SELECT * FROM read_parquet('{path}') -# """) -# logger.info("Loaded table %s from %s", name, path) -# return con - - -def _build() -> duckdb.DuckDBPyConnection: - print(DATA_DIR) - path = Path(proc_dir / "all_data.duckdb") - con = duckdb.connect(path, read_only=True) - _load_spatial(con) - return con - - -DB = _build() diff --git a/backend/query/production_db.py b/backend/query/production_db.py new file mode 100644 index 00000000..4e5105d5 --- /dev/null +++ b/backend/query/production_db.py @@ -0,0 +1,56 @@ +""" +**Author**: + Fitz Koch +**Created**: + 2026-06-01 +**Updated**: + 2026-09-01 +**Description**: + Provides a lazy connection to the finalized database, + `warehouse.duckdb`, which is derived from the CLEANED DuckLake tables. +""" + +import logging +import os +from pathlib import Path + +import duckdb + +logger = logging.getLogger(__name__) + +BACKEND_DIR = Path(__file__).resolve().parents[1] +DATA_DIR = Path(os.environ.get("DATA_DIR", BACKEND_DIR / "Data")) +WAREHOUSE_PATH = DATA_DIR / "warehouse.duckdb" + + +def _load_spatial(con: duckdb.DuckDBPyConnection) -> None: + """Load the spatial extension.""" + try: + con.execute("LOAD spatial") + except duckdb.Error as exc: + raise RuntimeError("The DuckDB spatial extension could not be loaded.") from exc + + +def get_db() -> duckdb.DuckDBPyConnection: + """ + Open a read-only connection to the production warehouse. + + The warehouse is generated by the ETL process. If it does not exist, + run `just etl` before starting the API. + """ + if not WAREHOUSE_PATH.exists(): + raise RuntimeError( + f"Production database not found at {WAREHOUSE_PATH}.\n" + "Run `just etl` first to build the production database." + ) + + try: + con = duckdb.connect(WAREHOUSE_PATH, read_only=True) + _load_spatial(con) + return con + + except Exception as exc: + raise RuntimeError( + f"Could not open the production database at {WAREHOUSE_PATH}.\n" + "Try running `just etl` to rebuild the production database." + ) from exc diff --git a/backend/query/sql/acs5/acs5_timeseries.sql b/backend/query/sql/acs5/acs5_timeseries.sql new file mode 100644 index 00000000..1b0fda1d --- /dev/null +++ b/backend/query/sql/acs5/acs5_timeseries.sql @@ -0,0 +1,4 @@ +SELECT * +FROM {{ table }} +{{ where_string }} +ORDER BY year \ No newline at end of file diff --git a/backend/query/sql/acs5/median_earnings.sql b/backend/query/sql/acs5/median_earnings.sql deleted file mode 100644 index 0ea7e7f4..00000000 --- a/backend/query/sql/acs5/median_earnings.sql +++ /dev/null @@ -1,9 +0,0 @@ -SELECT - year, - NAME, - Value, - -- quoted: the case change from `variable` is intentional (frontend key) - variable AS "Variable" -- noqa: RF06 -FROM acs5_median_earnings -{{ where_string }} -ORDER BY year diff --git a/backend/query/sql/acs5/snapshot.sql b/backend/query/sql/acs5/snapshot.sql deleted file mode 100644 index c300f449..00000000 --- a/backend/query/sql/acs5/snapshot.sql +++ /dev/null @@ -1,8 +0,0 @@ -SELECT - year, - NAME, - Value, - Variable -FROM acs5_snapshot -{{ where_string }} -ORDER BY year diff --git a/backend/query/sql/acs5/unemployment_rate.sql b/backend/query/sql/acs5/unemployment_rate.sql deleted file mode 100644 index 5413bcdd..00000000 --- a/backend/query/sql/acs5/unemployment_rate.sql +++ /dev/null @@ -1,8 +0,0 @@ -SELECT - year, - NAME, - Value, - Value AS Percent -FROM acs5_unemployment_rate -{{ where_string }} -ORDER BY year diff --git a/backend/query/sql/ambulance/ambulance_geo_query.sql b/backend/query/sql/ambulance/ambulance_geo_query.sql index 0d1338fb..4606e826 100644 --- a/backend/query/sql/ambulance/ambulance_geo_query.sql +++ b/backend/query/sql/ambulance/ambulance_geo_query.sql @@ -8,7 +8,9 @@ FROM ( SELECT json_object( 'type', 'Feature', - 'geometry', ST_AsGeoJSON(ST_Simplify(g.geometry, 0.0001))::JSON, + 'geometry', ST_AsGeoJSON( + ST_Simplify(ST_GeomFromWKB(g.geometry), 0.0001) + )::JSON, 'properties', json_object( 'Certification Level', i.Cert_Level, 'Acres', ROUND(g.Shape__Area, 2), @@ -22,9 +24,9 @@ FROM ( ) ) ) AS feature - FROM ambulance_ambulance_info AS i - INNER JOIN ambulance_ambulance_geom AS g USING (OBJECTID) - LEFT JOIN ambulance_ambulance_colors AS c + FROM VCGI_ambulanceService_info AS i + INNER JOIN VCGI_ambulanceService_geom AS g USING (OBJECTID) + LEFT JOIN VCGI_ambulanceService_colors AS c ON i.Cert_Level = c.certification_level {{ join_filter_block }} ) AS features \ No newline at end of file diff --git a/backend/query/sql/cdc/county_places.sql b/backend/query/sql/cdc/county_places.sql index da1cdb03..2debc1de 100644 --- a/backend/query/sql/cdc/county_places.sql +++ b/backend/query/sql/cdc/county_places.sql @@ -6,7 +6,9 @@ SELECT ROUND(p.natl_pct * 100, 2) AS natl_pct, c.CountyFIPS, c.CountyName, - ST_ASGEOJSON(c.geom) AS geometry -FROM cdc_county_places AS p -LEFT JOIN vermont_counties AS c ON p.LocationID = c.CountyFIPS -{{ where_string }} + ST_ASGEOJSON(ST_GeomFromWKB(c.geometry)) AS geometry +FROM cdc_places_county AS p +LEFT JOIN vt_county_lines_geom AS c + ON p.LocationID = c.CountyFIPS + +{{ where_string }} \ No newline at end of file diff --git a/backend/query/sql/cdc/tract_places.sql b/backend/query/sql/cdc/tract_places.sql index 7ff166f2..1273e1f8 100644 --- a/backend/query/sql/cdc/tract_places.sql +++ b/backend/query/sql/cdc/tract_places.sql @@ -4,8 +4,10 @@ SELECT p.Data_Value, p.bin, ROUND(p.natl_pct * 100, 2) AS natl_pct, - ST_ASGEOJSON(c.geometry) AS geometry, + ST_ASGEOJSON(ST_GeomFromWKB(c.geometry)) AS geometry, c.name -FROM cdc_tract_places AS p -LEFT JOIN vermont_tracts AS c ON p.LocationID = c.LocationID -{{ where_string }} +FROM cdc_places_tract AS p +LEFT JOIN vt_tract_lines_geom AS c + ON p.LocationID = c.LocationID + +{{ where_string }} \ No newline at end of file diff --git a/backend/query/sql/wastewater/service_area_geo_query.sql b/backend/query/sql/wastewater/service_area_geo_query.sql index bb47f2e7..aca3f2e7 100644 --- a/backend/query/sql/wastewater/service_area_geo_query.sql +++ b/backend/query/sql/wastewater/service_area_geo_query.sql @@ -20,7 +20,7 @@ FROM ( ) ) ) AS feature - FROM service_areas_service_area_info AS i - INNER JOIN service_areas_service_area_geom AS g USING (ID) + FROM VersoWastewater_serviceAreas_info AS i + INNER JOIN VersoWastewater_serviceAreas_geom AS g USING (Area_ID) {{ join_filter_block }} ) AS features diff --git a/backend/query/sql/wastewater/soil_suitability_geo_query.sql b/backend/query/sql/wastewater/soil_suitability_geo_query.sql index 80c10ff3..96a937f5 100644 --- a/backend/query/sql/wastewater/soil_suitability_geo_query.sql +++ b/backend/query/sql/wastewater/soil_suitability_geo_query.sql @@ -1,4 +1,3 @@ -{{ cte_filter_block }} SELECT json_object( 'type', 'FeatureCollection', @@ -8,7 +7,10 @@ FROM ( SELECT json_object( 'type', 'Feature', - 'geometry', ST_AsGeoJSON(ST_Simplify(g.geom, 0.0001))::JSON, + 'geometry', + ST_AsGeoJSON( + ST_Simplify(g.geometry, 0.0001) + )::JSON, 'properties', json_object( 'Suitability', i.Suitability, 'Acres', ROUND(i.Acres, 2), @@ -21,9 +23,10 @@ FROM ( ) ) ) AS feature - FROM soil_suitability_info_soil_suit AS i - INNER JOIN soil_suitability_geom_soil_suit AS g USING (ID) - LEFT JOIN soil_suitability_soil_suitability_colors AS c + FROM VersoWastewater_soilSuitability_info AS i + INNER JOIN VersoWastewater_soilSuitability_geom AS g + ON i.OGC_FID = g.OGC_FID + LEFT JOIN VersoWastewater_soilSuitability_colors AS c ON i.Suitability = c.soil_suitability - {{ join_filter_block }} -) AS features + {{ where_string }} +) AS features; \ No newline at end of file diff --git a/backend/query/sql/wastewater/waste_treatment_geo_query.sql b/backend/query/sql/wastewater/waste_treatment_geo_query.sql index a95d1ccf..9ad07833 100644 --- a/backend/query/sql/wastewater/waste_treatment_geo_query.sql +++ b/backend/query/sql/wastewater/waste_treatment_geo_query.sql @@ -23,7 +23,7 @@ FROM ( ) ) ) AS feature - FROM treatment_facilities_treatment_facility_info AS i - INNER JOIN treatment_facilities_treatment_facility_geom AS g USING (ID) + FROM VersoWastewater_treatmentFacilities_info AS i + INNER JOIN VersoWastewater_treatmentFacilities_geom AS g USING (Facility_ID) {{ join_filter_block }} ) AS features diff --git a/backend/query/sql/wastewater/waste_treatment_permit_table.sql b/backend/query/sql/wastewater/waste_treatment_permit_table.sql index 51de5228..d2746763 100644 --- a/backend/query/sql/wastewater/waste_treatment_permit_table.sql +++ b/backend/query/sql/wastewater/waste_treatment_permit_table.sql @@ -5,5 +5,5 @@ SELECT p.NPDESPermitID AS "NPDES Permit ID", p.PermitLink AS "Permit Link", p.PermitteeName AS "Permittee Name" -FROM treatment_facilities_treatment_facility_info AS i -INNER JOIN treatment_facilities_treatment_facility_permit_info AS p USING (ID) +FROM VersoWastewater_treatmentFacilities_info AS i +INNER JOIN VersoWastewater_treatmentFacilitiesPermits_info AS p USING (ID) diff --git a/backend/query/sql/zoning/agg_info_table.sql b/backend/query/sql/zoning/agg_info_table.sql index 25d241c4..357b2a5f 100644 --- a/backend/query/sql/zoning/agg_info_table.sql +++ b/backend/query/sql/zoning/agg_info_table.sql @@ -3,7 +3,7 @@ SELECT i.District_Type AS "District Type", SUM(i.Acres) AS Acres, ANY_VALUE(c.hex_color) AS hex_color -FROM zoning_info AS i -LEFT JOIN zoning_colors AS c ON i.District_Type = c.district_type +FROM VersoZoning_info AS i +LEFT JOIN VersoZoning_colors AS c ON i.District_Type = c.district_type {{ join_filter_block }} GROUP BY i.District_Type diff --git a/backend/query/sql/zoning/agg_rules_table.sql b/backend/query/sql/zoning/agg_rules_table.sql index a997d004..be1f659c 100644 --- a/backend/query/sql/zoning/agg_rules_table.sql +++ b/backend/query/sql/zoning/agg_rules_table.sql @@ -3,8 +3,8 @@ SELECT r.use_type, r.val, SUM(i.Acres) AS Acres -FROM zoning_rules AS r -INNER JOIN zoning_info AS i USING (OBJECT_ID) +FROM VersoZoning_rules AS r +INNER JOIN VersoZoning_info AS i USING (OBJECT_ID) {{ join_filter_block }} WHERE r.rule = 'Allowance' diff --git a/backend/query/sql/zoning/geo_query.sql b/backend/query/sql/zoning/geo_query.sql index e1daa13c..fa9df795 100644 --- a/backend/query/sql/zoning/geo_query.sql +++ b/backend/query/sql/zoning/geo_query.sql @@ -8,10 +8,10 @@ filtered AS ( i.Municipal_Name, i.District_Name, c.rgba, - g.geom - FROM zoning_info AS i - INNER JOIN zoning_geom AS g USING (OBJECT_ID) - LEFT JOIN zoning_colors AS c ON i.District_Type = c.district_type + g.geometry + FROM VersoZoning_info AS i + INNER JOIN VersoZoning_geom AS g USING (OBJECT_ID) + LEFT JOIN VersoZoning_colors AS c ON i.District_Type = c.district_type {{ join_filter_block }} ), @@ -19,7 +19,7 @@ features AS ( SELECT JSON_OBJECT( 'type', 'Feature', - 'geometry', ST_ASGEOJSON(ST_SIMPLIFY(geom, 0.0001))::JSON, + 'geometry', ST_ASGEOJSON(ST_SIMPLIFY(geometry, 0.0001))::JSON, 'properties', JSON_OBJECT( 'District Type', District_Type, 'Acres', Acres, @@ -40,7 +40,7 @@ features AS ( matched_area AS ( SELECT County, - ST_Area_Spheroid(ST_Union_Agg(geom)) / 4046.8564224 AS matched_acres + ST_Area_Spheroid(ST_Union_Agg(geometry)) / 4046.8564224 AS matched_acres FROM filtered GROUP BY County ), @@ -48,9 +48,9 @@ matched_area AS ( county_area AS ( SELECT i.County, - ST_Area_Spheroid(ST_Union_Agg(g.geom)) / 4046.8564224 AS total_acres - FROM zoning_info AS i - INNER JOIN zoning_geom AS g USING (OBJECT_ID) + ST_Area_Spheroid(ST_Union_Agg(g.geometry)) / 4046.8564224 AS total_acres + FROM VersoZoning_info AS i + INNER JOIN VersoZoning_geom AS g USING (OBJECT_ID) GROUP BY i.County ), diff --git a/backend/query/sql/zoning/info_table.sql b/backend/query/sql/zoning/info_table.sql index fa7b75aa..04e3f7b6 100644 --- a/backend/query/sql/zoning/info_table.sql +++ b/backend/query/sql/zoning/info_table.sql @@ -5,6 +5,6 @@ SELECT i.District_Type AS "District Type", ROUND(i.Acres, 2) AS Acres, c.hex_color -FROM zoning_info AS i -LEFT JOIN zoning_colors AS c ON i.District_Type = c.district_type +FROM VersoZoning_info AS i +LEFT JOIN VersoZoning_colors AS c ON i.District_Type = c.district_type {{ join_filter_block }} diff --git a/backend/query/sql/zoning/rules.sql b/backend/query/sql/zoning/rules.sql index 607676b5..58433d6d 100644 --- a/backend/query/sql/zoning/rules.sql +++ b/backend/query/sql/zoning/rules.sql @@ -21,8 +21,8 @@ FROM ( ) ) ) AS feature - FROM zoning_info AS i - INNER JOIN zoning_geom AS g USING (OBJECT_ID) - LEFT JOIN zoning_colors AS c ON i.District_Type = c.district_type + FROM VersoZoning_info AS i + INNER JOIN VersoZoning_geom AS g USING (OBJECT_ID) + LEFT JOIN VersoZoning_colors AS c ON i.District_Type = c.district_type {{ join_filter_block }} ) AS features diff --git a/backend/query/sql/zoning/rules_table.sql b/backend/query/sql/zoning/rules_table.sql index 86a5c843..e66a250b 100644 --- a/backend/query/sql/zoning/rules_table.sql +++ b/backend/query/sql/zoning/rules_table.sql @@ -4,5 +4,5 @@ SELECT r.use_type, r.rule, r.val -FROM zoning_rules AS r +FROM VersoZoning_rules AS r {{ join_filter_block }} diff --git a/backend/query/sql/zoning/unzoned.sql b/backend/query/sql/zoning/unzoned.sql index 23b62ea9..ec917b5e 100644 --- a/backend/query/sql/zoning/unzoned.sql +++ b/backend/query/sql/zoning/unzoned.sql @@ -13,7 +13,12 @@ FROM ( SELECT JSON_OBJECT( 'type', 'Feature', - 'geometry', ST_ASGEOJSON(ST_SIMPLIFY(geom, 0.0001))::JSON, + 'geometry', ST_ASGEOJSON( + ST_SIMPLIFY( + ST_GEOMFROMWKB(geom), + 0.0001::DOUBLE + ) + )::JSON, 'properties', JSON_OBJECT( 'rgba_color', JSON_ARRAY(170, 170, 170, 160), 'tooltip', JSON_OBJECT( @@ -22,5 +27,5 @@ FROM ( ) ) ) AS feature - FROM zoning_empty_geom + FROM VersoZoning_empty_geom ) AS features; diff --git a/backend/query/wastewater.py b/backend/query/wastewater.py index 5362d8d5..778ec7e5 100644 --- a/backend/query/wastewater.py +++ b/backend/query/wastewater.py @@ -4,7 +4,7 @@ **Created**: 2026-07-06 **Description**: - Functions for serving wastewater data to the API from the parquet files. + Functions for serving wastewater data to the API from the database tables. """ import logging @@ -14,7 +14,9 @@ from api.models import FilterSource from app_utils.sql_render import sql_filter_block -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) sql_dir = Path(__file__).resolve().parent / "sql" / "wastewater" @@ -32,6 +34,7 @@ def get_waste_service_areas_geojson(sources: list[FilterSource]): def get_waste_treatment_facility_geojson(sources: list[FilterSource]): sql, params = sql_filter_block(sql_dir / "waste_treatment_geo_query.sql", sources) result = DB.execute(sql, params).fetchone() + print(f"TREATMENT FACILITY GEOM RESULT: {result}") if result is None: logger.error("geo query returned no rows for filters: %s", sources) raise ValueError(f"no results for filters: {sources}") @@ -48,7 +51,11 @@ def get_waste_treatment_facility_permits(sources: list[FilterSource]) -> pd.Data def get_soil_suit_geojson(sources: list[FilterSource]): - sql, params = sql_filter_block(sql_dir / "soil_suitability_geo_query.sql", sources) + sql, params = sql_filter_block( + sql_dir / "soil_suitability_geo_query.sql", + sources, + ) + result = DB.execute(sql, params).fetchone() if result is None: logger.error("geo query returned no rows for filters: %s", sources) diff --git a/backend/query/zoning.py b/backend/query/zoning.py index 68f222d9..d31b75b2 100644 --- a/backend/query/zoning.py +++ b/backend/query/zoning.py @@ -3,8 +3,10 @@ Fitz Koch **Created**: 2026-06-01 +**Updated**: + 2026-08-20 **Description**: - Functions for serving zoning_info data to the API from the parquet files. + Functions for serving VersoZoning_info data to the API from the database. """ import logging @@ -14,7 +16,9 @@ from api.models import FilterSource from app_utils.sql_render import render_sql, sql_filter_block -from query.processed_db import DB +from query.production_db import get_db + +DB = get_db() logger = logging.getLogger(__name__) sql_dir = Path(__file__).resolve().parent / "sql" / "zoning" diff --git a/backend/run_data_cleaning.py b/backend/run_data_cleaning.py index a7f81058..87527d43 100644 --- a/backend/run_data_cleaning.py +++ b/backend/run_data_cleaning.py @@ -2,11 +2,12 @@ from importlib import import_module import data_cleaning as cleaning +from lake_build import get_connection def get_cleaners(): """ - Return all cleaning scripts in the backend/data_cleaning folder + Return all cleaning scripts in the backend/data_cleaning folder. """ cleaners = [] @@ -14,36 +15,45 @@ def get_cleaners(): if ispkg: continue - # Skip private/helper modules - if module_name.startswith("_"): + # Don't accidentally import the master runner itself. + if module_name == "run_cleaning": continue module = import_module(f"data_cleaning.{module_name}") - # Only include modules that expose a main() function if hasattr(module, "main"): cleaners.append(module) return cleaners -# def create_duckdb_version(): -# tables = con.execute( -# """--sql -# SELECT table_name -# FROM duckdb_tables -# WHERE database_name = 'lake' -# AND schema_name = 'CLEANED' -# """).fetchall() +def run_master_clean(): + failed = [] -# for table in tables + con = get_connection() + try: + for cleaner in get_cleaners(): + name = cleaner.__name__.split(".")[-1] + print(f"Running {name}...") -def run_master_clean(): - for cleaner in get_cleaners(): - print(f"Running {cleaner.__name__.split('.')[-1]}...") - cleaner.main() - print(f"Completed {cleaner.__name__.split('.')[-1]}") + try: + cleaner.main(con) + print(f"Completed {name}") + + except Exception as e: + failed.append(name) + print(f"FAILED {name}: {e}") + + finally: + con.close() + + print("\nCleaning ETL process completed.") + + if failed: + print(f"Failed cleaners: {', '.join(failed)}") + else: + print("All cleaners completed successfully.") def main(): diff --git a/backend/run_data_collection.py b/backend/run_data_collection.py index f5f03df8..76301873 100644 --- a/backend/run_data_collection.py +++ b/backend/run_data_collection.py @@ -1,20 +1,9 @@ -""" -**Author**: - Ian Sargent -**Created**: - 2026-07-10 -** Updated**: - 2026-08-10 -**Description**: - This is the master orchestrating data scraping script. - Running this document will call each individual category scraper - and populate tables into the DuckLake's RAW schema. -""" - import argparse +from datetime import datetime from data_collection import ( acs5, + ambulance, cdc, demographics, economic, @@ -27,13 +16,19 @@ wastewater, zoning, ) -from lake_build import insert_year, replace_table +from lake_build import get_connection, insert_year, replace_table -# Datasets WITH year columns (longitudinal) -YEARLY_SCRAPERS = [acs5, demographics, economic, education, housing, qcew] +YEARLY_SCRAPERS = [ + acs5, + demographics, + economic, + education, + housing, + qcew, +] -# Datasets WITHOUT year columns (static) STATIC_SCRAPERS = [ + ambulance, cdc, fips, flood, @@ -42,58 +37,96 @@ zoning, ] -YEARS = range(2009, 2025) +MAX_YEAR = datetime.now().year - 1 -def run_scraper(scraper, yearly: bool = False, years: range = YEARS): +def run_scraper( + scraper, + con, + yearly: bool = False, + years: range | None = None, +): + """Run a scraper and write its output to DuckLake.""" name = scraper.__name__.split(".")[-1] - try: - print(f"Running {name}...") + print(f"Running {name}...") - if yearly: - outputs = scraper.collect(years) - else: - outputs = scraper.collect() + if yearly: + if years is None: + raise ValueError(f"No years provided for yearly scraper {name}.") + outputs = scraper.collect(years) + else: + outputs = scraper.collect() - if not isinstance(outputs, dict): - outputs = {name: outputs} + if not isinstance(outputs, dict): + outputs = {name: outputs} - for table_name, df in outputs.items(): - full_name = f"RAW.{table_name}" - print(f"Loading {full_name}") - # If the dataset is longitudinal, replace or append that year's data - if yearly: - insert_year(full_name, df, years) - # If a static dataset, replace the whole table - else: - replace_table(full_name, df) + for table_name, df in outputs.items(): + full_name = f"RAW.{table_name}" + print(f"Loading {full_name}") - print(f"Completed {name}") + if yearly: + insert_year(full_name, df, years, con=con) + else: + replace_table(full_name, df, con=con) - except Exception as e: - print(f"Failed to write {name}: {e}") - raise + print(f"Completed {name}") -def run_master_scrape(start_year: int = 2009, end_year: int = 2024): - for scraper in YEARLY_SCRAPERS: - run_scraper(scraper, yearly=True, years=range(start_year, end_year + 1)) +def run_master_scrape( + start_year: int = 2009, + end_year: int = MAX_YEAR, +): + """Run all data collection scrapers.""" + years = range(start_year, end_year + 1) + con = get_connection() + failed = [] - for scraper in STATIC_SCRAPERS: - run_scraper(scraper, yearly=False) + try: + for scraper in YEARLY_SCRAPERS: + name = scraper.__name__.split(".")[-1] + try: + run_scraper(scraper, con=con, yearly=True, years=years) + except Exception as e: + failed.append(name) + print(f"FAILED {name}: {e}") + + for scraper in STATIC_SCRAPERS: + name = scraper.__name__.split(".")[-1] + try: + run_scraper(scraper, con=con, yearly=False) + except Exception as e: + failed.append(name) + print(f"FAILED {name}: {e}") + + finally: + con.close() + + print("\nData collection process completed.") + if failed: + print(f"Failed scrapers: {', '.join(failed)}") + else: + print("All scrapers completed successfully.") def main(): - # Accepts the year argument from justfile for collection + """Run the master scraper from the command line.""" parser = argparse.ArgumentParser() parser.add_argument("start_year", type=int) parser.add_argument("end_year", type=int) args = parser.parse_args() + if args.start_year > args.end_year: + raise ValueError( + f"start_year ({args.start_year}) cannot be greater than end_year ({args.end_year})." + ) + print(f"Collecting data from {args.start_year} to {args.end_year}") - run_master_scrape(args.start_year, args.end_year) + run_master_scrape( + start_year=args.start_year, + end_year=args.end_year, + ) if __name__ == "__main__": diff --git a/backend/tests/test_lake.py b/backend/tests/test_lake.py index 8a45e7f3..5fadbd5a 100644 --- a/backend/tests/test_lake.py +++ b/backend/tests/test_lake.py @@ -5,7 +5,9 @@ python -m tests.test_lake """ -from lake_build import con +from lake_build import get_connection + +con = get_connection() def inspect_schema(schema: str) -> None: diff --git a/backend/tests/test_sql_render.py b/backend/tests/test_sql_render.py index a2741753..512d5b0b 100644 --- a/backend/tests/test_sql_render.py +++ b/backend/tests/test_sql_render.py @@ -29,14 +29,14 @@ # --------------------------------------------------------------------------- WHERE_SOURCE = FilterSource( - filter_table="acs5_b10_census", + filter_table="acs5_demographics_tidy", filters={ "NAME": ["Vergennes", "Addison town"], "year": RangeFilter(min=2015, max=2020), }, ) CTE_SOURCE = FilterSource( - filter_table="zoning_info", + filter_table="VersoZoning_info", filters={"County": ["Addison"]}, join_key="OBJECT_ID", join_type="inner", @@ -50,7 +50,7 @@ "query/sql/acs5/unemployment_rate.sql": [WHERE_SOURCE], "query/sql/cdc/county_places.sql": [ FilterSource( - filter_table="cdc_county_places", + filter_table="cdc_places_county", filters={"Measure": ["Depression among adults"]}, ) ], @@ -176,7 +176,9 @@ def test_left_join(self): def test_spatial_join(self): src = FilterSource( - filter_table="zoning_geom", join_key="geom", join_type="spatial_intersect" + filter_table="VersoZoning_geom", + join_key="geom", + join_type="spatial_intersect", ) _, join = compile_filters([src], []) assert join == "JOIN f0 ON ST_Intersects(g.geom, f0.geom)" @@ -199,7 +201,7 @@ def test_where_string_injected(self): sql, params = sql_filter_block( BACKEND / "query/sql/acs5/acs5_tidy.sql", [WHERE_SOURCE] ) - assert "FROM acs5_b10_census" in sql + assert "FROM acs5_demographics_tidy" in sql assert 'WHERE "NAME" IN ($1, $2)' in sql assert params == ["Vergennes", "Addison town", 2015.0, 2020.0] assert "{{" not in sql and "{%" not in sql diff --git a/design/archive/notes/data_coverage.md b/design/archive/notes/data_coverage.md index 9030bc72..3c193e75 100644 --- a/design/archive/notes/data_coverage.md +++ b/design/archive/notes/data_coverage.md @@ -4,7 +4,7 @@ The scrapers in `backend/data_collection/` pull from the **ACS 5-year estimates** via the Census API. The ACS 5-year program began with the 2005–2009 dataset, released in December 2010. **2009 is the earliest year available in this product** and is the hard floor for all longitudinal tables (demographics, education, housing, labor force, income). -Current scraper config: `YEARS = list(range(2009, 2025))` in `data_collection/base.py`. +Current scraper config: `YEARS = list(range(2009, MAX_YEAR))` in `data_collection/base.py`. Education data starts at 2012 in practice (earlier tables used different variable structures). @@ -21,7 +21,7 @@ The ACS replaced the decennial Census long-form starting with the 2010 cycle. Be | Period | Source | Resolution | Town-level? | | ---------------- | ----------------------------- | ----------- | ----------------------- | -| 2009–2024 | ACS 5-year estimates | Annual | Yes | +| 2009–Present | ACS 5-year estimates | Annual | Yes | | 2005–2008 | ACS 1-year estimates | Annual | No (≥65k pop only) | | 2000, 2010, 2020 | Decennial Census (short form) | Every 10 yr | Yes (limited variables) | | 1970–2000 | Decennial Census long-form | Every 10 yr | Partial | diff --git a/design/archive/zoning-four-table-migration.md b/design/archive/zoning-four-table-migration.md index fe4084bd..cd57722c 100644 --- a/design/archive/zoning-four-table-migration.md +++ b/design/archive/zoning-four-table-migration.md @@ -201,9 +201,9 @@ def geojson(filters: dict | None = None) -> dict: c.hex_color AS hex_color, c.rgba AS rgba, ST_AsGeoJSON(ST_Simplify(g.geom, 0.0001)) AS geometry - FROM zoning_info i - JOIN zoning_geom g USING (OBJECT_ID) - LEFT JOIN zoning_colors c ON c.district_type = i.District_Type + FROM VersoZoning_info i + JOIN VersoZoning_geom g USING (OBJECT_ID) + LEFT JOIN VersoZoning_colors c ON c.district_type = i.District_Type {where} """, params, diff --git a/design/current/Data_Engineering.md b/design/current/Data_Engineering.md index fb9283fb..641e6676 100644 --- a/design/current/Data_Engineering.md +++ b/design/current/Data_Engineering.md @@ -17,15 +17,72 @@ # Overview of Steps -1. COLLECTION: data is collected, either in direct download or via an API or scrape. API is preferred. -2. BUILD: data is processed into a clean dataset and stored in SQL tables. - - how these decisions are arrived in light of the data should be well articulated in the corresponding `.qmd` in the notebooks folder, with ample code included. -3. QUERY and FILTER: - - queries are built for the data in SQL and wrapped up in python functions. This is how the data tables are manipulated to serve the precise data the frontend needs. - - Filtering is done in reference to `backend/api/schema.json` - - see the [schema](#schema) section below for an explanation of fields - - see `backend/api/routes/get_routes/get_filters.py` and `backend/api/routes/get_routes/get_filters.py` for how those fields are used in practice. -4. API: thin wrapper of fastapi stuff around the queries. +0. **LAKE CREATION** If the DuckLake is not yet instanciated, the `just build-lake` justfile recipe will create the DuckLake instance, install the spatial extension, and establish both the `RAW` and `CLEANED` table schemas. + +- Files called upon: [`backend/lake_build.py`](../../backend/lake_build.py) + +1. **COLLECTION:** Data is collected from external sources within the [`backend/data_collection/`](../../backend/data_collection/) folder, with APIs preferred whenever available. Direct downloads and locally stored tables are used when an API is unavailable or does not provide the required data. Raw data is loaded into the DuckLake `RAW` schema with minimal transformations so that the original source data is preserved. + +- The collection process is orchestrated using the `just get-data {start_year} {end_year}` justfile recipe, which collects data given the specified year range (inclusive). +- Files called upon: [`backend/run_data_collection.py`](../../backend/run_data_collection.py) +- Separate data file collectors live within [`backend/data_collection/`](../../backend/data_collection/) + - [`acs5.py`](../../backend/data_collection/acs5.py)