Apply requester-style privacy thresholding to request/download counts across all summaries and totals, with configurable minimum threshold (#272)

* Initial plan

* Threshold archive request and download totals for privacy

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Apply privacy thresholding to dataset summaries and totals

* Update pyproject.toml

* Parameterize privacy threshold for summary and total request/download counts

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Cody Baker <51133164+CodyCBakerPhD@users.noreply.github.com>

Author: 3 people

pyproject.toml

@@ -12,7 +12,7 @@ packages = ["src/s3_log_extraction"]
 
 [project]
 name = "s3-log-extraction"
-version="1.10.1"
+version="1.10.2"
 authors = [
   { name="Cody Baker", email="cody.c.baker.phd@gmail.com" },
 ]

src/s3_log_extraction/summarize/_generate_all_dataset_totals.py

@@ -3,12 +3,14 @@
 
 import pandas
 
+from ._generate_summaries import _round_requester_count
 from ..config import get_cache_subdirectory
 from ..ip_utils._globals import EXCLUDED_REGION_LABELS
 
 
 def generate_all_dataset_totals(
     cache_directory: str | pathlib.Path | None = None,
+    privacy_threshold_minimum: int = 50,
 ) -> None:
     """
     Generate top-level totals of summarized access activity for all datasets.
@@ -18,6 +20,9 @@ def generate_all_dataset_totals(
     cache_directory : path-like, optional
         The top-level cache directory from which the summary directory is derived.
         If not provided, the default cache directory is used.
+    privacy_threshold_minimum : int
+        Minimum disclosure threshold for privacy-rounded request/download totals.
+        Default is ``50``.
     """
     summary_directory = get_cache_subdirectory(cache_directory=cache_directory, name="summaries")
 
@@ -34,6 +39,8 @@ def generate_all_dataset_totals(
         if not summary_file_path.exists():
             continue
         summary = pandas.read_table(filepath_or_buffer=summary_file_path)
+        for column_name in ("number_of_requests", "number_of_downloads"):
+            summary[column_name] = pandas.to_numeric(summary[column_name], errors="coerce").fillna(0).astype("int64")
 
         unique_countries: set[str] = set()
         for region in summary["region"]:
@@ -60,8 +67,12 @@ def generate_all_dataset_totals(
             "total_bytes_sent": int(summary["bytes_sent"].sum()),
             "number_of_unique_regions": number_of_unique_regions,
             "number_of_unique_countries": number_of_unique_countries,
-            "total_number_of_requests": int(summary["number_of_requests"].sum()),
-            "total_number_of_downloads": int(summary["number_of_downloads"].sum()),
+            "total_number_of_requests": _round_requester_count(
+                count=int(summary["number_of_requests"].sum()), modulo=20, minimum=privacy_threshold_minimum
+            ),
+            "total_number_of_downloads": _round_requester_count(
+                count=int(summary["number_of_downloads"].sum()), modulo=20, minimum=privacy_threshold_minimum
+            ),
             "number_of_requesters": number_of_requesters,
         }
 

src/s3_log_extraction/summarize/_generate_archive_summaries.py

@@ -4,12 +4,15 @@
 import natsort
 import pandas
 
+from ._generate_summaries import _privacy_round_request_download_columns
 from ..config import get_cache_subdirectory
 
 
 @beartype.beartype
 def generate_archive_summaries(
-    cache_directory: str | pathlib.Path | None = None, asset_types_in_order: tuple[str, ...] | list[str] | None = None
+    cache_directory: str | pathlib.Path | None = None,
+    asset_types_in_order: tuple[str, ...] | list[str] | None = None,
+    privacy_threshold_minimum: int = 50,
 ) -> None:
     """
     Generate summaries by day and region for the entire archive from the mapped S3 logs.
@@ -22,6 +25,9 @@ def generate_archive_summaries(
     asset_types_in_order : sequence[str], optional
         Preferred output column ordering for known asset types in the archive
         ``by_asset_type_per_week.tsv`` summary.
+    privacy_threshold_minimum : int
+        Minimum disclosure threshold for privacy-rounded request/download
+        summary values. Default is ``50``.
     """
     asset_types_in_order = list(dict.fromkeys(asset_types_in_order)) if asset_types_in_order is not None else []
 
@@ -36,6 +42,9 @@ def generate_archive_summaries(
         for dataset_by_day_summary_file_path in summary_directory.rglob(pattern="by_day.tsv")
         if dataset_by_day_summary_file_path.parent.name != "archive"
     ]
+    for summary in all_dataset_summaries_by_day:
+        for column_name in ("number_of_requests", "number_of_downloads"):
+            summary[column_name] = pandas.to_numeric(summary[column_name], errors="coerce").fillna(0).astype("int64")
     aggregated_dataset_summaries_by_day = pandas.concat(objs=all_dataset_summaries_by_day, ignore_index=True)
 
     pre_aggregated = aggregated_dataset_summaries_by_day.groupby(by="date", as_index=False)[
@@ -49,6 +58,9 @@ def generate_archive_summaries(
     aggregated_activity_by_day = aggregated_activity_by_day.astype(
         dtype={"bytes_sent": "int64", "number_of_requests": "int64", "number_of_downloads": "int64"}
     )
+    aggregated_activity_by_day = _privacy_round_request_download_columns(
+        summary_table=aggregated_activity_by_day, minimum=privacy_threshold_minimum
+    )
 
     archive_summary_by_day_file_path = archive_directory / "by_day.tsv"
     aggregated_activity_by_day.to_csv(
@@ -61,6 +73,9 @@ def generate_archive_summaries(
         for dataset_by_region_summary_file_path in summary_directory.rglob(pattern="by_region.tsv")
         if dataset_by_region_summary_file_path.parent.name != "archive"
     ]
+    for summary in all_dataset_summaries_by_region:
+        for column_name in ("number_of_requests", "number_of_downloads"):
+            summary[column_name] = pandas.to_numeric(summary[column_name], errors="coerce").fillna(0).astype("int64")
     aggregated_dataset_summaries_by_region = pandas.concat(objs=all_dataset_summaries_by_region, ignore_index=True)
 
     pre_aggregated = aggregated_dataset_summaries_by_region.groupby(by="region", as_index=False)[
@@ -74,6 +89,9 @@ def generate_archive_summaries(
     aggregated_activity_by_region = aggregated_activity_by_region.astype(
         dtype={"bytes_sent": "int64", "number_of_requests": "int64", "number_of_downloads": "int64"}
     )
+    aggregated_activity_by_region = _privacy_round_request_download_columns(
+        summary_table=aggregated_activity_by_region, minimum=privacy_threshold_minimum
+    )
 
     archive_summary_by_region_file_path = archive_directory / "by_region.tsv"
     aggregated_activity_by_region.to_csv(
@@ -87,7 +105,11 @@ def generate_archive_summaries(
         if summary_file_path.parent.name != "archive" and "<" not in (value := summary_file_path.read_text().strip())
     ]
     total_requester_count: int = sum(requester_counts)
-    archive_requester_count: str = "<50" if total_requester_count < 50 else str(total_requester_count)
+    archive_requester_count: str = (
+        f"<{privacy_threshold_minimum}"
+        if total_requester_count < privacy_threshold_minimum
+        else str(total_requester_count)
+    )
 
     archive_requester_count_file_path = archive_directory / "requester_count.tsv"
     archive_requester_count_file_path.write_text(archive_requester_count)

src/s3_log_extraction/summarize/_generate_archive_totals.py

@@ -4,13 +4,15 @@
 import beartype
 import pandas
 
+from ._generate_summaries import _round_requester_count
 from ..config import get_cache_subdirectory
 from ..ip_utils._globals import EXCLUDED_REGION_LABELS
 
 
 @beartype.beartype
 def generate_archive_totals(
     cache_directory: str | pathlib.Path | None = None,
+    privacy_threshold_minimum: int = 50,
 ) -> None:
     """
     Generate top-level totals of the entire archive from the archive summaries in the mapped S3 logs folder.
@@ -20,13 +22,18 @@ def generate_archive_totals(
     cache_directory : path-like, optional
         The top-level cache directory from which the summary directory is derived.
         If not provided, the default cache directory is used.
+    privacy_threshold_minimum : int
+        Minimum disclosure threshold for privacy-rounded request/download totals.
+        Default is ``50``.
     """
     summary_directory = get_cache_subdirectory(cache_directory=cache_directory, name="summaries")
     archive_directory = summary_directory / "archive"
     archive_directory.mkdir(exist_ok=True)
 
     summary_file_path = archive_directory / "by_region.tsv"
     summary = pandas.read_table(filepath_or_buffer=summary_file_path)
+    for column_name in ("number_of_requests", "number_of_downloads"):
+        summary[column_name] = pandas.to_numeric(summary[column_name], errors="coerce").fillna(0).astype("int64")
 
     unique_countries: set[str] = set()
     for region in summary["region"]:
@@ -60,8 +67,12 @@ def generate_archive_totals(
         "total_bytes_sent": int(summary["bytes_sent"].sum()),
         "number_of_unique_regions": number_of_unique_regions,
         "number_of_unique_countries": number_of_unique_countries,
-        "total_number_of_requests": int(summary["number_of_requests"].sum()),
-        "total_number_of_downloads": int(summary["number_of_downloads"].sum()),
+        "total_number_of_requests": _round_requester_count(
+            count=int(summary["number_of_requests"].sum()), modulo=20, minimum=privacy_threshold_minimum
+        ),
+        "total_number_of_downloads": _round_requester_count(
+            count=int(summary["number_of_downloads"].sum()), modulo=20, minimum=privacy_threshold_minimum
+        ),
         "number_of_requesters": number_of_requesters,
     }
 

src/s3_log_extraction/summarize/_generate_summaries.py

@@ -39,6 +39,16 @@ def _round_requester_count(count: int, modulo: int, minimum: int) -> str | int:
     return round(count / modulo) * modulo
 
 
+def _privacy_round_request_download_columns(
+    summary_table: pandas.DataFrame, *, modulo: int = 20, minimum: int = 50
+) -> pandas.DataFrame:
+    for column_name in ("number_of_requests", "number_of_downloads"):
+        summary_table[column_name] = summary_table[column_name].map(
+            lambda count: _round_requester_count(count=int(count), modulo=modulo, minimum=minimum)
+        )
+    return summary_table
+
+
 def _collect_unique_ips(asset_directories: list[pathlib.Path], use_encryption: bool = True) -> set[str]:
     """
     Collect all unique IP addresses across the given asset directories.
@@ -106,7 +116,10 @@ def _summarize_dataset_requester_count(
 
 
 def generate_summaries(
-    level: int = 0, cache_directory: str | pathlib.Path | None = None, use_encryption: bool = True
+    level: int = 0,
+    cache_directory: str | pathlib.Path | None = None,
+    use_encryption: bool = True,
+    privacy_threshold_minimum: int = 50,
 ) -> None:
     """
     Generate summaries for each dataset in the extraction directory.
@@ -127,6 +140,9 @@ def generate_summaries(
     use_encryption : bool
         If ``True`` (default), ``ips.txt`` and IP cache files are decrypted when read.
         If ``False``, files are read as plaintext.
+    privacy_threshold_minimum : int
+        Minimum disclosure threshold for privacy-rounded requester/request/download
+        values. Default is ``50``.
     """
     if level != 0:
         message = (
@@ -164,6 +180,7 @@ def generate_summaries(
             summary_directory=summary_directory,
             ip_to_region=ip_to_region,
             use_encryption=use_encryption,
+            privacy_threshold_minimum=privacy_threshold_minimum,
         )
 
         all_archive_unique_ips.update(
@@ -172,7 +189,9 @@ def generate_summaries(
     if all_archive_unique_ips:
         archive_directory = summary_directory / "archive"
         archive_directory.mkdir(exist_ok=True)
-        rounded_archive_count = _round_requester_count(count=len(all_archive_unique_ips), modulo=20, minimum=50)
+        rounded_archive_count = _round_requester_count(
+            count=len(all_archive_unique_ips), modulo=20, minimum=privacy_threshold_minimum
+        )
         (archive_directory / "requester_count.tsv").write_text(str(rounded_archive_count))
 
 
@@ -183,29 +202,36 @@ def _summarize_dataset(
     summary_directory: pathlib.Path,
     ip_to_region: dict[str, str],
     use_encryption: bool = True,
+    privacy_threshold_minimum: int = 50,
 ) -> None:
     _summarize_dataset_by_day(
         asset_directories=asset_directories,
         summary_file_path=summary_directory / dataset_id / "by_day.tsv",
+        privacy_threshold_minimum=privacy_threshold_minimum,
     )
     _summarize_dataset_by_asset(
         asset_directories=asset_directories,
         summary_file_path=summary_directory / dataset_id / "by_asset.tsv",
+        privacy_threshold_minimum=privacy_threshold_minimum,
     )
     _summarize_dataset_by_region(
         asset_directories=asset_directories,
         summary_file_path=summary_directory / dataset_id / "by_region.tsv",
         ip_to_region=ip_to_region,
         use_encryption=use_encryption,
+        privacy_threshold_minimum=privacy_threshold_minimum,
     )
     _summarize_dataset_requester_count(
         asset_directories=asset_directories,
         summary_file_path=summary_directory / dataset_id / "requester_count.tsv",
+        minimum=privacy_threshold_minimum,
         use_encryption=use_encryption,
     )
 
 
-def _summarize_dataset_by_day(*, asset_directories: list[pathlib.Path], summary_file_path: pathlib.Path) -> None:
+def _summarize_dataset_by_day(
+    *, asset_directories: list[pathlib.Path], summary_file_path: pathlib.Path, privacy_threshold_minimum: int = 50
+) -> None:
     all_dates = []
     all_bytes_sent = []
     all_downloads = []
@@ -258,10 +284,15 @@ def _summarize_dataset_by_day(*, asset_directories: list[pathlib.Path], summary_
     )
     summary_table.sort_values(by="date", inplace=True)
     summary_table.index = range(len(summary_table))
+    summary_table = _privacy_round_request_download_columns(
+        summary_table=summary_table, minimum=privacy_threshold_minimum
+    )
     summary_table.to_csv(path_or_buf=summary_file_path, mode="w", sep="\t", header=True, index=False)
 
 
-def _summarize_dataset_by_asset(*, asset_directories: list[pathlib.Path], summary_file_path: pathlib.Path) -> None:
+def _summarize_dataset_by_asset(
+    *, asset_directories: list[pathlib.Path], summary_file_path: pathlib.Path, privacy_threshold_minimum: int = 50
+) -> None:
     dataset_id = summary_file_path.parent.name
     extraction_base_path = summary_file_path.parent.parent.parent / "extraction" / dataset_id  # Assumes same cache dir
 
@@ -302,6 +333,9 @@ def _summarize_dataset_by_asset(*, asset_directories: list[pathlib.Path], summar
             "number_of_downloads": [number_of_downloads_by_asset[path] for path in all_asset_paths],
         }
     )
+    summary_table = _privacy_round_request_download_columns(
+        summary_table=summary_table, minimum=privacy_threshold_minimum
+    )
     summary_table.to_csv(path_or_buf=summary_file_path, mode="w", sep="\t", header=True, index=False)
 
 
@@ -311,6 +345,7 @@ def _summarize_dataset_by_region(
     summary_file_path: pathlib.Path,
     ip_to_region: dict[str, str],
     use_encryption: bool = True,
+    privacy_threshold_minimum: int = 50,
 ) -> None:
     all_regions = []
     all_bytes_sent = []
@@ -359,4 +394,7 @@ def _summarize_dataset_by_region(
             "number_of_downloads": [number_of_downloads_by_region[region] for region in all_regions_ordered],
         }
     )
+    summary_table = _privacy_round_request_download_columns(
+        summary_table=summary_table, minimum=privacy_threshold_minimum
+    )
     summary_table.to_csv(path_or_buf=summary_file_path, mode="w", sep="\t", header=True, index=False)

tests/expected_output/summaries/archive/by_day.tsv

@@ -1,6 +1,6 @@
 date	bytes_sent	number_of_requests	number_of_downloads
-2020-01-01	6286489	3	2
-2022-04-06	12	1	1
-2022-08-03	0	1	0
-2023-11-13	0	1	1
-2024-11-13	1194552	1	1
+2020-01-01	6286489	<50	<50
+2022-04-06	12	<50	<50
+2022-08-03	0	<50	<50
+2023-11-13	0	<50	<50
+2024-11-13	1194552	<50	<50

tests/expected_output/summaries/archive/by_region.tsv

@@ -1,2 +1,2 @@
 region	bytes_sent	number_of_requests	number_of_downloads
-missing	7481053	7	5
+missing	7481053	<50	<50

tests/expected_output/summaries/archive_totals.json

@@ -1 +1 @@
-{"total_bytes_sent": 7481053, "number_of_unique_regions": 1, "number_of_unique_countries": 0, "total_number_of_requests": 7, "total_number_of_downloads": 5, "number_of_requesters": "<50"}
+{"total_bytes_sent": 7481053, "number_of_unique_regions": 1, "number_of_unique_countries": 0, "total_number_of_requests": "<50", "total_number_of_downloads": "<50", "number_of_requesters": "<50"}

tests/expected_output/summaries/ds001161/by_asset.tsv

@@ -1,5 +1,5 @@
 asset_path	bytes_sent	number_of_requests	number_of_downloads
-dataset_description.json	1194552	1	1
-sub-23/anat/sub-23_run-01_T1w.nii.gz	0	1	1
-sub-23/func/sub-23_task-effortmixedgambles_run-01_bold.nii.gz	0	1	0
-task-effortmixedgambles_bold.json	12	1	1
+dataset_description.json	1194552	<50	<50
+sub-23/anat/sub-23_run-01_T1w.nii.gz	0	<50	<50
+sub-23/func/sub-23_task-effortmixedgambles_run-01_bold.nii.gz	0	<50	<50
+task-effortmixedgambles_bold.json	12	<50	<50

tests/expected_output/summaries/ds001161/by_day.tsv

@@ -1,5 +1,5 @@
 date	bytes_sent	number_of_requests	number_of_downloads
-2022-04-06	12	1	1
-2022-08-03	0	1	0
-2023-11-13	0	1	1
-2024-11-13	1194552	1	1
+2022-04-06	12	<50	<50
+2022-08-03	0	<50	<50
+2023-11-13	0	<50	<50
+2024-11-13	1194552	<50	<50