Merge pull request #279 from SANDAG/feature-employment-industry-xref

[PULL REQUEST] Block to MGRA Employment Industry Cross Reference

Author: GregorSchroeder

python/employment.py

@@ -97,11 +97,13 @@ def _aggregate_lodes_to_mgra(
     This function allocates jobs from Census blocks to MGRAs using distributions from
     the California Employment Development Department (EDD) point-level dataset. Blocks
     with no EDD data available use a simple land area intersection to allocate jobs to
-    MGRAs.
+    MGRAs. The allocation first attempts to allocate within industry codes using EDD
+    data, then falls back to using EDD data without considering industry codes, and
+    finally falls back to using the land area intersection.
 
     Args:
         combined_data: LODES data with columns: year, block, industry_code, jobs
-        xref: Crosswalk with columns: block, mgra, pct_edd, pct_area, edd_flag
+        xref: Crosswalk with columns: block, mgra, pct_industry, pct_edd, pct_area, flag
         year: The year for which to aggregate data
 
     Returns:
@@ -111,14 +113,12 @@ def _aggregate_lodes_to_mgra(
     # Get MGRA data from SQL
     with utils.ESTIMATES_ENGINE.connect() as con:
         mgra_data = pd.read_sql_query(
-            sql=sql.text(
-                """
+            sql=sql.text("""
                 SELECT DISTINCT [mgra]
                 FROM [inputs].[mgra]
                 WHERE run_id = :run_id
                 ORDER BY [mgra]
-                """
-            ),
+                """),
             con=con,
             params={"run_id": utils.RUN_ID},
         )
@@ -129,10 +129,14 @@ def _aggregate_lodes_to_mgra(
         mgra_data.merge(pd.DataFrame({"industry_code": unique_industries}), how="cross")
         .assign(year=year)
         .merge(
-            combined_data.merge(xref, on="block", how="inner")
+            combined_data.merge(xref, on=["block", "industry_code"], how="inner")
             .assign(
                 value=lambda df: df["jobs"]
-                * np.where(df["edd_flag"] == 1, df["pct_edd"], df["pct_area"])
+                * np.where(
+                    df["flag"] == "pct_industry",
+                    df["pct_industry"],
+                    np.where(df["flag"] == "pct_edd", df["pct_edd"], df["pct_area"]),
+                )
             )
             .groupby(["year", "mgra", "industry_code"], as_index=False)["value"]
             .sum(),
@@ -170,7 +174,14 @@ def _distribute_self_emp_to_mgra(
     """
     # Check that required columns are present
     required_b24080_cols = {"year", "geography", "industry_code", "value"}
-    required_xref_cols = {"geography", "mgra", "flag", "pct_18_64", "pct_pop", "pct_split"}
+    required_xref_cols = {
+        "geography",
+        "mgra",
+        "flag",
+        "pct_18_64",
+        "pct_pop",
+        "pct_split",
+    }
     if not required_b24080_cols.issubset(b24080.columns):
         raise ValueError(
             f"B24080 DataFrame is missing required columns: {required_b24080_cols - set(b24080.columns)}"
@@ -179,7 +190,7 @@ def _distribute_self_emp_to_mgra(
         raise ValueError(
             f"xref DataFrame is missing required columns: {required_xref_cols - set(xref.columns)}"
         )
-    
+
     # Check that flag column only contains expected values
     expected_flags = {"pct_18_64", "pct_pop", "pct_split"}
     if not set(xref["flag"].unique()).issubset(expected_flags):
@@ -214,8 +225,7 @@ def _distribute_self_emp_to_mgra(
 
     # Sum weighted values to the MGRA level
     merged = (
-        merged
-        .groupby(["year", "mgra", "industry_code"])["weighted_value"]
+        merged.groupby(["year", "mgra", "industry_code"])["weighted_value"]
         .sum()
         .reset_index()
         .assign(run_id=utils.RUN_ID)
@@ -347,11 +357,11 @@ def _validate_jobs_inputs(jobs_inputs: dict[str, pd.DataFrame]) -> None:
         null={},
     )
     # No row count validation performed as xref is many-to-many
+    # NULLs are allowed in the result set
     tests.validate_data(
         "xref_block_to_mgra",
         jobs_inputs["xref_block_to_mgra"],
         negative={},
-        null={},
     )
     # No row count validation performed as xref is many-to-many
     tests.validate_data(
@@ -374,7 +384,6 @@ def _validate_jobs_inputs(jobs_inputs: dict[str, pd.DataFrame]) -> None:
         negative={},
         null={},
     )
-    
 
 
 def _create_jobs_output(

sql/employment/get_naics72_split.sql

@@ -7,7 +7,8 @@ Point-level data is gotten from the confidential EDD dataset, assigned to
 Notes: 
     1) This query assumes the connection is to the GIS server.
     2) Data prior to year 2017 is not present in the EDD view and must be
-    queried directly from the source database table.
+    queried directly from the source database table. Note there is no 2016
+    data available.
     3) If no split is present for a block, the regional percentage split is
     substituted. All 2020 Census blocks are represented, except three 
     water-only slivers see: https://github.com/SANDAG/Estimates-Program/issues/193
@@ -94,7 +95,7 @@ BEGIN
         [emp_valid] > 0
         AND [emp_total] > 0
 END
-ELSE IF @year = 2016 OR @year BETWEEN 2010 AND 2014
+ELSE IF @year BETWEEN 2010 AND 2013
 BEGIN
     INSERT INTO [#edd]
     SELECT
@@ -123,7 +124,7 @@ BEGIN
         AND [businesses].[emp_id] = [employment].[emp_id]
     WHERE LEFT([code], 3) IN ('721','722')
 END
-ELSE IF @year = 2015
+ELSE IF @year BETWEEN 2014 AND 2016
 BEGIN
     INSERT INTO [#edd]
     SELECT
@@ -176,6 +177,8 @@ BEGIN
     -- Calculate % split of NAICS 72 into 721 and 722 for 2020 Census Blocks -
     SELECT
         [GEOID20] AS [block],
+        -- Note these CASE statements default to regional average for the block
+        -- Using the Window functions with OVER() to calculate regional averages
         CASE 
             WHEN [72] = 0 THEN SUM([721]) OVER() / SUM([72]) OVER()
             ELSE [721] / [72]