Merge pull request #48

Update AutoGrowth upload pipeline for high-frequency OD pre-aggregation and flexible calibration inputs

Author: enryH

AutoGrowth/0_upload_data.py

@@ -6,6 +6,7 @@
     REQUIRED_COLUMNS_NAME_MAP,
     process_chibio_data,
     process_od_pioreactor,
+    read_od_adjustment_table,
 )
 from ui_components import page_header_with_help
 
@@ -210,7 +211,7 @@ def apply_linear_adjustments(
         with st.popover("See an Example", width="stretch"):
             st.markdown("**OD Calibration Table**")
             st.markdown(
-                "- CSV file with columns `reactor` and `od`.\n"
+                "- CSV/TXT (`,` or `;`) or Excel file with columns `reactor` and `od`.\n"
                 "- Used to adjust OD readings by reactor based on calibration data."
             )
             st.divider()
@@ -236,7 +237,7 @@ def apply_linear_adjustments(
             st.info(f"File previously uploaded: {_file_name}")
         od_adjustment_upload = st.file_uploader(
             "OD adjustment table",
-            type=["csv", "txt"],
+            type=["csv", "txt", "xlsx", "xls"],
             key="upload_page_od_adjustment_table",
         )
     with optional_upload_cols[1]:
@@ -422,26 +423,29 @@ def apply_linear_adjustments(
             iqr_range_value = st.slider(
                 "IQR factor for outlier removal",
                 1.0,
-                3.0,
+                5.0,
                 st.session_state.get("iqr_range_value", 1.5),
                 step=0.1,
                 help="Used when outlier method is IQR. Multiplier of the IQR.",
             )
             rolling_window = st.slider(
                 "Rolling window (of timepoints) for IQR outlier removal",
                 11,
-                61,
+                141,
                 st.session_state.get("rolling_window", 21),
                 step=2,
                 help="Used when outlier method is IQR.",
             )
             ecod_factor = st.slider(
                 "ECOD factor for outlier removal",
                 0.5,
-                8.0,
+                12.0,
                 st.session_state.get("ecod_factor", 4.0),
                 step=0.1,
-                help="Used when outlier method is ECOD. Anomaly detection sensitivity.",
+                help=(
+                    "Used when outlier method is ECOD. Lower values are more "
+                    "sensitive; higher values are less sensitive."
+                ),
             )
 
         st.divider()
@@ -460,10 +464,10 @@ def apply_linear_adjustments(
                 "Round time to nearest second (defining timesteps). "
                 "Used to align timeseries "
                 "with slight time offsets.",
-                1,
+                5,
                 300,
                 st.session_state.get("round_time", 5),
-                step=1,
+                step=5,
                 help=(
                     "Rounding helps pivot the data to wide format from the "
                     "long format. If you have multiple measurements for the same "
@@ -504,6 +508,15 @@ def apply_linear_adjustments(
                     "outliers and therefore the default."
                 ),
             )
+        aggregate_high_frequency_raw_data = st.checkbox(
+            "Aggregate raw OD data when sampled below every 15 seconds (PioReactor)",
+            value=st.session_state.get("aggregate_high_frequency_raw_data", False),
+            disabled=reactor_type != "PioReactor",
+            help=(
+                "If enabled, PioReactor raw OD data sampled faster than every 15 "
+                "seconds is aggregated to 15-second time bins before processing."
+            ),
+        )
         st.divider()
         button_pressed = st.form_submit_button(
             "Apply options to uploaded data", type="primary", width="stretch"
@@ -531,6 +544,9 @@ def apply_linear_adjustments(
 st.session_state["aggregate_duplicated_rounded_timepoint_method"] = (
     aggregate_duplicated_rounded_timepoint_method
 )
+st.session_state["aggregate_high_frequency_raw_data"] = (
+    aggregate_high_frequency_raw_data
+)
 
 # region: Process files
 ########################################################################################
@@ -626,6 +642,7 @@ def apply_linear_adjustments(
             keep_core_data=keep_core_data,
             aggregate_duplicated_rounded_timepoint=aggregate_duplicated_rounded_timepoint,
             aggregate_duplicated_rounded_timepoint_method=aggregate_duplicated_rounded_timepoint_method,
+            aggregate_high_frequency_raw_data=aggregate_high_frequency_raw_data,
         )
 
     rerun = st.session_state.get("df_raw_od_data") is None
@@ -806,7 +823,7 @@ def apply_linear_adjustments(
                 "OD adjustments have already been applied. "
                 "Re-applying will overwrite previous adjustments."
             )
-        df_adjustments = pd.read_csv(od_adjustment_upload).convert_dtypes()
+        df_adjustments = read_od_adjustment_table(od_adjustment_upload)
         try:
             df_rolling, adjustment_warnings = apply_linear_adjustments(
                 df_rolling, df_adjustments

AutoGrowth/process_data.py

@@ -1,3 +1,4 @@
+import csv
 from pathlib import Path
 
 import pandas as pd
@@ -37,6 +38,90 @@
 }
 
 
+def maybe_aggregate_high_frequency_raw_data(
+    df_raw_od_data: pd.DataFrame,
+    min_interval_seconds: int = 15,
+    aggregation_method: str = "median",
+) -> tuple[pd.DataFrame, bool, float | None]:
+    """Aggregate OD data if sampling is faster than the minimum interval.
+
+    Returns
+    -------
+    tuple[pd.DataFrame, bool, float | None]
+        (possibly aggregated data, whether aggregation was applied,
+        median sampling interval in seconds).
+    """
+    sampling_interval = (
+        df_raw_od_data.sort_values(["pioreactor_unit", "timestamp_localtime"])
+        .groupby("pioreactor_unit")["timestamp_localtime"]
+        .diff()
+        .dt.total_seconds()
+        .dropna()
+    )
+    median_interval_seconds = (
+        float(sampling_interval.median()) if not sampling_interval.empty else None
+    )
+    if (
+        median_interval_seconds is None
+        or median_interval_seconds >= min_interval_seconds
+    ):
+        return df_raw_od_data, False, median_interval_seconds
+
+    df_aggregated = df_raw_od_data.copy()
+    # flooring to 00, 15, 30, 45 seconds for example, to aggregate to 15s intervals
+    # ? Can it cause trouble with metadata?
+    df_aggregated["timestamp_localtime"] = df_aggregated[
+        "timestamp_localtime"
+    ].dt.floor(f"{min_interval_seconds}s")
+    group_columns = ["timestamp_localtime", "pioreactor_unit"]
+    agg_map = {
+        col: (aggregation_method if col == "od_reading" else "first")
+        for col in df_aggregated.columns
+        if col not in group_columns
+    }
+    df_aggregated = df_aggregated.groupby(
+        group_columns,
+        sort=False,
+        dropna=False,
+        as_index=False,
+    ).agg(agg_map)
+    return df_aggregated.convert_dtypes(), True, median_interval_seconds
+
+
+def read_od_adjustment_table(file) -> pd.DataFrame:
+    """Read OD adjustment table from CSV/TXT or Excel files.
+
+    Parameters
+    ----------
+    file
+        Uploaded file-like object with a ``name`` attribute.
+
+    Returns
+    -------
+    pd.DataFrame
+        Adjustment table expected to include ``reactor`` and ``od`` columns.
+    """
+    suffix = Path(getattr(file, "name", "")).suffix.lower()
+    if hasattr(file, "seek"):
+        file.seek(0)
+    if suffix in {".xlsx", ".xls"}:
+        return pd.read_excel(file).convert_dtypes()
+    if hasattr(file, "read"):
+        preview = file.read(4096)
+        if hasattr(file, "seek"):
+            file.seek(0)
+    else:
+        preview = ""
+    if isinstance(preview, bytes):
+        preview = preview.decode("utf-8", errors="ignore")
+    header = next((line for line in preview.splitlines() if line.strip()), "")
+    try:
+        delimiter = csv.Sniffer().sniff(header, delimiters=",;").delimiter
+    except csv.Error:
+        delimiter = ","
+    return pd.read_csv(file, sep=delimiter).convert_dtypes()
+
+
 def read_pioreactor_csv(file: str, round_time: int = 60):
     """Read raw OD data from a PioReactor export CSV file and round timestamps."""
     df_raw_od_data = pd.read_csv(file, converters=COLUMN_TYPES_PIO).convert_dtypes()
@@ -117,6 +202,8 @@ def process_od_pioreactor(
     keep_core_data: bool = True,
     aggregate_duplicated_rounded_timepoint: bool = True,
     aggregate_duplicated_rounded_timepoint_method: str = "mean",
+    aggregate_high_frequency_raw_data: bool = False,
+    min_raw_sampling_interval_seconds: int = 15,
 ):
     """Process raw OD data from a PioReactor export CSV file and return both the
     raw and wide formats of the data, along with a summary message and a boolean
@@ -142,6 +229,11 @@ def process_od_pioreactor(
         Method to use for aggregating duplicated rounded timepoints, by default "mean".
         Options are what pandas groupby.agg accepts, e.g. "mean", "median",
         "min", "max", etc.
+    aggregate_high_frequency_raw_data : bool, optional
+        Whether to aggregate raw OD data before processing when sampling is faster
+        than the minimum sampling interval.
+    min_raw_sampling_interval_seconds : int, optional
+        Minimum expected sampling interval in seconds, default 15.
 
     Returns
     -------
@@ -150,6 +242,29 @@ def process_od_pioreactor(
         the processing steps.
     """
     df_raw_od_data, msg = read_pioreactor_csv(file, round_time)
+    if aggregate_high_frequency_raw_data:
+        n_before = df_raw_od_data.shape[0]
+        df_raw_od_data, was_aggregated, median_interval_seconds = (
+            maybe_aggregate_high_frequency_raw_data(
+                df_raw_od_data=df_raw_od_data,
+                min_interval_seconds=min_raw_sampling_interval_seconds,
+                aggregation_method=aggregate_duplicated_rounded_timepoint_method,
+            )
+        )
+        if was_aggregated:
+            n_after = df_raw_od_data.shape[0]
+            msg += (
+                "- Aggregated high-frequency raw OD data sampled every "
+                f"{median_interval_seconds:.1f}s to {min_raw_sampling_interval_seconds}s "
+                f"(rows: {n_before:,d} -> {n_after:,d}).\n"
+            )
+        elif median_interval_seconds is not None:
+            msg += (
+                "- Raw OD data sampling interval is "
+                f"{median_interval_seconds:.1f}s (>= "
+                f"{min_raw_sampling_interval_seconds}s), so no pre-aggregation was "
+                "applied.\n"
+            )
     # use starttime to compute elapsed time
     start_time = df_raw_od_data["timestamp_rounded"].min()
     st.session_state["start_time"] = start_time