Update flagging for SD Card data and implement flag-aware resampling

quantaq_cli/cli.py

@@ -1,6 +1,6 @@
 from pathlib import Path
-
 import pkg_resources
+
 import rich_click as click
 from loguru import logger
 
@@ -10,7 +10,6 @@
 from quantaq_cli.log import configure_logging, LOG_LEVELS
 from quantaq_cli.toolkit.resample import WIND_COLUMNS
 from quantaq_cli.utilities import safe_load
-from quantaq_cli.variables import SUPPORTED_MODELS, SUPPORTED_SOURCES
 
 
 CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
@@ -112,6 +111,18 @@ def merge_command(files, output, tscol, log_level):
     default=("wx_u", "wx_v", "wx_ws", "wx_wd"),
     help="(u, v, speed, direction) column names. The u/v columns don't need to be present in the dataframe, but the column names need to be specified.",
 )
+@click.option(
+    "--flag-aware",
+    is_flag=True,
+    default=False,
+    help=(
+        "Whether to apply flag-aware row selection when resampling: bins "
+        "with >1 good rows --> average only those rows; bins with no good "
+        "rows -> average all rows and OR their flag values "
+        "together. If False (default), all rows are averaged regardless "
+        "of flags and the flag column is dropped."
+    ),
+)
 @click.option("--numeric_how", default="mean", help="Aggregation for numeric columns.")
 @click.option("--nonnumeric_how", default="first", help="Aggregation for non-numeric columns.")
 @click.option("--log-level", default="INFO",
@@ -124,6 +135,7 @@ def resample_command(file, rule, output, log_level, **kwargs):
     on = kwargs.pop("on", "timestamp")
     by = kwargs.pop("by", None)
     wind = kwargs.pop("wind", WIND_COLUMNS)
+    flag_aware = kwargs.pop("flag_aware", False)
     numeric_how = kwargs.pop("numeric_how", "mean")
     nonnumeric_how = kwargs.pop("nonnumeric_how", "first")
 
@@ -148,6 +160,7 @@ def resample_command(file, rule, output, log_level, **kwargs):
         on=on,
         by=by,
         wind=wind,
+        flag_aware=flag_aware,
         numeric_how=numeric_how,
         nonnumeric_how=nonnumeric_how,
     )

quantaq_cli/schema.py

@@ -56,20 +56,23 @@
     ('bin3MToF', np.float64),
     ('bin5MToF', np.float64),
     ('bin7MToF', np.float64),
-    ('sample_period', np.float64),
-    ('opc_sample_period', np.float64),
-    ('sample_flow', np.float64),
-    ('opc_sample_flow', np.float64),
     ('opc_temp', np.float64),
     ('opc_rh', np.float64),
     ('opc_pm1', np.float64),
     ('opc_pm25', np.float64),
     ('opc_pm10', np.float64),
     ('opcn3_pm1', np.float64),
-    ('opcn3_pm25', np.float64),
     ('opcn3_pm10', np.float64),
-    ('laser_status', np.int16),
-    ('opc_laser_status', np.int16),
+    ('opcn3_pm25', np.float64),
+    ('opc.pm1', np.float64),
+    ('opc.pm10', np.float64),
+    ('opc.pm25', np.float64),
+    ('sample_period', np.float64),
+    ('opc_sample_period', np.float64),
+    ('sample_flow', np.float64),
+    ('opc_sample_flow', np.float64),
+    ('laser_status', np.int64),
+    ('opc_laser_status', np.int64),
 
     # --- Nephelometer columns ---
     ('pm1_std', np.float64),
@@ -90,6 +93,10 @@
     ('neph_bin3', np.float64),
     ('neph_bin4', np.float64),
     ('neph_bin5', np.float64),
+    ('neph.bin0', np.float64),
+    ('neph.pm1', np.float64),
+    ('neph.pm10', np.float64),
+    ('neph.pm25', np.float64), 
 
     # --- RH / Temp / Pressure columns ---
     ('sample_rh', np.float64),
@@ -159,10 +166,12 @@
     ('vbat', np.float64),
 
     # --- Device / metadata columns ---
-    ('fw', np.int16),
+    ('fw', np.int64),
     ('flag', np.int64),
     ('connection_status', np.int16),
     ('iteration', np.int16),
+    ('dd_measurement_state', np.int64),
+    ('dd_operating_state', np.int64),
 ]
 
 def build_dtype_schema(column_definitions, nullable=True, required=False):

quantaq_cli/toolkit/expunge.py

@@ -1,5 +1,5 @@
-import numpy as np
 from loguru import logger
+import numpy as np
 
 from quantaq_cli.utilities import determine_timestamp_column
 from quantaq_cli.variables import FLAG_DEFINITIONS
@@ -24,7 +24,7 @@ def expunge_dataframe(df):
 
     for label, value, cols in list_of_flags:
         mask = df["flag"] & value == value
-        if not mask.any():
+        if not mask.any() or cols is None:
             continue
 
         # NaN the necessary columns

quantaq_cli/toolkit/flag.py

@@ -1,18 +1,107 @@
-from pathlib import Path
-
 from loguru import logger
 import pandas as pd
 import rich
 from rich.table import Table
 
-from quantaq_cli.variables import FLAG_DEFINITIONS, SUPPORTED_MODELS, SUPPORTED_SOURCES
-from quantaq_cli.variables import Range, Gap, flag_name_to_criteria
+from quantaq_cli.variables import FLAG_DEFINITIONS, FLAG_VALUES
+from quantaq_cli.variables import Range, Gap, Single, Multiple, Ratio, OPS, flag_name_to_criteria
 from quantaq_cli.utilities import fix_timestamps, determine_timestamp_column
 from quantaq_cli.utilities import infer_data_source, infer_data_model
 
+def evaluate_criterion(df, criterion):
+    """Create a mask that is True for every row meeting the flag criterion.
+
+    Args:
+        df (pd.DataFrame): DataFrame to mask.
+        criterion: The rule used to decide which rows should be flagged. 
+
+    Returns:
+        pd.Series: Boolean mask - True for rows that meet the criterion.
+    """
+
+    if isinstance(criterion, Range):
+        if criterion.column not in df.columns:
+            return pd.Series(False, index=df.index)
+        col = df[criterion.column]
+        mask = (col < criterion.lo) | (col > criterion.hi)
+        return mask
+
+    elif isinstance(criterion, Single):
+        if criterion.column not in df.columns:
+            return pd.Series(False, index=df.index)
+        col = df[criterion.column]
+        mask = OPS[criterion.op](col, criterion.value)
+        return mask
+
+    elif isinstance(criterion, Ratio):
+        if (
+            criterion.column_denominator not in df.columns
+            or criterion.column_numerator not in df.columns
+        ):
+            return pd.Series(False, index=df.index)
+        ratio = df[criterion.column_numerator] / df[criterion.column_denominator]
+        mask = OPS[criterion.op](ratio, criterion.value)
+        return mask
+
+    elif isinstance(criterion, Gap):
+        # to do: check if the dataframe is sorted here (this should be done before calling
+        # _evaluate_criterion() -- we don't want to sort the df in this block unless
+        # we also reindex the mask on the original df, so that we can combine masks
+
+        # this block might need to live somewhere else, but basically, we don't bother
+        # to set FLAG_STARTUP for devices without gas measurements since only the 
+        # gas columns get nan'd when FLAG_STARTUP is set --- we don't have a 
+        # startup flag for other devices yet so this is mostly to avoid confusion
+        model = infer_data_model(df)
+        if model not in {"modulair", "modulair-x"}:
+            return pd.Series(False, index=df.index)
+
+        tscol = determine_timestamp_column(df)
+        tdiff = df[tscol].diff().dt.total_seconds()
+
+        if criterion.warmup_min_lipo_soc is not None:
+            # if there's a warmup_min_lipo_soc criterion (as in RAWSD_CRITERIA),
+            # the gap_starts are defined if the device was offline for the past hour 
+            # AND if the LIPO SOC is < warmup_min_lipo_soc (10%)
+            gap_starts_mask = (tdiff > criterion.gap_in_seconds) & (df['soc'] < criterion.warmup_min_lipo_soc)
+            gap_starts = df.loc[gap_starts_mask, tscol]
+        else:
+            # if there's no warmup_min_lipo_soc criterion (as in DATABASE_CRITERIA),
+            # the gap_starts are defined only if the device was offine for the past hour
+            gap_starts = df.loc[tdiff > criterion.gap_in_seconds, tscol]
+
+        # set mask to True for every row that falls within post_gap_flag_length_seconds 
+        # after any detected timestamp gap larger than gap_in_seconds
+        postgap_delta = pd.Timedelta(seconds=criterion.post_gap_flag_length_seconds)
+        mask = pd.Series(False, index=df.index)
+        for ts in gap_starts: 
+            mask |= (df[tscol] >= ts) & (df[tscol] <= ts + postgap_delta)
+        return mask
+
+    elif isinstance(criterion, Multiple):
+        masks = [evaluate_criterion(df, c) for c in criterion.criteria]
+        if criterion.logical_operator == "AND":
+            # start with first mask and progressively AND the remaining ones
+            mask = masks[0]
+            for m in masks[1:]:
+                mask = mask & m
+            return mask
+        else:
+            # not bothering to implement logical_operator = "OR" because that's the
+            # default behavior for items in the top-level criteria lists. 
+            error = NotImplementedError(
+            f"Unsupported logical_operator: {criterion.logical_operator!r} "
+            f"(only 'AND' is implemented, use a regular criteria list for OR)"
+            )
+            logger.error(error)
+            raise error
+    else:
+        error = NotImplementedError(f"Unsupported criterion type: {type(criterion)}")
+        logger.error(error)
+        raise error
 
-def _add_flag(df, flag_name, flag_value, criterion):
-    """Add a specific flag to a DataFrame based on a given criterion.
+def add_flag(df, mask, flag_name, flag_value):
+    """Set the 'flag' column to the flag bitmask value if the mask evaluates to true.
 
     For each row that meets the specified criterion, this function modifies the 'flag' 
     column by applying a bitwise OR operation with the given flag value.
@@ -21,47 +110,17 @@ def _add_flag(df, flag_name, flag_value, criterion):
         df (pd.DataFrame): DataFrame to which the flag will be added.
         flag_name (str): Name of the flag to be added.
         flag_value (int): Bitmask value of the flag to be added.
-        criterion (Range or Gap): Logic based on which the flag will be added.
+        mask (pd.Series): True where the flag criterion are true.
 
     Returns:
         pd.DataFrame: DataFrame with the flag added.
     """
-    if isinstance(criterion, Range):
-        if criterion.column in df.columns:
-            col = df[criterion.column]
-            mask = (col < criterion.lo) | (col > criterion.hi)
-            if not mask.sum():
-                return df
-            df.loc[mask, "flag"] |= flag_value
-            logger.info(
-                f"Flagged: {flag_name} (flag {flag_value}) --> {mask.sum()} rows",
-            )
-
-    elif isinstance(criterion, Gap):
-        # sort based on a time column
-        df = fix_timestamps(df, sort_values=True)
-        tscol = determine_timestamp_column(df)
-
-        # Create a column to hold the time diff
-        df["tdiff"] = df[tscol].diff().dt.total_seconds()
-
-        # If we're missing enough data, apply the startup flag.
-        startup_mask = df["tdiff"] > criterion.gap_in_seconds
-        starts = df.loc[startup_mask]
-        postgap_delta = pd.Timedelta(seconds=criterion.post_gap_flag_length_seconds)
-
-        for _, row in starts.iterrows():
-            # Flag everything between the startup flag's timestamp up to the gap.
-            mask = (df[tscol] >= row[tscol]) & (
-                df[tscol] <= (row[tscol] + postgap_delta)
-            )
-            df.loc[mask, "flag"] |= flag_value  
-            logger.info(
-                f"Flagged: {flag_name} (flag {flag_value}) --> {mask.sum()} rows",
-            )
-
-        # Delete the tdiff col
-        del df["tdiff"]
+    if not mask.any(): # skip if every value in the mask is False
+        return df
+    df.loc[mask, "flag"] |= flag_value  
+    logger.info(
+        f"Flagged: {flag_name} (flag {flag_value}) --> {mask.sum()} rows",
+    )
 
     return df
 
@@ -83,7 +142,7 @@ def flag_summary(df):
     df["flag"] = df["flag"].astype(int, errors='ignore')
 
     rows = []
-    for name, value, cols in FLAG_DEFINITIONS:
+    for name, value, _ in FLAG_DEFINITIONS:
         mask = df["flag"] & value
         num_naffected = mask.astype(bool).sum()
         percent_affected = f"{100 * num_naffected / df.shape[0]:.1f}"
@@ -126,22 +185,22 @@ def flag_dataframe(df):
     """
     df = df.copy()
 
-    model = infer_data_model(df)
     source = infer_data_source(df)
 
-    # get flag criteria (this also checks if model and data source is valid)
-    name_to_criteria = flag_name_to_criteria(source, model).items()
+    # 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"] = 0
 
-    # get the flag values for each flag name
-    FLAG_VALUES = {flag.name: flag.value for flag in FLAG_DEFINITIONS}
+    # sort the dataframe once before adding flags
+    df = fix_timestamps(df, sort_values=True)
 
     # set the flag for each flag_name and their respective crtieria 
     for flag_name, criteria in name_to_criteria:
         flag_value = FLAG_VALUES[flag_name]                      
-        for criterion in criteria:              
-            df = _add_flag(df, flag_name, flag_value, criterion)
+        for criterion in criteria:
+            mask =  evaluate_criterion(df, criterion)             
+            df = add_flag(df, mask, flag_name, flag_value)
     return df

quantaq_cli/toolkit/munge.py

@@ -1,9 +1,5 @@
-import numpy as np
-from pathlib import Path
-import pandas as pd
 from loguru import logger
 
-from quantaq_cli.exceptions import InvalidFileExtension
 from quantaq_cli.utilities import fix_timestamps, drop_unnamed