Merge branch 'main' into 7dav_new_geo_ids

Author: qx-teo

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.2.5
+current_version = 0.2.19
 commit = True
 message = chore: bump covidcast-indicators to {new_version}
 tag = False

.github/workflows/python-ci.yml

@@ -32,7 +32,7 @@ jobs:
         pip install pylint pytest pydocstyle wheel
     - name: Install
       run: |
-        make install
+        make install-ci
     - name: Lint
       run: |
         make lint

Jenkinsfile

@@ -9,7 +9,7 @@
    - Keep in sync with '.github/workflows/python-ci.yml'.
    - TODO: #527 Get this list automatically from python-ci.yml at runtime.
  */
-def indicator_list = ["cdc_covidnet", "changehc", "claims_hosp", "combo_cases_and_deaths", "covid_act_now", "facebook", "google_symptoms", "hhs_hosp", "jhu", "nchs_mortality", "quidel", "quidel_covidtest", "safegraph_patterns", "sir_complainsalot", "usafacts"]
+def indicator_list = ["changehc", "claims_hosp", "covid_act_now", "facebook", "google_symptoms", "hhs_hosp", "jhu", "nchs_mortality", "quidel", "quidel_covidtest", "safegraph_patterns", "sir_complainsalot", "usafacts"]
 def build_package = [:]
 def deploy_staging = [:]
 def deploy_production = [:]

README.md

@@ -26,6 +26,33 @@ Each indicator pipeline includes its own documentation.
 * Consult REVIEW.md for the checklist to use for code reviews. 
 * Consult DETAILS.md (if present) for implementation details, including handling of corner cases.
 
+## Development
+
+`prod` reflects what is currently in production. `main` is the staging branch for the next release.
+
+1. Branch from `main` to develop a new change
+2. PR into `main` and assign a reviewer (or tag someone) to get feedback on your change. List the issue number under `Fixes` if your change resolves an existing [GitHub Issue](https://github.com/cmu-delphi/covidcast-indicators/issues).
+3. Add new commits to your branch in response to feedback.
+4. When approved, tag an admin to merge the PR. Let them know if this change should be released immediately, at a set future date, or if it can just go along for the ride whenever the next release happens.
+
+## Release Process
+
+The release process consists of multiple steps which can all be done via the GitHub website:
+
+1. Go to [create_release GitHub Action](https://github.com/cmu-delphi/covidcast-indicators/actions/workflows/create-release.yml) and click the `Run workflow` dropdown button. Leave branch as `main` unless you know what you're doing. Enter the type of release (patch: bugfixes, params file changes, new signals for existing indicators; minor: new indicators, new utilities; major: backwards-incompatible changes requiring substantial refactoring) and GitHub will automatically compute the next version number for you; alternately, specify the version number by hand. Hit the green `Run workflow` button.
+2. The action will prepare a new release and generate an associated [Pull Request](https://github.com/cmu-delphi/covidcast-indicators/pulls).
+3. Edit the PR description and **list all pull requests included in this release**. This is a manual step to make sure you are aware of 100% of the changes that will be deployed. You can use `#xxx` notation and GitHub will automatically render the title of each PR in Preview mode and when the edit is saved.
+4. Verify that CI passes for the PR as a whole and for the most-recent/bottom-most commit in the PR. We're currently having problems where [python-ci does not run on release PRs](https://github.com/cmu-delphi/covidcast-indicators/issues/1310), but if you see a green check next to the most-recent commit you should be fine.
+5. Approve the PR, merge it, and delete the branch.
+6. Jenkins will automatically deploy the most-recently-built indicator packages to the pipeline servers
+7. Another GitHub action will automatically
+   1. Create a git tag
+   2. Create another [Pull Request](https://github.com/cmu-delphi/covidcast-indicators/pulls) to merge the changes back into the `main` branch
+   3. (if `delphi-utils` was updated) Upload the new version of `delphi-utils` to PyPI
+8. Approve the sync PR, merge it, and delete the branch
+9. Done
+
+You may need to be an admin to perform some of the steps above.
 
 ## License
 

_delphi_utils_python/.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.2.4
+current_version = 0.2.8
 commit = True
 message = chore: bump delphi_utils to {new_version}
 tag = False

_delphi_utils_python/Makefile

@@ -8,6 +8,11 @@ install: venv
 	pip install wheel ; \
 	pip install -e .
 
+install-ci: venv
+	. env/bin/activate; \
+	pip install wheel ; \
+	pip install .
+
 lint:
 	. env/bin/activate; pylint delphi_utils
 	. env/bin/activate; pydocstyle delphi_utils

_delphi_utils_python/delphi_utils/__init__.py

@@ -15,4 +15,4 @@
 from .nancodes import Nans
 from .weekday import Weekday
 
-__version__ = "0.2.4"
+__version__ = "0.2.8"

_delphi_utils_python/delphi_utils/archive.py

@@ -307,8 +307,9 @@ def filter_exports(self, common_diffs: FileDiffMap):
             else:
                 replace(diff_file, exported_file)
 
-    def run(self):
+    def run(self, logger=None):
         """Run the differ and archive the changed and new files."""
+        start_time = time.time()
         self.update_cache()
 
         # Diff exports, and make incremental versions
@@ -317,6 +318,11 @@ def run(self):
         # Archive changed and new files only
         to_archive = [f for f, diff in common_diffs.items()
                       if diff is not None]
+        if logger:
+            logger.debug("Diffed exports",
+                         phase="archiving",
+                         new_files_count=len(new_files),
+                         common_diffs_count=len(to_archive))
         to_archive += new_files
         _, fails = self.archive_exports(to_archive)
 
@@ -329,6 +335,14 @@ def run(self):
         for exported_file in fails:
             print(f"Failed to archive '{exported_file}'")
 
+        elapsed_time_in_seconds = round(time.time() - start_time, 2)
+        if logger:
+            logger.info("Completed archive run",
+                        phase="archiving",
+                        elapsed_time_in_seconds=elapsed_time_in_seconds,
+                        new_changed_count=len(to_archive),
+                        fail_count=len(fails))
+
 
 class S3ArchiveDiffer(ArchiveDiffer):
     """
@@ -641,8 +655,23 @@ def archive_exports(self, exported_files):
             successes: All files from input
             fails: Empty list
         """
+        archive_success, archive_fail = [], []
+        for exported_file in exported_files:
+            archive_file = abspath(
+                join(self.cache_dir, basename(exported_file)))
+
+            # Copy export to cache
+            try:
+                # Archive
+                shutil.copyfile(exported_file, archive_file)
+                archive_success.append(exported_file)
+
+            except FileNotFoundError as ex:
+                print(ex)
+                archive_fail.append(exported_file)
+
         self._exports_archived = True
-        return exported_files, []
+        return archive_success, archive_fail
 
     def update_cache(self):
         """Handle cache updates with a no-op.
@@ -667,13 +696,6 @@ def update_cache(self):
     if "archive" not in _params:
         _params = {"archive": _params, "common": _params}
 
-    logger = get_structured_logger(
+    archiver_from_params(_params).run(get_structured_logger(
         __name__, filename=_params["common"].get("log_filename"),
-        log_exceptions=_params["common"].get("log_exceptions", True))
-    start_time = time.time()
-
-    archiver_from_params(_params).run()
-
-    elapsed_time_in_seconds = round(time.time() - start_time, 2)
-    logger.info("Completed archive run.",
-                elapsed_time_in_seconds=elapsed_time_in_seconds)
+        log_exceptions=_params["common"].get("log_exceptions", True)))

_delphi_utils_python/delphi_utils/runner.py

@@ -36,17 +36,19 @@ def run_indicator_pipeline(indicator_fn:  Callable[[Params], None],
         None if no archiving should be performed.
     """
     params = read_params()
+    logger = get_structured_logger(
+        name=indicator_fn.__module__,
+        filename=params["common"].get("log_filename", None),
+        log_exceptions=params["common"].get("log_exceptions", True))
     indicator_fn(params)
     validator = validator_fn(params)
     archiver = archiver_fn(params)
     if validator:
         validation_report = validator.validate()
-        validation_report.log(get_structured_logger(
-            name = indicator_fn.__module__,
-            filename=params["common"].get("log_filename", None)))
+        validation_report.log(logger)
     if (not validator or validation_report.success()):
         if archiver:
-            archiver.run()
+            archiver.run(logger)
         if "delivery" in params:
             transfer_files()
 

_delphi_utils_python/delphi_utils/validator/dynamic.py

@@ -345,7 +345,7 @@ def pad_reference_api_df(self, reference_api_df, geo_sig_df, reference_end_date)
         Returns:
             - reference_api_df: Supplemented version of original
         """
-        reference_api_df_max_date = reference_api_df.time_value.max()
+        reference_api_df_max_date = reference_api_df.time_value.max().date()
         if reference_api_df_max_date < reference_end_date:
             # Querying geo_sig_df, only taking relevant rows
             geo_sig_df_supplement = geo_sig_df.query(