Make "yesterday" mode configurable to be N days ago

config.EXAMPLE/exporter.env.EXAMPLE

@@ -23,12 +23,11 @@ INSTANCE_NAME=
 # progressing it down the pipeline?
 CSV_AGE_THRESHOLD_MINUTES=180
 
-# For production use only process CSV files that have a date matching to
-# yesterday. Setting this to FALSE will process all relevant csv, which may
-# be useful for development or restarting a stopped service.
-ONLY_USE_CSV_FROM_YESTERDAY=TRUE
+# If non-blank, only process CSV files with the date of N days ago. Must be an integer >=1.
+# For our particular data set, it must be set to >=2 for correct results (see #110).
+ONLY_USE_CSV_FROM_N_DAYS_AGO=2
 
-# if ONLY_USE_CSV_FROM_YESTERDAY set to FALSE, you can optionally
+# if ONLY_USE_CSV_FROM_N_DAYS_AGO is empty, you can optionally
 # specify a date to process format YYYY-MM-DD also accepts a regular
 # expression to match multiple date
 PROCESS_CSV_FROM_DATE=

docs/deployment.md

@@ -193,7 +193,7 @@ Filtering by variable is not currently possible.
 
 Variables you may wish to modify:
 ```
-ONLY_USE_CSV_FROM_YESTERDAY=FALSE
+ONLY_USE_CSV_FROM_N_DAYS_AGO=2
 # something shorter than the standard 180 may be needed if you only just processed the data
 CSV_AGE_THRESHOLD_MINUTES=???
 # use actual date you want to process

exporter-scripts/scheduled-script.sh

@@ -23,19 +23,15 @@ SNAKEMAKE_CORES="${SNAKEMAKE_CORES:-1}"
 echo "$0: invoking snakemake, cores=$SNAKEMAKE_CORES, logging to $outer_log_file"
 # For telling the pipeline not to go all the way
 SNAKEMAKE_RULE_UNTIL="${SNAKEMAKE_RULE_UNTIL:-all}"
-# We default to only matching files from yesterday
-ONLY_USE_CSV_FROM_YESTERDAY="${ONLY_USE_CSV_FROM_YESTERDAY:-True}"
-# Default will match any date containing numbers
-PROCESS_CSV_FROM_DATE="${PROCESS_CSV_FROM_DATE:-'[0-9]'}"
 set +e
 snakemake --snakefile /app/src/pipeline/Snakefile \
   --resources ftps_server=1 \
   --cores "$SNAKEMAKE_CORES" \
   --until "$SNAKEMAKE_RULE_UNTIL" \
   --config \
       CSV_AGE_THRESHOLD_MINUTES="${CSV_AGE_THRESHOLD_MINUTES}" \
-      ONLY_USE_CSV_FROM_YESTERDAY="${ONLY_USE_CSV_FROM_YESTERDAY}" \
-      PROCESS_CSV_FROM_DATE="${PROCESS_CSV_FROM_DATE}" \
+      ONLY_USE_CSV_FROM_N_DAYS_AGO="${ONLY_USE_CSV_FROM_N_DAYS_AGO:-}" \
+      PROCESS_CSV_FROM_DATE="${PROCESS_CSV_FROM_DATE:-}" \
       EHR_LOOKUP_BYPASS="${EHR_LOOKUP_BYPASS}" \
   >> "$outer_log_file" 2>&1
 ret_code=$?

src/pipeline/Snakefile

@@ -17,7 +17,7 @@ from locations import (
     ALL_UPLOADED_JSON,
     ALL_FTPS_LOG, EHR_STEM_PATTERN_HASHED,
 )
-from pipeline.utils import config_bool, determine_eventual_outputs, timestamp_for_paths
+from pipeline.utils import config_bool, determine_eventual_outputs, timestamp_for_paths, config_int, config_str
 from pseudon.pseudon import csv_to_parquets
 from electronic_health_records.ehr import ehr_for_csn
 
@@ -32,20 +32,20 @@ from electronic_health_records.ehr import ehr_for_csn
 # so being very cautious here may not be necessary.
 CSV_AGE_THRESHOLD_MINUTES = timedelta(minutes=float(config['CSV_AGE_THRESHOLD_MINUTES']))
 
-# To prevent Snakemake picking up old files in production, we can set a flag to to only
-# use files with a date prefix matching yesterday's date. Defaults to True
-ONLY_USE_CSV_FROM_YESTERDAY = config_bool(config['ONLY_USE_CSV_FROM_YESTERDAY'])
+# To prevent Snakemake picking up old files in production, we can set a flag to only
+# use files with a date prefix matching a certain date.
+ONLY_USE_CSV_FROM_N_DAYS_AGO = config_int(config['ONLY_USE_CSV_FROM_N_DAYS_AGO'])
 
 # We can optionally select a single date to process or a regular expression to match multiple dates.
-PROCESS_CSV_FROM_DATE = str(config['PROCESS_CSV_FROM_DATE'])
+PROCESS_CSV_FROM_DATE = config_str(config['PROCESS_CSV_FROM_DATE'])
 
 # The problem here is that snakemake works "backwards" in that you have to define which
 # files you want to be produced, yet we don't know what output files we need until the
 # input files (original CSVs) are present.
 # Therefore, look at the input files and work out the eventual output files so they can
 # be fed into snakemake.
 
-all_outputs, hash_to_csn = determine_eventual_outputs(CSV_AGE_THRESHOLD_MINUTES, ONLY_USE_CSV_FROM_YESTERDAY, PROCESS_CSV_FROM_DATE)
+all_outputs, hash_to_csn = determine_eventual_outputs(CSV_AGE_THRESHOLD_MINUTES, ONLY_USE_CSV_FROM_N_DAYS_AGO, PROCESS_CSV_FROM_DATE)
 ALL_FTPS_UPLOADED = sorted({ao.get_ftps_uploaded_all_file() for ao in all_outputs})
 ALL_DAILY_HASH_LOOKUPS = sorted({ao.get_daily_hash_lookup() for ao in all_outputs})
 ALL_EHR_LOOKUPS = sorted({ao.get_ehr_lookup() for ao in all_outputs})

src/pipeline/utils.py

@@ -3,6 +3,7 @@
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 import re
+from typing import Optional
 
 from snakemake.io import glob_wildcards
 
@@ -18,8 +19,32 @@
     ALL_UPLOADED_JSON,
 )
 
+# Snakemake seems to like guessing the types of env vars, so
+# here are some methods to force them to the type they're supposed to be.
 
-def config_bool(value):
+
+def config_str(value) -> Optional[str]:
+    """Convert a config value from env/CLI to a string."""
+    if value is None:
+        return None
+    return str(value).strip()
+
+
+def config_int(value) -> Optional[int]:
+    """Convert a config value from env/CLI to an int.
+
+    :raises ValueError: if the string cannot be interpreted as an int (including if it
+        has a decimal point).
+    """
+    if value is None:
+        return None
+    s = str(value).strip()
+    if not s:
+        return None
+    return int(s)
+
+
+def config_bool(value) -> bool:
     """Convert a config value from env/CLI to a bool."""
     s = str(value).strip().lower()
     if s in {"", "0", "false"}:
@@ -95,14 +120,20 @@ def timestamp_for_paths() -> str:
 
 
 def determine_eventual_outputs(
-    csv_wait_time: timedelta, process_only_yesterday: bool, process_datestring: str
+    csv_wait_time: timedelta,
+    process_only_n_days_ago: Optional[int] = None,
+    process_dates_regex: Optional[str] = None,
 ):
     """
     :param csv_wait_time: only process files older than this
-    :param process_only_yesterday: if false we process all dates, true only from yesterday
-    :param process_datestring: a regular expression to match datestrings. Has no effect if
-    process_only_yesterday is true
-    :returns: A list of InputCsvFile and a dictionary containing the hashed csn -> csn mappings.
+    :param process_only_n_days_ago: if None, process all dates,
+                                    if >=1 only from that many days ago (1 = yesterday),
+                                    else raise ValueError
+    :param process_dates_regex: Only has effect if process_only_n_days_ago is None.
+                                if process_dates_regex is None, all datestrings are included,
+                                 otherwise it is treated as a regular expression that must match datestrings
+                                 if they are to be included
+    :returns: A list of InputCsvFile and a dictionary containing the hashed csn -> csn mappings, according to the filtering rules.
     """
     # Discover all CSVs using the basic file name pattern
     before = time.perf_counter()
@@ -114,9 +145,12 @@ def determine_eventual_outputs(
     # Build reverse lookup using named wildcards
     _hash_to_csn: dict[str, str] = {}
 
-    if process_only_yesterday:
-        process_datestring = (
-            datetime.now(tz=timezone.utc).date() - timedelta(days=1)
+    if process_only_n_days_ago is not None:
+        if process_only_n_days_ago < 1:
+            raise ValueError("process_only_n_days_ago must be >= 1")
+        process_dates_regex = (
+            datetime.now(tz=timezone.utc).date()
+            - timedelta(days=process_only_n_days_ago)
         ).isoformat()
 
     for csn in all_wc.csn:
@@ -128,8 +162,11 @@ def determine_eventual_outputs(
     ):
         input_file_obj = InputCsvFile(date, csn, variable_id, channel_id, units)
         orig_file = input_file_obj.get_original_csv_path()
-        if re.search(process_datestring, date) is None:
-            print(f"Skipping file not from {process_datestring} {orig_file}")
+        if (
+            process_dates_regex is not None
+            and re.search(process_dates_regex, date) is None
+        ):
+            print(f"Skipping file not from {process_dates_regex} {orig_file}")
             continue
         if csn == "unmatched_csn":
             print(f"Skipping file with unmatched CSN: {orig_file}")

tests/test_snakemake.py

@@ -1,5 +1,6 @@
 from datetime import timedelta, datetime, timezone
 from pathlib import Path
+from typing import Optional
 
 import pytest
 
@@ -48,10 +49,26 @@ def _make_test_input_csv(csv_dir: Path):
             4,
         )
     )
-    # two files from 2 days ago
+
+    # from two days ago
     files.append(
         TestFileDescription(
             (today - timedelta(days=2)).isoformat(),
+            1735740783.0,
+            "SECRET_CSN_1235",
+            "SECRET_MRN_12346",
+            "SECRET_LOCATION_123",
+            "27",
+            "noCh",
+            50,
+            "uV",
+            4,
+        )
+    )
+    # two files from 3 days ago
+    files.append(
+        TestFileDescription(
+            (today - timedelta(days=3)).isoformat(),
             1735801965.0,
             "SECRET_CSN_1234",
             "SECRET_MRN_12345",
@@ -65,7 +82,7 @@ def _make_test_input_csv(csv_dir: Path):
     )
     files.append(
         TestFileDescription(
-            (today - timedelta(days=2)).isoformat(),
+            (today - timedelta(days=3)).isoformat(),
             1735801965.0,
             "SECRET_CSN_1234",
             "SECRET_MRN_12345",
@@ -88,33 +105,44 @@ def _make_test_input_csv(csv_dir: Path):
     return [f.get_orig_csv() for f in files]
 
 
+# Perhaps a relative date *range* will ultimately be what we need (eg. 7 to 2 days ago),
+# but let's test it as it stands for now.
 @pytest.mark.parametrize(
-    "process_only_yesterday, process_datestring, expected_file_indexes",
+    "process_only_n_days_ago, process_date_regex, expected_file_indexes",
     [
-        # with process only yesterday true we should return only the single file from yesterday
-        (True, "", [1]),
-        # process only yesterday overrides process_datestring
+        # with process only n days ago we should return only the file from N days ago
+        (1, None, [1]),
+        (2, None, [2]),
+        # process only n days ago overrides regex
         (
-            True,
-            (datetime.now(tz=timezone.utc).date() - timedelta(days=2)).isoformat(),
-            [1],
+            2,
+            (datetime.now(tz=timezone.utc).date() - timedelta(days=3)).isoformat(),
+            [2],
         ),
-        # with process only yesterday false and an empty process_datestring we should return all 4 files.
-        (False, "", [0, 1, 2, 3]),
-        # with process only yesterday false and process_datestring set to two days ago we should return the two files from two days ago
+        # with process only n days None and an empty process_datestring we should return all 4 files.
+        (None, None, [0, 1, 2, 3, 4]),
+        # empty regex would also work but is not really a requirement
+        (None, "", [0, 1, 2, 3, 4]),
+        # catchall regex
+        (None, ".", [0, 1, 2, 3, 4]),
+        # dates tend to have a 2 in them
+        (None, "2", [0, 1, 2, 3, 4]),
+        # but never a Z
+        (None, "Z", []),
+        # with process only n days None and regex set to three days ago we should return the two files from three days ago
         (
-            False,
-            (datetime.now(tz=timezone.utc).date() - timedelta(days=2)).isoformat(),
-            [2, 3],
+            None,
+            (datetime.now(tz=timezone.utc).date() - timedelta(days=3)).isoformat(),
+            [3, 4],
         ),
     ],
 )
 def test_determine_eventual_outputs(
     tmp_path: Path,
     monkeypatch,
-    process_only_yesterday,
-    process_datestring,
-    expected_file_indexes,
+    process_only_n_days_ago: int,
+    process_date_regex: str,
+    expected_file_indexes: list[int],
 ):
     original_csv_dir = tmp_path / locations.WAVEFORM_ORIGINAL_CSV.relative_to("/")
     expected_paths = _make_test_input_csv(original_csv_dir)
@@ -125,8 +153,89 @@ def test_determine_eventual_outputs(
     csv_wait_time = timedelta(0)
 
     files, hash_to_csn = utils.determine_eventual_outputs(
-        csv_wait_time, process_only_yesterday, process_datestring
+        csv_wait_time, process_only_n_days_ago, process_date_regex
     )
     assert {
         str(f.get_original_csv_path().relative_to(original_csv_dir)) for f in files
     } == {expected_paths[ex_i] for ex_i in expected_file_indexes}
+
+
+# Snakemake config type coercion methods. Snakemake has already done some
+# conversion from env var by the time we get to see it, so this is not testing the whole config pipeline.
+# We could possibly move to yaml config files and avoid some of this.
+@pytest.mark.parametrize(
+    ["value", "expected_output"],
+    [
+        ["", False],
+        [" ", False],
+        ["0", False],
+        [" 0 ", False],
+        [0, False],
+        [1, True],
+        [2, ValueError],
+        ["1", True],
+        ["2", ValueError],
+        [True, True],
+        [False, False],
+    ],
+)
+def test_config_bool(value, expected_output: bool | type[Exception]):
+    if isinstance(expected_output, type):
+        with pytest.raises(expected_output):
+            utils.config_bool(value)
+    else:
+        actual_val = utils.config_bool(value)
+        assert actual_val is expected_output
+
+
+@pytest.mark.parametrize(
+    ["value", "expected_output"],
+    [
+        ["", None],
+        [" ", None],
+        [None, None],
+        ["0", 0],
+        [" 0 ", 0],
+        [-1, -1],
+        [0, 0],
+        [1, 1],
+        [2, 2],
+        ["1", 1],
+        ["-2", -2],
+        ["cheese", ValueError],
+        [1.0, ValueError],
+        [True, ValueError],
+        [False, ValueError],
+    ],
+)
+def test_config_int(value, expected_output: Optional[int] | type[Exception]):
+    if isinstance(expected_output, type):
+        with pytest.raises(expected_output):
+            utils.config_int(value)
+    else:
+        actual_val = utils.config_int(value)
+        assert actual_val is None or isinstance(actual_val, int)
+        assert actual_val == expected_output
+
+
+@pytest.mark.parametrize(
+    ["value", "expected_output"],
+    [
+        ["", ""],
+        [" ", ""],
+        [None, None],
+        [" 0 ", "0"],
+        [1.1, "1.1"],
+        [-2, "-2"],
+        ["  cheese ", "cheese"],
+        [True, "True"],
+    ],
+)
+def test_config_str(value, expected_output: Optional[str] | type[Exception]):
+    if isinstance(expected_output, type):
+        with pytest.raises(expected_output):
+            utils.config_str(value)
+    else:
+        actual_val = utils.config_str(value)
+        assert actual_val is None or isinstance(actual_val, str)
+        assert actual_val == expected_output

tests/test_snakemake_integration.py

@@ -371,7 +371,7 @@ def _run_snakemake(tmp_path):
         "SNAKEMAKE_CORES=1\n"
         "INSTANCE_NAME=pytest\n"
         "CSV_AGE_THRESHOLD_MINUTES=5\n"
-        "ONLY_USE_CSV_FROM_YESTERDAY=False\n"
+        "ONLY_USE_CSV_FROM_N_DAYS_AGO=\n"
         "PROCESS_CSV_FROM_DATE=\n"
         "SCHEMA_NAME=\n"  # in testing mode, value doesn't matter but it has to exist
     )