Propagate #363 to safegraph (whoops)

.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)

_delphi_utils_python/delphi_utils/__init__.py

@@ -8,5 +8,6 @@
 from .export import create_export_csv
 from .utils import read_params
 from .geomap import GeoMapper
+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/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,7 +406,11 @@ 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):

quidel_covidtest/delphi_quidel_covidtest/handle_wip_sensor.py → _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")

ansible/templates/usafacts-params-prod.json.j2

@@ -0,0 +1,12 @@
+{
+  "export_start_date": "latest",
+  "static_file_dir": "./static",
+  "export_dir": "/common/covidcast/receiving/usa-facts",
+  "cache_dir": "./cache",
+  "base_url": "https://usafactsstatic.blob.core.windows.net/public/data/covid-19/covid_{metric}_usafacts.csv",
+  "aws_credentials": {
+    "aws_access_key_id": "{{ delphi_aws_access_key_id }}",
+    "aws_secret_access_key": "{{ delphi_aws_secret_access_key }}"
+  },
+  "bucket_name": "delphi-covidcast-indicator-output"
+}

cdc_covidnet/delphi_cdc_covidnet/__init__.py

@@ -10,7 +10,6 @@
 
 from . import run
 from . import api_config
-from . import geo_maps
 from . import update_sensor
 from . import covidnet