From 62005964c5e031455b57f2501577dd06816dcfbb Mon Sep 17 00:00:00 2001 From: Isarge05 Date: Tue, 18 Aug 2026 13:48:57 -0400 Subject: [PATCH 1/6] Added .env CENSUS_API_KEY to Justfile --- backend/data_collection/acs5.py | 9 +++++---- backend/data_collection/base.py | 5 +++-- backend/data_collection/qcew.py | 1 - justfile | 6 ++++-- 4 files changed, 12 insertions(+), 9 deletions(-) diff --git a/backend/data_collection/acs5.py b/backend/data_collection/acs5.py index 42618a0..e6e48ed 100644 --- a/backend/data_collection/acs5.py +++ b/backend/data_collection/acs5.py @@ -11,6 +11,7 @@ Use --append to merge new rows into existing files instead of overwriting. """ +import os import time import pandas as pd @@ -19,9 +20,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,7 +32,6 @@ "DP04": "Housing", "DP05": "Demographic", } -YEARS = list(range(2009, 2025)) STORAGE_LOCATION = "Data/Census/ACS_5" ID_VARS = ["year", "geo_type", "table", "NAME", "state", "county"] diff --git a/backend/data_collection/base.py b/backend/data_collection/base.py index 106fed8..6fb229c 100644 --- a/backend/data_collection/base.py +++ b/backend/data_collection/base.py @@ -21,6 +21,7 @@ replaces rather than duplicates those rows), then writes the merged result. """ +import os import time from dataclasses import dataclass @@ -29,10 +30,10 @@ 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" # --------------------------------------------------------------------------- diff --git a/backend/data_collection/qcew.py b/backend/data_collection/qcew.py index 9d5b0ef..7ab9695 100644 --- a/backend/data_collection/qcew.py +++ b/backend/data_collection/qcew.py @@ -86,7 +86,6 @@ ] BASE_URL = "https://data.bls.gov/cew/data/api/{year}/{q}/area/{fips}.csv" -# YEARS = list(range(2009, 2024)) QUARTERS = [1, 2, 3, 4] diff --git a/justfile b/justfile index faec1b9..e6a7eaa 100644 --- a/justfile +++ b/justfile @@ -1,6 +1,8 @@ ## Set up environment ## export DATA_DIR := justfile_directory() / "Data" +# Load environment variables +set dotenv-filename := ".env.local" ################ # CLI Development # @@ -95,8 +97,7 @@ check-frontend: npx tsc --noEmit -################ -# ETL (Pipeline) Container +## ETL (Pipeline) Container ################ # --------- 1. Data Collection (E) --------------------- @@ -108,6 +109,7 @@ build-collection: # Get the data for a specified year [working-directory("backend")] get-data year: build-collection + echo "Using API key: $CENSUS_API_KEY" podman run --rm -v "$(pwd)/Data:/data:z" -e DATA_DIR=/data localhost/vdc-collection {{year}} From 2208c34d65043de069369b19a8f3bde25793e1a9 Mon Sep 17 00:00:00 2001 From: Isarge05 Date: Tue, 18 Aug 2026 13:56:19 -0400 Subject: [PATCH 2/6] justfile documentation --- justfile | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/justfile b/justfile index e6a7eaa..66b4fc3 100644 --- a/justfile +++ b/justfile @@ -106,7 +106,7 @@ check-frontend: build-collection: podman build -t localhost/vdc-collection -f ETL/dockerfile.collect . -# Get the data for a specified year +# Collect the data for a specified year and add to lake.RAW tables [working-directory("backend")] get-data year: build-collection echo "Using API key: $CENSUS_API_KEY" @@ -114,7 +114,7 @@ get-data year: build-collection # --------- 2. Data Cleaning (T) --------------------- -# build and run the backend CLEANING image +# Run each RAW table through it's data cleaning script [working-directory("backend")] transform-data: podman build -t localhost/vdc-cleaning -f ETL/dockerfile.clean . @@ -122,15 +122,14 @@ transform-data: # --------- 3. Data Loading (L) --------------------- -# build and run the backend LOADING image (loads cleaned tables into a DuckDB) +# Load the lake.CLEANED tables into a DuckDB database [working-directory("backend")] load-data: podman build -t localhost/vdc-loading -f ETL/dockerfile.load . podman run --rm -v "$(pwd)/Data:/data:z" localhost/vdc-loading - -# --------- FULL PIPELINE RUN (ETL) --------------------- +# Collect (E), clean (T), and load (L) the data (Full pipeline run) [working-directory("backend")] run-etl year: # Collect the data for a certain year From eeeb78e50451ee6e1083abde1a94c0fcc0725437 Mon Sep 17 00:00:00 2001 From: Isarge05 Date: Tue, 18 Aug 2026 14:06:59 -0400 Subject: [PATCH 3/6] Added build-lake recipe to justfile --- backend/ETL/dockerfile.lake | 24 +++++++++++++++++++ backend/data_cleaning/clean_cdc.py | 2 +- backend/data_cleaning/clean_demographics.py | 2 +- .../data_cleaning/clean_dependency_ratio.py | 2 +- .../clean_derived_time_series.py | 2 +- backend/data_cleaning/clean_economic.py | 2 +- backend/data_cleaning/clean_education.py | 2 +- backend/data_cleaning/clean_flood.py | 2 +- .../clean_health_insurance_coverage.py | 2 +- .../clean_historic_population.py | 2 +- backend/data_cleaning/clean_housing.py | 2 +- backend/data_cleaning/clean_qcew.py | 2 +- backend/data_cleaning/clean_wastewater.py | 2 +- backend/data_cleaning/clean_zoning.py | 2 +- backend/{datastore => }/lake_build.py | 0 backend/run_data_collection.py | 2 +- backend/run_data_loading.py | 2 +- backend/tests/test_lake.py | 2 +- justfile | 7 ++++++ 19 files changed, 47 insertions(+), 16 deletions(-) create mode 100644 backend/ETL/dockerfile.lake rename backend/{datastore => }/lake_build.py (100%) diff --git a/backend/ETL/dockerfile.lake b/backend/ETL/dockerfile.lake new file mode 100644 index 0000000..63dde14 --- /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/data_cleaning/clean_cdc.py b/backend/data_cleaning/clean_cdc.py index fc6ea6a..edde5e3 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 diff --git a/backend/data_cleaning/clean_demographics.py b/backend/data_cleaning/clean_demographics.py index d75cfe4..d35b901 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 1142e6e..e11e858 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 2d616c7..3818ee4 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 diff --git a/backend/data_cleaning/clean_economic.py b/backend/data_cleaning/clean_economic.py index d83ff54..1cc43c8 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 248fe8d..65dcf3c 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_flood.py b/backend/data_cleaning/clean_flood.py index 67440d1..38d057c 100644 --- a/backend/data_cleaning/clean_flood.py +++ b/backend/data_cleaning/clean_flood.py @@ -9,7 +9,7 @@ python -m ETL.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 96a0993..d282915 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 91b3e57..fdc71b7 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: diff --git a/backend/data_cleaning/clean_housing.py b/backend/data_cleaning/clean_housing.py index fd23158..0063c6b 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_qcew.py b/backend/data_cleaning/clean_qcew.py index d697484..58bb490 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_wastewater.py b/backend/data_cleaning/clean_wastewater.py index 3754f74..ad39e94 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 diff --git a/backend/data_cleaning/clean_zoning.py b/backend/data_cleaning/clean_zoning.py index 1b72578..573396d 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/datastore/lake_build.py b/backend/lake_build.py similarity index 100% rename from backend/datastore/lake_build.py rename to backend/lake_build.py diff --git a/backend/run_data_collection.py b/backend/run_data_collection.py index c43525c..5a4142a 100644 --- a/backend/run_data_collection.py +++ b/backend/run_data_collection.py @@ -27,7 +27,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] diff --git a/backend/run_data_loading.py b/backend/run_data_loading.py index 0af2c2b..a9cd94e 100644 --- a/backend/run_data_loading.py +++ b/backend/run_data_loading.py @@ -15,7 +15,7 @@ import duckdb # DuckLake connection -from datastore.lake_build import con +from lake_build import con # New DuckDB connection (where CLEANED lake tables will go) db_con = duckdb.connect() diff --git a/backend/tests/test_lake.py b/backend/tests/test_lake.py index 5ccb831..8a45e7f 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/justfile b/justfile index 66b4fc3..96a8bc7 100644 --- a/justfile +++ b/justfile @@ -100,6 +100,13 @@ check-frontend: ## ETL (Pipeline) Container ################ +# --------- Pre-step: Lake Builder --------------------- +[working-directory("backend")] +build-lake: + podman build -t localhost/vdc-lake -f ETL/dockerfile.lake . + podman run --rm -v "$(pwd)/Data:/data:z" -e DATA_DIR=/data localhost/vdc-lake + + # --------- 1. Data Collection (E) --------------------- # build the backend COLLECTION image [working-directory("backend")] From b94837f8e450e77b03eec3cf6b313e2d2866c68a Mon Sep 17 00:00:00 2001 From: Isarge05 Date: Tue, 18 Aug 2026 14:28:35 -0400 Subject: [PATCH 4/6] fixed lake_build.py bug and updated dockerfiles --- backend/ETL/dockerfile.clean | 2 +- backend/ETL/dockerfile.collect | 2 +- backend/ETL/dockerfile.load | 2 +- backend/lake_build.py | 6 +++--- justfile | 8 ++++++-- 5 files changed, 12 insertions(+), 8 deletions(-) diff --git a/backend/ETL/dockerfile.clean b/backend/ETL/dockerfile.clean index 478d9c8..d400888 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 03df0a8..7b9bde5 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.load b/backend/ETL/dockerfile.load index 99d98a9..4d5962c 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/lake_build.py b/backend/lake_build.py index e6df53e..acc9612 100644 --- a/backend/lake_build.py +++ b/backend/lake_build.py @@ -25,7 +25,7 @@ # Attach DuckLake catalog con.execute( - f""" + f"""--sql ATTACH '{LAKE_PATH.as_posix()}' AS lake ( @@ -37,8 +37,8 @@ ) # 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): diff --git a/justfile b/justfile index 96a8bc7..c99c8b2 100644 --- a/justfile +++ b/justfile @@ -2,7 +2,7 @@ export DATA_DIR := justfile_directory() / "Data" # Load environment variables -set dotenv-filename := ".env.local" +set dotenv-filename := ".env" ################ # CLI Development # @@ -117,7 +117,11 @@ build-collection: [working-directory("backend")] get-data year: build-collection echo "Using API key: $CENSUS_API_KEY" - podman run --rm -v "$(pwd)/Data:/data:z" -e DATA_DIR=/data localhost/vdc-collection {{year}} + podman run --rm \ + -v "$(pwd)/Data:/data:z" \ + -e DATA_DIR=/data \ + -e CENSUS_API_KEY="$CENSUS_API_KEY" \ + localhost/vdc-collection {{year}} # --------- 2. Data Cleaning (T) --------------------- From 9d7e9eaac3d0b1753d842a3ce6289801bc42996e Mon Sep 17 00:00:00 2001 From: Isarge05 Date: Tue, 18 Aug 2026 15:06:38 -0400 Subject: [PATCH 5/6] Attached duckdb to DuckLake (warehouse.duckdb) --- .gitignore | 5 +++ backend/run_data_loading.py | 75 ++++++++++++++++++++++++++----------- 2 files changed, 58 insertions(+), 22 deletions(-) diff --git a/.gitignore b/.gitignore index 75ed795..5a84823 100644 --- a/.gitignore +++ b/.gitignore @@ -16,9 +16,14 @@ venv/ backend/Data/Parcels/ #DuckLake files + +# DuckLake catalogue backend/Data/lake backend/Data/lake.files/ +# DuckDB Database with CLEAN tables +backend/Data/warehouse.duckdb + # Local planning/notes files claude_todo.md claude-work-done.md diff --git a/backend/run_data_loading.py b/backend/run_data_loading.py index a9cd94e..cf66688 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 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(): From f417b988bb74b72d9c49cf4e82cf550bc4fa4ce1 Mon Sep 17 00:00:00 2001 From: Isarge05 Date: Tue, 18 Aug 2026 17:06:18 -0400 Subject: [PATCH 6/6] Added year RANGE for etl process instead of singular year --- backend/data_collection/acs5.py | 84 +++++++++++++++---------- backend/data_collection/demographics.py | 32 ++++++---- backend/data_collection/economic.py | 41 +++++++----- backend/data_collection/education.py | 41 +++++++----- backend/data_collection/housing.py | 33 +++++++--- backend/data_collection/qcew.py | 27 ++++---- backend/lake_build.py | 20 +++--- backend/run_data_collection.py | 19 +++--- justfile | 8 +-- 9 files changed, 187 insertions(+), 118 deletions(-) diff --git a/backend/data_collection/acs5.py b/backend/data_collection/acs5.py index e6e48ed..6509fd1 100644 --- a/backend/data_collection/acs5.py +++ b/backend/data_collection/acs5.py @@ -35,6 +35,8 @@ STORAGE_LOCATION = "Data/Census/ACS_5" ID_VARS = ["year", "geo_type", "table", "NAME", "state", "county"] +YEARS = range(2009, 2025) + # Default geos list in (label, for_clause, in_clause) format GEOS = [(k, *v) for k, v in ALL_GEOS.items()] @@ -71,21 +73,25 @@ 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) + + # Save wide + tidy per table results = {} for table, frames in all_frames.items(): if not frames: @@ -107,7 +113,7 @@ def run_acs5_scrape(year: int = 2024, geos: list = GEOS, append: bool = False): # wide_csv_path = f"{STORAGE_LOCATION}/{title}.csv" if append: - new_names = set(combined["NAME"].unique()) + new_names = set(combined["year", "geo_type", "NAME"].unique()) # --- Wide --- try: existing_wide = pd.read_parquet(wide_parquet_path) @@ -124,31 +130,41 @@ def run_acs5_scrape(year: int = 2024, geos: list = GEOS, append: bool = False): # print(f"Saved wide: {title} ({len(combined):,} rows)") # Tidy: run per-year so column labels are year-accurate - tidy_frames = [] + # Tidy: run per-year so column labels are year-accurate + tidy_frames = [] + + for year in sorted(combined["year"].unique()): year_df = combined[combined["year"] == year] - if not year_df.empty: - try: - tidy_year = tidy_census(year_df, year=year, id_vars=ID_VARS) - tidy_year["table"] = table - tidy_frames.append(tidy_year) - except Exception as e: - print(f" SKIP tidy {year} / {table}: {e}") - if tidy_frames: - tidy = pd.concat(tidy_frames, ignore_index=True) + if year_df.empty: + continue + + try: + tidy_year = tidy_census( + year_df, + year=year, + id_vars=ID_VARS, + ) + tidy_year["table"] = table + tidy_frames.append(tidy_year) - label = TABLES[table] + except Exception as e: + print(f" SKIP tidy {year} / {table}: {e}") - results[f"acs5_{label.lower()}"] = tidy - # tidy_parquet_path = f"{STORAGE_LOCATION}/{title}_tidy.parquet" - # tidy_csv_path = f"{STORAGE_LOCATION}/{title}_tidy.csv" + if tidy_frames: + tidy = pd.concat(tidy_frames, ignore_index=True) + + label = TABLES[table] + results[f"acs5_{label.lower()}"] = tidy + # tidy_parquet_path = f"{STORAGE_LOCATION}/{title}_tidy.parquet" + # tidy_csv_path = f"{STORAGE_LOCATION}/{title}_tidy.csv" - # No separate append needed for tidy: it's derived from the - # already-merged wide frame, so it naturally contains all geos. + # 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)") + # 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 @@ -172,13 +188,13 @@ def merge_tidy_tables(): 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/demographics.py b/backend/data_collection/demographics.py index ed6e02b..72d63b1 100644 --- a/backend/data_collection/demographics.py +++ b/backend/data_collection/demographics.py @@ -8,6 +8,8 @@ Output: vt_acs5_b_demographics_tidy.parquet """ +import pandas as pd + from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape # --------------------------------------------------------------------------- @@ -25,6 +27,8 @@ ("75 Plus", range(23, 26), range(47, 50)), ] +YEARS = range(2009, 2025) + def _b01001_codes(male_r, female_r): return [f"B01001_{str(i).zfill(3)}E" for i in male_r] + [ @@ -87,26 +91,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=2024) p.add_argument( "--geos", nargs="+", @@ -124,7 +134,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 de6a212..0f84a28 100644 --- a/backend/data_collection/economic.py +++ b/backend/data_collection/economic.py @@ -20,6 +20,8 @@ Output: vt_acs5_b_economic_tidy.parquet """ +import pandas as pd + from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape SL = "Labor Force" @@ -70,40 +72,49 @@ "B19301": ["B19301_001E"], } +YEARS = range(2009, 2025) + -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=2024) + 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 35f543c..076abf2 100644 --- a/backend/data_collection/education.py +++ b/backend/data_collection/education.py @@ -12,6 +12,8 @@ Output: vt_acs5_b_education_tidy.parquet """ +import pandas as pd + from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape TOTAL = "B15003_001E" @@ -37,40 +39,49 @@ "B15003": [TOTAL] + [f"B15003_{str(i).zfill(3)}E" for i in range(2, 26)], } +YEARS = range(2009, 2025) + -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=2024) + 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 0bcc58a..108fad9 100644 --- a/backend/data_collection/housing.py +++ b/backend/data_collection/housing.py @@ -15,6 +15,8 @@ Output: vt_acs5_b_housing_tidy.parquet """ +import pandas as pd + from data_collection.base import ALL_GEOS, VarGroup, run_acs_b_scrape S = "Housing" @@ -42,24 +44,35 @@ } -def collect(year: int = 2024, geos=None, append=False): +YEARS = range(2009, 2025) + + +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=2024) p.add_argument( "--geos", nargs="+", @@ -77,7 +90,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 7ab9695..d34eb57 100644 --- a/backend/data_collection/qcew.py +++ b/backend/data_collection/qcew.py @@ -88,6 +88,8 @@ BASE_URL = "https://data.bls.gov/cew/data/api/{year}/{q}/area/{fips}.csv" QUARTERS = [1, 2, 3, 4] +YEARS = range(2009, 2025) + # --------------------------------------------------------------------------- # Fetching @@ -214,18 +216,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.") @@ -235,13 +236,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/lake_build.py b/backend/lake_build.py index acc9612..8658548 100644 --- a/backend/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 @@ -41,14 +42,18 @@ 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/run_data_collection.py b/backend/run_data_collection.py index 5a4142a..f5f03df 100644 --- a/backend/run_data_collection.py +++ b/backend/run_data_collection.py @@ -42,15 +42,17 @@ zoning, ] +YEARS = range(2009, 2025) -def run_scraper(scraper, yearly=False, year=None): + +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 +64,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 +76,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 = 2024): 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 +87,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/justfile b/justfile index c99c8b2..375252d 100644 --- a/justfile +++ b/justfile @@ -115,13 +115,13 @@ build-collection: # Collect the data for a specified year and add to lake.RAW tables [working-directory("backend")] -get-data year: build-collection +get-data start_year end_year: build-collection echo "Using API key: $CENSUS_API_KEY" podman run --rm \ -v "$(pwd)/Data:/data:z" \ -e DATA_DIR=/data \ -e CENSUS_API_KEY="$CENSUS_API_KEY" \ - localhost/vdc-collection {{year}} + localhost/vdc-collection {{start_year}} {{end_year}} # --------- 2. Data Cleaning (T) --------------------- @@ -142,9 +142,9 @@ load-data: # Collect (E), clean (T), and load (L) the data (Full pipeline run) [working-directory("backend")] -run-etl year: +run-etl start_year end_year: # Collect the data for a certain year - just get-data {{year}} + just get-data {{start_year}} {{end_year}} # Clean the RAW populated lake tables into CLEANED just transform-data # Load CLEANED tables into DuckDB instance