diff --git a/README.md b/README.md index 02b3f7e..33fb3c1 100644 --- a/README.md +++ b/README.md @@ -19,64 +19,64 @@ Helper library for enabling data studies: utilities for study enablement such as ### Date shifting (basic example) ```python -from nuh_helper import shift_excel_dates -# or: from nuh_helper.date_shift import shift_excel_dates +from nuh_helper import shift_excel_dates_inplace + + + source_file = Path(__file__).parent / ( + "data/structural.with-blank.xlsx" if good else "data/structural.bad-blank.xlsx" + ) + output_path = tmp_path / "target.xlsx" + linking_table_old = tmp_path / "linking_table_old.csv" + linking_table_out = tmp_path / "linking_table_out.csv" # Configure which sheets and columns to shift sheet_configs = { - "patients": { - "patient_id_col": "patient_id", - "date_columns": ["dob", "date_of_diagnosis"], - "header_row": 1, # Optional: zero-based row index for column names - }, - "labs": { - "patient_id_col": "patient_id", - "date_columns": ["test_date"], - "header_row": 1, + + # shift the dates in the "paige" sheet + "paige": { + # patient IDs are in the "ptid" column + "patient_id_col": "ptid", + "date_columns": [ + # shift the dates in the DOB column + "dob", + ], + "text_columns": [ + # the food column has no dates + "food", + ], + "header_row": 0, # Optional: zero-based row index for column names. defaults to 0 + "skip_rows_after_header": [], }, + + # ignore the "stuff" page + "stuff": "skip", } + # Shift dates in the Excel file -shift_excel_dates( +shift_excel_dates_inplace( + # input / output files input_file="input.xlsx", output_file="output.xlsx", - patient_sheet="patients", - patient_id_col="patient_id", + + # columns used to compute a list of patients + patient_sheet="paige", + patient_id_col="ptid", + + sheet_configs=sheet_configs, min_shift_days=-15, # Lower range max_shift_days=15, # Upper range seed=42, # For reproducibility - date_format="YYYY-MM-DD", -) -``` - -### Excluding fixed study dates with `shift_exceptions` - -Some columns contain a mix of patient-specific dates (which should be shifted) and fixed study-wide dates (e.g. an end-of-study date) that must remain unchanged. Use `shift_exceptions` in `sheet_configs` to list any date values that should never be shifted: - -```python -sheet_configs = { - "patients": { - "patient_id_col": "patient_id", - "date_columns": ["last_alive"], - "shift_exceptions": { - "last_alive": ["2024-12-31"], # end-of-study date — never shift - }, - }, -} -shift_excel_dates( - input_file="input.xlsx", - output_file="output.xlsx", - patient_sheet="patients", - patient_id_col="patient_id", - sheet_configs=sheet_configs, - seed=42, + # optional - csv files saving how many days each patient was shifted + # ... do not distribute these + # ... and probably best to keep them somewhere + linking_table_path="linking_table_old.csv", + linking_table_output="linking_table_new.csv", ) ``` -The exception date strings are parsed with the same flexible parser used for all date values (supports multiple formats and placeholder strings). Exceptions are matched against the parsed date, so `"2024-12-31"` and `"31-12-2024"` both match the same calendar date. - ### Reproducible Shifts with Linking Table To use the same shifts across multiple runs, save and reuse a linking table: @@ -118,8 +118,30 @@ shift_excel_dates_inplace( output_file="output.xlsx", patient_sheet="patients", sheet_configs=sheet_configs, - seed=42, - linking_table_output="shift_mappings.csv", + min_shift_days=-15, # Lower range + max_shift_days=15, # Upper range + seed=42, # For reproducibility + + # optional - csv files saving how many days each patient was shifted + # ... do not distribute these + # ... and probably best to keep them somewhere specific + linking_table_output="linking_table_first.csv", +) + +# Second run: reuse the same shifts +shift_excel_dates_inplace( + input_file="input_final.xlsx", + output_file="output_second.xlsx", + patient_sheet="paige", + patient_id_col="ptid", + sheet_configs=sheet_configs, + min_shift_days=-15, # Lower range + max_shift_days=15, # Upper range + seed=42, # For reproducibility + + # write to a new file this time + linking_table_path="linking_table_first.csv", + linking_table_output="linking_table_second.csv", ) ``` diff --git a/nuh_helper/__init__.py b/nuh_helper/__init__.py index 6e863d0..e020d8b 100644 --- a/nuh_helper/__init__.py +++ b/nuh_helper/__init__.py @@ -11,19 +11,11 @@ logging.getLogger(__name__).addHandler(logging.NullHandler()) from nuh_helper.date_shift import ( # noqa: E402 - apply_date_shifts, - generate_shift_mappings, - load_shift_mappings, - shift_excel_dates, shift_excel_dates_inplace, ) from nuh_helper.profile import generate_scan_report # noqa: E402 __all__ = [ - "shift_excel_dates", - "shift_excel_dates_inplace", - "apply_date_shifts", - "generate_shift_mappings", - "load_shift_mappings", - "generate_scan_report", + generate_scan_report, + shift_excel_dates_inplace, ] diff --git a/nuh_helper/book_page/__init__.py b/nuh_helper/book_page/__init__.py index bf02025..62dd080 100644 --- a/nuh_helper/book_page/__init__.py +++ b/nuh_helper/book_page/__init__.py @@ -1,4 +1,4 @@ -from nuh_helper.book_page.book import BasePage, BaseWorkBook, normalize_row +from nuh_helper.book_page.book import BasePage, BaseWorkBook, normalize_row, row_val_t from nuh_helper.book_page.csv_book import CSVWorkBook from nuh_helper.book_page.generic import book_open from nuh_helper.book_page.xlsx_book import XLSXWorkBook @@ -10,4 +10,5 @@ BasePage, book_open, normalize_row, + row_val_t, ] diff --git a/nuh_helper/book_page/book.py b/nuh_helper/book_page/book.py index 427a4e1..c8ae838 100644 --- a/nuh_helper/book_page/book.py +++ b/nuh_helper/book_page/book.py @@ -1,6 +1,6 @@ from abc import ABC, abstractmethod from collections.abc import Generator as gen -from collections.abc import Iterable as iterator +from collections.abc import Iterable as iterable from datetime import date, datetime from functools import cached_property from pathlib import Path @@ -17,7 +17,7 @@ def __init__(self, name: str, page: object) -> None: @abstractmethod def _stream_rows_impl(self) -> gen[tuple[int, row_val_t]]: """performs the workbook specific yield/write of the row data""" - raise NotImplementedError() + ... def stream_rows(self) -> gen[tuple[int, row_val_t]]: """public method for reading a RO stream of rows from a specific page""" @@ -41,17 +41,15 @@ def __contains__(self, name: str) -> bool: return name in self.keys() @abstractmethod - def __getitem__(self, name: str) -> gen[BasePage]: - raise NotImplementedError() + def __getitem__(self, name: str) -> BasePage: ... @abstractmethod - def keys(self) -> iterator[str]: - raise NotImplementedError() + def keys(self) -> iterable[str]: ... @abstractmethod def _stream_copy_impl(self, into: Path) -> gen[tuple[str, int, row_val_t]]: """performs the workbook specific yield/write of the row data""" - raise NotImplementedError() + ... def stream_copy( self, @@ -71,6 +69,9 @@ def stream_copy( date_format = date_format if date_format else "%Y-%m-%d" for page_name, row_idx, row_data in self._stream_copy_impl(into): + if BaseWorkBook.normalize_values: + for raw_idx, raw_val in enumerate(row_data): + row_data[raw_idx] = normalize_val(raw_val) yield page_name, row_idx, row_data if BaseWorkBook.normalize_values: for raw_idx, raw_val in enumerate(row_data): @@ -81,11 +82,10 @@ def stream_copy( def __iter__(self) -> gen[BasePage]: for name in self.keys(): - with self[name] as page: - yield page + yield self[name] -def normalize_val(src: object) -> object: +def normalize_val(src: object) -> cell_val_t: """parse to a datetime, .strip(), turn "" into None, retain bool""" if ( @@ -131,7 +131,7 @@ def normalize_val(src: object) -> object: return src -def normalize_row(row: list[str]) -> list[str]: +def normalize_row(row: iterable[str]) -> row_val_t: """modifies (and returns) the passed array to drop extra whitespace and replace fully blank entries with None. doesn't touch non-strings""" diff --git a/nuh_helper/book_page/csv_book.py b/nuh_helper/book_page/csv_book.py index deb0f40..fbcf1e2 100644 --- a/nuh_helper/book_page/csv_book.py +++ b/nuh_helper/book_page/csv_book.py @@ -1,4 +1,3 @@ -import contextlib import csv from _collections_abc import dict_keys from collections.abc import Generator as gen @@ -16,8 +15,7 @@ def __init__(self, path: str | Path | list[Path]) -> None: def keys(self) -> dict_keys: return self._page_files.keys() - @contextlib.contextmanager - def __getitem__(self, name: str) -> gen[BasePage]: + def __getitem__(self, name: str) -> BasePage: class CSVPage(BasePage): def _stream_rows_impl(self) -> gen[tuple[int, row_val_t]]: @@ -32,6 +30,7 @@ def _stream_rows_impl(self) -> gen[tuple[int, row_val_t]]: # strip bom if ( row == 0 + and line and isinstance(line[0], str) and line[0][:1] == "\ufeff" ): @@ -40,7 +39,7 @@ def _stream_rows_impl(self) -> gen[tuple[int, row_val_t]]: yield (row, line) assert isinstance(self._page_files[name], Path) - yield CSVPage(name, self._page_files[name]) + return CSVPage(name, self._page_files[name]) def _stream_copy_impl(self, into: Path) -> gen[tuple[str, int, row_val_t]]: diff --git a/nuh_helper/book_page/xlsx_book.py b/nuh_helper/book_page/xlsx_book.py index 4d04ea0..16bb358 100644 --- a/nuh_helper/book_page/xlsx_book.py +++ b/nuh_helper/book_page/xlsx_book.py @@ -1,11 +1,11 @@ -import contextlib import shutil from collections.abc import Generator as gen from pathlib import Path import openpyxl +from openpyxl.cell.read_only import EmptyCell -from .book import BasePage, BaseWorkBook +from .book import BasePage, BaseWorkBook, row_val_t class XLSXWorkBook(BaseWorkBook): @@ -15,12 +15,12 @@ def __init__(self, file: str | Path, read_only: bool = True) -> None: if not file.is_file(): raise RuntimeError(f"xlsx file not found {file}") self._file = file - self._book = openpyxl.load_workbook(file, read_only=read_only) + self._book = openpyxl.load_workbook(file, read_only=read_only, data_only=True) def keys(self) -> list[str]: return [page.title for page in self._book.worksheets] - def _stream_copy_impl(self, into: str | Path) -> gen[tuple[str, int, list[str]]]: + def _stream_copy_impl(self, into: str | Path) -> gen[tuple[str, int, row_val_t]]: if not isinstance(into, Path): into = Path(into) @@ -41,7 +41,6 @@ def _stream_copy_impl(self, into: str | Path) -> gen[tuple[str, int, list[str]]] page.cell(r + 1, c + 1).value = val the_copy.save(into) - @contextlib.contextmanager def __getitem__(self, name: str) -> gen[BasePage]: if len(name) > 31: @@ -49,13 +48,13 @@ def __getitem__(self, name: str) -> gen[BasePage]: class XLSXPage(BasePage): def _stream_rows_impl(self) -> gen[tuple[int, list[str]]]: - for row in range(self._page.max_row): + for row_idx, row_data in enumerate(self._page.iter_rows()): yield ( - row, + row_idx, [ - self._page.cell(row + 1, col + 1).value - for col in range(self._page.max_column) + None if isinstance(cell, EmptyCell) else cell.internal_value + for cell in row_data ], ) - yield XLSXPage(name, self._book[name]) + return XLSXPage(name, self._book[name]) diff --git a/nuh_helper/date_shift/__init__.py b/nuh_helper/date_shift/__init__.py index c05223d..1dc2fd4 100644 --- a/nuh_helper/date_shift/__init__.py +++ b/nuh_helper/date_shift/__init__.py @@ -5,329 +5,11 @@ in an Excel file, with support for reproducible shifts using a linking table. """ -import logging -from datetime import date, datetime -from pathlib import Path -from typing import Any, cast - -import pandas as pd - -from nuh_helper.date_shift import _excel, _parse, mappings -from nuh_helper.date_shift.mappings import ( # noqa: E402 - generate_shift_mappings, - load_shift_mappings, -) from nuh_helper.date_shift.stream_shift import ( shift_excel_dates_inplace, ) -logger = logging.getLogger(__name__) - - -def _get_patient_ids_and_shift_mappings( - input_file: Path, - patient_sheet: str, - patient_id_col: str, - sheet_configs: dict[str, dict[str, Any]], - min_shift_days: int, - max_shift_days: int, - linking_table_path: str | None, - seed: int | None, - patient_header_row: int, - patient_skip_rows: list[int] | None, -) -> tuple[list[str], pd.DataFrame]: - """ - Read patient IDs from the patient sheet and resolve shift mappings - (load from CSV or generate). Returns (patient_ids, shift_mappings) - with shift_mappings already normalized and deduplicated. - """ - effective_patient_header_row = patient_header_row - effective_patient_skip_rows = patient_skip_rows - if patient_sheet in sheet_configs: - cfg = sheet_configs[patient_sheet] - effective_patient_header_row = cast( - int, cfg.get("header_row", patient_header_row) - ) - effective_patient_skip_rows = cfg.get( - "skip_rows_after_header", patient_skip_rows - ) - - patient_excel = pd.ExcelFile(input_file, engine="openpyxl") - patient_df, _, _, _ = _excel._read_sheet_with_structure( - patient_excel, - sheet_name=patient_sheet, - header_row=effective_patient_header_row, - input_file=input_file, - skip_rows_after_header=effective_patient_skip_rows, - ) - if patient_id_col not in patient_df.columns: - raise mappings.PatientColumnMissing(patient_sheet, patient_id_col) - - patient_ids = ( - patient_df[patient_id_col] - .apply(_parse._normalize_patient_id) - .dropna() - .unique() - .tolist() - ) - logger.info("Found %d patient(s) in sheet '%s'", len(patient_ids), patient_sheet) - - if linking_table_path and Path(linking_table_path).exists(): - logger.info("Loading shift mappings from '%s'", linking_table_path) - shift_mappings = mappings.load_shift_mappings(linking_table_path) - shift_mappings = shift_mappings[shift_mappings["patient_id"].isin(patient_ids)] - existing_ids = set(shift_mappings["patient_id"]) - missing_ids = [pid for pid in patient_ids if pid not in existing_ids] - if missing_ids: - logger.warning( - "%d patient(s) had no entry in the linking table; new shifts generated", - len(missing_ids), - ) - new_shifts = mappings.generate_shift_mappings( - missing_ids, min_shift_days, max_shift_days, seed - ) - shift_mappings = pd.concat([shift_mappings, new_shifts], ignore_index=True) - else: - logger.info("Generating shift mappings for %d patient(s)", len(patient_ids)) - shift_mappings = mappings.generate_shift_mappings( - patient_ids, min_shift_days, max_shift_days, seed - ) - - shift_mappings["patient_id"] = shift_mappings["patient_id"].apply( - _parse._normalize_patient_id - ) - shift_mappings = shift_mappings.dropna(subset=["patient_id"]).drop_duplicates( - subset=["patient_id"], keep="first" - ) - return (patient_ids, shift_mappings) - - -def apply_date_shifts( - df: pd.DataFrame, - patient_id_col: str, - date_columns: list[str], - shift_mappings: pd.DataFrame, - date_format: str | None = None, - shift_exceptions: dict[str, list[str]] | None = None, -) -> pd.DataFrame: - """ - Apply date shifts to specified columns in a DataFrame. - - Args: - df: pd.DataFrame containing patient data. - patient_id_col: Name of the column containing patient IDs. - date_columns: List of column names containing dates to shift. - shift_mappings: DataFrame with 'patient_id' and 'shift_days' columns. - date_format: Optional date format string (e.g., 'YYYY-MM-DD'). - Note: This parameter is kept for API compatibility but formatting - is applied at the Excel cell level, not in the DataFrame. - shift_exceptions: Optional dict mapping column names to lists of date strings - that should never be shifted (e.g. a fixed end-of-study date). - - Returns: - DataFrame with shifted dates. - """ - df = df.copy() - - # Normalize patient IDs in the working DataFrame to align with mapping keys - df[patient_id_col] = df[patient_id_col].apply(_parse._normalize_patient_id) - - shift_dict = dict( - zip( - shift_mappings["patient_id"], - shift_mappings["shift_days"], - strict=True, - ) - ) - - # Pre-parse exception dates once per column - parsed_exceptions: dict[str, set[date]] = {} - if shift_exceptions: - for col, exc_values in shift_exceptions.items(): - parsed_set: set[date] = set() - for v in exc_values: - ts = _parse._parse_date_value(v) - if ts is not None: - parsed_set.add(ts.date()) - if parsed_set: - parsed_exceptions[col] = parsed_set - - for date_col in date_columns: - if date_col not in df.columns: - logger.warning( - "Date column '%s' not found in DataFrame, skipping", date_col - ) - continue - - # Parse flexible date strings (handles YYYY-DD-MM and placeholders "Unknown") - non_null_before = df[date_col].notna().sum() - df[date_col] = df[date_col].apply(_parse._parse_date_value) - parse_failures = non_null_before - sum(x is not None for x in df[date_col]) - if parse_failures > 0: - logger.debug( - "Column '%s': %d value(s) could not be parsed as dates", - date_col, - parse_failures, - ) - - # Apply shifts - exc_dates = parsed_exceptions.get(date_col, set()) - df[date_col] = df.apply( - lambda row: ( - row[date_col] # noqa: B023 - + pd.Timedelta(days=shift_dict.get(row[patient_id_col], 0)) - if row[date_col] is not None # noqa: B023 - and row[patient_id_col] in shift_dict # noqa: B023 - and row[date_col].date() not in exc_dates # noqa: B023 - else row[date_col] # noqa: B023 - ), - axis=1, - ) - - # Convert back to date-only format (removes time component) - df[date_col] = df[date_col].apply( - lambda x: ( - x.date() if isinstance(x, pd.Timestamp | datetime | date) else None - ), - ) - - return df - - -def shift_excel_dates( - input_file: str, - output_file: str, - patient_sheet: str, - patient_id_col: str, - sheet_configs: dict[str, dict[str, Any]], - min_shift_days: int = -15, - max_shift_days: int = 15, - linking_table_path: str | None = None, - linking_table_output: str | None = None, - seed: int | None = None, - patient_header_row: int = 0, - patient_skip_rows: list[int] | None = None, - date_format: str | None = None, -) -> None: - """ - Shift dates in an Excel file for patient IDs consistently across sheets. - - Args: - input_file: Path to input Excel file. - output_file: Path to output Excel file with shifted dates. - patient_sheet: Name of the sheet containing patient IDs. - patient_id_col: Name of the column containing patient IDs in the patient sheet. - sheet_configs: Dictionary mapping sheet names to configuration dicts. - Each config dict should have: - - 'patient_id_col': Name of patient ID column in that sheet - - 'date_columns': List of date column names to shift - Optional per-sheet header handling: - - 'header_row': zero-based row index of the column names (default 0) - - 'skip_rows_after_header': list of zero-based row indices to exclude - from data (e.g. a data-type row immediately below the header) - min_shift_days: Minimum number of days to shift (default: -15). - max_shift_days: Maximum number of days to shift (default: 15). - linking_table_path: Optional path to existing linking table CSV for reproducibility. - linking_table_output: Path to save the linking table CSV (default: 'shift_mappings.csv'). - seed: Optional random seed for generating shifts. - patient_header_row: Zero-based header row index for the patient sheet (default: 0). - patient_skip_rows: Optional zero-based row indices to exclude from patient data - (e.g. a data-type row immediately below the header). - date_format: Optional Excel date format string (e.g., 'YYYY-MM-DD', 'yyyy-mm-dd'). - If None, Excel's default date format is used. - Common formats: 'YYYY-MM-DD', 'MM/DD/YYYY', 'DD-MM-YYYY', etc. - """ # noqa: E501 - logger.info("Shifting dates: '%s' → '%s'", input_file, output_file) - logger.debug( - "Shift range: %d to %d days, seed=%s", - min_shift_days, - max_shift_days, - seed, - ) - - _patient_ids, shift_mappings = _get_patient_ids_and_shift_mappings( - input_file=input_file, - patient_sheet=patient_sheet, - patient_id_col=patient_id_col, - sheet_configs=sheet_configs, - min_shift_days=min_shift_days, - max_shift_days=max_shift_days, - linking_table_path=linking_table_path, - seed=seed, - patient_header_row=patient_header_row, - patient_skip_rows=patient_skip_rows, - ) - - with pd.ExcelWriter(output_file, engine="openpyxl") as writer: - excel_file = pd.ExcelFile(input_file, engine="openpyxl") - - for sheet_name in excel_file.sheet_names: - default_header_row = 0 - header_row = default_header_row - sheet_date_columns: list[str] | None = None - skip_rows_after_header: list[int] | None = None - sheet_shift_exceptions: dict[str, list[str]] | None = None - - if sheet_name in sheet_configs: - config = sheet_configs[cast(str, sheet_name)] - sheet_patient_id_col: str = cast(str, config["patient_id_col"]) - date_columns: list[str] = cast(list[str], config["date_columns"]) - header_row = cast(int, config.get("header_row", header_row)) - skip_rows_after_header = config.get("skip_rows_after_header") - sheet_shift_exceptions = config.get("shift_exceptions") - sheet_date_columns = date_columns - logger.info( - "Shifting %d date column(s) in sheet '%s'", - len(date_columns), - sheet_name, - ) - - df, _description_df, description_rows, description_merged_ranges = ( - _excel._read_sheet_with_structure( - excel_file, - sheet_name=cast(str, sheet_name), - header_row=header_row, - input_file=input_file, - skip_rows_after_header=skip_rows_after_header, - ) - ) - - if sheet_name in sheet_configs: - if sheet_patient_id_col not in df.columns: - raise ValueError( - f"Patient ID column '{sheet_patient_id_col}' not found in sheet '{sheet_name}'" # noqa: E501 - ) - - df = apply_date_shifts( - df, - sheet_patient_id_col, - date_columns, - shift_mappings, - date_format=None, - shift_exceptions=sheet_shift_exceptions, - ) - - _excel._write_sheet_with_structure( - writer, - sheet_name=cast(str, sheet_name), - data_df=df, - description_rows=description_rows, - header_row=header_row, - date_columns=sheet_date_columns, - date_format=date_format, - description_merged_ranges=description_merged_ranges, - ) - - logger.info("Output written to '%s'", output_file) - - linking_path = linking_table_output or "shift_mappings.csv" - shift_mappings.to_csv(linking_path, index=False) - logger.info("Linking table saved to '%s'", linking_path) - - # Re-export public API __all__ = [ - generate_shift_mappings, - load_shift_mappings, shift_excel_dates_inplace, ] diff --git a/nuh_helper/date_shift/_excel.py b/nuh_helper/date_shift/_excel.py deleted file mode 100644 index 204b4a5..0000000 --- a/nuh_helper/date_shift/_excel.py +++ /dev/null @@ -1,231 +0,0 @@ -"""Excel sheet structure helpers for date-shift.""" - -import contextlib -from typing import Any, cast - -import pandas as pd -from openpyxl import load_workbook -from openpyxl.cell.cell import MergedCell -from openpyxl.worksheet.worksheet import Worksheet - - -def _get_row_values_resolving_merged( - sheet: Worksheet, row_1based: int, max_col: int -) -> list[Any]: - """ - Return values for one row, resolving merged cells to the top-left cell value. - - openpyxl stores the value only in the first cell of a merged range; - other cells are MergedCell and have no value. This walks the row and - fills in the value from the merge range's top-left for each column. - """ - result: list[Any] = [] - for col in range(1, max_col + 1): - cell = sheet.cell(row=row_1based, column=col) - if isinstance(cell, MergedCell): - for merged_range in sheet.merged_cells.ranges: - if cell.coordinate in merged_range: - top_left = sheet.cell( - row=merged_range.min_row, column=merged_range.min_col - ) - result.append(top_left.value) - break - else: - result.append(None) - else: - result.append(cell.value) - return result - - -def _description_merged_ranges( - sheet: Worksheet, num_description_rows: int -) -> list[str]: - """Merged range refs (e.g. 'A1:D1') in the first num_description_rows.""" - result: list[str] = [] - for merged_range in sheet.merged_cells.ranges: - if merged_range.max_row <= num_description_rows: - result.append(str(merged_range)) - return result - - -def _read_sheet_with_structure( - excel_file: pd.ExcelFile, - sheet_name: str, - header_row: int = 0, - input_file: str | None = None, - skip_rows_after_header: list[int] | None = None, -) -> tuple[pd.DataFrame, pd.DataFrame, list[list[Any]], list[str]]: - """ - Read a sheet preserving description rows and structure. - - Uses openpyxl to resolve merged cells in the header row so column names - are correct when the sheet has merged cells. Rows whose indices are in - skip_rows_after_header (0-based) are excluded from the data (e.g. a - data-type row immediately below the header). - - Returns: - Tuple of (data_df, description_df, description_rows, - description_merged_ranges) - - data_df: DataFrame with header row as column names and data rows - - description_df: DataFrame with description rows (if any) - - description_rows: List of description row data (for writing back) - - description_merged_ranges: Merged range refs to preserve when writing - """ - # Read entire sheet without header to preserve all rows - full_df = pd.read_excel(excel_file, sheet_name=sheet_name, header=None) - max_col = full_df.shape[1] - description_merged_ranges: list[str] = [] - - # Resolve header row (and optional description merged ranges) via openpyxl - if input_file and max_col: - wb = load_workbook(input_file, read_only=False, data_only=True) - if sheet_name in wb.sheetnames: - ws = wb[sheet_name] - # Header row in 1-based openpyxl - header_row_1based = header_row + 1 - header_values = _get_row_values_resolving_merged( - ws, header_row_1based, max_col - ) - # Pandas-friendly column names (no None) - columns = [] - for i, v in enumerate(header_values): - if v is None or (isinstance(v, float) and pd.isna(cast("float", v))): - columns.append(f"Unnamed: {i}") - else: - columns.append(str(v).strip() or f"Unnamed: {i}") - - if header_row > 0: - description_merged_ranges = _description_merged_ranges(ws, header_row) - - # Description rows (rows before header) - description_rows = ( - full_df.iloc[:header_row].values.tolist() if header_row > 0 else [] - ) - # Data rows: everything after header, optionally excluding some rows - data_block = full_df.iloc[header_row + 1 :] - if skip_rows_after_header: - # Drop by 0-based index (data_block index = original row) - to_drop = [i for i in data_block.index if i in skip_rows_after_header] - data_block = data_block.drop(index=to_drop, errors="ignore") - - data_df = pd.DataFrame(data_block.values, columns=columns) - description_df = ( - full_df.iloc[:header_row].copy() if header_row > 0 else pd.DataFrame() - ) - wb.close() - return ( - data_df, - description_df, - description_rows, - description_merged_ranges, - ) - - # Fallback when no openpyxl path or empty sheet - if header_row == 0: - description_rows = [] - description_df = pd.DataFrame() - data_df = pd.read_excel(excel_file, sheet_name=sheet_name, header=0) - if skip_rows_after_header: - data_df = data_df.drop(index=skip_rows_after_header, errors="ignore") - else: - description_rows = full_df.iloc[:header_row].values.tolist() - description_df = full_df.iloc[:header_row].copy() - data_df = pd.read_excel(excel_file, sheet_name=sheet_name, header=header_row) - if skip_rows_after_header: - data_df = data_df.drop(index=skip_rows_after_header, errors="ignore") - - return ( - data_df, - description_df, - description_rows, - description_merged_ranges, - ) - - -def _write_sheet_with_structure( - writer: pd.ExcelWriter, - sheet_name: str, - data_df: pd.DataFrame, - description_rows: list[list[Any]], - header_row: int, - date_columns: list[str] | None = None, - date_format: str | None = None, - description_merged_ranges: list[str] | None = None, -) -> None: - """ - Write a sheet preserving description rows and structure. - - Args: - writer: ExcelWriter instance. - sheet_name: Name of the sheet to write. - data_df: DataFrame with data to write. - description_rows: List of description rows to write at the top. - header_row: Row index where header should be written. - date_columns: Optional list of date column names to format. - date_format: Optional Excel date format string (e.g., 'yyyy-mm-dd'). - description_merged_ranges: Optional merged cell refs to restore - (e.g. 'A1:D1'). - """ - # Convert Python date format to Excel format - excel_date_format = None - if date_format: - # Convert common Python format strings to Excel format - # Excel uses lowercase: yyyy (year), mm (month), dd (day) - excel_date_format = ( - date_format.replace("YYYY", "yyyy") - .replace("YY", "yy") - .replace("DD", "dd") - .replace("MM", "mm") - ) - - # Calculate where to start writing data (after description rows + header row) - data_start_row = len(description_rows) + 1 if description_rows else 1 - - # Write data without header first (header=False), then we'll add header manually - data_df.to_excel( - writer, - sheet_name=sheet_name, - index=False, - header=False, - startrow=data_start_row, - ) - - # Get the workbook and worksheet - workbook = writer.book - worksheet = workbook[sheet_name] - - # Write description rows at the top - if description_rows: - for row_idx, desc_row in enumerate(description_rows, start=1): - for col_idx, value in enumerate(desc_row, start=1): - cell_value = value - # Handle NaN values - if pd.isna(cell_value): - cell_value = None - worksheet.cell(row=row_idx, column=col_idx, value=cell_value) - # Restore merged cells in the description area - if description_merged_ranges: - for range_ref in description_merged_ranges: - with contextlib.suppress(Exception): - worksheet.merge_cells(range_ref) - - # Write header row (after description rows) - header_row_idx = len(description_rows) + 1 if description_rows else 1 - for col_idx, col_name in enumerate(data_df.columns, start=1): - worksheet.cell(row=header_row_idx, column=col_idx, value=col_name) - - # Apply date formatting to date columns if specified - if excel_date_format and date_columns: - # Find column indices for date columns - col_map = {col: idx + 1 for idx, col in enumerate(data_df.columns)} - for date_col in date_columns: - if date_col in col_map: - col_idx = col_map[date_col] - # Apply format to all data rows (skip description and header rows) - for row_idx in range( - data_start_row + 1, data_start_row + len(data_df) + 1 - ): - cell = worksheet.cell(row=row_idx, column=col_idx) - # Only format if cell has a date value - if cell.value is not None: - cell.number_format = excel_date_format diff --git a/nuh_helper/date_shift/_parse.py b/nuh_helper/date_shift/_parse.py index ee8c494..d71433c 100644 --- a/nuh_helper/date_shift/_parse.py +++ b/nuh_helper/date_shift/_parse.py @@ -18,14 +18,12 @@ def _parse_date_value( return result if pd.notna(result) else None if isinstance(value, str): - v = value.strip() - if not v or v.lower() in {"unknown", "unk", "unkown", "n/a", "none", "null"}: - return None + value = value.strip() # Try a handful of common formats, including YYYY-DD-MM found in some feeds for fmt in ("%Y-%m-%d", "%Y-%d-%m", "%d-%m-%Y", "%m-%d-%Y"): try: - parsed = pd.to_datetime(v, format=fmt, errors="coerce") + parsed = pd.to_datetime(value, format=fmt, errors="coerce") if pd.notna(parsed): return parsed except (TypeError, ValueError): @@ -33,7 +31,7 @@ def _parse_date_value( pass # Fallback: let pandas try with dayfirst to handle ambiguous strings - parsed = pd.to_datetime(v, errors="coerce", dayfirst=True) + parsed = pd.to_datetime(value, errors="coerce", dayfirst=True) return parsed if pd.notna(parsed) else None # Anything else: no parse @@ -45,9 +43,10 @@ def _normalize_patient_id(value: object) -> str | None: if value is None or (isinstance(value, float) and pd.isna(value)): return None - if isinstance(value, str): - v = value.strip() - return v if v else None + if value and not isinstance(value, str): + value = str(value) + + if value: + value = value.strip() - v = str(value).strip() - return v if v else None + return value if value else None diff --git a/nuh_helper/date_shift/mappings.py b/nuh_helper/date_shift/mappings.py index 2090c63..7828a97 100644 --- a/nuh_helper/date_shift/mappings.py +++ b/nuh_helper/date_shift/mappings.py @@ -20,49 +20,6 @@ logger = logging.getLogger(__name__) -def generate_shift_mappings( - patient_ids: list[str], - min_shift_days: int = -15, - max_shift_days: int = 15, - seed: int | None = None, -) -> pd.DataFrame: - """ - Generate random shift mappings for patient IDs. - - Args: - patient_ids: List of patient IDs to generate shifts for. - min_shift_days: Minimum number of days to shift (default: -15). - max_shift_days: Maximum number of days to shift (default: 15). - seed: Optional random seed for reproducibility. - - Returns: - DataFrame with columns 'patient_id' and 'shift_days'. - """ - if seed is not None: - random.seed(seed) - - shifts = [random.randint(min_shift_days, max_shift_days) for _ in patient_ids] - return pd.DataFrame({"patient_id": patient_ids, "shift_days": shifts}) - - -def load_shift_mappings(csv_path: str) -> pd.DataFrame: - """ - Load shift mappings from a CSV file. - - Args: - csv_path: Path to the CSV file containing shift mappings. - Expected columns: 'patient_id' and 'shift_days'. - - Returns: - DataFrame with shift mappings. - """ - df = pd.read_csv(csv_path) - if "patient_id" not in df.columns or "shift_days" not in df.columns: - raise ValueError("CSV must contain 'patient_id' and 'shift_days' columns") - logger.info("Loaded %d shift mapping(s) from '%s'", len(df), csv_path) - return df - - def load_shift_deltas( book: BaseWorkBook, patient_sheet: str, @@ -102,16 +59,16 @@ def load_shift_deltas( elif not isinstance(out_linking_table, Path): out_linking_table = Path(out_linking_table) - with book[patient_sheet] as page: - scanner = PatientRowStateMachine(sheet_configs) - patient_ids: list[str] = [ - pid.patient_id - for pid in [ - scanner.next_details(page.name, row, data) - for row, data in page.stream_rows() - ] - if pid + page = book[patient_sheet] + scanner = PatientRowStateMachine(sheet_configs) + patient_ids: list[str] = [ + pid.patient_id + for pid in [ + scanner.next_details(page.name, row, data) + for row, data in page.stream_rows() ] + if pid + ] # load the old ids (should we remove IDs that are "gone"?) if not old_linking_table or not old_linking_table.is_file(): diff --git a/nuh_helper/date_shift/row_machine.py b/nuh_helper/date_shift/row_machine.py index 3fe6a47..3d7c29d 100644 --- a/nuh_helper/date_shift/row_machine.py +++ b/nuh_helper/date_shift/row_machine.py @@ -56,19 +56,42 @@ def column_type( ) -> bool: """determine what we should do for a given column""" + # values we ignore in this column ignored: set[None | str | dt.date | dt.datetime] = shift_ignore.get( self.page_name, {} ).get(col_name, {None, "", "NULL", "None"}) - return ( - ColumnType.IGNORE - if cell_value in ignored - else ColumnType.TEXT - if col_name in self.text_columns - # always allow NULL in date columns - even when it's not in the list - else (ColumnType.DATE if cell_value != "NULL" else ColumnType.IGNORE) + # expand to include the dates when datetime=00:00:00 and datetime for dates + ignored = ( + list(ignored) + + [ + # add date -> datetime(00:00) + dt.datetime(each.year, each.month, each.day) + for each in ignored + if isinstance(each, dt.date) + ] + + [ + # add datetime(00:00) -> date + dt.date(each.year, each.month, each.day) + for each in ignored + if isinstance(each, dt.datetime) + and each.hour == 0 + and each.minute == 0 + and each.second == 0 + ] ) + if cell_value in ignored: + return ColumnType.IGNORE + + if col_name in self.text_columns: + return ColumnType.TEXT + + if cell_value == "NULL": + return ColumnType.IGNORE + + return ColumnType.DATE + class PatientRowStateMachine: """builds upon the callbacks from book_page to handle row/page switching. does all diff --git a/tests/conftest.py b/tests/conftest.py index 32f4cc6..e20eb6e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,6 +3,9 @@ import pytest +from nuh_helper import book_page +from nuh_helper.book_page import normalize_row, row_val_t + OUTPUT_DIR = Path(__file__).parent / "output" @@ -30,3 +33,12 @@ def output_path(request: pytest.FixtureRequest, tmp_path: Path) -> Path: path.mkdir(parents=True) return path return tmp_path + + +def load_workbook( + path: str | Path | list[Path], +) -> dict[str, list[row_val_t]]: + return { + page.name: [normalize_row(cells) for _, cells in page.stream_rows()] + for page in book_page.book_open(path) + } diff --git a/tests/data/test_date_shift/shift-ignore.yaml b/tests/data/test_date_shift/shift-ignore.yaml new file mode 100644 index 0000000..62ede34 --- /dev/null +++ b/tests/data/test_date_shift/shift-ignore.yaml @@ -0,0 +1,3 @@ +patients: + last_alive: + - "2024-12-31" diff --git a/tests/data/test_e2e_date_shift/shift-ignore.yaml b/tests/data/test_e2e_date_shift/shift-ignore.yaml new file mode 100644 index 0000000..4ebcc47 --- /dev/null +++ b/tests/data/test_e2e_date_shift/shift-ignore.yaml @@ -0,0 +1,3 @@ +results: + date_result: + - unknown diff --git a/tests/test_book_page.py b/tests/test_book_page.py index 624ad8a..d757d65 100644 --- a/tests/test_book_page.py +++ b/tests/test_book_page.py @@ -4,7 +4,12 @@ import pytest -from nuh_helper.book_page import BaseWorkBook, CSVWorkBook, XLSXWorkBook, book_open +from nuh_helper.book_page import ( + BaseWorkBook, + CSVWorkBook, + XLSXWorkBook, + book_open, +) test_data = ( # i feel like test data should be in a folder following this pattern @@ -89,12 +94,12 @@ def test_read_data( # assert assert isinstance(book, kind) assert name in book - with book[name] as page: - copy = [] - for _, row in page.stream_rows(): - copy.append([str(c) for c in row]) + page = book[name] + copy = [] + for _, row in page.stream_rows(): + copy.append([str(c) for c in row]) - assert data == copy + assert data == copy @pytest.mark.parametrize( @@ -168,21 +173,21 @@ def test_copy_book( assert isinstance(source, kind) assert isinstance(target, kind) - with target[name] as page: - copy = [] - for row, cell in page.stream_rows(): - assert len(copy) == row - copy.append(cell) - assert copy == changed_data + page = target[name] + copy = [] + for row, cell in page.stream_rows(): + assert len(copy) == row + copy.append(cell) + assert copy == changed_data def test_stream_csv_rows() -> None: book = CSVWorkBook([test_book / "data_3x4.csv", test_book / "data_3x6.tsv"]) copy = [] - with book["data_3x4"] as page: - for _, row in page.stream_rows(): - copy.append(row.copy()) + page = book["data_3x4"] + for _, row in page.stream_rows(): + copy.append(row.copy()) assert copy == data_3x4 @@ -257,10 +262,10 @@ def test_bom_is_removed() -> None: book = CSVWorkBook([test_data / "page-desc.csv"]) obtained = [] - with book["page-desc"] as page: - for idx, row in page.stream_rows(): - assert idx == len(obtained) - obtained.append(row.copy()) + page = book["page-desc"] + for idx, row in page.stream_rows(): + assert idx == len(obtained) + obtained.append(row.copy()) expected = [ ["foo", "pid", "glitter"], diff --git a/tests/test_date_shift.py b/tests/test_date_shift.py index b4d3241..a120265 100644 --- a/tests/test_date_shift.py +++ b/tests/test_date_shift.py @@ -5,20 +5,17 @@ import pandas as pd import pytest +from conftest import load_workbook from openpyxl import Workbook +from nuh_helper.book_page.book import normalize_row from nuh_helper.date_shift import ( - apply_date_shifts, - generate_shift_mappings, - load_shift_mappings, - shift_excel_dates, -) -from nuh_helper.date_shift._excel import ( - _description_merged_ranges, - _get_row_values_resolving_merged, + shift_excel_dates_inplace, ) from nuh_helper.date_shift._parse import _normalize_patient_id, _parse_date_value +test_data = Path(__file__).parent / "data" / __name__ + class TestNormalizePatientId: def test_none_returns_none(self) -> None: @@ -85,132 +82,6 @@ def test_mm_dd_yyyy_string(self) -> None: assert _parse_date_value("01-15-2023") == pd.Timestamp("2023-01-15") -class TestGenerateShiftMappings: - def test_returns_correct_columns(self) -> None: - result = generate_shift_mappings(["P001"]) - assert list(result.columns) == ["patient_id", "shift_days"] - - def test_one_row_per_patient(self) -> None: - ids = ["P001", "P002", "P003"] - result = generate_shift_mappings(ids) - assert len(result) == 3 - assert list(result["patient_id"]) == ids - - def test_reproducible_with_seed(self) -> None: - ids = ["P001", "P002", "P003"] - result1 = generate_shift_mappings(ids, seed=42) - result2 = generate_shift_mappings(ids, seed=42) - assert list(result1["shift_days"]) == list(result2["shift_days"]) - - def test_different_seeds_differ(self) -> None: - ids = ["P001", "P002", "P003"] - result1 = generate_shift_mappings(ids, seed=1) - result2 = generate_shift_mappings(ids, seed=2) - assert list(result1["shift_days"]) != list(result2["shift_days"]) - - def test_shifts_within_specified_range(self) -> None: - ids = [f"P{i:03d}" for i in range(100)] - result = generate_shift_mappings( - ids, min_shift_days=-7, max_shift_days=7, seed=42 - ) - assert result["shift_days"].between(-7, 7).all() - - def test_empty_patient_list(self) -> None: - result = generate_shift_mappings([]) - assert len(result) == 0 - - -class TestLoadShiftMappings: - def test_loads_valid_csv(self, tmp_path: Path) -> None: - csv_file = tmp_path / "shifts.csv" - csv_file.write_text("patient_id,shift_days\nP001,5\nP002,-3\n") - result = load_shift_mappings(str(csv_file)) - assert list(result["patient_id"]) == ["P001", "P002"] - assert list(result["shift_days"]) == [5, -3] - - def test_raises_on_missing_columns(self, tmp_path: Path) -> None: - csv_file = tmp_path / "bad.csv" - csv_file.write_text("id,days\nP001,5\n") - with pytest.raises(ValueError, match="patient_id.*shift_days"): - load_shift_mappings(str(csv_file)) - - -class TestGetRowValuesResolvingMerged: - """Unit tests for _get_row_values_resolving_merged (merged cell header handling).""" - - def test_merged_row_returns_value_repeated_for_each_column(self) -> None: - wb = Workbook() - ws = wb.active - assert ws is not None - ws["A1"] = "crazy" - ws.merge_cells("A1:D1") - result = _get_row_values_resolving_merged(ws, 1, 4) - assert result == ["crazy", "crazy", "crazy", "crazy"] - - def test_normal_row_returns_cell_values(self) -> None: - wb = Workbook() - ws = wb.active - assert ws is not None - ws["A2"] = "patient_id" - ws["B2"] = "measurement" - ws["C2"] = "date_result" - ws["D2"] = "type" - result = _get_row_values_resolving_merged(ws, 2, 4) - assert result == ["patient_id", "measurement", "date_result", "type"] - - def test_partial_merge_second_column_gets_top_left_value(self) -> None: - wb = Workbook() - ws = wb.active - assert ws is not None - ws["A1"] = "a" - ws["B1"] = "b" - ws.merge_cells("B1:C1") - ws["D1"] = "d" - result = _get_row_values_resolving_merged(ws, 1, 4) - assert result == ["a", "b", "b", "d"] - - -class TestDescriptionMergedRanges: - """Unit tests for _description_merged_ranges.""" - - def test_returns_merged_ranges_within_description_rows(self) -> None: - wb = Workbook() - ws = wb.active - assert ws is not None - ws["A1"] = "title" - ws.merge_cells("A1:D1") - result = _description_merged_ranges(ws, 1) - assert result == ["A1:D1"] - - def test_excludes_merged_ranges_below_description_rows(self) -> None: - wb = Workbook() - ws = wb.active - assert ws is not None - ws["A1"] = "title" - ws.merge_cells("A1:D1") - ws["A2"] = "x" - ws.merge_cells("A2:B2") - result = _description_merged_ranges(ws, 1) - assert result == ["A1:D1"] - - def test_empty_when_no_merged_cells(self) -> None: - wb = Workbook() - ws = wb.active - assert ws is not None - ws["A1"] = "a" - result = _description_merged_ranges(ws, 2) - assert result == [] - - def test_empty_when_num_description_rows_zero(self) -> None: - wb = Workbook() - ws = wb.active - assert ws is not None - ws["A1"] = "x" - ws.merge_cells("A1:B1") - result = _description_merged_ranges(ws, 0) - assert result == [] - - class TestShiftExcelDatesWithComplexLayout: """Integration tests: header_row, skip_rows_after_header, merged cells, patient sheet from config.""" @@ -260,24 +131,33 @@ def test_header_row_and_skip_rows_after_header_produce_correct_columns( "patients": { "patient_id_col": "patient_id", "date_columns": ["date_result"], + "text_columns": [ + "type", + "measurement", + ], "header_row": 2, "skip_rows_after_header": [3], }, } - shift_excel_dates( + shift_excel_dates_inplace( input_file=str(xlsx), output_file=str(out), patient_sheet="patients", - patient_id_col="patient_id", sheet_configs=sheet_configs, + linking_table_path=tmp_path / "old-linking.csv", linking_table_output=str(tmp_path / "linking.csv"), seed=42, ) df = pd.read_excel(str(out), sheet_name="patients", header=2) assert list(df.columns) == ["patient_id", "measurement", "date_result", "type"] - # Data should be 2 rows (P001, P002), not 3 (no "stupid" row) - assert len(df) == 2 - assert list(df["patient_id"]) == ["P001", "P002"] + # Data should now be 3 rows (stupid, P001, P002) + # ... Pandas can see "stupid" but date shifting doesn't + assert ( + # this should be 3 these days + # ... it used to be "2" because "skip_rows_after_header" removed the rows? + len(df) == 3 + ) + assert list(df["patient_id"]) == ["stupid", "P001", "P002"] def test_patient_sheet_uses_config_header_when_sheet_in_sheet_configs( self, tmp_path: Path @@ -291,6 +171,10 @@ def test_patient_sheet_uses_config_header_when_sheet_in_sheet_configs( "patients": { "patient_id_col": "patient_id", "date_columns": ["date_result"], + "text_columns": [ + "measurement", + "type", + ], "header_row": 2, "skip_rows_after_header": [3], }, @@ -299,12 +183,12 @@ def test_patient_sheet_uses_config_header_when_sheet_in_sheet_configs( Do not pass patient_header_row; patient sheet "patients" is in config with header_row=2 """ - shift_excel_dates( + shift_excel_dates_inplace( input_file=str(xlsx), output_file=str(out), patient_sheet="patients", - patient_id_col="patient_id", sheet_configs=sheet_configs, + linking_table_path=tmp_path / "old-linking.csv", linking_table_output=str(tmp_path / "linking.csv"), seed=42, ) @@ -312,182 +196,11 @@ def test_patient_sheet_uses_config_header_when_sheet_in_sheet_configs( df = pd.read_excel(str(out), sheet_name="patients", header=2) assert "patient_id" in df.columns assert "date_result" in df.columns - assert len(df) == 2 - - -class TestApplyDateShifts: - def _make_mappings(self, patient_id: str, shift_days: int) -> pd.DataFrame: - return pd.DataFrame({"patient_id": [patient_id], "shift_days": [shift_days]}) - - def test_shifts_date_forward(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2023-01-15"]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 10) - ) - assert result["visit_date"].iloc[0] == date(2023, 1, 25) - - def test_shifts_date_backward(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2023-06-01"]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", -5) - ) - assert result["visit_date"].iloc[0] == date(2023, 5, 27) - - def test_zero_shift_leaves_date_unchanged(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2023-01-15"]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 0) - ) - assert result["visit_date"].iloc[0] == date(2023, 1, 15) - - def test_unknown_patient_leaves_date_unchanged(self) -> None: - df = pd.DataFrame({"patient_id": ["P999"], "visit_date": ["2023-01-15"]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 10) - ) - assert result["visit_date"].iloc[0] == date(2023, 1, 15) - - def test_placeholder_date_becomes_none(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["Unknown"]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 10) - ) - assert result["visit_date"].iloc[0] is None - - def test_none_date_stays_none(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": [None]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 10) - ) - assert result["visit_date"].iloc[0] is None - - def test_missing_date_column_is_skipped(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2023-01-15"]}) - mappings = self._make_mappings("P001", 10) - # "nonexistent" should be silently skipped; "visit_date" still shifted - result = apply_date_shifts( - df, "patient_id", ["nonexistent", "visit_date"], mappings - ) - assert result["visit_date"].iloc[0] == date(2023, 1, 25) - - def test_strips_whitespace_from_patient_ids(self) -> None: - df = pd.DataFrame({"patient_id": [" P001 "], "visit_date": ["2023-01-15"]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 10) - ) - assert result["visit_date"].iloc[0] == date(2023, 1, 25) - - def test_multiple_patients_shifted_independently(self) -> None: - df = pd.DataFrame( - { - "patient_id": ["P001", "P002"], - "visit_date": ["2023-01-01", "2023-01-01"], - } + assert ( + # this should be 3 these days + # ... it used to be "2" because "skip_rows_after_header" removed the rows? + len(df) == 3 ) - mappings = pd.DataFrame({"patient_id": ["P001", "P002"], "shift_days": [5, -5]}) - result = apply_date_shifts(df, "patient_id", ["visit_date"], mappings) - assert result["visit_date"].iloc[0] == date(2023, 1, 6) - assert result["visit_date"].iloc[1] == date(2022, 12, 27) - - def test_result_dates_are_date_not_datetime(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2023-01-15"]}) - result = apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 0) - ) - val = result["visit_date"].iloc[0] - assert isinstance(val, date) - assert not isinstance(val, datetime) - - def test_does_not_mutate_input(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2023-01-15"]}) - original_visit = df["visit_date"].iloc[0] - apply_date_shifts( - df, "patient_id", ["visit_date"], self._make_mappings("P001", 10) - ) - assert df["visit_date"].iloc[0] == original_visit - - -class TestApplyDateShiftsExceptions: - def _make_mappings(self, patient_id: str, shift_days: int) -> pd.DataFrame: - return pd.DataFrame({"patient_id": [patient_id], "shift_days": [shift_days]}) - - def test_exception_date_is_not_shifted(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2024-12-31"]}) - result = apply_date_shifts( - df, - "patient_id", - ["visit_date"], - self._make_mappings("P001", 10), - shift_exceptions={"visit_date": ["2024-12-31"]}, - ) - assert result["visit_date"].iloc[0] == date(2024, 12, 31) - - def test_non_exception_date_in_same_column_is_shifted(self) -> None: - df = pd.DataFrame( - { - "patient_id": ["P001", "P002"], - "visit_date": ["2024-12-31", "2023-06-01"], - } - ) - mappings = pd.DataFrame( - {"patient_id": ["P001", "P002"], "shift_days": [10, 10]} - ) - result = apply_date_shifts( - df, - "patient_id", - ["visit_date"], - mappings, - shift_exceptions={"visit_date": ["2024-12-31"]}, - ) - assert result["visit_date"].iloc[0] == date(2024, 12, 31) - assert result["visit_date"].iloc[1] == date(2023, 6, 11) - - def test_exception_only_applies_to_specified_column(self) -> None: - df = pd.DataFrame( - { - "patient_id": ["P001"], - "col_a": ["2024-12-31"], - "col_b": ["2024-12-31"], - } - ) - result = apply_date_shifts( - df, - "patient_id", - ["col_a", "col_b"], - self._make_mappings("P001", 5), - shift_exceptions={"col_a": ["2024-12-31"]}, - ) - assert result["col_a"].iloc[0] == date(2024, 12, 31) - assert result["col_b"].iloc[0] == date(2025, 1, 5) - - def test_multiple_exception_dates(self) -> None: - df = pd.DataFrame( - { - "patient_id": ["P001", "P001", "P001"], - "visit_date": ["2024-12-31", "2024-06-30", "2023-01-15"], - } - ) - result = apply_date_shifts( - df, - "patient_id", - ["visit_date"], - self._make_mappings("P001", 7), - shift_exceptions={"visit_date": ["2024-12-31", "2024-06-30"]}, - ) - assert result["visit_date"].iloc[0] == date(2024, 12, 31) - assert result["visit_date"].iloc[1] == date(2024, 6, 30) - assert result["visit_date"].iloc[2] == date(2023, 1, 22) - - def test_no_exceptions_kwarg_shifts_normally(self) -> None: - df = pd.DataFrame({"patient_id": ["P001"], "visit_date": ["2024-12-31"]}) - result = apply_date_shifts( - df, - "patient_id", - ["visit_date"], - self._make_mappings("P001", 10), - shift_exceptions=None, - ) - assert result["visit_date"].iloc[0] == date(2025, 1, 10) class TestShiftExcelDatesExceptionsIntegration: @@ -510,26 +223,32 @@ def test_exception_date_unchanged_other_dates_shifted(self, tmp_path: Path) -> N xlsx = tmp_path / "in.xlsx" self._make_excel(xlsx) out = tmp_path / "out.xlsx" + + # raise Exception(f'{__name__=}') sheet_configs = { "patients": { "patient_id_col": "patient_id", "date_columns": ["last_alive"], - "shift_exceptions": {"last_alive": ["2024-12-31"]}, + "text_columns": [], } } - shift_excel_dates( + shift_excel_dates_inplace( input_file=str(xlsx), output_file=str(out), patient_sheet="patients", - patient_id_col="patient_id", sheet_configs=sheet_configs, + linking_table_path=tmp_path / "does-not-exist.csv", linking_table_output=str(tmp_path / "linking.csv"), + shift_ignore_yaml=test_data / "shift-ignore.yaml", seed=42, ) - df = pd.read_excel(str(out), sheet_name="patients") - p001_date = df.loc[df["patient_id"] == "P001", "last_alive"].iloc[0] - p002_date = df.loc[df["patient_id"] == "P002", "last_alive"].iloc[0] - # P002's date is an exception — must be unchanged - assert pd.Timestamp(p002_date).date() == date(2024, 12, 31) - # P001's date must have been shifted (seed=42 gives a non-zero shift) - assert pd.Timestamp(p001_date).date() != date(2023, 6, 1) + expected = { + "patients": [ + normalize_row(["patient_id", "last_alive"]), + normalize_row(["P001", "2023-06-06"]), + normalize_row(["P002", "2024-12-31"]), + ], + } + obtained = load_workbook(out) + + assert obtained == expected diff --git a/tests/test_e2e_date_shift.py b/tests/test_e2e_date_shift.py index ed8c999..5cb6fbc 100644 --- a/tests/test_e2e_date_shift.py +++ b/tests/test_e2e_date_shift.py @@ -13,7 +13,7 @@ import pandas as pd -from nuh_helper import shift_excel_dates +from nuh_helper import shift_excel_dates_inplace class _ShiftKwargs(TypedDict, total=False): @@ -27,15 +27,25 @@ class _ShiftKwargs(TypedDict, total=False): DATA_DIR = Path(__file__).parent / "data" INPUT_FILE = DATA_DIR / "patients.xlsx" +test_data = Path(__file__).parent / "data" / __name__ + # Config matching the structure of patients.xlsx SHEET_CONFIGS = { "patients": { "patient_id_col": "patient_id", "date_columns": ["dob", "last_alive"], + "text_columns": ["name"], }, "results": { "patient_id_col": "patient_id", "date_columns": ["date_result"], + "text_columns": [ + "measurement", + "type", + ], + "pass_as_is": { + "date_result": ["unknown"], + }, }, } @@ -48,13 +58,14 @@ def run_shift(base: Path, **kwargs: Unpack[_ShiftKwargs]) -> tuple[Path, Path]: """ output = base / "output.xlsx" linking = base / "linking.csv" - shift_excel_dates( + shift_excel_dates_inplace( str(INPUT_FILE), str(output), patient_sheet="patients", - patient_id_col="patient_id", sheet_configs=SHEET_CONFIGS, + linking_table_path=linking.parent / "old-linking.csv", linking_table_output=str(linking), + shift_ignore_yaml=test_data / ("shift-ignore.yaml"), **kwargs, ) return output, linking @@ -167,7 +178,7 @@ def test_placeholder_date_becomes_null_in_output(self, output_path: Path) -> Non test5_date = output_results.loc[ output_results["patient_id"] == "Test5", "date_result" ].iloc[0] - assert pd.isna(test5_date) + assert test5_date == "unknown" class TestReproducibility: @@ -192,14 +203,14 @@ def test_linking_table_reproduces_same_shifts_on_new_file( output1, linking = run_shift(output_path, seed=42) output2 = output_path / "output2.xlsx" - shift_excel_dates( + shift_excel_dates_inplace( str(INPUT_FILE), str(output2), patient_sheet="patients", - patient_id_col="patient_id", sheet_configs=SHEET_CONFIGS, linking_table_path=str(linking), linking_table_output=str(output_path / "linking2.csv"), + shift_ignore_yaml=test_data / ("shift-ignore.yaml"), ) for sheet in SHEET_CONFIGS: diff --git a/tests/test_parse.py b/tests/test_parse.py new file mode 100644 index 0000000..bdcaf16 --- /dev/null +++ b/tests/test_parse.py @@ -0,0 +1,51 @@ +import datetime +from typing import Never + +import pandas as pd +import pytest + +from nuh_helper.date_shift._parse import _normalize_patient_id, _parse_date_value + + +@pytest.mark.parametrize( + "source, expected", + [ + ("12-31-2004", "2004-12-31 00:00:00"), + ("12/03/2016", "2016-03-12 00:00:00"), + ("2016-12-03", "2016-12-03 00:00:00"), + ("2016-13-03", "2016-03-13 00:00:00"), + ("31-12-2004", "2004-12-31 00:00:00"), + ("not-a-date", "None"), + (3.9, "None"), + (None, "None"), + (datetime.date(2026, 9, 12), "2026-09-12 00:00:00"), + (datetime.datetime(2026, 9, 12, 13, 14, 15), "2026-09-12 13:14:15"), + (float("nan"), "None"), + (pd.NaT, "None"), + (pd.Timestamp("2026-09-12 13:14:15"), "2026-09-12 13:14:15"), + ], +) +def test_date_value(source: str, expected: str) -> Never: + obtained: str = str(_parse_date_value(source)) + + assert obtained == expected + + +@pytest.mark.parametrize( + "source, expected", + [ + (" ", "None"), + (" nuh007 ", "nuh007"), + ("", "None"), + ("2016-13-03", "2016-13-03"), + (0, "None"), + (3.9, "3.9"), + (None, "None"), + (datetime.date(2026, 9, 12), "2026-09-12"), + (float("nan"), "None"), + ], +) +def test_patient_id(source: str, expected: str) -> Never: + obtained: str = str(_normalize_patient_id(source)) + + assert obtained == expected diff --git a/tests/test_shift_ignore.py b/tests/test_shift_ignore.py index 41db13d..649a07a 100644 --- a/tests/test_shift_ignore.py +++ b/tests/test_shift_ignore.py @@ -1,8 +1,9 @@ from pathlib import Path import pytest +from conftest import load_workbook -from nuh_helper import book_page, shift_excel_dates_inplace +from nuh_helper import shift_excel_dates_inplace from nuh_helper.book_page import normalize_row from nuh_helper.date_shift.defs import ( HiddenDate, @@ -447,10 +448,3 @@ def src_and_out_names( source_file = test_data / "workbook.xlsx" output_path = tmp_path / "target.xlsx" return source_file, output_path - - -def load_workbook(path: str | Path) -> dict[str, list[list[str | None]]]: - return { - page.name: [normalize_row(cells) for _, cells in page.stream_rows()] - for page in book_page.book_open(path) - }