diff --git a/.github/workflows/etl.yml b/.github/workflows/etl.yml new file mode 100644 index 00000000..7b262e32 --- /dev/null +++ b/.github/workflows/etl.yml @@ -0,0 +1,49 @@ +name: Vermont Data Collaborative ETL + +on: + schedule: + # Runs once a month on the 30th day (@2:00 AM) + - cron: '0 2 30 * *' + workflow_dispatch: + +jobs: + run-etl: + runs-on: ubuntu-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Install uv + uses: astral-sh/setup-uv@v6 + with: + version: '0.11.21' + enable-cache: true + + - name: Install Podman + run: | + sudo apt-get update + sudo apt-get install -y podman + podman --version + + - name: Install Just and DuckDB + run: | + curl --proto '=https' --tlsv1.2 -sSf https://just.systems | bash -s -- --to /usr/local/bin + wget https://github.com + + - name: Run ETL Pipeline + # Grab data just from the latest year to add to the database (2024) + run: just run-etl 2024 2024 + + - name: Commit and Push Updated DuckDB Database + run: | + git config --global user.name "github-actions[bot]" + git config --global user.email "41898282+github-actions[bot]@users.noreply.github.com" + git add backend/Data/warehouse.duckdb + git diff-index --quiet HEAD || git commit -m "Automated ETL: Update DuckDB data [skip ci]" + git push \ No newline at end of file diff --git a/.gitignore b/.gitignore index 89b5f19c..20b819cd 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,8 @@ venv/ backend/Data/Parcels/ #DuckLake files + +# DuckLake catalogue backend/Data/lake backend/Data/lake.files/ diff --git a/README.md b/README.md index 408e3956..b1030658 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. @@ -137,11 +137,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/ETL/dockerfile.clean b/backend/ETL/dockerfile.clean index 478d9c88..d4008887 100644 --- a/backend/ETL/dockerfile.clean +++ b/backend/ETL/dockerfile.clean @@ -22,7 +22,7 @@ COPY --chown=app:app ./api /code/api COPY --chown=app:app ./logger /code/logger # TODO: Remove the dependency on app_utils COPY --chown=app:app ./app_utils /code/app_utils -COPY --chown=app:app ./datastore /code/datastore +COPY --chown=app:app ./lake_build.py /code/lake_build.py ENV DATA_DIR=/data diff --git a/backend/ETL/dockerfile.collect b/backend/ETL/dockerfile.collect index 03df0a89..7b9bde59 100644 --- a/backend/ETL/dockerfile.collect +++ b/backend/ETL/dockerfile.collect @@ -21,7 +21,7 @@ COPY --chown=app:app ./run_data_collection.py /code/run_data_collection.py # TODO: Remove the dependency on app_utils COPY --chown=app:app ./app_utils /code/app_utils -COPY --chown=app:app ./datastore /code/datastore +COPY --chown=app:app ./lake_build.py /code/lake_build.py ENV DATA_DIR=/data diff --git a/backend/ETL/dockerfile.lake b/backend/ETL/dockerfile.lake new file mode 100644 index 00000000..63dde147 --- /dev/null +++ b/backend/ETL/dockerfile.lake @@ -0,0 +1,24 @@ +FROM python:3.12-slim-trixie +COPY --from=ghcr.io/astral-sh/uv:0.11.21 /uv /uvx /bin/ + +# Set working directory +WORKDIR /code +RUN useradd -m -u 1000 app && chown app:app /code /home/app +USER app + +# Dependencies +COPY --chown=app:app pyproject.toml uv.lock /code/ +RUN --mount=type=cache,target=/home/app/.cache/uv,uid=1000,gid=1000 \ + uv sync --frozen --no-dev --no-install-project + +# Spatial extension installation +RUN uv run python -c "import duckdb; duckdb.connect().execute('INSTALL spatial')" + +# The source data collection scripts (source --> destination) +COPY --chown=app:app ./lake_build.py /code/lake_build.py + +ENV DATA_DIR=/data +ENV PYTHONPATH=/code + +# Command to run the DuckLake builder script (container entrypoint) +ENTRYPOINT ["uv", "run", "python", "-u", "-m", "lake_build"] \ No newline at end of file diff --git a/backend/ETL/dockerfile.load b/backend/ETL/dockerfile.load index 99d98a9c..4d5962ca 100644 --- a/backend/ETL/dockerfile.load +++ b/backend/ETL/dockerfile.load @@ -19,7 +19,7 @@ COPY --chown=app:app ./run_data_loading.py /code/run_data_loading.py # TODO: Remove the dependency on app_utils COPY --chown=app:app ./app_utils /code/app_utils # For lake connection import -COPY --chown=app:app ./datastore /code/datastore +COPY --chown=app:app ./lake_build.py /code/lake_build.py ENV DATA_DIR=/data 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..011b9ea6 100644 --- a/backend/api/routes/get_routes/get_wholedata.py +++ b/backend/api/routes/get_routes/get_wholedata.py @@ -5,6 +5,7 @@ from app_utils import data_loading from app_utils.flooding import add_flood_color +from query.production_db import DB logger = logging.getLogger(__name__) @@ -16,14 +17,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..baf2421a 100644 --- a/backend/api/routes/post_routes/post_acs5_db.py +++ b/backend/api/routes/post_routes/post_acs5_db.py @@ -4,15 +4,13 @@ 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 DB logger = logging.getLogger(__name__) router = APIRouter() @@ -21,6 +19,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 +38,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 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")) +# 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")) + + +# 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 +204,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 +220,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_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..55bf7e2d 100644 --- a/backend/api/routes/post_routes/post_census.py +++ b/backend/api/routes/post_routes/post_census.py @@ -3,12 +3,11 @@ 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 DB router = APIRouter() @@ -16,81 +15,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", -} - -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"}, + ("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", } -# 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 +53,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 +69,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..db0a95cf 100644 --- a/backend/api/routes/post_routes/post_qcew.py +++ b/backend/api/routes/post_routes/post_qcew.py @@ -3,7 +3,7 @@ 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 DB router = APIRouter() @@ -36,7 +36,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 +44,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 e019a8e3..49891cd0 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", diff --git a/backend/app_utils/timeseries_db.py b/backend/app_utils/timeseries_db.py index 7d8df186..ef84d57f 100644 --- a/backend/app_utils/timeseries_db.py +++ b/backend/app_utils/timeseries_db.py @@ -6,7 +6,7 @@ import logging -from query.processed_db import DB +from query.production_db import 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..c713422b --- /dev/null +++ b/backend/data_cleaning/clean_acs5.py @@ -0,0 +1,164 @@ +""" +**Author**: + Ian Sargent +**Created**: + 2026-08-24 +**Description**: + Clean the raw ACS-5 Data Profile (DP) tables from DuckLake and build: + - Individual cleaned DP tables + - A combined DP table + - A county GEOID lookup table + + DP02 – Social + DP03 – Economic + DP04 – Housing + DP05 – Demographic + +**Run with**: +python -m data_cleaning.clean_acs5 +""" + +import pandas as pd + +from lake_build import con + +# Data Profile tables +DP_TABLES = { + "DP02": ("acs5_social", "dp_social"), + "DP03": ("acs5_economic", "dp_economic"), + "DP04": ("acs5_housing", "dp_housing"), + "DP05": ("acs5_demographic", "dp_demographic"), +} + +# County GEOIDs for the county_geom table +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 read_raw_data() -> dict[str, pd.DataFrame]: + """ + Read the raw ACS-5 DP tables from DuckLake. + """ + tables = {} + for dp, (raw_table_name, dp_table_name) in DP_TABLES.items(): + query = f"SELECT * FROM lake.RAW.{raw_table_name}" + tables[dp] = con.execute(query).df() + + return tables + + +def clean_dp_tables(raw_tables: dict[str, pd.DataFrame]) -> dict[str, pd.DataFrame]: + """ + Clean the individual ACS-5 DP tables. + """ + cleaned = {} + + for dp, df in raw_tables.items(): + # Future cleaning steps can go in here! + cleaned[dp] = df + + return cleaned + + +def build_county_geoids(): + """ + Create the county GEOID table. + """ + values = ", ".join(f"('{name}', {geoid})" for name, geoid in COUNTY_GEOIDS.items()) + + con.execute( + f"""--sql + CREATE OR REPLACE TABLE lake.CLEANED.vt_county_geoids AS + SELECT * FROM (VALUES {values}) AS t(NAME, GEOID) + """ + ) + + +def add_dp_tables(cleaned: dict[str, pd.DataFrame]): + """ + Write each DP table to the CLEANED DuckLake schema. + + Adds a `table` column containing the DP table identifier + (e.g. DP02, DP03, DP04, DP05). + """ + for dp, df in cleaned.items(): + table_name = DP_TABLES[dp][1] + + df = df.copy() + df["table"] = dp + + con.register("tmp_df", df) + + try: + con.execute( + f""" + CREATE OR REPLACE TABLE lake.CLEANED.{table_name} AS + SELECT + * + FROM tmp_df + """ + ) + finally: + con.unregister("tmp_df") + + +def build_dp_combined(): + """ + Combine the four cleaned DP tables into one tidy table. + """ + + tables = [table_name for _, table_name in DP_TABLES.values()] + + union = "\nUNION ALL\n".join( + f""" + SELECT + NAME, + "table", + Category, + Subcategory, + Variable, + Measure, + year, + Value + FROM lake.CLEANED.{table_name} + """ + for table_name in tables + ) + + con.execute( + f""" + CREATE OR REPLACE TABLE lake.CLEANED.acs5_dp_combined_tidy AS + {union} + """ + ) + + +def clean(): + raw_tables = read_raw_data() + cleaned = clean_dp_tables(raw_tables) + + add_dp_tables(cleaned) + build_dp_combined() + build_county_geoids() + + +def main(): + clean() + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_cdc.py b/backend/data_cleaning/clean_cdc.py index fc6ea6a5..8d101fec 100644 --- a/backend/data_cleaning/clean_cdc.py +++ b/backend/data_cleaning/clean_cdc.py @@ -14,7 +14,7 @@ from sklearn.decomposition import PCA from build.core_functions import bin_measures -from datastore.lake_build import con +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 @@ -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: @@ -154,7 +195,7 @@ def clean() -> dict[str, pd.DataFrame]: ) # 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( @@ -168,7 +209,7 @@ 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, } diff --git a/backend/data_cleaning/clean_demographics.py b/backend/data_cleaning/clean_demographics.py index d75cfe46..d35b9011 100644 --- a/backend/data_cleaning/clean_demographics.py +++ b/backend/data_cleaning/clean_demographics.py @@ -12,7 +12,7 @@ import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: diff --git a/backend/data_cleaning/clean_dependency_ratio.py b/backend/data_cleaning/clean_dependency_ratio.py index 1142e6e0..e11e858d 100644 --- a/backend/data_cleaning/clean_dependency_ratio.py +++ b/backend/data_cleaning/clean_dependency_ratio.py @@ -18,7 +18,7 @@ import numpy as np import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: diff --git a/backend/data_cleaning/clean_derived_time_series.py b/backend/data_cleaning/clean_derived_time_series.py index 2d616c7a..e00fd290 100644 --- a/backend/data_cleaning/clean_derived_time_series.py +++ b/backend/data_cleaning/clean_derived_time_series.py @@ -29,7 +29,7 @@ import numpy as np import pandas as pd -from datastore.lake_build import con +from lake_build import con UNAVAILABLE_SENTINEL = -666666666.0 @@ -101,14 +101,6 @@ 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'", - ), } diff --git a/backend/data_cleaning/clean_economic.py b/backend/data_cleaning/clean_economic.py index d83ff540..1cc43c84 100644 --- a/backend/data_cleaning/clean_economic.py +++ b/backend/data_cleaning/clean_economic.py @@ -12,7 +12,7 @@ import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: diff --git a/backend/data_cleaning/clean_education.py b/backend/data_cleaning/clean_education.py index 248fe8d7..65dcf3cc 100644 --- a/backend/data_cleaning/clean_education.py +++ b/backend/data_cleaning/clean_education.py @@ -12,7 +12,7 @@ import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: diff --git a/backend/data_cleaning/clean_fips.py b/backend/data_cleaning/clean_fips.py new file mode 100644 index 00000000..92ad49b6 --- /dev/null +++ b/backend/data_cleaning/clean_fips.py @@ -0,0 +1,146 @@ +""" +**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 +""" + +from lake_build import con + +## LOAD SPATIAL EXTENSION FUNCTION -------------------- + + +def _load_spatial() -> 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") + + +## BUILD CLEANED VIEWS -------------------- + + +def build_county_lines(): + """ + 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(): + """ + 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(): + """ + 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(): + 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(): + _load_spatial() + + build_county_lines() + build_town_lines() + build_tract_lines() + + +def main(): + clean() + add_to_lake() + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_flood.py b/backend/data_cleaning/clean_flood.py index 67440d17..bcbb3337 100644 --- a/backend/data_cleaning/clean_flood.py +++ b/backend/data_cleaning/clean_flood.py @@ -6,10 +6,10 @@ **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 datastore.lake_build import con +from lake_build import con ## LOAD SPATIAL EXTENSION FUNCTION -------------------- diff --git a/backend/data_cleaning/clean_health_insurance_coverage.py b/backend/data_cleaning/clean_health_insurance_coverage.py index 96a09932..d2829152 100644 --- a/backend/data_cleaning/clean_health_insurance_coverage.py +++ b/backend/data_cleaning/clean_health_insurance_coverage.py @@ -13,7 +13,7 @@ import numpy as np import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: diff --git a/backend/data_cleaning/clean_historic_population.py b/backend/data_cleaning/clean_historic_population.py index 91b3e571..a287c149 100644 --- a/backend/data_cleaning/clean_historic_population.py +++ b/backend/data_cleaning/clean_historic_population.py @@ -11,7 +11,7 @@ import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: @@ -49,10 +49,10 @@ 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 @@ -82,7 +82,7 @@ def add_population_aggregations(df: pd.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 +90,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] @@ -112,12 +112,12 @@ def clean(): raw_df = read_raw_data() # 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) diff --git a/backend/data_cleaning/clean_housing.py b/backend/data_cleaning/clean_housing.py index fd23158e..0063c6b1 100644 --- a/backend/data_cleaning/clean_housing.py +++ b/backend/data_cleaning/clean_housing.py @@ -12,7 +12,7 @@ import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: diff --git a/backend/data_cleaning/clean_median_earnings.py b/backend/data_cleaning/clean_median_earnings.py new file mode 100644 index 00000000..c5b89ad3 --- /dev/null +++ b/backend/data_cleaning/clean_median_earnings.py @@ -0,0 +1,82 @@ +""" +**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 numpy as np +import pandas as pd + +from lake_build import con + +INFLATION_YEAR = datetime.now().year - 2 + + +def read_raw_data() -> 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): + df["Value"] = pd.to_numeric(df["Value"], errors="coerce") + + return df + + +def replace_unavailable_data(df: pd.DataFrame): + df["Value"] = df["Value"].replace(-666666666.0, np.nan) + + return df + + +def clean(): + raw_df = read_raw_data() + df = change_dtype(raw_df) + df = replace_unavailable_data(df) + + return df + + +def add_to_lake(clean_df: pd.DataFrame): + """ + 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(): + clean_df = clean() + add_to_lake(clean_df) + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_qcew.py b/backend/data_cleaning/clean_qcew.py index d6974849..58bb490f 100644 --- a/backend/data_cleaning/clean_qcew.py +++ b/backend/data_cleaning/clean_qcew.py @@ -12,7 +12,7 @@ import pandas as pd -from datastore.lake_build import con +from lake_build import con def read_raw_data() -> pd.DataFrame: diff --git a/backend/data_cleaning/clean_snapshot.py b/backend/data_cleaning/clean_snapshot.py new file mode 100644 index 00000000..41211c91 --- /dev/null +++ b/backend/data_cleaning/clean_snapshot.py @@ -0,0 +1,89 @@ +""" +**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 pandas as pd + +from lake_build import con + + +def read_raw_data() -> 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() -> 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() + + 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(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(): + clean_df = clean() + add_to_lake(clean_df) + + +if __name__ == "__main__": + main() diff --git a/backend/data_cleaning/clean_wastewater.py b/backend/data_cleaning/clean_wastewater.py index 3754f746..b238378b 100644 --- a/backend/data_cleaning/clean_wastewater.py +++ b/backend/data_cleaning/clean_wastewater.py @@ -10,7 +10,7 @@ python -m data_cleaning.clean_wastewater """ -from datastore.lake_build import con +from lake_build import con # TODO: Path is useful when sql files are created! # from build import BACKEND @@ -23,10 +23,10 @@ def _load_spatial() -> None: Load the spatial extension, installing it first if necessary. """ try: - con.execute("""--sql LOAD spatial""") + con.execute("""LOAD spatial""") except Exception: - con.execute("""--sql INSTALL spatial""") - con.execute("""--sql LOAD spatial""") + con.execute("""INSTALL spatial""") + con.execute("""LOAD spatial""") ## ADD UNIQUE ID COLUMNS -------------------- @@ -241,6 +241,25 @@ def build_suitability_geom() -> None: ) +def build_suitability_colors() -> 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: # "Type" labels derived from VERSO WIM GitHub pages @@ -314,6 +333,7 @@ def clean(): # Soil suitability Tables build_suitability_info() build_suitability_geom() + build_suitability_colors() # Stormwater Management Tables build_stormwater_info() @@ -332,6 +352,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", ] diff --git a/backend/data_cleaning/clean_zoning.py b/backend/data_cleaning/clean_zoning.py index 1b72578f..573396d0 100644 --- a/backend/data_cleaning/clean_zoning.py +++ b/backend/data_cleaning/clean_zoning.py @@ -14,7 +14,7 @@ import pandas as pd from app_utils.sql_render import render_sql -from datastore.lake_build import con +from lake_build import con SQL_PATH = Path(__file__).resolve().parent / "sql" # Town and zoning-district boundaries were digitised separately, so subtracting diff --git a/backend/data_collection/acs5.py b/backend/data_collection/acs5.py index 42618a01..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 @@ -11,7 +11,9 @@ Use --append to merge new rows into existing files instead of overwriting. """ +import os import time +from datetime import datetime import pandas as pd import requests @@ -19,9 +21,10 @@ from app_utils.census import tidy_census from data_collection.base import ALL_GEOS -API_KEY = ( - "29af5488bbdb8c7d9f67b7f4ff9c9151e8c2bd0a" # TODO: Get this as a .env variable!!! -) +# Define API key through the .env file +API_KEY = os.environ.get("CENSUS_API_KEY") + + BASE_URL = "https://api.census.gov/data/{year}/acs/acs5/profile" STATE_FIPS = "50" # Vermont TABLES = { @@ -30,10 +33,13 @@ "DP04": "Housing", "DP05": "Demographic", } -YEARS = list(range(2009, 2025)) STORAGE_LOCATION = "Data/Census/ACS_5" ID_VARS = ["year", "geo_type", "table", "NAME", "state", "county"] +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()] @@ -70,84 +76,91 @@ def fetch_table(year, table, for_clause, in_clause): return None -def run_acs5_scrape(year: int = 2024, geos: list = GEOS, append: bool = False): +def run_acs5_scrape(years: range = YEARS, geos: list = GEOS, append: bool = False): + """ + Collect ACS 5-year profile tables for multiple years. + """ # Collect raw frames per table all_frames = {table: [] for table in TABLES} - 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) - if df is not None: - df["geo_type"] = geo_label - all_frames[table].append(df) - time.sleep(0.1) - - # Save wide + tidy per table + 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, + ) + + if df is not None: + df["geo_type"] = geo_label + all_frames[table].append(df) + + time.sleep(0.1) + 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["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_frames = [] - year_df = combined[combined["year"] == year] - if not year_df.empty: + + for year in sorted(combined["year"].unique()): + year_df = combined[combined["year"] == year] + + if year_df.empty: + continue + try: - tidy_year = tidy_census(year_df, year=year, id_vars=ID_VARS) + 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) - - label = TABLES[table] + tidy = pd.concat( + tidy_frames, + ignore_index=True, + ) 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)") return results @@ -165,19 +178,18 @@ 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 -def collect(year: int = 2024, geos=GEOS, append=False): +def collect(years: range = YEARS, geos=GEOS, append=False): """ Collect ACS profile tables and return tidy datasets. """ tables = run_acs5_scrape( - year=year, + years=years, geos=geos, append=append, ) diff --git a/backend/data_collection/base.py b/backend/data_collection/base.py index 106fed85..5e854beb 100644 --- a/backend/data_collection/base.py +++ b/backend/data_collection/base.py @@ -21,18 +21,20 @@ replaces rather than duplicates those rows), then writes the merged result. """ +import os import time from dataclasses import dataclass +from datetime import datetime import pandas as pd import requests from app_utils.census import split_name_col -API_KEY = "29af5488bbdb8c7d9f67b7f4ff9c9151e8c2bd0a" +# Define API key through the .env file +API_KEY = os.environ.get("CENSUS_API_KEY") BASE_URL = "https://api.census.gov/data/{year}/acs/acs5" STATE_FIPS = "50" -# YEARS = list(range(2009, 2025)) STORAGE_LOCATION = "Data/Census/ACS_5" # --------------------------------------------------------------------------- @@ -50,6 +52,8 @@ GEOS = [(k, *v) for k, v in ALL_GEOS.items()] +MAX_YEAR = datetime.now().year - 2 + # --------------------------------------------------------------------------- # Data structures # --------------------------------------------------------------------------- @@ -150,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 ed6e02b0..c8fef05d 100644 --- a/backend/data_collection/demographics.py +++ b/backend/data_collection/demographics.py @@ -8,6 +8,10 @@ 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 # --------------------------------------------------------------------------- @@ -25,6 +29,10 @@ ("75 Plus", range(23, 26), range(47, 50)), ] +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) + def _b01001_codes(male_r, female_r): return [f"B01001_{str(i).zfill(3)}E" for i in male_r] + [ @@ -87,26 +95,32 @@ def _all_b01001_vars(): } -def collect(year: int = 2024, geos=None, append=False): +def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: if geos is None: geos = [(k, *ALL_GEOS[k]) for k in ALL_GEOS] - df = run_acs_b_scrape( - fetch_specs, - var_groups, - "vt_acs5_b_demographics_tidy.parquet", - year=year, - geos=geos, - append=append, - ) + frames = [] + for year in years: + df = run_acs_b_scrape( + fetch_specs, + var_groups, + "vt_acs5_b_demographics_tidy.parquet", + year=year, + geos=geos, + append=append, + ) + if df is not None: + frames.append(df) - return df + return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame() if __name__ == "__main__": import argparse 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=MAX_YEAR - 1) p.add_argument( "--geos", nargs="+", @@ -124,7 +138,7 @@ def collect(year: int = 2024, geos=None, append=False): selected_geos = [(k, *ALL_GEOS[k]) for k in args.geos] df = collect( - year=args.year, + years=range(args.start_year, args.end_year + 1), geos=selected_geos, append=args.append, ) diff --git a/backend/data_collection/economic.py b/backend/data_collection/economic.py index de6a212a..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,10 @@ 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 SL = "Labor Force" @@ -35,6 +39,7 @@ "B23001_125E", "B23001_132E", # female 25-54 ] + _PRIME_TOTAL = [ "B23001_024E", "B23001_031E", @@ -53,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, @@ -64,46 +75,62 @@ ] 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"], } +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) + -def collect(year: int = 2024, geos=None, append=False): +def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: if geos is None: geos = [(k, *ALL_GEOS[k]) for k in ALL_GEOS] - return run_acs_b_scrape( - fetch_specs, - var_groups, - "vt_acs5_b_economic_tidy.parquet", - year=year, - geos=geos, - append=append, - ) + frames = [] + for year in years: + df = run_acs_b_scrape( + fetch_specs, + var_groups, + "vt_acs5_b_economic_tidy.parquet", + year=year, + geos=geos, + append=append, + ) + if df is not None: + frames.append(df) + + return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame() if __name__ == "__main__": import argparse - parser = argparse.ArgumentParser(description="Scrape ACS B-table economic data.") - parser.add_argument("year", type=int, nargs="?", default=2024) - parser.add_argument( + 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=MAX_YEAR - 1) + p.add_argument( "--geos", nargs="+", choices=list(ALL_GEOS), default=list(ALL_GEOS), ) - parser.add_argument("--append", action="store_true") + p.add_argument("--append", action="store_true") - args = parser.parse_args() + args = p.parse_args() selected_geos = [(k, *ALL_GEOS[k]) for k in args.geos] df = collect( - year=args.year, + years=range(args.start_year, args.end_year + 1), geos=selected_geos, append=args.append, ) diff --git a/backend/data_collection/education.py b/backend/data_collection/education.py index 35f543c7..4ceb6b18 100644 --- a/backend/data_collection/education.py +++ b/backend/data_collection/education.py @@ -12,6 +12,10 @@ 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 TOTAL = "B15003_001E" @@ -37,40 +41,51 @@ "B15003": [TOTAL] + [f"B15003_{str(i).zfill(3)}E" for i in range(2, 26)], } +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) -def collect(year: int = 2024, geos=None, append=False): + +def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: if geos is None: geos = [(k, *ALL_GEOS[k]) for k in ALL_GEOS] - return run_acs_b_scrape( - fetch_specs, - var_groups, - "vt_acs5_b_education_tidy.parquet", - year=year, - geos=geos, - append=append, - ) + frames = [] + for year in years: + df = run_acs_b_scrape( + fetch_specs, + var_groups, + "vt_acs5_b_education_tidy.parquet", + year=year, + geos=geos, + append=append, + ) + if df is not None: + frames.append(df) + + return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame() if __name__ == "__main__": import argparse - parser = argparse.ArgumentParser(description="Scrape ACS B-table education data.") - parser.add_argument("year", type=int, nargs="?", default=2024) - parser.add_argument( + 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=MAX_YEAR - 1) + p.add_argument( "--geos", nargs="+", choices=list(ALL_GEOS), default=list(ALL_GEOS), ) - parser.add_argument("--append", action="store_true") + p.add_argument("--append", action="store_true") - args = parser.parse_args() + args = p.parse_args() selected_geos = [(k, *ALL_GEOS[k]) for k in args.geos] df = collect( - year=args.year, + years=range(args.start_year, args.end_year + 1), geos=selected_geos, append=args.append, ) diff --git a/backend/data_collection/housing.py b/backend/data_collection/housing.py index 0bcc58ad..d5784cd6 100644 --- a/backend/data_collection/housing.py +++ b/backend/data_collection/housing.py @@ -15,6 +15,10 @@ 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 S = "Housing" @@ -41,25 +45,37 @@ "B25077": ["B25077_001E"], } +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) -def collect(year: int = 2024, geos=None, append=False): + +def collect(years: range = YEARS, geos=None, append=False) -> pd.DataFrame: if geos is None: geos = [(k, *ALL_GEOS[k]) for k in ALL_GEOS] - return run_acs_b_scrape( - fetch_specs, - var_groups, - "vt_acs5_b_housing_tidy.parquet", - year=year, - geos=geos, - append=append, - ) + frames = [] + for year in years: + df = run_acs_b_scrape( + fetch_specs, + var_groups, + "vt_acs5_b_housing_tidy.parquet", + year=year, + geos=geos, + append=append, + ) + if df is not None: + frames.append(df) + + return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame() if __name__ == "__main__": import argparse 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=MAX_YEAR - 1) p.add_argument( "--geos", nargs="+", @@ -77,7 +93,7 @@ def collect(year: int = 2024, geos=None, append=False): selected_geos = [(k, *ALL_GEOS[k]) for k in args.geos] df = collect( - year=args.year, + years=range(args.start_year, args.end_year + 1), geos=selected_geos, append=args.append, ) diff --git a/backend/data_collection/qcew.py b/backend/data_collection/qcew.py index 9d5b0ef8..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 @@ -86,9 +87,12 @@ ] BASE_URL = "https://data.bls.gov/cew/data/api/{year}/{q}/area/{fips}.csv" -# YEARS = list(range(2009, 2024)) QUARTERS = [1, 2, 3, 4] +MAX_YEAR = datetime.now().year - 1 + +YEARS = range(2009, MAX_YEAR) + # --------------------------------------------------------------------------- # Fetching @@ -215,18 +219,17 @@ def process_county(area_fips: str, county_name: str, year: int) -> pd.DataFrame: # --------------------------------------------------------------------------- -def run_qcew_scrape(year: int) -> pd.DataFrame: +def run_qcew_scrape(years: range = YEARS) -> pd.DataFrame: STORAGE_PATH.mkdir(parents=True, exist_ok=True) all_frames = [] - - for fips, name in VT_COUNTIES.items(): - print(f"\n=== {name} County ({fips}) ===") - df = process_county(fips, name, year) - if not df.empty: - all_frames.append(df) - print(f" {len(df):,} rows") - else: - print(" No data") + for year in years: + for fips, name in VT_COUNTIES.items(): + print(f"\n=== {name} County ({fips}) ===") + df = process_county(fips, name, year) + if not df.empty: + all_frames.append(df) + else: + print("No data") if not all_frames: print("No data fetched.") @@ -236,13 +239,11 @@ def run_qcew_scrape(year: int) -> pd.DataFrame: combined.sort_values(["County", "year", "quarter", "sector"], inplace=True) combined.reset_index(drop=True, inplace=True) - # combined.to_parquet(OUTPUT_FILE, index=False) - # print(f"\nDone. {len(combined):,} rows → {OUTPUT_FILE}") return combined -def collect(year: int = 2024): - df = run_qcew_scrape(year) +def collect(years: range = YEARS): + df = run_qcew_scrape(years) return df diff --git a/backend/datastore/lake_build.py b/backend/lake_build.py similarity index 81% rename from backend/datastore/lake_build.py rename to backend/lake_build.py index e6df53e8..86585487 100644 --- a/backend/datastore/lake_build.py +++ b/backend/lake_build.py @@ -1,5 +1,6 @@ import os from pathlib import Path +from typing import Iterable, Union import duckdb import geopandas as gpd @@ -25,7 +26,7 @@ # Attach DuckLake catalog con.execute( - f""" + f"""--sql ATTACH '{LAKE_PATH.as_posix()}' AS lake ( @@ -37,18 +38,22 @@ ) # Create schemas in the lake catalog -con.execute("""--sql CREATE SCHEMA IF NOT EXISTS lake.RAW""") -con.execute("""--sql CREATE SCHEMA IF NOT EXISTS lake.CLEANED""") +con.execute("""CREATE SCHEMA IF NOT EXISTS lake.RAW""") +con.execute("""CREATE SCHEMA IF NOT EXISTS lake.CLEANED""") -def insert_year(name: str, df: pd.DataFrame, year: int): +def insert_year(name: str, df: pd.DataFrame, years: Union[int, Iterable[int]]): """ - Insert or replace one year's data in a DuckLake table. + Insert or replace data for specific year(s) in a DuckLake table. """ - if "year" not in map(str.lower, df.columns): raise ValueError(f"DataFrame for {name!r} does not contain a 'year' column.") + if isinstance(years, int): + years_list = [years] + else: + years_list = list(years) + if isinstance(df, gpd.GeoDataFrame): df = df.copy() df["geometry"] = df.geometry.to_wkb() @@ -81,19 +86,20 @@ def insert_year(name: str, df: pd.DataFrame, year: int): ) return - # Remove this year's existing data. + # Remove existing data for the target years con.execute( f""" DELETE FROM lake.{schema}.{table} - WHERE year = ? + WHERE year IN ({",".join("?" for _ in years_list)}) """, - [year], + years_list, ) - # Insert the replacement. + # Insert the replacement data con.execute( f""" INSERT INTO lake.{schema}.{table} + BY NAME SELECT * FROM tmp_df """ ) 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 f4ed354d..ebd50565 100644 --- a/backend/query/__init__.py +++ b/backend/query/__init__.py @@ -2,7 +2,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 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..c883e8c3 100644 --- a/backend/query/acs5.py +++ b/backend/query/acs5.py @@ -15,7 +15,7 @@ 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 DB logger = logging.getLogger(__name__) sql_path = Path(__file__).resolve().parent / "sql" / "acs5" @@ -23,19 +23,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": {}, + }, + "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": {}, + }, + }, }, - "income": {"table": "acs5_b_economic", "fixed_filters": {"Section": ["Income"]}}, - "median_age": { - "table": "acs5_b10_census", - "fixed_filters": {"Variable": ["Median Age"]}, + "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 +146,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 +157,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 +210,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/cdc.py b/backend/query/cdc.py index d00454dd..1a29a2d4 100644 --- a/backend/query/cdc.py +++ b/backend/query/cdc.py @@ -17,7 +17,7 @@ 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 DB logger = logging.getLogger(__name__) sql_dir = Path(__file__).resolve().parent / "sql" / "cdc" @@ -46,7 +46,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 +54,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 +81,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 +106,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 +122,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 +150,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..4256f493 100644 --- a/backend/query/core_functions.py +++ b/backend/query/core_functions.py @@ -13,7 +13,7 @@ import logging from api.models import FilterResponse, RangeDescriptor -from query.processed_db import DB +from query.production_db import 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..27e4b60f --- /dev/null +++ b/backend/query/production_db.py @@ -0,0 +1,43 @@ +""" +**Author**: + Fitz Koch +**Created**: + 2026-06-01 +**Updated**: + 2026-08-26 +**Description**: + Establishes the DB 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__) +proc_dir = Path(__file__).resolve().parent.parent / "Data" / "_Processed" + +BACKEND_DIR = Path(__file__).resolve().parents[1] +DATA_DIR = Path(os.environ.get("DATA_DIR", BACKEND_DIR / "Data")) + + +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_etl_db() -> duckdb.DuckDBPyConnection: + path = Path(DATA_DIR / "warehouse.duckdb") + con = duckdb.connect(path, read_only=True) + _load_spatial(con) + return con + + +DB = _build_etl_db() 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/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..5a2050df 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,7 @@ 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 DB logger = logging.getLogger(__name__) sql_dir = Path(__file__).resolve().parent / "sql" / "wastewater" @@ -32,6 +32,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 +49,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..3febee7c 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,7 @@ 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 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..89823f6c 100644 --- a/backend/run_data_cleaning.py +++ b/backend/run_data_cleaning.py @@ -14,10 +14,6 @@ def get_cleaners(): if ispkg: continue - # Skip private/helper modules - if module_name.startswith("_"): - continue - module = import_module(f"data_cleaning.{module_name}") # Only include modules that expose a main() function @@ -27,18 +23,6 @@ def get_cleaners(): 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() - -# for table in tables - - def run_master_clean(): for cleaner in get_cleaners(): print(f"Running {cleaner.__name__.split('.')[-1]}...") diff --git a/backend/run_data_collection.py b/backend/run_data_collection.py index c43525c9..7b9f239f 100644 --- a/backend/run_data_collection.py +++ b/backend/run_data_collection.py @@ -12,6 +12,7 @@ """ import argparse +from datetime import datetime from data_collection import ( acs5, @@ -27,7 +28,7 @@ wastewater, zoning, ) -from datastore.lake_build import insert_year, replace_table +from lake_build import insert_year, replace_table # Datasets WITH year columns (longitudinal) YEARLY_SCRAPERS = [acs5, demographics, economic, education, housing, qcew] @@ -41,16 +42,20 @@ wastewater, zoning, ] +MAX_YEAR = datetime.now().year - 1 -def run_scraper(scraper, yearly=False, year=None): +YEARS = range(2009, MAX_YEAR) + + +def run_scraper(scraper, yearly: bool = False, years: range = YEARS): name = scraper.__name__.split(".")[-1] try: print(f"Running {name}...") if yearly: - outputs = scraper.collect(year=year) + outputs = scraper.collect(years) else: outputs = scraper.collect() @@ -62,7 +67,7 @@ def run_scraper(scraper, yearly=False, year=None): print(f"Loading {full_name}") # If the dataset is longitudinal, replace or append that year's data if yearly: - insert_year(full_name, df, year) + insert_year(full_name, df, years) # If a static dataset, replace the whole table else: replace_table(full_name, df) @@ -74,9 +79,9 @@ def run_scraper(scraper, yearly=False, year=None): raise -def run_master_scrape(year: int): +def run_master_scrape(start_year: int = 2009, end_year: int = MAX_YEAR - 1): for scraper in YEARLY_SCRAPERS: - run_scraper(scraper, yearly=True, year=year) + run_scraper(scraper, yearly=True, years=range(start_year, end_year + 1)) for scraper in STATIC_SCRAPERS: run_scraper(scraper, yearly=False) @@ -85,12 +90,13 @@ def run_master_scrape(year: int): def main(): # Accepts the year argument from justfile for collection parser = argparse.ArgumentParser() - parser.add_argument("year", type=int) + parser.add_argument("start_year", type=int) + parser.add_argument("end_year", type=int) args = parser.parse_args() - print(f"Collecting data for {args.year}") + print(f"Collecting data from {args.start_year} to {args.end_year}") - run_master_scrape(args.year) + run_master_scrape(args.start_year, args.end_year) if __name__ == "__main__": diff --git a/backend/run_data_loading.py b/backend/run_data_loading.py index 0af2c2b0..cf66688d 100644 --- a/backend/run_data_loading.py +++ b/backend/run_data_loading.py @@ -12,16 +12,16 @@ python -m run_data_loading """ -import duckdb +import os +from pathlib import Path -# DuckLake connection -from datastore.lake_build import con +import duckdb -# New DuckDB connection (where CLEANED lake tables will go) -db_con = duckdb.connect() +ROOT = Path(__file__).resolve().parent.parent +DATA_DIR = Path(os.getenv("DATA_DIR", ROOT / "Data")) -def get_cleaned_tables() -> list[str]: +def get_cleaned_tables(db_con: duckdb.DuckDBPyConnection) -> list[str]: """ List all active DuckLake tables within the 'CLEANED' lake schema. """ @@ -34,33 +34,64 @@ def get_cleaned_tables() -> list[str]: WHERE s.schema_name = 'CLEANED' AND t.end_snapshot IS NULL AND s.end_snapshot IS NULL + ORDER BY table_name """ - tables = [row[0] for row in con.execute(query).fetchall()] + tables = [row[0] for row in db_con.execute(query).fetchall()] return tables def create_duckdb(): - # Fetch CLEANED table names - tables = get_cleaned_tables() + warehouse_path = DATA_DIR / "warehouse.duckdb" + lake_db_path = DATA_DIR / "lake" + storage_path = DATA_DIR / "lake.files" + + db_con = duckdb.connect(str(warehouse_path)) + + db_con.execute( + f""" + ATTACH '{lake_db_path.as_posix()}' + AS lake ( + TYPE ducklake, + DATA_PATH '{storage_path.as_posix()}', + OVERRIDE_DATA_PATH TRUE + ) + """ + ) + + tables = get_cleaned_tables(db_con) total_tables = len(tables) - # Each table gets written to the DuckDB as its own table - try: - for i, table in enumerate(tables): - df = con.sql(f"SELECT * FROM lake.CLEANED.{table}").df() # noqa: F841 - db_con.execute(f"CREATE OR REPLACE TABLE {table} AS SELECT * FROM df") - - print( - f"\r{i + 1} / {total_tables} tables added to the database", - end="", - flush=True, + failed = [] + + for i, table in enumerate(tables): + try: + db_con.execute( + f''' + CREATE OR REPLACE TABLE "{table}" + AS + SELECT * + FROM lake.CLEANED."{table}" + ''' ) + except Exception as e: + failed.append((table, str(e))) + + print( + f"\r{i + 1}/{total_tables} processed", + end="", + flush=True, + ) + + if failed: + print(f"\n{len(failed)} tables failed:") + for table, error in failed: + print(f" {table}: {error}") - except Exception as e: - print(f"Database creation failed: {e}") + db_con.execute("DETACH lake") + db_con.close() - print("\nDATABASE COMPLETED") + print("\nDATABASE COMPLETED!") def main(): diff --git a/backend/tests/test_lake.py b/backend/tests/test_lake.py index 5ccb8310..8a45e7f3 100644 --- a/backend/tests/test_lake.py +++ b/backend/tests/test_lake.py @@ -5,7 +5,7 @@ python -m tests.test_lake """ -from datastore.lake_build import con +from lake_build import con def inspect_schema(schema: str) -> None: diff --git a/backend/tests/test_sql_render.py b/backend/tests/test_sql_render.py index 76fa137b..ce19a3f4 100644 --- a/backend/tests/test_sql_render.py +++ b/backend/tests/test_sql_render.py @@ -12,7 +12,7 @@ from sqlfluff.core import FluffConfig, Linter from api.models import FilterSource, RangeFilter -from sql_render import ( +from app_utils.sql_render import ( compile_filters, compile_where, filter_clauses, @@ -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"]}, ) ], @@ -182,7 +182,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)" @@ -205,7 +207,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)