Add Michigan EJ Screen into data-pipeline's ETL and provide automated scoring and statistics outputs (#1091)

* draft wip * initial commit * clear output from notebook * revert to 65ceb7900f * draft wip * initial commit * clear output from notebook * revert to 65ceb7900f * make michigan prefix for readable * standardize Michigan names and move all constants from class into field names module * standardize Michigan names and move all constants from class into field names module * include only pertinent columns for scoring comparison tool * michigan EJSCREEN standardization * final PR feedback * added exposition and summary of Michigan EJSCREEN * added exposition and summary of Michigan EJSCREEN * fix typo Co-authored-by: Saran Ahluwalia <ahlusar.ahluwalia@gmail.com>
2025-07-29 03:31:17 -07:00 · 2021-12-31 15:38:52 -05:00 · 2021-12-31 15:38:52 -05:00 · a4137fdc98
commit a4137fdc98
parent 24f8eb93c4
6 changed files with 142 additions and 3 deletions
--- a/data/data-pipeline/data_pipeline/etl/sources/michigan_ejscreen/etl.py
+++ b/data/data-pipeline/data_pipeline/etl/sources/michigan_ejscreen/etl.py
@ -0,0 +1,69 @@
+import pandas as pd
+
+from data_pipeline.etl.base import ExtractTransformLoad
+from data_pipeline.utils import get_module_logger
+from data_pipeline.score import field_names
+from data_pipeline.config import settings
+
+logger = get_module_logger(__name__)
+
+
+class MichiganEnviroScreenETL(ExtractTransformLoad):
+    """Michigan EJ Screen class that ingests dataset represented
+    here: https://www.arcgis.com/apps/webappviewer/index.html?id=dc4f0647dda34959963488d3f519fd24
+    This class ingests the data presented in "Assessing the State of Environmental
+    Justice in Michigan." Please see the README in this module for further details.
+    """
+
+    def __init__(self):
+        self.MICHIGAN_EJSCREEN_S3_URL = (
+            settings.AWS_JUSTICE40_DATASOURCES_URL
+            + "/michigan_ejscore_12212021.csv"
+        )
+
+        self.CSV_PATH = self.DATA_PATH / "dataset" / "michigan_ejscreen"
+        self.MICHIGAN_EJSCREEN_PRIORITY_COMMUNITY_THRESHOLD: float = 0.75
+
+        self.COLUMNS_TO_KEEP = [
+            self.GEOID_TRACT_FIELD_NAME,
+            field_names.MICHIGAN_EJSCREEN_SCORE_FIELD,
+            field_names.MICHIGAN_EJSCREEN_PERCENTILE_FIELD,
+            field_names.MICHIGAN_EJSCREEN_PRIORITY_COMMUNITY_FIELD,
+        ]
+
+        self.df: pd.DataFrame
+
+    def extract(self) -> None:
+        logger.info("Downloading Michigan EJSCREEN Data")
+        self.df = pd.read_csv(
+            filepath_or_buffer=self.MICHIGAN_EJSCREEN_S3_URL,
+            dtype={"GEO_ID": "string"},
+            low_memory=False,
+        )
+
+    def transform(self) -> None:
+        logger.info("Transforming Michigan EJSCREEN Data")
+
+        self.df.rename(
+            columns={
+                "GEO_ID": self.GEOID_TRACT_FIELD_NAME,
+                "EJ_Score_Cal_Min": field_names.MICHIGAN_EJSCREEN_SCORE_FIELD,
+                "Pct_CalMin": field_names.MICHIGAN_EJSCREEN_PERCENTILE_FIELD,
+            },
+            inplace=True,
+        )
+        # Calculate the top quartile of prioritized communities
+        # Please see pg. 104 - 109 from source:
+        # pg. https://deepblue.lib.umich.edu/bitstream/handle/2027.42/149105/AssessingtheStateofEnvironmentalJusticeinMichigan_344.pdf
+        self.df[field_names.MICHIGAN_EJSCREEN_PRIORITY_COMMUNITY_FIELD] = (
+            self.df[field_names.MICHIGAN_EJSCREEN_PERCENTILE_FIELD]
+            >= self.MICHIGAN_EJSCREEN_PRIORITY_COMMUNITY_THRESHOLD
+        )
+
+    def load(self) -> None:
+        logger.info("Saving Michigan Environmental Screening Tool to CSV")
+        # write nationwide csv
+        self.CSV_PATH.mkdir(parents=True, exist_ok=True)
+        self.df[self.COLUMNS_TO_KEEP].to_csv(
+            self.CSV_PATH / "michigan_ejscreen.csv", index=False
+        )