Bug fix: update infer_data_source with a tdiff tolerance, and update resample_dataframe

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "quantaq-cli"
-version = "1.0.0rc3"
+version = "1.0.0rc4"
 description = "The QuantAQ Python Toolkit and CLI"
 authors = ["David H Hagan <david.hagan@quant-aq.com>"]
 license = "Apache-2.0"

quantaq_cli/schema.py

@@ -110,12 +110,12 @@
     ('vbat', np.float64),
 
     # --- Device / metadata columns ---
-    ('fw', np.int64),
+    ('fw', np.float64), # needs to be nullable for older data
     ('flag', np.int64),
     ('connection_status', np.int16),
     ('iteration', np.int16),
-    ('dd_measurement_state', np.int64),
-    ('dd_operating_state', np.int64),
+    ('dd_measurement_state', np.float64), # needs to be nullable for older data
+    ('dd_operating_state', np.float64), # needs to be nullable for older data
 ]
 
 STATIC_COLUMN_RENAMES = {
@@ -126,6 +126,8 @@
     "opcn3_pm1": "opc_pm1",
     "opcn3_pm25": "opc_pm25",
     "opcn3_pm10": "opc_pm10",
+    "opcn3_temp": "opc_temp",
+    "opcn3_rh": "opc_rh",
     "sample_period": "opc_sample_period",
     "sample_flow": "opc_sample_flow",
     "laser_status": "opc_laser_status",
@@ -151,6 +153,9 @@
     # --- RHT columns ---
     "temp": "sample_temp",
     "rh": "sample_rh",
+
+    # --- Device / metadata columns ---
+    "operating_state": "dd_operating_state"
 }
 
 # Prefixes for unstandardized column names

quantaq_cli/toolkit/resample.py

@@ -51,61 +51,75 @@ def _components_from_polar(
     df[v_col] = df[ws_col] * np.cos(wd_rad)
     return df
 
-def _aggregate_group(group, agg):
-    """Aggregate a single resample bin using flag-aware row selection.
+def _min_required_count(timestamps, rule, threshold):
+    """The minimum number of non-null samples needed in one resampling bin to 
+    meet the data completeness threshold.
 
-    This attempts to mirror the averaging logic in the firmware:
-    If the bin has at least one good row (flag == 0), 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 instead, and the resulting flag is the
-    bitwise OR of every flag value that was present in the bin.
+    We infer the native sampling interval from the median tdiff (spacing between
+    consecutive timestamps in ``timestamps``), then compared to the resampling
+    bin width to estimate how many native-resolution rows should fall in
+    one bin. E.g. 1-min data resampled to "1hr" expects ~60 rows/bin;
+    at ``threshold=0.75`` a bin needs >= 45 non-null rows to survive.
 
     Args:
-        group (pd.DataFrame): All rows falling in one resample bin
-        agg (dict): Mapping of column name -> aggregation method, defined inside
-            resample_dataframe.
+        timestamps: The timestamp series (pre-resampling)
+        rule: The pandas offset alias passed to ``resample_dataframe``.
+        threshold: The required fraction of expected samples, e.g. ``0.75``.
 
-    Returns:
-        pd.Series: One aggregated row for this bin.
     """
-    if 'flag' not in group.columns: 
-        error = ValueError("No 'flag' column found in dataframe! Cannot implement"
-        "flag-aware resampling. Consider calling flag_dataframe() first.")
+    bin_width = pd.Timedelta(pd.tseries.frequencies.to_offset(rule))
+    median_tdiff = pd.to_datetime(timestamps).sort_values().diff().median()
+    if pd.isna(median_tdiff) or median_tdiff <= pd.Timedelta(0):
+        return 0
+    expected_samples = bin_width / median_tdiff
+    return int(np.ceil(threshold * expected_samples))
+
+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
+    "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.
+
+    Args:
+        df (pd.DataFrame): the input dataframe to resample
+        rule: Any pandas offset alias, e.g. ``"1min"``, ``"1h"``, ``"1D"``.
+        keys (list[str]): Column(s) to group by before resampling (e.g.
+            ``["sn"]``), so each device/location is resampled independently.
+            Pass an empty list to resample the whole frame as one series
+            of bins.
+        agg (dict): Mapping of column name -> aggregation method, defined inside
+                    resample_dataframe.
+    """
+
+    if "flag" not in df.columns:
+        error = ValueError(
+            "No 'flag' column found in dataframe! Cannot implement "
+            "flag-aware resampling. Consider calling flag_dataframe() first."
+        )
         logger.error(error)
         raise error
 
-    # resample can produce empty bins so we check len(group)
-    if len(group): 
-        clean_mask = group['flag'] == 0 # true for every good row
+    def _resampler(frame):
+        return frame.groupby(keys).resample(rule) if keys else frame.resample(rule)
 
-        # there's at least 1 good row --> aggregate only the good rows, and set the new flag to 0
-        if clean_mask.any():
-            subset = group.loc[clean_mask]
-            group_flag = 0 
+    clean_agg = _resampler(df[df["flag"] == 0]).agg(agg)
 
-        # there are no good rows --> aggregate all rows, and combine flags using bitwise OR
-        else:
-            subset = group 
-            group_flag = int(np.bitwise_or.reduce(group['flag'].to_numpy()))
-    else:
-        # empty bin (produced by resampe) --> all columns will be nan, include the flag
-        subset = group
-        group_flag = np.nan
+    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()))
+    flag_or = flag_col.agg(lambda s: int(np.bitwise_or.reduce(s.to_numpy())) if len(s) else np.nan)
 
-    # hack for naming collision issue with "first" and "last" agg methods
-    # (TypeError: NDFrame.first() missing 1 required positional argument: 'offset')
-    values = {}
-    for col, how in agg.items():
-        if how == "first":
-            values[col] = subset[col].iloc[0] if len(subset) else np.nan
-        elif how == "last":
-            values[col] = subset[col].iloc[-1] if len(subset) else np.nan
-        else:
-            values[col] = subset[col].agg(how)
+    clean_agg = clean_agg.reindex(all_agg.index)
+    has_clean = has_clean.reindex(all_agg.index)
+    flag_or = flag_or.reindex(all_agg.index)
 
-    values['flag'] = group_flag
+    out = clean_agg.where(has_clean, all_agg)
+    out["flag"] = np.where(has_clean, 0, flag_or)
 
-    return pd.Series(values)
+    return out.reset_index()
 
 def resample_dataframe(
     df: pd.DataFrame,
@@ -117,6 +131,8 @@ def resample_dataframe(
     flag_aware: bool = False,
     numeric_how: str = "mean",
     nonnumeric_how: str = "first",
+    force_nonnumeric: str | list[str] | None = None,
+    completeness_threshold: float | None = None,
 ) -> pd.DataFrame:
     """Resample a time-indexed frame, handling mixed dtypes safely.
 
@@ -153,6 +169,17 @@ def resample_dataframe(
             output.
         numeric_how: Aggregation for numeric columns.
         nonnumeric_how: Aggregation for non-numeric columns.
+        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)
+              - If ``flag_aware=False``: the bin needs this fraction of its expected 
+                sample count to be non-null
 
     Returns:
         A new frame with ``on`` (and any ``by`` keys) as columns.
@@ -161,6 +188,11 @@ def resample_dataframe(
         df[on] = pd.to_datetime(df[on])
 
     keys = [by] if isinstance(by, str) else list(by or [])
+    force_nonnumeric_set = (
+        {force_nonnumeric}
+        if isinstance(force_nonnumeric, str)
+        else set(force_nonnumeric or [])
+    )
 
     # When vector-averaging wind, the speed/direction columns must never be
     # scalar-aggregated -- drop them from the agg pass and derive them from the
@@ -201,32 +233,94 @@ def resample_dataframe(
     value_cols = [
         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')
     agg = {
-        c: (numeric_how if is_numeric_dtype(df[c]) else nonnumeric_how)
+        c: (
+            numeric_how if is_numeric_dtype(df[c]) and c not in force_nonnumeric_set
+            else (lambda s: s.iloc[0] if len(s) else np.nan) if nonnumeric_how == "first"
+            else (lambda s: s.iloc[-1] if len(s) else np.nan) if nonnumeric_how == "last"
+            else nonnumeric_how
+        )
         for c in value_cols
     }
-
     indexed = df.set_index(on)
     if flag_aware:
-        if keys:
-            out = (
-                indexed.groupby(keys)
-                .resample(rule)
-                .apply(_aggregate_group, agg=agg)
-                .reset_index()
-            )
-        else:
-            out = (
-                indexed.resample(rule)
-                .apply(_aggregate_group, agg=agg)
-                .reset_index()
-            )
+        out = _flag_aware_resample(indexed, rule, keys, agg)
     else:
         if keys:
             out = indexed.groupby(keys).resample(rule).agg(agg).reset_index()
         else:
             out = indexed.resample(rule).agg(agg).reset_index()
 
+    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)
+            clean_indexed = indexed[indexed["flag"] == 0]
+            if keys:
+                clean_counts = (
+                    clean_indexed.groupby(keys).resample(rule)["flag"]
+                    .count().rename("_clean_count").reset_index()
+                )
+                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")
+            else:
+                clean_counts = (
+                    clean_indexed["flag"].resample(rule)
+                    .count().rename("_clean_count").reset_index()
+                )
+                clean_counts["_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:
+                if c not in out.columns:
+                    continue
+                out.loc[fail_mask.to_numpy(), c] = np.nan
+        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
+                )
+
+            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)",
+            completeness_threshold, len(numeric_value_cols),
+        )
+
     if do_wind:
         out = _vector_wind(out, *wind)
     elif wind is not None:

quantaq_cli/utilities.py

@@ -42,9 +42,10 @@ def infer_data_source(df, tscol=None):
 
     dominant_tdiff = tdiff_counts.idxmax()
 
-    mostly_1min = dominant_tdiff == 60.0
-    mostly_5sec = dominant_tdiff == 5.0
-    mostly_10sec = dominant_tdiff == 10.0
+    tolerance = 2.0
+    mostly_1min = abs(dominant_tdiff - 60.0) <= tolerance
+    mostly_5sec = abs(dominant_tdiff - 5.0) <= tolerance
+    mostly_10sec = abs(dominant_tdiff - 10.0) <= tolerance
 
     if mostly_1min:
         logger.info(f"Reading mostly {dominant_tdiff}s data --> inferring database")

quantaq_cli/variables.py

@@ -155,7 +155,7 @@
         # Check 2: ratio between the OPC and nephelometer is within spec
         Multiple(
             criteria=(
-                Single(column="opc_bin0", op=">=", value=10.0),
+                Single(column="neph_bin0", op=">=", value=10.0),
                 Ratio(column_numerator="neph_bin0", 
                       column_denominator="opc_bin0",
                       op=">", 
@@ -183,7 +183,7 @@
         # Check 5: ratio between the OPC and nephelometer is within spec
         Multiple(
             criteria=(
-                Single(column="opc_bin0", op=">=", value=10.0),
+                Single(column="neph_bin0", op=">=", value=10.0),
                 Ratio(column_numerator="neph_bin0", 
                       column_denominator="opc_bin0",
                       op=">",