diff --git a/config.EXAMPLE/exporter.env.EXAMPLE b/config.EXAMPLE/exporter.env.EXAMPLE index b2fd99e..8300c33 100644 --- a/config.EXAMPLE/exporter.env.EXAMPLE +++ b/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= diff --git a/docs/deployment.md b/docs/deployment.md index 8773460..2e07288 100644 --- a/docs/deployment.md +++ b/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 diff --git a/exporter-scripts/scheduled-script.sh b/exporter-scripts/scheduled-script.sh index b7af8c5..d69d0e2 100755 --- a/exporter-scripts/scheduled-script.sh +++ b/exporter-scripts/scheduled-script.sh @@ -23,10 +23,6 @@ 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 \ @@ -34,8 +30,8 @@ snakemake --snakefile /app/src/pipeline/Snakefile \ --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=$? diff --git a/src/pipeline/Snakefile b/src/pipeline/Snakefile index fd9bffc..f469fd9 100644 --- a/src/pipeline/Snakefile +++ b/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,12 +32,12 @@ 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 @@ -45,7 +45,7 @@ PROCESS_CSV_FROM_DATE = str(config['PROCESS_CSV_FROM_DATE']) # 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}) diff --git a/src/pipeline/utils.py b/src/pipeline/utils.py index ba32b3b..63c5008 100644 --- a/src/pipeline/utils.py +++ b/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}") diff --git a/tests/test_snakemake.py b/tests/test_snakemake.py index c32b07d..1d88060 100644 --- a/tests/test_snakemake.py +++ b/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 diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 9cc9f04..dc7266e 100644 --- a/tests/test_snakemake_integration.py +++ b/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 )