Merge branch 'main' into dev-smoothing

Author: krivard

.github/ISSUE_TEMPLATE/feature_release.md

@@ -0,0 +1,30 @@
+---
+name: Feature release 
+about: Begin the finishing work for features ready to be included in a release
+title: 'Release NEW_THING'
+labels: 'release'
+assignees: 'benjaminysmith'
+---
+
+- [Link to issue]()
+- [Link to PR]()
+- Proposed release version: <!-- eg 1.12 -->
+
+<!-- Additional information about the feature: -->
+
+
+<!-- relevant for most work -->
+
+- [ ] API [documentation](https://github.com/cmu-delphi/delphi-epidata/tree/main/docs/api) and/or [changelog](https://github.com/cmu-delphi/delphi-epidata/blob/main/docs/api/covidcast_changelog.md)
+- [ ] API mailing list notification
+
+<!-- relevant for new signals -->
+
+- [ ] Statistical review (usually [correlations](https://github.com/cmu-delphi/covidcast/tree/main/docs/R-notebooks))
+- [ ] Signal / source name review (usually [Roni](https://docs.google.com/document/d/10hGd4Evce4lJ4VkWaQEKFQxvmw2P4xyYGtIAWF52Sf8/edit?usp=sharing))
+
+<!-- relevant for new map signals -->
+
+- [ ] Visual review
+- [ ] [Signal description pop-up text](https://docs.google.com/document/d/1kDqRg8EaI4WQXMaUUbbCGPlsUqEql8kgXCNt6AvMA9I/edit?usp=sharing) review
+- [ ] [Map release notes](https://docs.google.com/document/d/1BpxGgIma_Lkd2kxtwEo2DBdHQ3zk6dHRz-leUIRlOIA/edit?usp=sharing)

.gitignore

@@ -126,3 +126,7 @@ venv.bak/
 .retry
 .indicators-ansible-vault-pass
 indicators-ansible-vault-pass
+
+# testing_utils
+testing_utils/cache
+testing_utils/*.csv

_delphi_utils_python/.pylintrc

@@ -1,8 +1,21 @@
-[DESIGN]
 
-min-public-methods=1
+[MESSAGES CONTROL]
 
+disable=logging-format-interpolation,
+    too-many-locals,
+    too-many-arguments,
+    # Allow pytest functions to be part of a class.
+    no-self-use,
+    # Allow pytest classes to have one test.
+    too-few-public-methods
 
-[MESSAGES CONTROL]
+[BASIC]
+
+# Allow arbitrarily short-named variables.
+variable-rgx=[a-z_][a-z0-9_]*
+argument-rgx=[a-z_][a-z0-9_]*
+
+[DESIGN]
 
-disable=R0801, C0330, E1101, E0611, C0114, C0116, C0103, R0913, R0914, W0702
+# Don't complain about pytest "unused" arguments.
+ignored-argument-names=(_.*|run_as_module)

_delphi_utils_python/data_proc/geomap/geo_data_proc.py

@@ -217,6 +217,12 @@ def create_jhu_uid_fips_crosswalk():
             {"jhu_uid": "63072999", "fips": "72000", "weight": 1.0},
         ]
     )
+    cruise_ships = pd.DataFrame(
+        [
+            {"jhu_uid": "84088888", "fips": "88888", "weight": 1.0},
+            {"jhu_uid": "84099999", "fips": "99999", "weight": 1.0},
+        ]
+    )
 
     jhu_df = (
         pd.read_csv(JHU_FIPS_URL, dtype={"UID": str, "FIPS": str})
@@ -234,7 +240,7 @@ def create_jhu_uid_fips_crosswalk():
     # Drop the JHU UIDs that were hand-modified
     dup_ind = jhu_df["jhu_uid"].isin(
         pd.concat(
-            [hand_additions, unassigned_states, out_of_state, puerto_rico_unassigned]
+            [hand_additions, unassigned_states, out_of_state, puerto_rico_unassigned, cruise_ships]
         )["jhu_uid"].values
     )
     jhu_df.drop(jhu_df.index[dup_ind], inplace=True)
@@ -391,6 +397,7 @@ def create_fips_population_table():
     df_pr = df_pr.groupby("fips").sum().reset_index()
     df_pr = df_pr[~df_pr["fips"].isin(census_pop["fips"])]
     census_pop_pr = pd.concat([census_pop, df_pr])
+
     census_pop_pr.to_csv(join(OUTPUT_DIR, FIPS_POPULATION_OUT_FILENAME), index=False)
 
 

_delphi_utils_python/delphi_utils/__init__.py

@@ -10,5 +10,6 @@
 
 from .geomap import GeoMapper
 from .smooth import Smoother
+from .signal import add_prefix, public_signal
 
 __version__ = "0.1.0"

_delphi_utils_python/delphi_utils/archive.py

@@ -351,7 +351,11 @@ def update_cache(self):
 
         self._cache_updated = True
 
-    def archive_exports(self, exported_files: Files) -> Tuple[Files, Files]:
+    def archive_exports(self,
+        exported_files: Files,
+        update_cache: bool = True,
+        update_s3: bool = True
+    ) -> Tuple[Files, Files]:
         """
         Handles actual archiving of files to the S3 bucket.
 
@@ -375,10 +379,12 @@ def archive_exports(self, exported_files: Files) -> Tuple[Files, Files]:
             archive_key = join(self.indicator_prefix, basename(exported_file))
 
             try:
-                # Update local cache
-                shutil.copyfile(exported_file, cached_file)
+                if update_cache:
+                    # Update local cache
+                    shutil.copyfile(exported_file, cached_file)
 
-                self.bucket.Object(archive_key).upload_file(exported_file)
+                if update_s3:
+                    self.bucket.Object(archive_key).upload_file(exported_file)
 
                 archive_success.append(exported_file)
             except FileNotFoundError:

_delphi_utils_python/delphi_utils/data/jhu_uid_fips_table.csv

@@ -82,8 +82,6 @@ jhu_uid,fips,weight
 63072149,72149,1.0
 63072151,72151,1.0
 63072153,72153,1.0
-84088888,88888,1.0
-84099999,99999,1.0
 84000001,01000,1.0
 84000002,02000,1.0
 84000004,04000,1.0

_delphi_utils_python/delphi_utils/geomap.py

@@ -77,6 +77,11 @@ class GeoMapper:
     ==========
     The main GeoMapper object loads and stores crosswalk dataframes on-demand.
 
+    When replacing geocodes with a new one an aggregation step is performed on the data columns
+    to merge entries  (i.e. in the case of a many to one mapping or a weighted mapping). This
+    requires a specification of the data columns, which are assumed to be all the columns that
+    are not the geocodes or the date column specified in date_col.
+
     Example 1: to add a new column with a new geocode, possibly with weights:
     > gmpr = GeoMapper()
     > df = gmpr.add_geocode(df, "fips", "zip", from_col="fips", new_col="geo_id",
@@ -305,7 +310,12 @@ def add_geocode(
             )
 
         # state codes are all stored in one table
-        if new_code in state_codes:
+        if from_code in state_codes and new_code in state_codes:
+            crosswalk = self._load_crosswalk(from_code="state", to_code="state")
+            crosswalk = crosswalk.rename(
+                columns={from_code: from_col, new_code: new_col}
+            )
+        elif new_code in state_codes:
             crosswalk = self._load_crosswalk(from_code=from_code, to_code="state")
             crosswalk = crosswalk.rename(
                 columns={from_code: from_col, new_code: new_col}
@@ -322,9 +332,13 @@ def add_geocode(
             df = df.merge(crosswalk, left_on=from_col, right_on=from_col, how="left")
 
         # Drop extra state columns
-        if new_code in state_codes:
+        if new_code in state_codes and not from_code in state_codes:
             state_codes.remove(new_code)
             df.drop(columns=state_codes, inplace=True)
+        elif new_code in state_codes and from_code in state_codes:
+            state_codes.remove(new_code)
+            state_codes.remove(from_code)
+            df.drop(columns=state_codes, inplace=True)
 
         return df
 
@@ -361,6 +375,9 @@ def replace_geocode(
         new_code: {'fips', 'zip', 'state_code', 'state_id', 'state_name', 'hrr', 'msa',
                    'hhs_region_number'}
             Specifies the geocode type of the data in new_col.
+        date_col: str or None, default "date"
+            Specify which column contains the date values. Used for value aggregation.
+            If None, then the aggregation is done only on geo_id.
         data_cols: list, default None
             A list of data column names to aggregate when doing a weighted coding. If set to
             None, then all the columns are used except for date_col and new_col.
@@ -389,12 +406,17 @@ def replace_geocode(
             # Multiply and aggregate (this automatically zeros NAs)
             df[data_cols] = df[data_cols].multiply(df["weight"], axis=0)
             df.drop("weight", axis=1, inplace=True)
-        df = df.groupby([date_col, new_col]).sum().reset_index()
+
+        if not date_col is None:
+            df = df.groupby([date_col, new_col]).sum().reset_index()
+        else:
+            df = df.groupby([new_col]).sum().reset_index()
         return df
 
-    def add_population_column(self, data, geocode_type, geocode_col=None):
+    def add_population_column(self, data, geocode_type, geocode_col=None, dropna=True):
         """
-        Appends a population column to a dateframe, based on the FIPS or ZIP code.
+        Appends a population column to a dataframe, based on the FIPS or ZIP code. If no
+        dataframe is provided, the full crosswalk from geocode to population is returned.
 
         Parameters
         ---------
@@ -412,24 +434,26 @@ def add_population_column(self, data, geocode_type, geocode_col=None):
             A dataframe with a population column appended.
         """
         geocode_col = geocode_type if geocode_col is None else geocode_col
+        data = data.copy()
 
         if geocode_type not in ["fips", "zip"]:
             raise ValueError(
                 "Only fips and zip geocodes supported. \
                 For other codes, aggregate those."
             )
 
+        pop_df = self._load_crosswalk(from_code=geocode_type, to_code="pop")
+
         if not is_string_dtype(data[geocode_col]):
             data[geocode_col] = data[geocode_col].astype(str).str.zfill(5)
 
-        pop_df = self._load_crosswalk(from_code=geocode_type, to_code="pop")
-
+        merge_type = "inner" if dropna else "left"
         data_with_pop = (
-            data.copy()
-            .merge(pop_df, left_on=geocode_col, right_on=geocode_type, how="inner")
+            data
+            .merge(pop_df, left_on=geocode_col, right_on=geocode_type, how=merge_type)
             .rename(columns={"pop": "population"})
         )
-        data_with_pop["population"] = data_with_pop["population"].astype(int)
+
         return data_with_pop
 
     @staticmethod

quidel_covidtest/delphi_quidel_covidtest/handle_wip_sensor.py renamed to _delphi_utils_python/delphi_utils/signal.py

@@ -1,13 +1,13 @@
-"""This file checks the wip status of signals"""
+"""Functions for understanding and creating signal names."""
 import covidcast
 
-def add_prefix(signal_names, wip_signal, prefix):
+def add_prefix(signal_names, wip_signal, prefix="wip_"):
     """Adds prefix to signal if there is a WIP signal
     Parameters
     ----------
     signal_names: List[str]
         Names of signals to be exported
-    prefix : 'wip_'
+    prefix : "wip_"
         prefix for new/non public signals
     wip_signal : List[str] or bool
         a list of wip signals: [], OR
@@ -36,11 +36,11 @@ def add_prefix(signal_names, wip_signal, prefix):
     raise ValueError("Supply True | False or '' or [] | list()")
 
 
-def public_signal(signal_):
+def public_signal(signal):
     """Checks if the signal name is already public using COVIDcast
     Parameters
     ----------
-    signal_ : str
+    signal : str
         Name of the signal
     Returns
     -------
@@ -50,6 +50,6 @@ def public_signal(signal_):
     """
     epidata_df = covidcast.metadata()
     for index in range(len(epidata_df)):
-        if epidata_df['signal'][index] == signal_:
+        if epidata_df["signal"][index] == signal:
             return True
     return False

_delphi_utils_python/setup.py

@@ -3,6 +3,7 @@
 
 required = [
     "boto3",
+    "covidcast",
     "gitpython",
     "moto",
     "numpy",

_delphi_utils_python/tests/test_geomap.py

@@ -278,11 +278,13 @@ def test_zip_to_state_id(self):
     def test_add_population_column(self):
         gmpr = GeoMapper()
         new_data = gmpr.add_population_column(self.fips_data_3, "fips")
-        assert new_data["population"].sum() == 274963
+        assert new_data.shape == (5, 5)
         new_data = gmpr.add_population_column(self.zip_data, "zip")
-        assert new_data["population"].sum() == 274902
+        assert new_data.shape == (6, 5)
         with pytest.raises(ValueError):
             new_data = gmpr.add_population_column(self.zip_data, "hrr")
+        new_data = gmpr.add_population_column(self.fips_data_5, "fips")
+        assert new_data.shape == (4, 5)
 
     def test_add_geocode(self):
         gmpr = GeoMapper()
@@ -382,13 +384,20 @@ def test_add_geocode(self):
         new_data2 = gmpr.add_geocode(new_data, "state_code", "hhs_region_number")
         assert new_data2["hhs_region_number"].unique().size == 2
 
+        # state_name -> state_id
+        new_data = gmpr.replace_geocode(self.zip_data, "zip", "state_name")
+        new_data2 = gmpr.add_geocode(new_data, "state_name", "state_id")
+        assert new_data2.shape == (4, 5)
+        new_data2 = gmpr.replace_geocode(new_data, "state_name", "state_id", new_col="abbr")
+        assert "abbr" in new_data2.columns
+
         # fips -> nation
-        new_data = gmpr.replace_geocode(self.fips_data_5, "fips", "nation")
+        new_data = gmpr.replace_geocode(self.fips_data_5, "fips", "nation", new_col="NATION")
         assert new_data.equals(
             pd.DataFrame().from_dict(
                 {
                     "date": {0: pd.Timestamp("2018-01-01 00:00:00")},
-                    "nation": {0: "us"},
+                    "NATION": {0: "us"},
                     "count": {0: 10024.0},
                     "total": {0: 100006.0},
                 }
@@ -411,6 +420,23 @@ def test_add_geocode(self):
             )
         )
 
+        # hrr -> nation
+        with pytest.raises(ValueError):    
+            new_data = gmpr.replace_geocode(self.zip_data, "zip", "hrr")
+            new_data2 = gmpr.replace_geocode(new_data, "hrr", "nation")
+
         # fips -> hrr (dropna=True/False check)
         assert not gmpr.add_geocode(self.fips_data_3, "fips", "hrr").isna().any().any()
         assert gmpr.add_geocode(self.fips_data_3, "fips", "hrr", dropna=False).isna().any().any()
+
+        # fips -> zip (date_col=None chech)
+        new_data = gmpr.replace_geocode(self.fips_data_5.drop(columns=["date"]), "fips", "hrr", date_col=None)
+        assert new_data.equals(
+            pd.DataFrame().from_dict(
+                {
+                    'hrr': {0: '1', 1: '183', 2: '184', 3: '382', 4: '7'},
+                    'count': {0: 1.772347174163783, 1: 7157.392403522299, 2: 2863.607596477701, 3: 1.0, 4: 0.22765282583621685},
+                    'total': {0: 3.544694348327566, 1: 71424.64801363471, 2: 28576.35198636529, 3: 1.0, 4: 0.4553056516724337}
+                }
+            )
+        )

_delphi_utils_python/tests/test_signal.py

@@ -0,0 +1,34 @@
+"""Tests for delphi_utils.signal."""
+from unittest.mock import patch
+import pandas as pd
+
+from delphi_utils.signal import add_prefix, public_signal
+
+# Constants for mocking out the call to `covidcast.metadata` within `public_signal()`.
+PUBLIC_SIGNALS = ["sig1", "sig2", "sig3"]
+PUBLIC_SIGNALS_FRAME = pd.DataFrame(data={"signal": PUBLIC_SIGNALS})
+
+class TestSignal:
+    """Tests for signal.py."""
+
+    def test_add_prefix_to_all(self):
+        """Tests that `add_prefix()` derives work-in-progress names for all input signals."""
+        assert add_prefix(["sig1", "sig3"], True, prefix="wip_") == ["wip_sig1", "wip_sig3"]
+    
+    def test_add_prefix_to_specified(self):
+        """Tests that `add_prefix()` derives work-in-progress names for specified signals."""
+        assert add_prefix(["sig1", "sig2", "sig3"], ["sig2"], prefix="wip_") ==\
+            ["sig1", "wip_sig2", "sig3"]
+    
+    @patch("covidcast.metadata")
+    def test_add_prefix_to_non_public(self, metadata):
+        """Tests that `add_prefix()` derives work-in-progress names for non-public signals."""
+        metadata.return_value = PUBLIC_SIGNALS_FRAME
+        assert add_prefix(["sig0", "sig1"], False, prefix="wip_") == ["wip_sig0", "sig1"]
+
+    @patch("covidcast.metadata")
+    def test_public_signal(self, metadata):
+        """Tests that `public_signal()` identifies public vs. private signals."""
+        metadata.return_value = PUBLIC_SIGNALS_FRAME
+        assert not public_signal("sig0")
+        assert public_signal("sig2")