From 3a914cf4d3e0e2af050fe71b440c3ff96a441536 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 28 Aug 2026 14:20:06 -0400
Subject: [PATCH 01/18] 3) clean_dataframe's args should pass through to
 validate_schema

---
 quantaq_cli/toolkit/munge.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/quantaq_cli/toolkit/munge.py b/quantaq_cli/toolkit/munge.py
index 4a3c4be..c712bf5 100644
--- a/quantaq_cli/toolkit/munge.py
+++ b/quantaq_cli/toolkit/munge.py
@@ -24,6 +24,6 @@ def clean_dataframe(df, coerce_dtypes=True, coerce_rename=True):
     
     # Validate the schema
     # by default this also coerces dtypes and column names, but that can be overrided
-    df = validate_schema(df, coerce_dtypes=True, coerce_rename=True)
+    df = validate_schema(df, coerce_dtypes=coerce_dtypes, coerce_rename=coerce_rename)
 
     return df

From eac97672dccc895cd109993fad3a2ac3f43f11bc Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 28 Aug 2026 14:20:38 -0400
Subject: [PATCH 02/18] 1) print schema validation errors per line

---
 quantaq_cli/schema.py | 63 +++++++++++++++++++++++++++++++++++--------
 1 file changed, 52 insertions(+), 11 deletions(-)

diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py
index 9bf0e51..659d8bb 100644
--- a/quantaq_cli/schema.py
+++ b/quantaq_cli/schema.py
@@ -1,4 +1,5 @@
 import contextlib
+import json
 
 from loguru import logger
 import numpy as np
@@ -179,16 +180,23 @@ def validate_schema(df, nullable=True, required=False, coerce_dtypes=True, coerc
     Returns:
         df (pd.DataFrame): the DataFrame with validated dtypes and column names
     """
-    df = df.copy()
 
     columns = {
-        name: pa.Column(dtype, nullable=nullable, required=required)
+        name: pa.Column(dtype, nullable=nullable, required=required, coerce=coerce_dtypes)
         for name, dtype in COLUMN_DEFINITIONS
     }
 
+    expected_dtypes = dict(COLUMN_DEFINITIONS)
     legacy_names = set(STATIC_COLUMN_RENAMES.keys())
     prefix_patterns = COLUMN_RENAME_PREFIXES
 
+    def _wrong_dtype_columns(df):
+        """Return columns whose dtype doesn't match their expected dtype."""
+        return sorted(
+            col for col in df.columns
+            if col in expected_dtypes and df[col].dtype != np.dtype(expected_dtypes[col])
+        )
+
     def _has_legacy_column_names(df):
         """Return True if any legacy names/prefixes remain."""
         has_legacy_name = bool(legacy_names.intersection(df.columns))
@@ -201,6 +209,42 @@ def _no_legacy_column_names(df):
         """Pandera check: True if no legacy names/prefixes remain."""
         return not _has_legacy_column_names(df)
 
+    def _expected_rename(col):
+        """Predict what standardize_columns() would rename this column to."""
+        if col in STATIC_COLUMN_RENAMES:
+            return STATIC_COLUMN_RENAMES[col]
+        if col.startswith("bin"):
+            return col.replace("bin", "opc_bin")
+        if col.startswith("opc.bin"):
+            return col.replace(".", "_")
+        if col.startswith("met."):
+            return col.replace("met.", "")
+        if col.startswith("gases."):
+            return "ox_diff" if col == "gases.o3.diff" else col.removeprefix("gases.").replace(".", "_")
+        if col.startswith("geo."):
+            return col.removeprefix("geo.")
+        return "unknown rename rule"
+
+    def _log_failures(df):
+        """Log one line per failure: dtype mismatches and legacy column names."""
+        wrong_dtype_cols = _wrong_dtype_columns(df)
+        for col in wrong_dtype_cols:
+            logger.info(
+                "Schema validation failed - wrong dtype: '{}' should be {}, got {}",
+                col, np.dtype(expected_dtypes[col]), df[col].dtype,
+            )
+
+        if _has_legacy_column_names(df):
+            legacy_cols = (
+                sorted(legacy_names.intersection(df.columns))
+                + sorted(col for col in df.columns for p in prefix_patterns if col.startswith(p))
+            )
+            for col in legacy_cols:
+                logger.info(
+                    "Schema validation failed - unstandardized column name: should be '{}', got '{}'",
+                    _expected_rename(col), col,
+                )
+
     # index = None means no index is specified
     # strict = False allows missing and extra columns in the DataFrame
     schema = pa.DataFrameSchema(
@@ -214,20 +258,16 @@ def _no_legacy_column_names(df):
     )
 
     try:
-        # print all schema errors instead of raising on the first error
         schema.validate(df, lazy=True)
     except pa.errors.SchemaErrors as err:
-        logger.error("Schema validation failed.")
-        logger.error("{}", err.failure_cases.to_string())
+        logger.bind(schema_errors=err.message).error("Schema validation failed")
+        _log_failures(df)
 
-        # Check the actual condition directly rather than parsing
-        # failure_cases["check"], since pandera's naming of anonymous
-        # check functions in that column isn't a stable contract.
         if coerce_rename and _has_legacy_column_names(df):
             logger.warning("Standardizing unstandardized column names.")
             df = standardize_columns(df)
 
-        if coerce_dtypes:
+        if coerce_dtypes and _wrong_dtype_columns(df):
             logger.warning("Coercing dtypes to expected types.")
             dtype_map = {
                 col: dtype for col, dtype in COLUMN_DEFINITIONS
@@ -240,8 +280,9 @@ def _no_legacy_column_names(df):
                 schema.validate(df, lazy=True)
                 logger.info("Schema validation passed after coercion.")
             except pa.errors.SchemaErrors as second_err:
-                logger.error("Schema validation still failing after coercion.")
-                logger.error("{}", second_err.failure_cases.to_string())
+                logger.bind(schema_errors=second_err.message).error("Schema validation still failing after coercion.")
+                _log_failures(df)
+                raise ValueError(f"Schema validation failed: {json.dumps(second_err.message)}") from second_err
 
     return df
 

From 07649dc1de37321b8e5eed7f5fd95d8be2e4ab17 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 28 Aug 2026 14:28:21 -0400
Subject: [PATCH 03/18] 6) undo adding of opc_ prefix to bin column names to
 match what's used in the pm models

---
 quantaq_cli/schema.py |  71 +++++++++++++---------------
 tests/test_schema.py  | 104 +++++++++++++++++++++---------------------
 2 files changed, 84 insertions(+), 91 deletions(-)

diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py
index 659d8bb..0806ec0 100644
--- a/quantaq_cli/schema.py
+++ b/quantaq_cli/schema.py
@@ -9,34 +9,34 @@
 # Default dtypes
 COLUMN_DEFINITIONS = [
     # --- OPC colunms ---
-    ('opc_bin0', np.float64),
-    ('opc_bin1', np.float64),
-    ('opc_bin2', np.float64),
-    ('opc_bin3', np.float64),
-    ('opc_bin4', np.float64),
-    ('opc_bin5', np.float64),
-    ('opc_bin6', np.float64),
-    ('opc_bin7', np.float64),
-    ('opc_bin8', np.float64),
-    ('opc_bin9', np.float64),
-    ('opc_bin10', np.float64),
-    ('opc_bin11', np.float64),
-    ('opc_bin12', np.float64),
-    ('opc_bin13', np.float64),
-    ('opc_bin14', np.float64),
-    ('opc_bin15', np.float64),
-    ('opc_bin16', np.float64),
-    ('opc_bin17', np.float64),
-    ('opc_bin18', np.float64),
-    ('opc_bin19', np.float64),
-    ('opc_bin20', np.float64),
-    ('opc_bin21', np.float64),
-    ('opc_bin22', np.float64),
-    ('opc_bin23', np.float64),
-    ('opc_bin1MToF', np.float64),
-    ('opc_bin3MToF', np.float64),
-    ('opc_bin5MToF', np.float64),
-    ('opc_bin7MToF', np.float64),
+    ('bin0', np.float64),
+    ('bin1', np.float64),
+    ('bin2', np.float64),
+    ('bin3', np.float64),
+    ('bin4', np.float64),
+    ('bin5', np.float64),
+    ('bin6', np.float64),
+    ('bin7', np.float64),
+    ('bin8', np.float64),
+    ('bin9', np.float64),
+    ('bin10', np.float64),
+    ('bin11', np.float64),
+    ('bin12', np.float64),
+    ('bin13', np.float64),
+    ('bin14', np.float64),
+    ('bin15', np.float64),
+    ('bin16', np.float64),
+    ('bin17', np.float64),
+    ('bin18', np.float64),
+    ('bin19', np.float64),
+    ('bin20', np.float64),
+    ('bin21', np.float64),
+    ('bin22', np.float64),
+    ('bin23', np.float64),
+    ('bin1MToF', np.float64),
+    ('bin3MToF', np.float64),
+    ('bin5MToF', np.float64),
+    ('bin7MToF', np.float64),
     ('opc_temp', np.float64),
     ('opc_rh', np.float64),
     ('opc_pm1', np.float64),
@@ -160,7 +160,7 @@
 }
 
 # Prefixes for unstandardized column names
-COLUMN_RENAME_PREFIXES = ("bin", "opc.bin", "met.", "gases.", "geo.")
+COLUMN_RENAME_PREFIXES = ("opc.bin", "met.", "gases.", "geo.")
 
 
 def validate_schema(df, nullable=True, required=False, coerce_dtypes=True, coerce_rename=True):
@@ -213,10 +213,8 @@ def _expected_rename(col):
         """Predict what standardize_columns() would rename this column to."""
         if col in STATIC_COLUMN_RENAMES:
             return STATIC_COLUMN_RENAMES[col]
-        if col.startswith("bin"):
-            return col.replace("bin", "opc_bin")
         if col.startswith("opc.bin"):
-            return col.replace(".", "_")
+            return col.replace("opc.", "")
         if col.startswith("met."):
             return col.replace("met.", "")
         if col.startswith("gases."):
@@ -314,11 +312,6 @@ def standardize_columns(df):
     # STATIC_COLUMN_RENAMES dict used by validate_schema
     column_renames = dict(STATIC_COLUMN_RENAMES)
 
-    # bin0 --> opc_bin0, etc..
-    for column in df.columns:
-        if column.startswith("bin"):  # opc_bin0, opc_bin23
-            column_renames[column] = column.replace("bin", "opc_bin")
-
     # Add diff columns (we minus ae) if they don't already exist
     if not any('diff' in col for col in df.columns):
         for pollutant in ("co", "no", "no2"):
@@ -332,8 +325,8 @@ def standardize_columns(df):
 
     # CloudAPI schema to database schema
     for column in df.columns:
-        if column.startswith("opc.bin"):  # opc_bin0, opc_bin23
-            column_renames[column] = column.replace(".", "_")
+        if column.startswith("opc.bin"): 
+            column_renames[column] = column.replace("opc.", "")
         elif column.startswith("met."):
             column_renames[column] = column.replace("met.", "")
         elif column.startswith("gases."):
diff --git a/tests/test_schema.py b/tests/test_schema.py
index 121be84..b84db7a 100644
--- a/tests/test_schema.py
+++ b/tests/test_schema.py
@@ -85,34 +85,34 @@ def test_schema_migration_modpm_rawsd(self):
             'sample_rh',
             'sample_temp',
             'sample_pres',
-            'opc_bin0',
-            'opc_bin1',
-            'opc_bin2',
-            'opc_bin3',
-            'opc_bin4',
-            'opc_bin5',
-            'opc_bin6',
-            'opc_bin7',
-            'opc_bin8',
-            'opc_bin9',
-            'opc_bin10',
-            'opc_bin11',
-            'opc_bin12',
-            'opc_bin13',
-            'opc_bin14',
-            'opc_bin15',
-            'opc_bin16',
-            'opc_bin17',
-            'opc_bin18',
-            'opc_bin19',
-            'opc_bin20',
-            'opc_bin21',
-            'opc_bin22',
-            'opc_bin23',
-            'opc_bin1MToF',
-            'opc_bin3MToF',
-            'opc_bin5MToF',
-            'opc_bin7MToF',
+            'bin0',
+            'bin1',
+            'bin2',
+            'bin3',
+            'bin4',
+            'bin5',
+            'bin6',
+            'bin7',
+            'bin8',
+            'bin9',
+            'bin10',
+            'bin11',
+            'bin12',
+            'bin13',
+            'bin14',
+            'bin15',
+            'bin16',
+            'bin17',
+            'bin18',
+            'bin19',
+            'bin20',
+            'bin21',
+            'bin22',
+            'bin23',
+            'bin1MToF',
+            'bin3MToF',
+            'bin5MToF',
+            'bin7MToF',
             'opc_sample_period',
             'opc_sample_flow',
             'opc_temp',
@@ -236,30 +236,30 @@ def test_schema_migration_modx_api(self):
             'neph_pm1_env',
             'neph_pm10_env',
             'neph_pm25_env',
-            'opc_bin0',
-            'opc_bin1',
-            'opc_bin10',
-            'opc_bin11',
-            'opc_bin12',
-            'opc_bin13',
-            'opc_bin14',
-            'opc_bin15',
-            'opc_bin16',
-            'opc_bin17',
-            'opc_bin18',
-            'opc_bin19',
-            'opc_bin2',
-            'opc_bin20',
-            'opc_bin21',
-            'opc_bin22',
-            'opc_bin23',
-            'opc_bin3',
-            'opc_bin4',
-            'opc_bin5',
-            'opc_bin6',
-            'opc_bin7',
-            'opc_bin8',
-            'opc_bin9',
+            'bin0',
+            'bin1',
+            'bin10',
+            'bin11',
+            'bin12',
+            'bin13',
+            'bin14',
+            'bin15',
+            'bin16',
+            'bin17',
+            'bin18',
+            'bin19',
+            'bin2',
+            'bin20',
+            'bin21',
+            'bin22',
+            'bin23',
+            'bin3',
+            'bin4',
+            'bin5',
+            'bin6',
+            'bin7',
+            'bin8',
+            'bin9',
             'opc_pm1',
             'opc_pm10',
             'opc_pm25'

From 2d92f851342aa17066424dfd018027516f066be0 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 28 Aug 2026 14:37:41 -0400
Subject: [PATCH 04/18] 2) bad records in int-column should be coerced to NaN

---
 quantaq_cli/schema.py | 20 +++++++++++++++-----
 1 file changed, 15 insertions(+), 5 deletions(-)

diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py
index 0806ec0..bb99f04 100644
--- a/quantaq_cli/schema.py
+++ b/quantaq_cli/schema.py
@@ -3,6 +3,7 @@
 
 from loguru import logger
 import numpy as np
+import pandas as pd
 import pandera.pandas as pa
 
 
@@ -267,11 +268,20 @@ def _log_failures(df):
 
         if coerce_dtypes and _wrong_dtype_columns(df):
             logger.warning("Coercing dtypes to expected types.")
-            dtype_map = {
-                col: dtype for col, dtype in COLUMN_DEFINITIONS
-                if col in df.columns
-            }
-            df = df.astype(dtype_map)
+            for col, dtype in COLUMN_DEFINITIONS:
+                if col not in df.columns:
+                    continue
+                if np.issubdtype(np.dtype(dtype), np.number):
+                    before_na = df[col].isna().sum()
+                    df[col] = pd.to_numeric(df[col], errors="coerce").astype(dtype)
+                    new_na = df[col].isna().sum() - before_na
+                    if new_na > 0:
+                        logger.warning(
+                            "Coerced {} unparseable value(s) in '{}' to NaN.",
+                            new_na, col,
+                        )
+                else:
+                    df[col] = df[col].astype(dtype)
 
         if coerce_rename or coerce_dtypes:
             try:

From c50c390573a4af087935b18b9a563877da851674 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 28 Aug 2026 15:17:48 -0400
Subject: [PATCH 05/18] 5) rename old wind column names

---
 quantaq_cli/schema.py           | 17 ++++++++++++++++-
 quantaq_cli/toolkit/resample.py |  2 +-
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py
index bb99f04..6cd627d 100644
--- a/quantaq_cli/schema.py
+++ b/quantaq_cli/schema.py
@@ -118,6 +118,14 @@
     ('iteration', np.int16),
     ('dd_measurement_state', np.float64), # needs to be nullable for older data
     ('dd_operating_state', np.float64), # needs to be nullable for older data
+
+    # -- Wind columns --
+    ('wx_u', np.float64),
+    ('wx_u', np.float64),
+    ('wx_wd', np.float64),
+    ('wx_ws', np.float64),
+    ('wx_ws_scalar', np.float64),
+
 ]
 
 STATIC_COLUMN_RENAMES = {
@@ -157,7 +165,14 @@
     "rh": "sample_rh",
 
     # --- Device / metadata columns ---
-    "operating_state": "dd_operating_state"
+    "operating_state": "dd_operating_state",
+
+    # -- Wind columns ---
+    "w_u": "wx_u", 
+    "w_v": "wx_v", 
+    "wd": "wx_wd",
+    "ws_vector": "wx_ws",  
+    "ws" : "wx_ws_scalar"
 }
 
 # Prefixes for unstandardized column names
diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py
index 75fddf3..bf54847 100644
--- a/quantaq_cli/toolkit/resample.py
+++ b/quantaq_cli/toolkit/resample.py
@@ -205,7 +205,7 @@ def resample_dataframe(
         if have_uv and (df[u_col].isna().all() or df[v_col].isna().all()):
             logger.debug(
                     "All wind components contain NaNs ({}: {}, {}: {}); "
-                    "Deriving them from the averaged u/v components",
+                    "cannot vector-average from u/v - will derive from speed/direction instead",
                     u_col, int(df[u_col].isna().sum()),
                     v_col, int(df[v_col].isna().sum()),
                 )

From d134274bb34a24b12b5d2ec0518026403d48e75e Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 28 Aug 2026 15:45:07 -0400
Subject: [PATCH 06/18] add old sdcard u/v

---
 quantaq_cli/schema.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py
index 6cd627d..06c6ce6 100644
--- a/quantaq_cli/schema.py
+++ b/quantaq_cli/schema.py
@@ -170,6 +170,8 @@
     # -- Wind columns ---
     "w_u": "wx_u", 
     "w_v": "wx_v", 
+    "u": "wx_u", 
+    "v": "wx_v", 
     "wd": "wx_wd",
     "ws_vector": "wx_ws",  
     "ws" : "wx_ws_scalar"

From f537f941c3fe05adeda7a8461a356e6689c2b35b Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Wed, 16 Sep 2026 15:33:45 -0400
Subject: [PATCH 07/18] recast flag values to in64

---
 quantaq_cli/toolkit/resample.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py
index bf54847..03e0609 100644
--- a/quantaq_cli/toolkit/resample.py
+++ b/quantaq_cli/toolkit/resample.py
@@ -118,6 +118,7 @@ def _resampler(frame):
 
     out = clean_agg.where(has_clean, all_agg)
     out["flag"] = np.where(has_clean, 0, flag_or)
+    out["flag"] = out["flag"].astype("int64")
 
     return out.reset_index()
 

From 04ea403b95dc9cde06f891d1774c643e7e44c228 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Wed, 16 Sep 2026 15:40:49 -0400
Subject: [PATCH 08/18] drop empty bins

---
 quantaq_cli/toolkit/resample.py | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py
index 03e0609..94b56b0 100644
--- a/quantaq_cli/toolkit/resample.py
+++ b/quantaq_cli/toolkit/resample.py
@@ -118,6 +118,13 @@ def _resampler(frame):
 
     out = clean_agg.where(has_clean, all_agg)
     out["flag"] = np.where(has_clean, 0, flag_or)
+    
+    # drop empty bins
+    is_empty_bin = flag_or.isna()
+    if is_empty_bin.any():
+        logger.debug("Dropping {} empty resample bin(s)", int(is_empty_bin.sum()))
+        out = out.loc[~is_empty_bin]
+
     out["flag"] = out["flag"].astype("int64")
 
     return out.reset_index()

From d73e9fbb7d01f22841c26156bb2350b0c455151a Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Thu, 17 Sep 2026 11:24:50 -0400
Subject: [PATCH 09/18] remove opc_ prefix from _OPC_COLUMNS

---
 quantaq_cli/variables.py | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py
index aac610f..918a51e 100644
--- a/quantaq_cli/variables.py
+++ b/quantaq_cli/variables.py
@@ -8,10 +8,10 @@
 
 # Columns to Nan -- only for standardized schema types
 _OPC_COLUMNS = [
-    "opc_bin0", "opc_bin1", "opc_bin2", "opc_bin3", "opc_bin4", "opc_bin5", "opc_bin6", "opc_bin7", "opc_bin8", "opc_bin9",
-    "opc_bin10", "opc_bin11", "opc_bin12", "opc_bin13", "opc_bin14", "opc_bin15", "opc_bin16", "opc_bin17", "opc_bin18",
-    "opc_bin19", "opc_bin20", "opc_bin21", "opc_bin22", "opc_bin23", 
-    "opc_bin1MToF", "opc_bin3MToF", "opc_bin5MToF", "opc_bin7MToF",
+    "bin0", "bin1", "bin2", "bin3", "bin4", "bin5", "bin6", "bin7", "bin8", "bin9",
+    "bin10", "bin11", "bin12", "bin13", "bin14", "bin15", "bin16", "bin17", "bin18",
+    "bin19", "bin20", "bin21", "bin22", "bin23", 
+    "bin1MToF", "bin3MToF", "bin5MToF", "bin7MToF",
     "opc_sample_period", "opc_sample_flow",
     "opc_pm1", "opc_pm25", "opc_pm10",
     "opc_laser_status", "opc_temp", "opc_rh", 
@@ -135,9 +135,9 @@
         # Check: ratio between the OPC and nephelometer is within spec
         Multiple(
             criteria=(
-                Single(column="opc_bin0", op=">=", value=10.0),
+                Single(column="bin0", op=">=", value=10.0),
                 Ratio(column_numerator="neph_bin0", 
-                      column_denominator="opc_bin0",
+                      column_denominator="bin0",
                       op=">", 
                       value=2000.0),
             ), logical_operator="AND",
@@ -147,7 +147,7 @@
         # Check 1: ensure that the neph isn't reading 0's when it shouldn't be
         Multiple(
             criteria=(
-                Single(column="opc_bin0", op=">=", value=10.0),
+                Single(column="bin0", op=">=", value=10.0),
                 Single(column="neph_bin0", op="==", value=0.0)
             ),
             logical_operator="AND",
@@ -157,7 +157,7 @@
             criteria=(
                 Single(column="neph_bin0", op=">=", value=10.0),
                 Ratio(column_numerator="neph_bin0", 
-                      column_denominator="opc_bin0",
+                      column_denominator="bin0",
                       op=">", 
                       value=2000.0),
             ), logical_operator="AND",
@@ -185,7 +185,7 @@
             criteria=(
                 Single(column="neph_bin0", op=">=", value=10.0),
                 Ratio(column_numerator="neph_bin0", 
-                      column_denominator="opc_bin0",
+                      column_denominator="bin0",
                       op=">", 
                       value=2000.0),
             ), logical_operator="AND",

From d371403193e0ff534ead67138483e2a39e821394 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 2 Oct 2026 15:02:11 -0400
Subject: [PATCH 10/18] Set flags to zero before re-flagging

---
 quantaq_cli/toolkit/flag.py | 13 ++-------
 tests/test_flag.py          | 54 -------------------------------------
 2 files changed, 2 insertions(+), 65 deletions(-)

diff --git a/quantaq_cli/toolkit/flag.py b/quantaq_cli/toolkit/flag.py
index ca6cd29..03dfe57 100644
--- a/quantaq_cli/toolkit/flag.py
+++ b/quantaq_cli/toolkit/flag.py
@@ -188,12 +188,8 @@ def flag_dataframe(df):
     """
     df = df.copy()
 
-    # Drop nan flags (could happen after a merge)
-    if "flag" not in df.columns:
-        df["flag"] = 0
-    elif df["flag"].isna().any():
-        logger.warning("Dropping {} rows with NaN flags", df["flag"].isna().sum())
-        df = df.dropna(how='any', subset=["flag"])
+    # set flags to zero before re-flagging
+    df["flag"] = np.int64(0)
 
     # only need column names to be valid for flagging
     # we don't coerce dtypes so that merged files can be flagged
@@ -205,11 +201,6 @@ def flag_dataframe(df):
     # get flag criteria (this also checks if the data source is valid)
     name_to_criteria = flag_name_to_criteria(source).items()
 
-    # create flag column if it doesn't exist
-    if "flag" not in df.columns:
-        df["flag"] = np.int64(0)
-
-
     # sort the dataframe once before adding flags
     df = fix_timestamps(df, sort_values=True)
 
diff --git a/tests/test_flag.py b/tests/test_flag.py
index c9c0ac0..c6ecf41 100644
--- a/tests/test_flag.py
+++ b/tests/test_flag.py
@@ -144,57 +144,3 @@ def test_flag_files_modulair_db_parquet(self):
         
         # is it a parquet?
         self.assertEqual(p.suffix, ".parquet")
-
-def test_flag_dataframe_nanflag():
-    sample_df = pd.DataFrame(
-        [
-            {
-                "timestamp": pd.to_datetime("2023-03-10T17:41:24Z"),
-                "sample_rh": 77.6,
-                "sample_temp": -10.4,
-                "neph_pm1_env": 10.6,
-                "neph_pm25_env": 25.0,
-                "neph_pm10_env": 32.4,
-                "neph_bin0": 1590.375,
-                "co_we": 758.8,
-                "co_ae": 656.8,
-                "co_diff": 102.0,
-                "flag": np.nan,  # this row should be dropped
-                "sn": "MOD-00001",
-            },
-            {
-                "timestamp": pd.to_datetime("2023-03-10T17:42:24Z"),
-                "sample_rh": 77.9,
-                "sample_temp": -10.3,
-                "neph_pm1_env": 10.4,
-                "neph_pm25_env": 24.8,
-                "neph_pm10_env": 32.1,
-                "neph_bin0": 1588.0,
-                "co_we": 759.1,
-                "co_ae": 657.0,
-                "co_diff": 102.1,
-                "flag": 0.,  # this row should survive
-                "sn": "MOD-00001",
-            },
-            {
-                "timestamp": pd.to_datetime("2023-03-10T17:43:24Z"),
-                "sample_rh": 78.1,
-                "sample_temp": -10.2,
-                "neph_pm1_env": 10.5,
-                "neph_pm25_env": 24.9,
-                "neph_pm10_env": 32.0,
-                "neph_bin0": 1589.0,
-                "co_we": 759.3,
-                "co_ae": 657.2,
-                "co_diff": 102.0,
-                "flag": 0.,  # this row should survive
-                "sn": "MOD-00001",
-            },
-        ],
-    )
-
-    expected = sample_df.dropna(how="any")
-    
-    flagged = flag_dataframe(sample_df)
-
-    assert_frame_equal(flagged, expected, check_dtype=False)

From 0e5987e5ab0a31f2acd60c8cdff4350b19756b11 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Sat, 3 Oct 2026 01:08:14 -0400
Subject: [PATCH 11/18] Make resetting flags optional

---
 quantaq_cli/toolkit/flag.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/quantaq_cli/toolkit/flag.py b/quantaq_cli/toolkit/flag.py
index 03dfe57..bcd82c7 100644
--- a/quantaq_cli/toolkit/flag.py
+++ b/quantaq_cli/toolkit/flag.py
@@ -176,7 +176,7 @@ def echo_flag_table(df):
         table.add_row(*map(str, row))
     rich.print(table)
 
-def flag_dataframe(df):
+def flag_dataframe(df, reset_flags=False):
     """Re-flags a DataFrame by iterating through the FLAG_DEFINITIONS and calling
     the _add_flag() function one-by-one.
 
@@ -189,7 +189,11 @@ def flag_dataframe(df):
     df = df.copy()
 
     # set flags to zero before re-flagging
-    df["flag"] = np.int64(0)
+    if reset_flags or 'flag' not in df.columns:
+        df["flag"] = np.int64(0)
+    elif df["flag"].isna().any():
+        logger.warning("Dropping {} rows with NaN flags", df["flag"].isna().sum())
+        df = df.dropna(how='any', subset=["flag"])
 
     # only need column names to be valid for flagging
     # we don't coerce dtypes so that merged files can be flagged

From 5207d84a1246f9068d2ebf16184ce62d7063d77a Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Thu, 8 Oct 2026 13:22:24 -0400
Subject: [PATCH 12/18] Add FLAG_OPC and FLAG_NEPH when both are reading 0.0
 (this happens when these flags are set in the firmware but the raw data is
 kept at 0.0 with the flags carried through --> we can't reflag this in the
 database but we can catch when both are 0.0)

---
 quantaq_cli/cli.py          |  6 +++---
 quantaq_cli/toolkit/flag.py |  3 ++-
 quantaq_cli/variables.py    | 20 ++++++++++++++++++++
 3 files changed, 25 insertions(+), 4 deletions(-)

diff --git a/quantaq_cli/cli.py b/quantaq_cli/cli.py
index 1177fc2..5fb37a6 100644
--- a/quantaq_cli/cli.py
+++ b/quantaq_cli/cli.py
@@ -1,5 +1,6 @@
 from pathlib import Path
-import pkg_resources
+from pathlib import Path
+from importlib.metadata import version
 
 import rich_click as click
 from loguru import logger
@@ -14,8 +15,7 @@
 
 CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
 
-__version__ = pkg_resources.get_distribution('quantaq_cli').version
-
+__version__ = version("quantaq_cli")
 
 @click.group(context_settings=CONTEXT_SETTINGS)
 @click.version_option(__version__)
diff --git a/quantaq_cli/toolkit/flag.py b/quantaq_cli/toolkit/flag.py
index bcd82c7..cd026c5 100644
--- a/quantaq_cli/toolkit/flag.py
+++ b/quantaq_cli/toolkit/flag.py
@@ -176,12 +176,13 @@ def echo_flag_table(df):
         table.add_row(*map(str, row))
     rich.print(table)
 
-def flag_dataframe(df, reset_flags=False):
+def flag_dataframe(df, reset_flags=True):
     """Re-flags a DataFrame by iterating through the FLAG_DEFINITIONS and calling
     the _add_flag() function one-by-one.
 
     Args:
         df (pd.DataFrame): DataFrame to be flagged (or re-flagged).
+        reset_flags (bool): If True, the 'flag' column is reset to zero. Default is True.
 
     Returns:
         pd.DataFrame: The flagged DataFrame.
diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py
index 918a51e..4ff0d9e 100644
--- a/quantaq_cli/variables.py
+++ b/quantaq_cli/variables.py
@@ -142,6 +142,16 @@
                       value=2000.0),
             ), logical_operator="AND",
             ),
+        # Check: ensure that the neph and opc are not both reading 0s
+        # (this happens when an entire 1-min window has FLAG_OPC and FLAG_NEPH
+        # set by the firmware at 5-s intervals)
+        Multiple(
+            criteria=(
+                Single(column="bin0", op="==", value=0.0),
+                Single(column="neph_bin0", op="==", value=0.0)
+            ),
+            logical_operator="AND",
+        ),
     ],
      "FLAG_NEPH": [
         # Check 1: ensure that the neph isn't reading 0's when it shouldn't be
@@ -162,6 +172,16 @@
                       value=2000.0),
             ), logical_operator="AND",
         ),
+        # Check 3: ensure that the neph and opc are not both reading 0s
+        # (this happens when an entire 1-min window has FLAG_OPC and FLAG_NEPH
+        # set by the firmware at 5-s intervals)
+        Multiple(
+            criteria=(
+                Single(column="bin0", op="==", value=0.0),
+                Single(column="neph_bin0", op="==", value=0.0)
+            ),
+            logical_operator="AND",
+        ),
     ],
 }
 

From c0761b5705314cb61739c377d0ede2658e8d8f53 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Tue, 20 Oct 2026 14:17:52 -0400
Subject: [PATCH 13/18] Relax pandas constraint for compatibility with
 quantaq-gases

---
 poetry.lock    | 4 ++--
 pyproject.toml | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/poetry.lock b/poetry.lock
index c74e4f5..d9930c4 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.8.4 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.8.5 and should not be changed by hand.
 
 [[package]]
 name = "alabaster"
@@ -3486,4 +3486,4 @@ type = ["pytest-mypy"]
 [metadata]
 lock-version = "2.0"
 python-versions = ">=3.9, <3.14"
-content-hash = "774d525b3fb63fc9dc29ac028ecb186dba419bc685fe67b5bbe1701dfb4d71b2"
+content-hash = "f89f75696d9b24bf6326bea467d5cd3a0fafc98460c3cdcc5af42875a2bf5540"
diff --git a/pyproject.toml b/pyproject.toml
index be73d95..25b05d0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -12,7 +12,7 @@ classifiers = []
 
 [tool.poetry.dependencies]
 python = ">=3.9, <3.14"
-pandas = ">=2.1.4,<2.2"
+pandas = ">=2.1.4,<3"
 click = "^8.1"
 pyarrow = ">=14.0.2"
 terminaltables = "^3.1.0"

From e777d5e1f7b155634693474da54b6b32df5a9427 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Tue, 20 Oct 2026 14:20:33 -0400
Subject: [PATCH 14/18] Relax numpy constraint for compatibility with
 quantaq-gases

---
 poetry.lock    | 2 +-
 pyproject.toml | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/poetry.lock b/poetry.lock
index d9930c4..0534055 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -3486,4 +3486,4 @@ type = ["pytest-mypy"]
 [metadata]
 lock-version = "2.0"
 python-versions = ">=3.9, <3.14"
-content-hash = "f89f75696d9b24bf6326bea467d5cd3a0fafc98460c3cdcc5af42875a2bf5540"
+content-hash = "bb1f7b0a0541e02a04ffbf88b42c373e159902965aa56a5ec71e0b0b9d598578"
diff --git a/pyproject.toml b/pyproject.toml
index 25b05d0..d56fab5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -17,7 +17,7 @@ click = "^8.1"
 pyarrow = ">=14.0.2"
 terminaltables = "^3.1.0"
 urllib3 = "^1.26.10"
-numpy = "^1.23.2"
+numpy = ">=1.23.2,<3"
 rich-click = ">=1.7.2"
 loguru = "^0.7.3"
 pandera = ">=0.24,<0.29"

From 0c3eab3cb0ee627752e3ae362d2d5100872a0d31 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Fri, 23 Oct 2026 11:06:41 -0400
Subject: [PATCH 15/18] Add schema utils for undoing column name
 standardization

---
 quantaq_cli/schema.py | 41 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 41 insertions(+)

diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py
index 06c6ce6..ac7391e 100644
--- a/quantaq_cli/schema.py
+++ b/quantaq_cli/schema.py
@@ -368,3 +368,44 @@ def standardize_columns(df):
     df = df.rename(columns=column_renames)
     df = drop_unnamed(df)
     return df
+
+# utilites to undo column name standardization
+def map_new_to_old_columns(old_cols, new_cols):
+    """Map renamed columns back to their original names.
+
+    Args:
+        old_cols (Iterable[str]): Column names before standardizing.
+        new_cols (Iterable[str]): Column names after standardizing.
+
+    Returns:
+        dict: {new_name: old_name} for every column whose name changed.
+    """
+    new_cols = set(new_cols)
+    new_to_old = {}
+    for old in old_cols:
+        new = standardize_columns(pd.DataFrame(columns=[old])).columns[0]
+        if new == old:
+            continue
+        if new not in new_cols:
+            error = ValueError(f"'{old}' standardizes to '{new}', which isn't in new_cols")
+            continue
+        if new in new_to_old:
+            error = ValueError(f"'{new_to_old[new]}' and '{old}' both standardize to '{new}'")
+            logger.error(error)
+            raise error
+        new_to_old[new] = old
+    return new_to_old
+
+def undo_standardize_columns(original_cols, standardized_df):
+    """Restore the original column names on a standardized DataFrame.
+    """
+    new_to_old = map_new_to_old_columns(original_cols, standardized_df.columns)
+    restored_df = standardized_df.rename(columns=new_to_old)
+
+    duplicates = sorted(set(restored_df.columns[restored_df.columns.duplicated()]))
+    if duplicates:
+        error = ValueError(f"Restoring original names creates duplicate columns: {duplicates}")
+        logger.error(error)
+        raise error
+
+    return restored_df

From 6e95dc35eb46c3ec9bd8f15c3255fa676c196332 Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Mon, 26 Oct 2026 14:34:44 -0400
Subject: [PATCH 16/18] Count clean flags with flag ==0 or flag ==4096
 (overheat)

---
 quantaq_cli/toolkit/resample.py | 44 ++++++++++++++++++++-------------
 1 file changed, 27 insertions(+), 17 deletions(-)

diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py
index 94b56b0..079b728 100644
--- a/quantaq_cli/toolkit/resample.py
+++ b/quantaq_cli/toolkit/resample.py
@@ -9,6 +9,9 @@
 # Default (u, v, speed, direction) column names for vector wind averaging.
 WIND_COLUMNS = ("wx_u", "wx_v", "wx_ws", "wx_wd")
 
+# Flag values that count as clean (good data). Any other flag value makes a row bad.
+CLEAN_FLAGS = (0, 4096) # 4096 = overheat (we don't nan any rows based on this)
+
 def _vector_wind(
     df: pd.DataFrame, u_col: str, v_col: str, ws_col: str, wd_col: str
 ) -> pd.DataFrame:
@@ -77,7 +80,7 @@ def _min_required_count(timestamps, rule, threshold):
 def _flag_aware_resample(df, rule, keys, agg):
     """Vectorized flag-aware resampling.
 
-    For each resample bin: if any row has flag == 0, aggregate only those
+    For each resample bin: if any row has flag == 0 or flag == 4096, aggregate only those
     "clean" rows and set the output flag to 0. Otherwise aggregate all rows
     in the bin and set the output flag to the bitwise OR of every flag value
     present.
@@ -104,20 +107,24 @@ def _flag_aware_resample(df, rule, keys, agg):
     def _resampler(frame):
         return frame.groupby(keys).resample(rule) if keys else frame.resample(rule)
 
-    clean_agg = _resampler(df[df["flag"] == 0]).agg(agg)
+    is_clean = df["flag"].isin(CLEAN_FLAGS)
+    clean_agg = _resampler(df[is_clean]).agg(agg)
+    # The clean flags are 0 and 4096, so the largest clean flag in a bin is their bitwise OR
+    clean_flag = _resampler(df[is_clean])["flag"].max()
 
     base_resampler = _resampler(df)
     all_agg = base_resampler.agg(agg)
     flag_col = base_resampler["flag"]
-    has_clean = flag_col.agg(lambda s: bool((s == 0).any()))
+    has_clean = flag_col.agg(lambda s: bool(s.isin(CLEAN_FLAGS).any()))
     flag_or = flag_col.agg(lambda s: int(np.bitwise_or.reduce(s.to_numpy())) if len(s) else np.nan)
 
     clean_agg = clean_agg.reindex(all_agg.index)
     has_clean = has_clean.reindex(all_agg.index)
     flag_or = flag_or.reindex(all_agg.index)
+    clean_flag = clean_flag.reindex(all_agg.index)
 
     out = clean_agg.where(has_clean, all_agg)
-    out["flag"] = np.where(has_clean, 0, flag_or)
+    out["flag"] = np.where(has_clean, clean_flag, flag_or)
     
     # drop empty bins
     is_empty_bin = flag_or.isna()
@@ -166,7 +173,7 @@ def resample_dataframe(
         flag_aware: Whether to apply flag-aware row selection when resampling.
             If True, each resample bin is aggregated as follows:
 
-              - If the bin has one or more good rows (non-flagged rows, flag == 0),
+              - If the bin has one or more good rows (non-flagged rows, flag == 0 or 4096),
                 only those rows are aggregated, and the resulting flag is 0.
               - If there are no good rows (every row in the bin is flagged), all
                 rows are aggregated, and the resulting flag is the bitwise OR of every
@@ -185,7 +192,7 @@ def resample_dataframe(
             The definition of "complete" depends on ``flag_aware``:
 
               - If ``flag_aware=True``: the bin needs this fraction of its expected 
-                sample count to be clean (flag==0)
+                sample count to be clean (flag==0 or 4096)
               - If ``flag_aware=False``: the bin needs this fraction of its expected 
                 sample count to be non-null
 
@@ -268,30 +275,33 @@ def resample_dataframe(
 
         if flag_aware:
             # require >= completeness_threshold fraction of the expected
-            # sample count to be clean (flag == 0)
-            clean_indexed = indexed[indexed["flag"] == 0]
+            # sample count to be clean (flag == 0 or 4096)
+            clean_indexed = indexed[indexed["flag"].isin(CLEAN_FLAGS)]
             if keys:
                 clean_counts = (
                     clean_indexed.groupby(keys).resample(rule)["flag"]
                     .count().rename("_clean_count").reset_index()
                 )
+            else:
+                clean_counts = (
+                    clean_indexed["flag"].resample(rule)
+                    .count().rename("_clean_count").reset_index()
+                )
+
+            # Bins with no clean rows at the start or end are missing from clean_counts; count them as 0.
+            merged = out[merge_on].merge(clean_counts, on=merge_on, how="left")
+            merged["_clean_count"] = merged["_clean_count"].fillna(0)
+            if keys:
                 min_count_by_group = (
                     df.groupby(keys)[on]
                     .apply(lambda s: _min_required_count(s, rule, completeness_threshold))
                     .rename("_min_count")
                 )
-                clean_counts = clean_counts.merge(min_count_by_group, on=keys, how="left")
+                merged = merged.merge(min_count_by_group, on=keys, how="left")
             else:
-                clean_counts = (
-                    clean_indexed["flag"].resample(rule)
-                    .count().rename("_clean_count").reset_index()
-                )
-                clean_counts["_min_count"] = _min_required_count(
+                merged["_min_count"] = _min_required_count(
                     df[on], rule, completeness_threshold
                 )
-            clean_counts["_clean_count"] = clean_counts["_clean_count"].fillna(0)
-
-            merged = out[merge_on].merge(clean_counts, on=merge_on, how="left")
             fail_mask = merged["_clean_count"] < merged["_min_count"]
 
             for c in numeric_value_cols:

From a7bae82bc978da3325daa485b3d9001409f845aa Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Mon, 26 Oct 2026 15:00:10 -0400
Subject: [PATCH 17/18] Fix to make flag-aware resampling work with the
 completeness_checks (but note on usage: flag-aware resampling of 5s data to
 1min should not use completeness checks to match the firmware logic (if all
 rows are clean, keep all rows; if some rows are clean, keep only clean rows;
 if no rows are clean, flag the entire row)

---
 quantaq_cli/toolkit/resample.py | 20 ++++++--------------
 1 file changed, 6 insertions(+), 14 deletions(-)

diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py
index 079b728..583c2b4 100644
--- a/quantaq_cli/toolkit/resample.py
+++ b/quantaq_cli/toolkit/resample.py
@@ -81,8 +81,8 @@ def _flag_aware_resample(df, rule, keys, agg):
     """Vectorized flag-aware resampling.
 
     For each resample bin: if any row has flag == 0 or flag == 4096, aggregate only those
-    "clean" rows and set the output flag to 0. Otherwise aggregate all rows
-    in the bin and set the output flag to the bitwise OR of every flag value
+    "clean" rows and set the output flag to the bitwise OR of their flags (0 or 4096). 
+    Otherwise aggregate all rows in the bin and set the output flag to the bitwise OR of every flag value
     present.
 
     Args:
@@ -174,7 +174,7 @@ def resample_dataframe(
             If True, each resample bin is aggregated as follows:
 
               - If the bin has one or more good rows (non-flagged rows, flag == 0 or 4096),
-                only those rows are aggregated, and the resulting flag is 0.
+                only those rows are aggregated, and the resulting flag is the bitwise OR of their flags (0 or 4096)
               - If there are no good rows (every row in the bin is flagged), all
                 rows are aggregated, and the resulting flag is the bitwise OR of every
                 flag value present in the bin.
@@ -276,21 +276,13 @@ def resample_dataframe(
         if flag_aware:
             # require >= completeness_threshold fraction of the expected
             # sample count to be clean (flag == 0 or 4096)
-            clean_indexed = indexed[indexed["flag"].isin(CLEAN_FLAGS)]
+            is_clean = indexed[keys].assign(_clean_count=indexed["flag"].isin(CLEAN_FLAGS).astype("int64"))
             if keys:
-                clean_counts = (
-                    clean_indexed.groupby(keys).resample(rule)["flag"]
-                    .count().rename("_clean_count").reset_index()
-                )
+                clean_counts = is_clean.groupby(keys).resample(rule)["_clean_count"].sum().reset_index()
             else:
-                clean_counts = (
-                    clean_indexed["flag"].resample(rule)
-                    .count().rename("_clean_count").reset_index()
-                )
+                clean_counts = is_clean["_clean_count"].resample(rule).sum().reset_index()
 
-            # Bins with no clean rows at the start or end are missing from clean_counts; count them as 0.
             merged = out[merge_on].merge(clean_counts, on=merge_on, how="left")
-            merged["_clean_count"] = merged["_clean_count"].fillna(0)
             if keys:
                 min_count_by_group = (
                     df.groupby(keys)[on]

From 86edef3027f15cc46f221759dc1615243444729b Mon Sep 17 00:00:00 2001
From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com>
Date: Mon, 26 Oct 2026 15:35:20 -0400
Subject: [PATCH 18/18] Simplify resample.py: only allow completeness_threshold
 check to work if flag_aware=False (if flag_aware=True, mirror the logic in
 the firmware)

---
 quantaq_cli/toolkit/resample.py | 127 +++++++++++++-------------------
 1 file changed, 52 insertions(+), 75 deletions(-)

diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py
index 583c2b4..34b40f7 100644
--- a/quantaq_cli/toolkit/resample.py
+++ b/quantaq_cli/toolkit/resample.py
@@ -10,7 +10,7 @@
 WIND_COLUMNS = ("wx_u", "wx_v", "wx_ws", "wx_wd")
 
 # Flag values that count as clean (good data). Any other flag value makes a row bad.
-CLEAN_FLAGS = (0, 4096) # 4096 = overheat (we don't nan any rows based on this)
+CLEAN_FLAGS = (0, 4096) # 4096 = overheat --> we don't nan any rows based on this so 4096 is just informational.
 
 def _vector_wind(
     df: pd.DataFrame, u_col: str, v_col: str, ws_col: str, wd_col: str
@@ -55,7 +55,7 @@ def _components_from_polar(
     return df
 
 def _min_required_count(timestamps, rule, threshold):
-    """The minimum number of non-null samples needed in one resampling bin to 
+    """The minimum number of non-null samples needed in one resampling bin to
     meet the data completeness threshold.
 
     We infer the native sampling interval from the median tdiff (spacing between
@@ -107,24 +107,24 @@ def _flag_aware_resample(df, rule, keys, agg):
     def _resampler(frame):
         return frame.groupby(keys).resample(rule) if keys else frame.resample(rule)
 
-    is_clean = df["flag"].isin(CLEAN_FLAGS)
-    clean_agg = _resampler(df[is_clean]).agg(agg)
+    is_clean = df["flag"].isin(CLEAN_FLAGS) # rows we trust: flag 0 or 4096
+    clean_agg = _resampler(df[is_clean]).agg(agg) # each bin averaged over its clean rows only (bins with no clean rows are missing here)
     # The clean flags are 0 and 4096, so the largest clean flag in a bin is their bitwise OR
     clean_flag = _resampler(df[is_clean])["flag"].max()
 
     base_resampler = _resampler(df)
-    all_agg = base_resampler.agg(agg)
+    all_agg = base_resampler.agg(agg) # each bin averaged over all its rows (used for bins with no clean rows)
     flag_col = base_resampler["flag"]
-    has_clean = flag_col.agg(lambda s: bool(s.isin(CLEAN_FLAGS).any()))
-    flag_or = flag_col.agg(lambda s: int(np.bitwise_or.reduce(s.to_numpy())) if len(s) else np.nan)
+    has_clean = flag_col.agg(lambda s: bool(s.isin(CLEAN_FLAGS).any())) # does the bin have at least one clean row?
+    flag_or = flag_col.agg(lambda s: int(np.bitwise_or.reduce(s.to_numpy())) if len(s) else np.nan) # OR of every flag in the bin --> NaN for bins with no rows at all (dropped below)
 
-    clean_agg = clean_agg.reindex(all_agg.index)
+    clean_agg = clean_agg.reindex(all_agg.index) # clean_agg and clean_flag only have bins that contain clean rows so we align everyting to the full set of bins
     has_clean = has_clean.reindex(all_agg.index)
     flag_or = flag_or.reindex(all_agg.index)
     clean_flag = clean_flag.reindex(all_agg.index)
 
-    out = clean_agg.where(has_clean, all_agg)
-    out["flag"] = np.where(has_clean, clean_flag, flag_or)
+    out = clean_agg.where(has_clean, all_agg) # per-bin: the clean-row average if it has clean rows, otherwise the all-rows average
+    out["flag"] = np.where(has_clean, clean_flag, flag_or) # per-bin: OR of its clean flags (0 or 4096) if it has clean rows, otherwise OR of all its flags
     
     # drop empty bins
     is_empty_bin = flag_or.isna()
@@ -173,7 +173,7 @@ def resample_dataframe(
         flag_aware: Whether to apply flag-aware row selection when resampling.
             If True, each resample bin is aggregated as follows:
 
-              - If the bin has one or more good rows (non-flagged rows, flag == 0 or 4096),
+              - If the bin has one or more good rows (clean rows, flag == 0 or 4096),
                 only those rows are aggregated, and the resulting flag is the bitwise OR of their flags (0 or 4096)
               - If there are no good rows (every row in the bin is flagged), all
                 rows are aggregated, and the resulting flag is the bitwise OR of every
@@ -187,22 +187,26 @@ def resample_dataframe(
         force_nonnumeric: Column name(s) to always aggregate with
             ``nonnumeric_how`` regardless of dtype, e.g. ``"fw"``,
             which is numeric but shouldn't be averaged.
-        completeness_threshold: If given (e.g. ``0.75``), bins that don't
-            meet the data-completeness requirement are set to NaN. 
-            The definition of "complete" depends on ``flag_aware``:
-
-              - If ``flag_aware=True``: the bin needs this fraction of its expected 
-                sample count to be clean (flag==0 or 4096)
-              - If ``flag_aware=False``: the bin needs this fraction of its expected 
-                sample count to be non-null
+        completeness_threshold: If given (e.g. ``0.75``), each numeric column is set
+            to NaN in bins where it has fewer non-null values than this fraction of
+            the expected sample count. Only allowed with ``flag_aware=False``; with
+            ``flag_aware=True`` it must be None, or a ValueError is raised.
 
     Returns:
         A new frame with ``on`` (and any ``by`` keys) as columns.
     """
+    if flag_aware and completeness_threshold is not None:
+        error = ValueError(
+            "completeness_threshold only works with flag_aware=False; "
+            "pass completeness_threshold=None when flag_aware=True"
+        )
+        logger.error(error)
+        raise error
+
     if not is_datetime64_any_dtype(df[on]):
         df[on] = pd.to_datetime(df[on])
 
-    keys = [by] if isinstance(by, str) else list(by or [])
+    keys = [by] if isinstance(by, str) else list(by or []) # Group-by keys as a list ("sn" --> ["sn"], None --> [])
     force_nonnumeric_set = (
         {force_nonnumeric}
         if isinstance(force_nonnumeric, str)
@@ -245,12 +249,12 @@ def resample_dataframe(
         if do_wind:
             derived = {ws_col, wd_col}
     
-    value_cols = [
+    value_cols = [ # columns to aggregate: everything except the timestamp, flag, group keys and wind speed/direction (rebuilt from u/v at the end)
         c for c in df.columns if c != on and c != 'flag' and c not in keys and c not in derived
     ]
 
-    # hack for naming collision issue with "first" and "last" agg methods
-    # (TypeError: NDFrame.first() missing 1 required positional argument: 'offset')
+    # numeric cols use numeric_how. other columns and those in force_nonnumeric, take the first (or last) value in the bin
+    # "first" and "last" are lambdas because passing those strings to resample().agg() raises a TypeError
     agg = {
         c: (
             numeric_how if is_numeric_dtype(df[c]) and c not in force_nonnumeric_set
@@ -260,7 +264,7 @@ def resample_dataframe(
         )
         for c in value_cols
     }
-    indexed = df.set_index(on)
+    indexed = df.set_index(on) # resample() needs the timestamps as the index
     if flag_aware:
         out = _flag_aware_resample(indexed, rule, keys, agg)
     else:
@@ -269,62 +273,34 @@ def resample_dataframe(
         else:
             out = indexed.resample(rule).agg(agg).reset_index()
 
+    # Completeness check (flag_aware=False only): for each numeric column, blank the bins where it
+    # has fewer non-null values than completeness_threshold x the expected number of samples
     if completeness_threshold is not None:
         merge_on = keys + [on]
         numeric_value_cols = [c for c in value_cols if is_numeric_dtype(df[c])]
 
-        if flag_aware:
-            # require >= completeness_threshold fraction of the expected
-            # sample count to be clean (flag == 0 or 4096)
-            is_clean = indexed[keys].assign(_clean_count=indexed["flag"].isin(CLEAN_FLAGS).astype("int64"))
-            if keys:
-                clean_counts = is_clean.groupby(keys).resample(rule)["_clean_count"].sum().reset_index()
-            else:
-                clean_counts = is_clean["_clean_count"].resample(rule).sum().reset_index()
-
-            merged = out[merge_on].merge(clean_counts, on=merge_on, how="left")
-            if keys:
-                min_count_by_group = (
-                    df.groupby(keys)[on]
-                    .apply(lambda s: _min_required_count(s, rule, completeness_threshold))
-                    .rename("_min_count")
-                )
-                merged = merged.merge(min_count_by_group, on=keys, how="left")
-            else:
-                merged["_min_count"] = _min_required_count(
-                    df[on], rule, completeness_threshold
-                )
-            fail_mask = merged["_clean_count"] < merged["_min_count"]
-
-            for c in numeric_value_cols:
-                if c not in out.columns:
-                    continue
-                out.loc[fail_mask.to_numpy(), c] = np.nan
+        if keys:
+            raw_counts = (
+                indexed.groupby(keys).resample(rule)[numeric_value_cols].count().reset_index()
+            )
+            min_count_by_group = (
+                df.groupby(keys)[on]
+                .apply(lambda s: _min_required_count(s, rule, completeness_threshold))
+                .rename("_min_count")
+            )
+            raw_counts = raw_counts.merge(min_count_by_group, on=keys, how="left")
         else:
-            # require >= completeness_threshold fraction of the expected
-            # sample count to be non-null, for each numeric value column
-            if keys:
-                raw_counts = (
-                    indexed.groupby(keys).resample(rule)[numeric_value_cols].count().reset_index()
-                )
-                min_count_by_group = (
-                    df.groupby(keys)[on]
-                    .apply(lambda s: _min_required_count(s, rule, completeness_threshold))
-                    .rename("_min_count")
-                )
-                raw_counts = raw_counts.merge(min_count_by_group, on=keys, how="left")
-            else:
-                raw_counts = indexed[numeric_value_cols].resample(rule).count().reset_index()
-                raw_counts["_min_count"] = _min_required_count(
-                    df[on], rule, completeness_threshold
-                )
+            raw_counts = indexed[numeric_value_cols].resample(rule).count().reset_index()
+            raw_counts["_min_count"] = _min_required_count(
+                df[on], rule, completeness_threshold
+            )
 
-            merged_counts = out[merge_on].merge(raw_counts, on=merge_on, how="left")
-            for c in numeric_value_cols:
-                if c not in out.columns:
-                    continue
-                short = merged_counts[c] < merged_counts["_min_count"]
-                out.loc[short.to_numpy(), c] = np.nan
+        merged_counts = out[merge_on].merge(raw_counts, on=merge_on, how="left")
+        for c in numeric_value_cols:
+            if c not in out.columns:
+                continue
+            short = merged_counts[c] < merged_counts["_min_count"]
+            out.loc[short.to_numpy(), c] = np.nan
 
         logger.debug(
             "Applied {:.0%} completeness threshold to {} numeric value column(s)",
@@ -332,6 +308,7 @@ def resample_dataframe(
         )
 
     if do_wind:
+        # Rebuild wind speed/direction from the averaged u/v
         out = _vector_wind(out, *wind)
     elif wind is not None:
         logger.trace("wind columns {} absent; skipping vector average", wind)
@@ -346,4 +323,4 @@ def resample_dataframe(
         "Resampled {} -> {} rows at '{}'{}",
         len(df), len(out), rule, f" grouped by {keys}" if keys else "",
     )
-    return out
+    return out
\ No newline at end of file