Make CEMS extraction handle new listed year_quarter partitions

.github/workflows/pytest.yml

@@ -172,6 +172,7 @@ jobs:
       - name: Run integration tests, trying to use GCS cache if possible
         run: |
           pip install --no-deps --editable .
+          pudl_datastore --dataset epacems --partition year_quarter=2022q1
           make pytest-integration
 
       - name: Upload coverage

src/pudl/extract/epacems.py

@@ -117,7 +117,7 @@ def get_filters(self):
     def get_quarterly_file(self) -> Path:
         """Return the name of the CSV file that holds annual hourly data."""
         return Path(
-            f"epacems-{self.year}-{pd.to_datetime(self.year_quarter).quarter}.csv"
+            f"epacems-{self.year}q{pd.to_datetime(self.year_quarter).quarter}.csv"
         )
 
 
@@ -138,7 +138,6 @@ def get_data_frame(self, partition: EpaCemsPartition) -> pd.DataFrame:
         archive = self.datastore.get_zipfile_resource(
             "epacems", **partition.get_filters()
         )
-
         with archive.open(str(partition.get_quarterly_file()), "r") as csv_file:
             df = self._csv_to_dataframe(
                 csv_file, ignore_cols=API_IGNORE_COLS, rename_dict=API_RENAME_DICT

src/pudl/workspace/datastore.py

@@ -104,6 +104,13 @@ def _matches(self, res: dict, **filters: Any):
                     f"Resource filter values should be all lowercase: {k}={v}"
                 )
         parts = res.get("parts", {})
+        # If partitions are list, match whole list if it contains desired element
+        if set(map(type, parts.values())) == {list}:
+            return all(
+                any(part.lower() == str(v).lower() for part in parts.get(k))
+                for k, v in filters.items()
+            )
+        # Otherwise return matches to int/str partitions
         return all(
             str(parts.get(k)).lower() == str(v).lower() for k, v in filters.items()
         )
@@ -134,7 +141,10 @@ def get_partitions(self, name: str = None) -> dict[str, set[str]]:
             if name and res["name"] != name:
                 continue
             for k, v in res.get("parts", {}).items():
-                partitions[k].add(v)
+                if isinstance(v, list):
+                    partitions[k] |= set(v)  # Add all items from list
+                else:
+                    partitions[k].add(v)
         return partitions
 
     def get_partition_filters(self, **filters: Any) -> Iterator[dict[str, str]]:
@@ -172,12 +182,12 @@ class ZenodoDoiSettings(BaseSettings):
 
     censusdp1tract: ZenodoDoi = "10.5281/zenodo.4127049"
     eia860: ZenodoDoi = "10.5281/zenodo.10067566"
-    eia860m: ZenodoDoi = "10.5281/zenodo.10204686"
+    eia860m: ZenodoDoi = "10.5281/zenodo.10423813"
     eia861: ZenodoDoi = "10.5281/zenodo.10204708"
     eia923: ZenodoDoi = "10.5281/zenodo.10067550"
     eia_bulk_elec: ZenodoDoi = "10.5281/zenodo.7067367"
     epacamd_eia: ZenodoDoi = "10.5281/zenodo.7900974"
-    epacems: ZenodoDoi = "10.5281/zenodo.10306114"
+    epacems: ZenodoDoi = "10.5281/zenodo.10425497"
     ferc1: ZenodoDoi = "10.5281/zenodo.8326634"
     ferc2: ZenodoDoi = "10.5281/zenodo.8326697"
     ferc6: ZenodoDoi = "10.5281/zenodo.8326696"