Skip to main content

functions

Pure functions for CSV report generation.

Extracted from bitfount.federated.algorithms.csv_report_algorithm._WorkerSide so that the same logic can be called from composable Prefect steps without depending on class inheritance.

Module​

Functions​

apply_csv_extensions​

def apply_csv_extensions(    df: pd.DataFrame, csv_extensions: list[str] | None = None,) ‑> tuple[pandas.core.frame.DataFrame, list[str]]:

Apply named DataFrame generation extensions.

Returns Tuple of (extended DataFrame, list of newly generated column names).

apply_filters​

def apply_filters(    df: pd.DataFrame,    filters: list[ColumnFilter | MethodFilter],    rename_columns: Mapping[str, str] | None = None,) ‑> tuple[pandas.core.frame.DataFrame, list[str]]:

Apply eligibility filters to a DataFrame.

Adds FILTER_MATCHING_COLUMN and FILTER_FAILED_REASON_COLUMN.

Returns Tuple of (filtered DataFrame, list of new filter-generated column names).

combine_results_dataframes​

def combine_results_dataframes(    results_df: pd.DataFrame | list[pd.DataFrame], filenames: list[str] | None = None,) ‑> pandas.core.frame.DataFrame:

Merge or concatenate multiple result DataFrames.

If results_df is a list and filenames is provided, merges all DataFrames on the filename column. Otherwise concatenates column-wise.

drop_columns​

def drop_columns(    df: pd.DataFrame,    columns_to_drop: list[str] | None = None,    columns_to_drop_prefix: list[str] | None = None,) ‑> pandas.core.frame.DataFrame:

Drop specified columns and columns matching prefix patterns.

fold_filter_result_columns​

def fold_filter_result_columns(    original_cols: list[str] | None, csv_report_df: pd.DataFrame,) ‑> Optional[list[str]]:

Extend original_cols with csv_report_df's dynamic filter-result columns.

tabulate_criteria_outcomes.evaluations_to_csv_report_df projects each active eligibility filter's result onto csv_report_df under dynamically-named columns: a ColumnFilter's own "{col} {op} {value}" boolean column, a code-list MethodFilter's own boolean column (and, conditionally, its "Context for {name}" column), plus the sentinel FILTER_MATCHING_COLUMN / FILTER_FAILED_REASON_COLUMN / FILTER_CONTEXT_COLUMN columns.

csv_report never re-runs filters (process_csv_dataframe is called with filters=None), so the legacy if filters: branch that used to fold these into the column selection (via apply_filters's filter_new_cols) never runs. Without this, select_and_rename_columns would silently drop every one of those dynamic columns whenever original_cols is non-empty — which it is for every Heidelberg / DICOM / TopconSource template (see get_default_columns_for_datasource).

The newly-folded columns are appended in csv_report_df's own column order, which reproduces legacy's resulting column order byte-for-byte for a ColumnFilter-only configuration (the only kind any current ga_trial_bronze_v9.yaml-style template activates — a ColumnFilter only ever contributes a single new column, so there is no per-filter ordering to reproduce). A code-list MethodFilter's own multi-column contribution may land in a different relative order than legacy (which sorts a filter's own new columns alphabetically via pd.Index.difference before appending them); this is unreached until a template configures conditions_inclusion/conditions_exclusion/procedures_inclusion/ procedures_exclusion (Phase C).

Arguments

  • original_cols: The datasource's default column list, or None (an unrestricted select_and_rename_columns selection already keeps every column, so there is nothing to fold).
  • csv_report_df: The tabulate_criteria_outcomes eligibility projection merged onto the report frame.

Returns original_cols extended with any csv_report_df columns not already present (excluding the filename column), or None if original_cols was None.

format_eligibility_column​

def format_eligibility_column(    df: pd.DataFrame, rename_columns: Mapping[str, str] | None = None,) ‑> pandas.core.frame.DataFrame:

Convert boolean FILTER_MATCHING_COLUMN to Eligible/Not eligible strings.

generate_csv_filename​

def generate_csv_filename(    trial_name: str | None = None, task_id: str | None = None, suffix: str = '.csv',) ‑> str:

Generate a CSV output filename from trial name and task ID.

get_default_columns_for_datasource​

def get_default_columns_for_datasource(datasource_type_name: str) ‑> list[str]:

Return the default original columns list based on datasource type name.

process_csv_dataframe​

def process_csv_dataframe(    df: pd.DataFrame,    filters: list[ColumnFilter | MethodFilter] | None = None,    original_cols: list[str] | None = None,    columns_to_include: list[str] | None = None,    rename_columns: Mapping[str, str] | None = None,    columns_to_drop: list[str] | None = None,    columns_to_drop_prefix: list[str] | None = None,    csv_extensions: list[str] | None = None,    trial_name: str | None = None,    decimal_places: int = 2,    produce_matched_only: bool = False,) ‑> pandas.core.frame.DataFrame:

Full CSV processing pipeline: filter → extend → drop → select → rename → round.

Arguments

  • df: Input DataFrame (source data merged with results).
  • filters: Eligibility filters to apply.
  • original_cols: Default column set for the datasource type.
  • columns_to_include: Explicit column whitelist (overrides original_cols).
  • rename_columns: Column rename mapping for final output.
  • columns_to_drop: Columns to explicitly remove.
  • columns_to_drop_prefix: Column prefixes to remove.
  • csv_extensions: Named extensions to generate additional columns.
  • trial_name: Trial name to add as a column.
  • decimal_places: Precision for float rounding.
  • produce_matched_only: If True, only include rows matching all filters.

Returns Processed DataFrame ready for CSV output.

read_csv​

def read_csv(    path: Path, encryption_key: bytes | None = None, **kwargs: Any,) ‑> pandas.core.frame.DataFrame:

Read a CSV file, handling encrypted files if needed.

round_numeric_columns​

def round_numeric_columns(    df: pd.DataFrame, precision: int = 2,) ‑> pandas.core.frame.DataFrame:

Round all float columns to the given precision.

select_and_rename_columns​

def select_and_rename_columns(    df: pd.DataFrame,    original_cols: list[str] | None = None,    columns_to_include: list[str] | None = None,    rename_columns: Mapping[str, str] | None = None,    extension_generated_cols: list[str] | None = None,) ‑> pandas.core.frame.DataFrame:

Select relevant columns and rename for final output.

Priority: columns_to_include > original_cols > all columns. Extension-generated columns are always included.

sort_csv​

def sort_csv(    path: Path,    sort_by: list[str],    ascending: bool = True,    encryption_key: bytes | None = None,) ‑> None:

Read a CSV file, sort it, and write it back.

If none of the sort_by columns exist in the file, the function returns early without performing any I/O to avoid a needless read-write cycle.

write_csv​

def write_csv(    df: pd.DataFrame,    path: Path,    encryption_key: bytes | None = None,    append: bool = False,    **kwargs: Any,) ‑> pathlib.Path:

Write a DataFrame to CSV, optionally encrypted.

Arguments

  • df: DataFrame to write.
  • path: Output file path.
  • encryption_key: If provided, uses encrypted CSV output.
  • append: If True, appends to existing file.
  • **kwargs: Additional args passed to to_csv / encryption functions.

Returns The path to the written file.