diff --git a/pyproject.toml b/pyproject.toml index df524be..be73d95 100644 --- a/pyproject.toml +++ b/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 "] license = "Apache-2.0" diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py index 7db7038..9bf0e51 100644 --- a/quantaq_cli/schema.py +++ b/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 diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py index 9f5a6b8..85cd581 100644 --- a/quantaq_cli/toolkit/resample.py +++ b/quantaq_cli/toolkit/resample.py @@ -51,61 +51,52 @@ 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 _flag_aware_resample(df, rule, keys, agg): + """Vectorized flag-aware resampling. - 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. + 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: - group (pd.DataFrame): All rows falling in one resample bin + 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. - - Returns: - pd.Series: One aggregated row for this bin. + resample_dataframe. """ - if 'flag' not in group.columns: - error = ValueError("No 'flag' column found in dataframe! Cannot implement" - "flag-aware resampling. Consider calling flag_dataframe() first.") + + 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, @@ -201,26 +192,21 @@ 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]) + 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() diff --git a/quantaq_cli/utilities.py b/quantaq_cli/utilities.py index 6f5536d..438d16c 100644 --- a/quantaq_cli/utilities.py +++ b/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") diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py index 7d92b96..aac610f 100644 --- a/quantaq_cli/variables.py +++ b/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=">",