feat!: add --select-by-* options to upload-notes, remove downloading

This adds the --select-by-word, -regex, -athena-table, -csv, and
-anon-csv options to upload-notes, deprecating the previous --docrefs
and --anon-docrefs options (which now map to by-csv and by-anon-csv).

This also removes the rarely used "download notes from your EHR on
the fly" feature which meant you didn't need local copies of the notes.
It could even do a bulk export if you didn't provide a --docrefs flag.
It is too hard to square with the new set of options and is a lot of
complexity for a feature we don't really use and don't really recommend
to sites to use (you should be archiving your notes for easy access
anyway).

Downloading the wrapped clinical notes referred to by your NDJSON is
still supported! This is just removing the ability to go direct to your
EHR for the DocRefs and DxReports themselves.

Author: mikix

cumulus_etl/deid/__init__.py

@@ -1,6 +1,6 @@
 """De-identification support"""
 
-from .codebook import Codebook
+from .codebook import Codebook, FilterFunc
 from .mstool import MSTOOL_CMD
 from .philter import Philter
 from .scrubber import Scrubber

cumulus_etl/deid/codebook.py

@@ -2,11 +2,10 @@
 
 import binascii
 import hmac
-import logging
 import os
 import secrets
 import uuid
-from collections.abc import Iterable, Iterator
+from collections.abc import Awaitable, Callable
 
 from cumulus_etl import common
 
@@ -80,22 +79,6 @@ def fake_id(self, resource_type: str | None, real_id: str, caching_allowed: bool
         else:
             return self.db.resource_hash(real_id)
 
-    def real_ids(self, resource_type: str, fake_ids: Iterable[str]) -> Iterator[str]:
-        """
-        Reverse-maps a list of fake IDs into real IDs.
-
-        This is an expensive operation, so only a bulk API is provided.
-        """
-        mapping = self.db.get_reverse_mapping(resource_type)
-        for fake_id in fake_ids:
-            real_id = mapping.get(fake_id)
-            if real_id:
-                yield real_id
-            else:
-                logging.warning(
-                    "Real ID not found for anonymous %s ID %s. Ignoring.", resource_type, fake_id
-                )
-
 
 ###############################################################################
 #
@@ -225,21 +208,6 @@ def _preserved_resource_hash(
 
         return fake_id
 
-    def get_reverse_mapping(self, resource_type: str) -> dict[str, str]:
-        """
-        Returns reversed cached mappings for a given resource.
-
-        This is used for reverse-engineering anonymous IDs to the original real IDs, for the resources we cache.
-        """
-        mapping = self.cached_mapping.get(resource_type, {})
-        reverse_mapping = {v: k for k, v in mapping.items()}
-
-        # Add any legacy mappings from settings (iteratively, to avoid a spare version in memory)
-        for k, v in self.settings.get(resource_type, {}).items():
-            reverse_mapping[v] = k
-
-        return reverse_mapping
-
     def resource_hash(self, real_id: str) -> str:
         """
         Get a fake ID for an arbitrary FHIR resource ID
@@ -305,3 +273,7 @@ def save(self) -> bool:
             saved = True
 
         return saved
+
+
+# Used for filtering note resource types (like DocRefs or DxReports)
+FilterFunc = Callable[[Codebook, dict], Awaitable[bool]] | None

cumulus_etl/etl/config.py

@@ -2,15 +2,12 @@
 
 import datetime
 import os
-from collections.abc import Awaitable, Callable
 from socket import gethostname
 
 import cumulus_fhir_support as cfs
 
 from cumulus_etl import common, deid, errors, formats, store
 
-FilterFunc = Callable[[deid.Codebook, dict], Awaitable[bool]] | None
-
 
 class JobConfig:
     """
@@ -42,7 +39,7 @@ def __init__(
         export_datetime: datetime.datetime | None = None,
         export_url: str | None = None,
         deleted_ids: dict[str, set[str]] | None = None,
-        resource_filter: FilterFunc = None,
+        resource_filter: deid.FilterFunc = None,
     ):
         self._dir_input_orig = dir_input_orig
         self.dir_input = dir_input_deid

cumulus_etl/nlp/selection.py

@@ -8,10 +8,9 @@
 import pyathena
 
 from cumulus_etl import cli_utils, deid, errors, fhir, id_handling
-from cumulus_etl.etl import config
 
 
-def add_note_selection(parser: argparse.ArgumentParser) -> None:
+def add_note_selection(parser: argparse.ArgumentParser):
     group = parser.add_argument_group("note selection")
     group.add_argument(
         "--select-by-word",
@@ -47,6 +46,7 @@ def add_note_selection(parser: argparse.ArgumentParser) -> None:
         action="store_true",
         help="allow a larger-than-normal selection",
     )
+    return group
 
 
 def query_athena_table(table: str, args) -> str:
@@ -134,7 +134,7 @@ def get_match(
         return self._id_pools[res_type].get(res_id)
 
 
-def _define_csv_filter(csv_file: str, is_anon: bool) -> config.FilterFunc:
+def _define_csv_filter(csv_file: str, is_anon: bool) -> deid.FilterFunc:
     matcher = CsvMatcher(csv_file, is_anon=is_anon)
 
     async def check_match(codebook, res):
@@ -145,7 +145,7 @@ async def check_match(codebook, res):
 
 def _define_regex_filter(
     client: cfs.FhirClient, words: list[str] | None, regexes: list[str] | None
-) -> config.FilterFunc:
+) -> deid.FilterFunc:
     patterns = []
     if regexes:
         patterns.extend(cli_utils.user_regex_to_pattern(regex).pattern for regex in regexes)
@@ -165,7 +165,7 @@ async def res_filter(codebook: deid.Codebook, resource: dict) -> bool:
     return res_filter
 
 
-def get_note_filter(client: cfs.FhirClient, args: argparse.Namespace) -> config.FilterFunc:
+def get_note_filter(client: cfs.FhirClient, args: argparse.Namespace) -> deid.FilterFunc:
     """Returns (patient refs to match, resource refs to match)"""
     # Confirm we don't have conflicting arguments. Which we could maybe combine, as a future
     # improvement, but is too much hassle right now)

cumulus_etl/upload_notes/cli.py

@@ -11,7 +11,7 @@
 from ctakesclient.typesystem import Polarity
 
 from cumulus_etl import cli_utils, common, deid, errors, fhir, nlp, store
-from cumulus_etl.upload_notes import downloader, labeling, selector
+from cumulus_etl.upload_notes import labeling, selector
 from cumulus_etl.upload_notes.labelstudio import LabelStudioClient, LabelStudioNote
 
 PHILTER_DISABLE = "disable"
@@ -44,31 +44,14 @@ async def gather_resources(
     """Selects and downloads just the docrefs we need to an export folder."""
     common.print_header("Gathering documents...")
 
-    # There are three possibilities: we have real IDs, fake IDs, or neither.
-    # Note that we don't support providing both real & fake IDs right now. It's not clear that would be useful.
-    if args.docrefs and args.anon_docrefs:
-        errors.fatal(
-            "You cannot use both --docrefs and --anon-docrefs at the same time.",
-            errors.ARGS_CONFLICT,
-        )
+    note_filter = nlp.get_note_filter(client, args)
 
-    if root_input.protocol == "https":  # is this a FHIR server?
-        return await downloader.download_resources_from_fhir_server(
-            client,
-            root_input,
-            codebook,
-            id_file=args.docrefs,
-            anon_id_file=args.anon_docrefs,
-            export_to=args.export_to,
-        )
-    else:
-        return selector.select_resources_from_files(
-            root_input,
-            codebook,
-            id_file=args.docrefs,
-            anon_id_file=args.anon_docrefs,
-            export_to=args.export_to,
-        )
+    return await selector.select_resources_from_files(
+        root_input,
+        codebook,
+        note_filter=note_filter,
+        export_to=args.export_to,
+    )
 
 
 def datetime_from_resource(resource: dict) -> datetime.datetime | None:
@@ -418,15 +401,10 @@ def define_upload_notes_parser(parser: argparse.ArgumentParser) -> None:
         "(must have note ID, label, and span columns)",
     )
 
-    docs = parser.add_argument_group("note selection")
-    docs.add_argument(
-        "--anon-docrefs",
-        metavar="PATH",
-        help="CSV file with anonymized patient_id,docref_id columns",
-    )
-    docs.add_argument(
-        "--docrefs", metavar="PATH", help="CSV file with a docref_id column of original IDs"
-    )
+    group = nlp.add_note_selection(parser)
+    # Add some deprecated aliases for some note selection options. Deprecated since Sep 2025.
+    group.add_argument("--anon-docrefs", dest="select_by_anon_csv", help=argparse.SUPPRESS)
+    group.add_argument("--docrefs", dest="select_by_csv", help=argparse.SUPPRESS)
 
     group = parser.add_argument_group("NLP")
     cli_utils.add_ctakes_override(group)
@@ -478,6 +456,7 @@ async def upload_notes_main(args: argparse.Namespace) -> None:
 
     args.dir_input = cli_utils.process_input_dir(args.dir_input)
     root_input = store.Root(args.dir_input)
+    store.Root(args.dir_phi, create=True)  # create PHI if needed (very edge case)
 
     # Auth & read files early for quick error feedback
     client = fhir.create_fhir_client_for_cli(