From 6c74da7dda6ac0429e856f0173101b8678a4f5c5 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Fri, 26 Jun 2026 10:27:56 -0400 Subject: [PATCH 01/23] Use namedtuple for FLAG and FLAG_CRITERIA --- quantaq_cli/variables.py | 153 ++++++++++++++++++++++++++++----------- 1 file changed, 110 insertions(+), 43 deletions(-) diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py index ff9a5eb..4f1be37 100644 --- a/quantaq_cli/variables.py +++ b/quantaq_cli/variables.py @@ -1,56 +1,123 @@ +############################################################################### + +"""variables.py + +Per-model data-quality flags. + +Each flag has a name, an integer bitmask value, and the set of columns to set +to NaN when the flag is raised: + + * a list of column names -> NaN just those columns + * an empty list -> flag carries no column-level action + * None -> NaN the entire row """ -Flags are in format (name, value, columns to NaN) -""" -FLAGS = dict() +from collections import namedtuple + +Flag = namedtuple("Flag", ["name", "value", "nan_columns"]) + +# Shared column groups, factored out so the per-model tables stay readable and +# the lists can't silently drift out of sync. +_OPC_COLUMNS_V1 = ["bin0", "bin1", "bin2", "bin3", "bin4", "bin5", "opc_flow"] + +_OPC_COLUMNS = [ + "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", + "sample_period", "sample_flow", "opc_temp", "opc_rh", "opc_pm1", "opc_pm25", "opc_pm10", + "laser_status", +] + +_NEPH_COLUMNS = [ + "pm1_std", "pm25_std", "pm10_std", "pm1_env", "pm25_env", "pm10_env", + "neph_bin0", "neph_bin1", "neph_bin2", "neph_bin3", "neph_bin4", "neph_bin5", +] + +FLAGS = {} FLAGS["v100"] = [ - ("FLAG_STARTUP", 1, []), - ("FLAG_OPC", 2, ["bin0", "bin1", "bin2", "bin3", "bin4", "bin5", "opc_flow"]), - ("FLAG_TOTAL_COUNTS", 4, ["voc_raw", "pressure", "temp_manifold", "rh_manifold", "temp_box", - "dew_point", "noise", "solar", "wind_dir", "wind_speed", - "sample_time", "opc_flow", "manifold_temp", "manifold_rh"]), - ("FLAG_CO", 8, ["co_we", "co_ae"]), - ("FLAG_NO", 16, ["no_we", "no_ae"]), - ("FLAG_NO2", 32, ["no2_we", "no2_ae"]), - ("FLAG_O3", 64, ["o3_we", "o3_ae"]), - ("FLAG_OPC_RECORD_NUM", 128, ["bin0", "bin1", "bin2", "bin3", "bin4", "bin5", "opc_flow"]), - ("FLAG_CO2", 256 , ["co2_raw"]), - # ("FLAG_PP", 512, ), - ("FLAG_ROW", 1024, None) + Flag("FLAG_STARTUP", 1, []), + Flag("FLAG_OPC", 2, _OPC_COLUMNS_V1), + Flag("FLAG_TOTAL_COUNTS", 4, [ + "voc_raw", "pressure", "temp_manifold", "rh_manifold", "temp_box", + "dew_point", "noise", "solar", "wind_dir", "wind_speed", + "sample_time", "opc_flow", "manifold_temp", "manifold_rh", + ]), + Flag("FLAG_CO", 8, ["co_we", "co_ae"]), + Flag("FLAG_NO", 16, ["no_we", "no_ae"]), + Flag("FLAG_NO2", 32, ["no2_we", "no2_ae"]), + Flag("FLAG_O3", 64, ["o3_we", "o3_ae"]), + Flag("FLAG_OPC_RECORD_NUM", 128, _OPC_COLUMNS_V1), + Flag("FLAG_CO2", 256, ["co2_raw"]), + # Flag("FLAG_PP", 512, [...]), + Flag("FLAG_ROW", 1024, None), ] -# flags for the v200 are the same as the v100 +# v200 flags are identical to the v100 flags. FLAGS["v200"] = FLAGS["v100"] FLAGS["modulair_pm"] = [ - ("FLAG_STARTUP", 1, []), - ("FLAG_OPC", 2, ["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", - "sample_period", "sample_flow", "opc_temp", "opc_rh", "opc_pm1", "opc_pm25", "opc_pm10", "laser_status"]), - ("FLAG_NEPH", 4, ["pm1_std", "pm25_std", "pm10_std", "pm1_env", "pm25_env", "pm10_env", - "neph_bin0", "neph_bin1", "neph_bin2", "neph_bin3", "neph_bin4", "neph_bin5"]), - ("FLAG_RHTP", 8, ["sample_rh", "sample_temp", "sample_pres"]), - ("FLAG_ROW", 1024, None) + Flag("FLAG_STARTUP", 1, []), + Flag("FLAG_OPC", 2, _OPC_COLUMNS), + Flag("FLAG_NEPH", 4, _NEPH_COLUMNS), + Flag("FLAG_RHTP", 8, ["sample_rh", "sample_temp", "sample_pres"]), + Flag("FLAG_ROW", 1024, None), ] FLAGS["modulair"] = [ - ("FLAG_STARTUP", 1, []), - ("FLAG_OPC", 2, ["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", - "sample_period", "sample_flow", "opc_temp", "opc_rh", "opc_pm1", "opc_pm25", "opc_pm10", "laser_status"]), - ("FLAG_NEPH", 4, ["pm1_std", "pm25_std", "pm10_std", "pm1_env", "pm25_env", "pm10_env", - "neph_bin0", "neph_bin1", "neph_bin2", "neph_bin3", "neph_bin4", "neph_bin5"]), - ("FLAG_RHT", 8, ["sample_rh", "sample_temp"]), - ("FLAG_CO", 16, ["co_we", "co_ae", "co_diff"]), - ("FLAG_NO", 32, ["no_we", "no_ae", "no_diff"]), - ("FLAG_NO2", 64, ["no2_we", "no2_ae", "no2_diff"]), - ("FLAG_O3", 128, ["ox_we", "ox_ae", "ox_diff", "o3_diff"]), - ("FLAG_CO2", 256, ["co2_raw"]), - ("FLAG_SO2", 512, ["so2_we", "so2_ae", "so2_diff"]), - ("FLAG_H2S", 1024, ["h2s_we", "h2s_ae", "h2s_diff"]), - ("FLAG_BAT", 2048, ["bat_voltage", "soc", "vbat"]), + Flag("FLAG_STARTUP", 1, []), + Flag("FLAG_OPC", 2, _OPC_COLUMNS), + Flag("FLAG_NEPH", 4, _NEPH_COLUMNS), + Flag("FLAG_RHT", 8, ["sample_rh", "sample_temp"]), + Flag("FLAG_CO", 16, ["co_we", "co_ae", "co_diff"]), + Flag("FLAG_NO", 32, ["no_we", "no_ae", "no_diff"]), + Flag("FLAG_NO2", 64, ["no2_we", "no2_ae", "no2_diff"]), + Flag("FLAG_O3", 128, ["ox_we", "ox_ae", "ox_diff", "o3_diff"]), + Flag("FLAG_CO2", 256, ["co2_raw"]), + Flag("FLAG_SO2", 512, ["so2_we", "so2_ae", "so2_diff"]), + Flag("FLAG_H2S", 1024, ["h2s_we", "h2s_ae", "h2s_diff"]), + Flag("FLAG_BAT", 2048, ["bat_voltage", "soc", "vbat"]), ] -SUPPORTED_MODELS = FLAGS.keys() \ No newline at end of file +SUPPORTED_MODELS = FLAGS.keys() + +############################################################################### + +Range = namedtuple("Range", ["column", "lo", "hi"]) +Gap = namedtuple("Gap", ["gap_in_seconds", "post_gap_flag_length_seconds"]) + +FLAG_CRITERIA = { + "database": { + "default": { # for DB flags, criteria are consistent across all QuantAQ products + "FLAG_STARTUP": [Gap(60 * 60, 60 * 60)], + "FLAG_CO": [Range("co_ae", 535.0, 800.0)], + "FLAG_NO": [Range("no_ae", 640.0, 900.0)], + "FLAG_NO2": [Range("no2_ae", 1600.0, 1700.0)], + "FLAG_O3": [Range("o3_ae", 1600.0, 1700.0)], + "FLAG_CO2": [Range("co2_raw", 1000.0, 5000.0)], + "FLAG_OPC": [Range("bin0", 0.0, 1e6)], + "FLAG_RHT": [ + Range("sample_rh", 0.0, 100.0), + Range("sample_temp", -60.0, 85.0), + ], + }, + }, + "rawsd": { # for rawSD flags, criteria are NOT consistent + #"modulair": { + # "FLAG_STARTUP": [Gap(60 * 60, 60 * 60)], # placeholder + #}, + #"modulair-x": { + # "NO_FLAG_STARTUP": [Gap(60 * 60, 60 * 60)], # placeholder + #}, + }, +} + +SUPPORTED_SOURCES = FLAG_CRITERIA.keys() + + +def get_flag_criteria(source, model): + """ + source = 'rawsd' or 'database' + model = 'default', 'modulair', 'modulair-x' + """ + by_model = FLAG_CRITERIA[source] + return by_model[model] From 02606fb80458b36306d17be74bd9cb09e71470fd Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Fri, 26 Jun 2026 10:33:35 -0400 Subject: [PATCH 02/23] Replace single-filter flag command with multi-filter flagging --- quantaq_cli/console/commands/flag.py | 127 ++++++++++++++++++--------- 1 file changed, 85 insertions(+), 42 deletions(-) diff --git a/quantaq_cli/console/commands/flag.py b/quantaq_cli/console/commands/flag.py index 9b5e17f..10ec4fa 100644 --- a/quantaq_cli/console/commands/flag.py +++ b/quantaq_cli/console/commands/flag.py @@ -1,27 +1,95 @@ from pathlib import Path + +import click import pandas as pd import numpy as np -import click +from loguru import logger from terminaltables import SingleTable +from ...variables import FLAGS, get_flag_criteria, SUPPORTED_MODELS, SUPPORTED_SOURCES +from ...variables import Range, Gap +from ...utilities import determine_timestamp_column, safe_load from ...exceptions import InvalidFileExtension, InvalidArgument, InvalidDeviceModel -from ...utilities import safe_load -from ...variables import FLAGS, SUPPORTED_MODELS -import operator -ops = { - 'eq': operator.eq, - 'lt': operator.lt, - 'le': operator.le, - 'gt': operator.gt, - 'ge': operator.ge - } +def add_flag(df, flag_name, criterion, bit): + if isinstance(criterion, Range): + col = df[criterion.column] + mask = (col < criterion.lo) | (col > criterion.hi) + if not mask.sum(): + return df + df.loc[mask, "flag"] |= bit + logger.info( + f"Flagged: {flag_name} (flag {bit}) --> {mask.sum()} rows", + ) + return df + + if isinstance(criterion, Gap): + # Find best timestamp column. + tscol = determine_timestamp_column(df) + + # Force timestamp type + df[tscol] = df[tscol].map(pd.to_datetime) + + # Sort by timestamp + df = df.sort_values(tscol) + + # 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"] |= bit + logger.info( + f"Flagged: {flag_name} (flag {bit}) --> {mask.sum()} rows", + ) + + # Delete the tdiff col + del df["tdiff"] + + return df + +def flag_dataframe(df, model, source): + """ + df = pandas dataframe to be flagged (or re-flagged) + model = the sensor model (in SUPPORTED_MODELS) + source = the data source (database or rawsd, eventually cloudAPI as well) + """ + df = df.copy() + + # ensure the model is valid + if model not in SUPPORTED_MODELS: + raise InvalidDeviceModel("Invalid device model. Must be one of {}".format(SUPPORTED_MODELS)) + + # ensure the data source is valid + if source not in SUPPORTED_SOURCES: + raise NotImplementedError # add this to exceptions + + # create flag column if it doesn't exist + if "flag" not in df.columns: + df["flag"] = 0 + + # get the bit values for each flag name + values = {flag.name: flag.value for flag in FLAGS[model]} + # set the flag for each flag_name and their respective crtieria + for flag_name, criteria in get_flag_criteria(source, model).items(): + bit = values[flag_name] + for criterion in criteria: + df = add_flag(df, flag_name, criterion, bit) + return df -def flag_command(file, column, comparator, value, output, **kwargs): +def flag_command(file, output, **kwargs): verbose = kwargs.pop("verbose", False) - flag = kwargs.pop("flag", "FLAG_ROW") + source = kwargs.pop("source", "rawsd") model = kwargs.pop("model", "modulair_pm") # make sure the extension is either a csv or feather format @@ -29,42 +97,16 @@ def flag_command(file, column, comparator, value, output, **kwargs): if output.suffix not in (".csv", ".feather"): raise InvalidFileExtension("Invalid file extension") - # ensure the model is valid - if model not in SUPPORTED_MODELS: - raise InvalidDeviceModel("Invalid device model. Must be one of {}".format(SUPPORTED_MODELS)) - save_as_csv = True if output.suffix == ".csv" else False - # concat everything in filepath if verbose: click.secho("File to read: {}".format(file), fg='green') # load the file df = safe_load(file) - # is the in df.columns? - if not column in df.columns: - raise InvalidArgument("Bad column name") - - # is the comparator valid? - if not comparator in ops.keys(): - raise InvalidArgument("Bad comparator") - - # create flag column if it doesn't exist - if "flag" not in df.columns: - df["flag"] = 0 - - # get the flag value - flag_value = 0 - flag_list = FLAGS.get(model) - for label, v, _ in flag_list: - if label == flag: - flag_value = v - break - - # create a mask and set the flag accordingly - mask = ops[comparator](df[column], value) - df.loc[mask, "flag"] = df[mask]["flag"] | flag_value + # flag the dataframe + df = flag_dataframe(df, model, source) # save the file if verbose: @@ -73,4 +115,5 @@ def flag_command(file, column, comparator, value, output, **kwargs): if save_as_csv: df.to_csv(output) else: - df.reset_index().to_feather(output) \ No newline at end of file + df.reset_index().to_feather(output) + \ No newline at end of file From 28197e10008cd0c9039032a170bd90fd66676370 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Fri, 26 Jun 2026 10:34:29 -0400 Subject: [PATCH 03/23] Add util for determining the timestamp col --- quantaq_cli/utilities.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/quantaq_cli/utilities.py b/quantaq_cli/utilities.py index ee49395..1c441d1 100644 --- a/quantaq_cli/utilities.py +++ b/quantaq_cli/utilities.py @@ -30,4 +30,12 @@ def safe_load(fpath, **kwargs): if unnamed: tmp.drop(columns=unnamed, inplace=True) - return tmp \ No newline at end of file + return tmp + +def determine_timestamp_column(sensor_df: pd.DataFrame) -> str: + """Find the best column to use for timestamps.""" + for col in ("timestamp_iso", "timestamp", "timestamp_local"): + if col in sensor_df.columns: + return col + + raise ValueError(f"Couldn't find a timestamp column: {sensor_df.columns}") From 1f8f9ba2ca723d7da52a95329ea1f6f0b8cb2584 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:21:06 -0400 Subject: [PATCH 04/23] Refactor - rename bit to flag_value, fix click commands --- quantaq_cli/console/__init__.py | 26 ++++++++----------- quantaq_cli/console/commands/flag.py | 39 ++++++++++++++-------------- quantaq_cli/variables.py | 19 +++++++++++--- 3 files changed, 46 insertions(+), 38 deletions(-) diff --git a/quantaq_cli/console/__init__.py b/quantaq_cli/console/__init__.py index 3106175..6c26008 100644 --- a/quantaq_cli/console/__init__.py +++ b/quantaq_cli/console/__init__.py @@ -1,6 +1,6 @@ import pkg_resources import rich_click as click -from quantaq_cli.variables import SUPPORTED_MODELS +from quantaq_cli.variables import SUPPORTED_MODELS, SUPPORTED_SOURCES CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help']) @@ -82,25 +82,21 @@ def expunge(file, dry_run, table, output, flag, verbose, model, **kwargs): @click.command("flag", short_help="flag data based on specific criteria") @click.argument("file", nargs=1, type=click.Path()) -@click.argument("column", nargs=1, type=str) -@click.argument("comparator", nargs=1, type=str) -@click.argument("value", nargs=1, type=float) -@click.option("-f", "--flag", default="FLAG_ROW", help="One of [FLAG_OPC, FLAG_CO, FLAG_NO, FLAG_NO2, FLAG_O3, FLAG_CO2, FLAG_ROW]") +@click.argument("model", type=click.Choice(SUPPORTED_MODELS)) +@click.argument("source", type=click.Choice(SUPPORTED_SOURCES)) @click.option("-o", "--output", default="output.csv", help="The filepath where you would like to save the file", type=str) @click.option("-v", "--verbose", is_flag=True, help="Enable verbose mode (debugging)") -@click.option("-m", "--model", default="modulair_pm", help="The device model type. One of {}".format(SUPPORTED_MODELS)) -def flag(file, column, comparator, value, flag, output, verbose, model, **kwargs): - """Set a FLAG based on user input or statistical method. - - Four arguments are required: - 1. FILE -> the path to the file of interest - 2. COLUMN -> the exact name of the column - 3. COMPARATOR -> one of ['lt', 'gt', 'eq', 'le', 'ge'] - 4. VALUE -> the value by which to filter +def flag(file, output, verbose, model, source, **kwargs): + """Reflag a data file based on data model and data source. + + Three arguments are required: + 1. FILE -> the path to the file of interest + 2. MODEL -> the device model type (one of SUPPORTED_MODELS) + 3. SOURCE -> the data source (one of SUPPORTED_SOURCES) """ from .commands.flag import flag_command - flag_command(file, column, comparator, value, output, flag=flag, verbose=verbose, model=model) + flag_command(file, output, model=model, source=source, verbose=verbose) @click.command("clean") diff --git a/quantaq_cli/console/commands/flag.py b/quantaq_cli/console/commands/flag.py index 10ec4fa..953772b 100644 --- a/quantaq_cli/console/commands/flag.py +++ b/quantaq_cli/console/commands/flag.py @@ -12,19 +12,20 @@ from ...exceptions import InvalidFileExtension, InvalidArgument, InvalidDeviceModel -def add_flag(df, flag_name, criterion, bit): +def add_flag(df, flag_name, flag_value, criterion): if isinstance(criterion, Range): - col = df[criterion.column] - mask = (col < criterion.lo) | (col > criterion.hi) - if not mask.sum(): - return df - df.loc[mask, "flag"] |= bit - logger.info( - f"Flagged: {flag_name} (flag {bit}) --> {mask.sum()} rows", + 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", ) - return df + return df - if isinstance(criterion, Gap): + elif isinstance(criterion, Gap): # Find best timestamp column. tscol = determine_timestamp_column(df) @@ -47,15 +48,15 @@ def add_flag(df, flag_name, criterion, bit): mask = (df[tscol] >= row[tscol]) & ( df[tscol] <= (row[tscol] + postgap_delta) ) - df.loc[mask, "flag"] |= bit + df.loc[mask, "flag"] |= flag_value logger.info( - f"Flagged: {flag_name} (flag {bit}) --> {mask.sum()} rows", + f"Flagged: {flag_name} (flag {flag_value}) --> {mask.sum()} rows", ) # Delete the tdiff col del df["tdiff"] - return df + return df def flag_dataframe(df, model, source): """ @@ -77,20 +78,18 @@ def flag_dataframe(df, model, source): if "flag" not in df.columns: df["flag"] = 0 - # get the bit values for each flag name - values = {flag.name: flag.value for flag in FLAGS[model]} + # get the flag values for each flag name + flag_values = {flag.name: flag.value for flag in FLAGS[model]} # set the flag for each flag_name and their respective crtieria for flag_name, criteria in get_flag_criteria(source, model).items(): - bit = values[flag_name] + flag_value = flag_values[flag_name] for criterion in criteria: - df = add_flag(df, flag_name, criterion, bit) + df = add_flag(df, flag_name, flag_value, criterion) return df -def flag_command(file, output, **kwargs): +def flag_command(file, output, model, source, **kwargs): verbose = kwargs.pop("verbose", False) - source = kwargs.pop("source", "rawsd") - model = kwargs.pop("model", "modulair_pm") # make sure the extension is either a csv or feather format output = Path(output) diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py index 4f1be37..32b714b 100644 --- a/quantaq_cli/variables.py +++ b/quantaq_cli/variables.py @@ -32,6 +32,10 @@ "neph_bin0", "neph_bin1", "neph_bin2", "neph_bin3", "neph_bin4", "neph_bin5", ] + +# Should FLAGS also be a nested dict like FLAG_CRITERIA below +# (with keys = source, nested keys = model)? +# i.e. are flag names and values different for database vs rawSD data? FLAGS = {} FLAGS["v100"] = [ @@ -67,7 +71,7 @@ Flag("FLAG_STARTUP", 1, []), Flag("FLAG_OPC", 2, _OPC_COLUMNS), Flag("FLAG_NEPH", 4, _NEPH_COLUMNS), - Flag("FLAG_RHT", 8, ["sample_rh", "sample_temp"]), + Flag("FLAG_RHTP", 8, ["sample_rh", "sample_temp", "sample_pres"]), Flag("FLAG_CO", 16, ["co_we", "co_ae", "co_diff"]), Flag("FLAG_NO", 32, ["no_we", "no_ae", "no_diff"]), Flag("FLAG_NO2", 64, ["no2_we", "no2_ae", "no2_diff"]), @@ -78,6 +82,9 @@ Flag("FLAG_BAT", 2048, ["bat_voltage", "soc", "vbat"]), ] +# modulair-x DB flags are identical to modulair? +FLAGS["modulair-x"] = FLAGS["modulair"] + SUPPORTED_MODELS = FLAGS.keys() ############################################################################### @@ -87,7 +94,7 @@ FLAG_CRITERIA = { "database": { - "default": { # for DB flags, criteria are consistent across all QuantAQ products + "modulair": { "FLAG_STARTUP": [Gap(60 * 60, 60 * 60)], "FLAG_CO": [Range("co_ae", 535.0, 800.0)], "FLAG_NO": [Range("no_ae", 640.0, 900.0)], @@ -95,9 +102,12 @@ "FLAG_O3": [Range("o3_ae", 1600.0, 1700.0)], "FLAG_CO2": [Range("co2_raw", 1000.0, 5000.0)], "FLAG_OPC": [Range("bin0", 0.0, 1e6)], - "FLAG_RHT": [ + "FLAG_RHTP": [ Range("sample_rh", 0.0, 100.0), Range("sample_temp", -60.0, 85.0), + Range("rh", 0.0, 100.0), + Range("temp", -60.0, 85.0), + #Range("sample_pres", lo, hi), # find these values ], }, }, @@ -111,6 +121,9 @@ }, } +# modulair-x DB flag criteria are identical to modulair? +FLAG_CRITERIA["database"]["modulair-x"] = FLAG_CRITERIA["database"]["modulair"] + SUPPORTED_SOURCES = FLAG_CRITERIA.keys() From 79e26bf635ff68fe906ac41e6b88f507fcb17e28 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:46:45 -0400 Subject: [PATCH 05/23] Update tests --- quantaq_cli/console/commands/flag.py | 1 - tests/test_flag.py | 39 ++++++++++++---------------- 2 files changed, 17 insertions(+), 23 deletions(-) diff --git a/quantaq_cli/console/commands/flag.py b/quantaq_cli/console/commands/flag.py index 953772b..efa5e6a 100644 --- a/quantaq_cli/console/commands/flag.py +++ b/quantaq_cli/console/commands/flag.py @@ -23,7 +23,6 @@ def add_flag(df, flag_name, flag_value, criterion): logger.info( f"Flagged: {flag_name} (flag {flag_value}) --> {mask.sum()} rows", ) - return df elif isinstance(criterion, Gap): # Find best timestamp column. diff --git a/tests/test_flag.py b/tests/test_flag.py index 6a48d2c..ead75d6 100644 --- a/tests/test_flag.py +++ b/tests/test_flag.py @@ -24,10 +24,9 @@ def test_flag_files_arisense_db(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - "co_we", - "lt", - "205.0" - ] + "v100", # or v200? + "database" + ], catch_exceptions=False ) # did it succeed? @@ -51,10 +50,9 @@ def test_flag_files_feather(self): os.path.join(self.test_dir, "output.feather"), "-v", os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - "co_we", - "gt", - "505.0" - ] + "v100", # or v200? + "database" + ], catch_exceptions=False ) # did it succeed? @@ -78,10 +76,9 @@ def test_flag_files_modulair_db(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair/MOD-00014-db-raw.csv"), - "co_we", - "lt", - "205.0" - ] + "modulair", + "database" + ], catch_exceptions=False ) # did it succeed? @@ -105,10 +102,9 @@ def test_flag_files_modulairx_rawsd(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair-x/MOD-X-00891-rawsd-file1.csv"), - "co_we", - "lt", - "205.0" - ] + "modulair-x", + "rawsd" + ], catch_exceptions=False ) # did it succeed? @@ -125,17 +121,16 @@ def test_flag_files_modulairx_rawsd(self): self.assertEqual(p.suffix, ".csv") def test_flag_files_modulairx_cloudapi(self): - runner = CliRunner() - result = runner.invoke(flag, + runner = CliRunner() + result = runner.invoke(flag, [ "-o", os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair-x/MOD-X-00993-cloudapi-file1.csv"), - "co_we", - "lt", - "205.0" - ] + "modulair-x", + "cloudapi" + ], catch_exceptions=False ) # did it succeed? From 003d5ad7bd473be8c3dfc265103db25b620c1468 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:55:34 -0400 Subject: [PATCH 06/23] Fix indentation --- tests/test_flag.py | 6 +++--- tests/test_merge.py | 2 -- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/tests/test_flag.py b/tests/test_flag.py index ead75d6..5c57d02 100644 --- a/tests/test_flag.py +++ b/tests/test_flag.py @@ -121,9 +121,9 @@ def test_flag_files_modulairx_rawsd(self): self.assertEqual(p.suffix, ".csv") def test_flag_files_modulairx_cloudapi(self): - runner = CliRunner() - result = runner.invoke(flag, - [ + runner = CliRunner() + result = runner.invoke(flag, + [ "-o", os.path.join(self.test_dir, "output.csv"), "-v", diff --git a/tests/test_merge.py b/tests/test_merge.py index 2e8d1c3..d208a9d 100644 --- a/tests/test_merge.py +++ b/tests/test_merge.py @@ -147,7 +147,6 @@ def test_merge_files_modulairx_rawsd(self): df2 = safe_load(os.path.join(self.test_files_dir, "modulair-x/MOD-X-00891-rawsd-file2.csv")) df3 = safe_load(os.path.join(self.test_dir, "output.csv")) - print(df1.shape[1], df2.shape[1], df3.shape[1], flush=True) self.assertEqual(df1.shape[1] + df2.shape[1] - 1, df3.shape[1]) def test_merge_files_modulairx_cloudapi(self): @@ -181,6 +180,5 @@ def test_merge_files_modulairx_cloudapi(self): df2 = safe_load(os.path.join(self.test_files_dir, "modulair-x/MOD-X-00993-cloudapi-file2.csv")) df3 = safe_load(os.path.join(self.test_dir, "output.csv")) - print(df1.shape[1], df2.shape[1], df3.shape[1], flush=True) self.assertEqual(df1.shape[1] + df2.shape[1] - 1, df3.shape[1]) \ No newline at end of file From d4b2c97b9a306c8e981289984fbac022824adb6d Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 09:50:11 -0400 Subject: [PATCH 07/23] Remove relative imports --- quantaq_cli/console/commands/flag.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/quantaq_cli/console/commands/flag.py b/quantaq_cli/console/commands/flag.py index efa5e6a..a74751c 100644 --- a/quantaq_cli/console/commands/flag.py +++ b/quantaq_cli/console/commands/flag.py @@ -6,10 +6,10 @@ from loguru import logger from terminaltables import SingleTable -from ...variables import FLAGS, get_flag_criteria, SUPPORTED_MODELS, SUPPORTED_SOURCES -from ...variables import Range, Gap -from ...utilities import determine_timestamp_column, safe_load -from ...exceptions import InvalidFileExtension, InvalidArgument, InvalidDeviceModel +from quantaq_cli.variables import FLAGS, get_flag_criteria, SUPPORTED_MODELS, SUPPORTED_SOURCES +from quantaq_cli.variables import Range, Gap +from quantaq_cli.utilities import determine_timestamp_column, safe_load +from quantaq_cli.exceptions import InvalidFileExtension, InvalidArgument, InvalidDeviceModel def add_flag(df, flag_name, flag_value, criterion): From 22b756031654a42f9d30482996dfe8ba38d825cd Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 09:56:16 -0400 Subject: [PATCH 08/23] Remove arisense flagging functionality --- quantaq_cli/variables.py | 21 ---------------- tests/test_flag.py | 52 ---------------------------------------- 2 files changed, 73 deletions(-) diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py index 32b714b..adfc75f 100644 --- a/quantaq_cli/variables.py +++ b/quantaq_cli/variables.py @@ -38,27 +38,6 @@ # i.e. are flag names and values different for database vs rawSD data? FLAGS = {} -FLAGS["v100"] = [ - Flag("FLAG_STARTUP", 1, []), - Flag("FLAG_OPC", 2, _OPC_COLUMNS_V1), - Flag("FLAG_TOTAL_COUNTS", 4, [ - "voc_raw", "pressure", "temp_manifold", "rh_manifold", "temp_box", - "dew_point", "noise", "solar", "wind_dir", "wind_speed", - "sample_time", "opc_flow", "manifold_temp", "manifold_rh", - ]), - Flag("FLAG_CO", 8, ["co_we", "co_ae"]), - Flag("FLAG_NO", 16, ["no_we", "no_ae"]), - Flag("FLAG_NO2", 32, ["no2_we", "no2_ae"]), - Flag("FLAG_O3", 64, ["o3_we", "o3_ae"]), - Flag("FLAG_OPC_RECORD_NUM", 128, _OPC_COLUMNS_V1), - Flag("FLAG_CO2", 256, ["co2_raw"]), - # Flag("FLAG_PP", 512, [...]), - Flag("FLAG_ROW", 1024, None), -] - -# v200 flags are identical to the v100 flags. -FLAGS["v200"] = FLAGS["v100"] - FLAGS["modulair_pm"] = [ Flag("FLAG_STARTUP", 1, []), Flag("FLAG_OPC", 2, _OPC_COLUMNS), diff --git a/tests/test_flag.py b/tests/test_flag.py index 5c57d02..9d97d71 100644 --- a/tests/test_flag.py +++ b/tests/test_flag.py @@ -16,58 +16,6 @@ def setUp(self): def tearDown(self): shutil.rmtree(self.test_dir) - def test_flag_files_arisense_db(self): - runner = CliRunner() - result = runner.invoke(flag, - [ - "-o", - os.path.join(self.test_dir, "output.csv"), - "-v", - os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - "v100", # or v200? - "database" - ], catch_exceptions=False - ) - - # did it succeed? - self.assertEqual(result.exit_code, 0) - - # did it output the correct text? - self.assertTrue("File to read" in result.output) - - # make sure the file exists - p = Path(self.test_dir + "/output.csv") - self.assertTrue(p.exists()) - - # is it a csv? - self.assertEqual(p.suffix, ".csv") - - def test_flag_files_feather(self): - runner = CliRunner() - result = runner.invoke(flag, - [ - "-o", - os.path.join(self.test_dir, "output.feather"), - "-v", - os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - "v100", # or v200? - "database" - ], catch_exceptions=False - ) - - # did it succeed? - self.assertEqual(result.exit_code, 0) - - # did it output the correct text? - self.assertTrue("File to read" in result.output) - - # make sure the file exists - p = Path(self.test_dir + "/output.feather") - self.assertTrue(p.exists()) - - # is it a csv? - self.assertEqual(p.suffix, ".feather") - def test_flag_files_modulair_db(self): runner = CliRunner() result = runner.invoke(flag, From 61f6d34c57732a2560bcfb4e70c05940e049d0ba Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:22:16 -0400 Subject: [PATCH 09/23] Modify safe_loader() to add device serial number to rawSD columns --- quantaq_cli/utilities.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/quantaq_cli/utilities.py b/quantaq_cli/utilities.py index 1c441d1..914a0ff 100644 --- a/quantaq_cli/utilities.py +++ b/quantaq_cli/utilities.py @@ -1,11 +1,16 @@ -import pandas as pd from pathlib import Path +from loguru import logger +import pandas as pd + from quantaq_cli.exceptions import InvalidFileExtension def safe_load(fpath, **kwargs): """Load and return a file + + TO DO: replace feather functionality with parquet (will do as part of sc-20240) + also note that pd.read_feather doesn't take a skiprows option """ p = Path(fpath) @@ -19,7 +24,17 @@ def safe_load(fpath, **kwargs): tmp = pd.read_csv(fpath, nrows=1, header=None) if as_csv else pd.read_feather(fpath) if tmp.iloc[0, 0] == "deviceModel": # hack to deal with modulair format + logger.info("Reading rawSD card data {}", fpath) + + # Grab the serial number from the header + tmp2 = pd.read_csv(fpath, nrows=3, header=None) + serial_number = tmp2.iloc[2, 1] + + # Add the sn as a column tmp = pd.read_csv(fpath, skiprows=3) if as_csv else pd.read_feather(fpath, skiprows=3) + tmp['sn'] = serial_number + logger.info("Added serial number {} to column `sn` ", serial_number) + elif tmp.shape[1] == 2: # hack to deal with bad header format tmp = pd.read_csv(fpath, skiprows=1) if as_csv else pd.read_feather(fpath, skiprows=1) else: From f0b40aa7126331646478e07ee14756ecc881eeb5 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 10:55:32 -0400 Subject: [PATCH 10/23] Add func to infer data model from the sn column --- quantaq_cli/utilities.py | 21 +++++++++++++++++++++ tests/test_utilities.py | 37 +++++++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+) create mode 100644 tests/test_utilities.py diff --git a/quantaq_cli/utilities.py b/quantaq_cli/utilities.py index 914a0ff..fbdeb19 100644 --- a/quantaq_cli/utilities.py +++ b/quantaq_cli/utilities.py @@ -6,6 +6,27 @@ from quantaq_cli.exceptions import InvalidFileExtension +def infer_data_model(df): + sn_array = df['sn'].unique() + if len(sn_array) > 1: + logger.debug(f"Found {len(sn_array)} unique serial numbers: {sn_array}") + raise ValueError( + f"More than one unique serial number found in dataframe: {sn_array}" + ) + device_sn = sn_array.item() + device_model = sn_to_model(device_sn) + return device_model + +def sn_to_model(device_sn): + """Map a device serial number to its model string. + MOD-00246 -> modulair + MOD-PM-00933 -> modulair-pm + MOD-X-00993 -> modulair-x + MOD-X-PM-01685 -> modulair-x-pm + """ + prefix, _, _ = device_sn.rpartition("-") + return prefix.lower().replace("mod", "modulair", 1) + def safe_load(fpath, **kwargs): """Load and return a file diff --git a/tests/test_utilities.py b/tests/test_utilities.py new file mode 100644 index 0000000..60ce0ec --- /dev/null +++ b/tests/test_utilities.py @@ -0,0 +1,37 @@ +import os +import shutil +import tempfile +import unittest + +import numpy as np + +from quantaq_cli.utilities import sn_to_model, safe_load, infer_data_model + +class SetupTestCase(unittest.TestCase): + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.test_files_dir = os.path.join(os.getcwd(), "tests/files") + + def tearDown(self): + shutil.rmtree(self.test_dir) + + def test_sn_to_model(self): + cases = [ + ("MOD-00246", "modulair"), + ("MOD-PM-00933", "modulair-pm"), + ("MOD-X-00993", "modulair-x"), + ("MOD-X-PM-01685", "modulair-x-pm"), + ] + for serial, expected in cases: + with self.subTest(serial=serial): + self.assertEqual(sn_to_model(serial), expected) + + def test_infer_data_model(self): + file = os.path.join( + self.test_files_dir, "modulair-x/MOD-X-00891-rawsd-file1.csv" + ) + expected = "modulair-x" + df = safe_load(file) + result = infer_data_model(df) + + self.assertEqual(result, expected) From 5ad1d1febc4da8d64c04e5153e22fcfd0b647408 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 11:15:57 -0400 Subject: [PATCH 11/23] Add func to infer data source from the dataframe sampling frequency (tdiff) --- quantaq_cli/utilities.py | 29 +++++++++++++++++++++++++++++ tests/test_utilities.py | 28 ++++++++++++++++++++++++---- 2 files changed, 53 insertions(+), 4 deletions(-) diff --git a/quantaq_cli/utilities.py b/quantaq_cli/utilities.py index fbdeb19..b8a2494 100644 --- a/quantaq_cli/utilities.py +++ b/quantaq_cli/utilities.py @@ -6,6 +6,35 @@ from quantaq_cli.exceptions import InvalidFileExtension +def infer_data_source(df): + # Determine the best timestamp column and sort + tscol = determine_timestamp_column(df) + df[tscol] = pd.to_datetime(df[tscol]) + df = df.sort_values(tscol) + + # Grab the tdiff between consecutive rows + df["tdiff"] = df[tscol].diff().dt.total_seconds() + tdiffs = set(df["tdiff"].dropna().unique()) + + has_1min = 60.0 in tdiffs + has_5sec = 5.0 in tdiffs + + if has_1min and has_5sec: + logger.debug("Mixed 1-min and 5-sec sampling not supported") + raise NotImplementedError + elif has_1min: + logger.info("Reading 1-min data --> inferring database or cloud API") + if "api_received_at" in df.columns: + return "cloudapi" + else: + return "database" + elif has_5sec: + logger.info("Reading 5-sec data --> inferring rawSD") + return "rawsd" + else: + logger.debug(f"Unrecognized sampling intervals: {sorted(tdiffs)}") + raise NotImplementedError + def infer_data_model(df): sn_array = df['sn'].unique() if len(sn_array) > 1: diff --git a/tests/test_utilities.py b/tests/test_utilities.py index 60ce0ec..f6bcaa5 100644 --- a/tests/test_utilities.py +++ b/tests/test_utilities.py @@ -5,7 +5,8 @@ import numpy as np -from quantaq_cli.utilities import sn_to_model, safe_load, infer_data_model +from quantaq_cli.utilities import safe_load +from quantaq_cli.utilities import sn_to_model, infer_data_model, infer_data_source class SetupTestCase(unittest.TestCase): def setUp(self): @@ -22,9 +23,9 @@ def test_sn_to_model(self): ("MOD-X-00993", "modulair-x"), ("MOD-X-PM-01685", "modulair-x-pm"), ] - for serial, expected in cases: - with self.subTest(serial=serial): - self.assertEqual(sn_to_model(serial), expected) + for sn, expected in cases: + with self.subTest(sn=sn): + self.assertEqual(sn_to_model(sn), expected) def test_infer_data_model(self): file = os.path.join( @@ -35,3 +36,22 @@ def test_infer_data_model(self): result = infer_data_model(df) self.assertEqual(result, expected) + + def test_infer_data_source(self): + df1 = safe_load(os.path.join( + self.test_files_dir, "modulair-pm/MOD-PM-00001-rawsd-file1.csv")) + df2 = safe_load(os.path.join( + self.test_files_dir, "modulair/MOD-00014-db-raw.csv")) + df3 = safe_load(os.path.join( + self.test_files_dir, "modulair-x/MOD-X-00993-cloudapi-file1.csv")) + + cases = [ + (df1, "rawsd"), + (df2, "database"), + (df3, "cloudapi"), + ] + + for df, expected in cases: + with self.subTest(expected=expected): + self.assertEqual(infer_data_source(df), expected) + From 045e34bdc39fb3a0840ca33b4f5ceeb7aee94917 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 11:19:41 -0400 Subject: [PATCH 12/23] Infer data model and source in flag_dataframe() instead of taking args --- quantaq_cli/console/commands/flag.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/quantaq_cli/console/commands/flag.py b/quantaq_cli/console/commands/flag.py index a74751c..4a5592f 100644 --- a/quantaq_cli/console/commands/flag.py +++ b/quantaq_cli/console/commands/flag.py @@ -9,6 +9,7 @@ from quantaq_cli.variables import FLAGS, get_flag_criteria, SUPPORTED_MODELS, SUPPORTED_SOURCES from quantaq_cli.variables import Range, Gap from quantaq_cli.utilities import determine_timestamp_column, safe_load +from quantaq_cli.utilities import infer_data_source, infer_data_model from quantaq_cli.exceptions import InvalidFileExtension, InvalidArgument, InvalidDeviceModel @@ -57,7 +58,7 @@ def add_flag(df, flag_name, flag_value, criterion): return df -def flag_dataframe(df, model, source): +def flag_dataframe(df): """ df = pandas dataframe to be flagged (or re-flagged) model = the sensor model (in SUPPORTED_MODELS) @@ -65,6 +66,9 @@ def flag_dataframe(df, model, source): """ df = df.copy() + model = infer_data_model(df) + source = infer_data_source(df) + # ensure the model is valid if model not in SUPPORTED_MODELS: raise InvalidDeviceModel("Invalid device model. Must be one of {}".format(SUPPORTED_MODELS)) @@ -87,7 +91,7 @@ def flag_dataframe(df, model, source): df = add_flag(df, flag_name, flag_value, criterion) return df -def flag_command(file, output, model, source, **kwargs): +def flag_command(file, output, **kwargs): verbose = kwargs.pop("verbose", False) # make sure the extension is either a csv or feather format @@ -104,7 +108,7 @@ def flag_command(file, output, model, source, **kwargs): df = safe_load(file) # flag the dataframe - df = flag_dataframe(df, model, source) + df = flag_dataframe(df) # save the file if verbose: From d04686a7868a0849daa0dbbfc52e4cd55e3b5aa0 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 13:48:08 -0400 Subject: [PATCH 13/23] Define identical FLAG_DEFINITIONS for QuantAQ products and modifiable FLAG_CRITERIA --- quantaq_cli/variables.py | 168 ++++++++++++++++++++++----------------- 1 file changed, 97 insertions(+), 71 deletions(-) diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py index adfc75f..d0ac23f 100644 --- a/quantaq_cli/variables.py +++ b/quantaq_cli/variables.py @@ -1,115 +1,141 @@ -############################################################################### - -"""variables.py - -Per-model data-quality flags. - -Each flag has a name, an integer bitmask value, and the set of columns to set -to NaN when the flag is raised: - - * a list of column names -> NaN just those columns - * an empty list -> flag carries no column-level action - * None -> NaN the entire row -""" from collections import namedtuple -Flag = namedtuple("Flag", ["name", "value", "nan_columns"]) +from loguru import logger -# Shared column groups, factored out so the per-model tables stay readable and -# the lists can't silently drift out of sync. -_OPC_COLUMNS_V1 = ["bin0", "bin1", "bin2", "bin3", "bin4", "bin5", "opc_flow"] +Flag = namedtuple("Flag", ["name", "value", "nan_columns"]) +# Columns to Nan across all schema types (db, rawsd, cloudapi) _OPC_COLUMNS = [ "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", - "sample_period", "sample_flow", "opc_temp", "opc_rh", "opc_pm1", "opc_pm25", "opc_pm10", - "laser_status", + "bin10", "bin11", "bin12", "bin13", "bin14", "bin15", "bin16", "bin17", "bin18", + "bin19", "bin20", "bin21", "bin22", "bin23", + "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", + "bin1MToF", "bin3MToF", "bin5MToF", "bin7MToF", + "sample_period", "opc_sample_period", + "sample_flow", "opc_sample_flow", + "opc_pm1", "opc_pm25", "opc_pm10", + "opcn3_pm1", "opcn3_pm25", "opcn3_pm10", + "laser_status", "opc_laser_status", + "opc_temp", "opc_rh", ] _NEPH_COLUMNS = [ - "pm1_std", "pm25_std", "pm10_std", "pm1_env", "pm25_env", "pm10_env", + "pm1_std", "pm25_std", "pm10_std", + "pm1_env", "pm25_env", "pm10_env", + "neph_pm1_std", "neph_pm25_std", "neph_pm10_std", + "neph_pm1_env", "neph_pm25_env", "neph_pm10_env" "neph_bin0", "neph_bin1", "neph_bin2", "neph_bin3", "neph_bin4", "neph_bin5", ] +_RHTP_COLUMNS = [ + "sample_rh", "sample_temp", "sample_pres", "rh", "temp" +] -# Should FLAGS also be a nested dict like FLAG_CRITERIA below -# (with keys = source, nested keys = model)? -# i.e. are flag names and values different for database vs rawSD data? -FLAGS = {} +_CO_COLUMNS = [ + "co", + "co_we", "co_ae", "co_diff", + "gases.co.we", "gases.co.ae", "gases.co.diff" +] -FLAGS["modulair_pm"] = [ - Flag("FLAG_STARTUP", 1, []), - Flag("FLAG_OPC", 2, _OPC_COLUMNS), - Flag("FLAG_NEPH", 4, _NEPH_COLUMNS), - Flag("FLAG_RHTP", 8, ["sample_rh", "sample_temp", "sample_pres"]), - Flag("FLAG_ROW", 1024, None), +_NO_COLUMNS = [ + "no", + "no_we", "no_ae", "no_diff", + "gases.no.we", "gases.no.ae", "gases.no.diff" +] + +_NO2_COLUMNS = [ + "no2", + "no2_we", "no2_ae", "no2_diff", + "gases.no2.we", "gases.no2.ae", "gases.no2.diff" +] + +_O3_COLUMNS = [ + "o3", + "ox_we", "ox_ae", "ox_diff", + "o3_we", "o3_ae", "o3_diff", + "gases.o3.we", "gases.o3.ae", "gases.o3.diff" ] -FLAGS["modulair"] = [ +# Flag Name, Flag Bitmask Value, Columns to Nan +FLAG_DEFINITIONS = [ Flag("FLAG_STARTUP", 1, []), Flag("FLAG_OPC", 2, _OPC_COLUMNS), Flag("FLAG_NEPH", 4, _NEPH_COLUMNS), - Flag("FLAG_RHTP", 8, ["sample_rh", "sample_temp", "sample_pres"]), - Flag("FLAG_CO", 16, ["co_we", "co_ae", "co_diff"]), - Flag("FLAG_NO", 32, ["no_we", "no_ae", "no_diff"]), - Flag("FLAG_NO2", 64, ["no2_we", "no2_ae", "no2_diff"]), - Flag("FLAG_O3", 128, ["ox_we", "ox_ae", "ox_diff", "o3_diff"]), - Flag("FLAG_CO2", 256, ["co2_raw"]), - Flag("FLAG_SO2", 512, ["so2_we", "so2_ae", "so2_diff"]), - Flag("FLAG_H2S", 1024, ["h2s_we", "h2s_ae", "h2s_diff"]), + Flag("FLAG_RHTP", 8, _RHTP_COLUMNS), + Flag("FLAG_CO", 16, _CO_COLUMNS), + Flag("FLAG_NO", 32, _NO_COLUMNS), + Flag("FLAG_NO2", 64, _NO2_COLUMNS), + Flag("FLAG_O3", 128, _O3_COLUMNS), + Flag("FLAG_CO2", 256, ["co2_raw", "co2"]), + Flag("FLAG_SO2", 512, ["so2_we", "so2_ae", "so2_diff", "so2"]), + Flag("FLAG_H2S", 1024, ["h2s_we", "h2s_ae", "h2s_diff", "h2s"]), Flag("FLAG_BAT", 2048, ["bat_voltage", "soc", "vbat"]), ] -# modulair-x DB flags are identical to modulair? -FLAGS["modulair-x"] = FLAGS["modulair"] - -SUPPORTED_MODELS = FLAGS.keys() - -############################################################################### +SUPPORTED_MODELS = ("modulair", "modulair-x", "modulair-pm", "modulair-x-pm") +# Types of flag criteria Range = namedtuple("Range", ["column", "lo", "hi"]) Gap = namedtuple("Gap", ["gap_in_seconds", "post_gap_flag_length_seconds"]) FLAG_CRITERIA = { "database": { - "modulair": { - "FLAG_STARTUP": [Gap(60 * 60, 60 * 60)], - "FLAG_CO": [Range("co_ae", 535.0, 800.0)], - "FLAG_NO": [Range("no_ae", 640.0, 900.0)], - "FLAG_NO2": [Range("no2_ae", 1600.0, 1700.0)], - "FLAG_O3": [Range("o3_ae", 1600.0, 1700.0)], - "FLAG_CO2": [Range("co2_raw", 1000.0, 5000.0)], - "FLAG_OPC": [Range("bin0", 0.0, 1e6)], + "default": { + "FLAG_STARTUP": [ + Gap(60 * 60, 60 * 60)], + "FLAG_CO": [ + Range("co_ae", 535.0, 800.0), + Range("gases.co.ae", 535.0, 800.0) + ], + "FLAG_NO": [ + Range("no_ae", 640.0, 900.0), + Range("gases.no.ae", 640.0, 900.0) + ], + "FLAG_NO2": [ + Range("no2_ae", 1600.0, 1700.0), + Range("gases.no2.ae", 1600.0, 1700.0) + ], + "FLAG_O3": [ + Range("o3_ae", 1600.0, 1700.0), + Range("gases.o3.ae", 1600.0, 1700.0), + ], + "FLAG_CO2": [ + Range("co2_raw", 1000.0, 5000.0)], + "FLAG_OPC": [ + Range("bin0", 0.0, 1e6), + Range("opc.bin0", 0.0, 1e6) + ], + #"FLAG_NEPH": [], "FLAG_RHTP": [ Range("sample_rh", 0.0, 100.0), Range("sample_temp", -60.0, 85.0), Range("rh", 0.0, 100.0), Range("temp", -60.0, 85.0), - #Range("sample_pres", lo, hi), # find these values - ], + ]}, }, - }, - "rawsd": { # for rawSD flags, criteria are NOT consistent - #"modulair": { - # "FLAG_STARTUP": [Gap(60 * 60, 60 * 60)], # placeholder - #}, - #"modulair-x": { - # "NO_FLAG_STARTUP": [Gap(60 * 60, 60 * 60)], # placeholder - #}, - }, -} - -# modulair-x DB flag criteria are identical to modulair? -FLAG_CRITERIA["database"]["modulair-x"] = FLAG_CRITERIA["database"]["modulair"] + "rawsd": {} # different logic applies + } -SUPPORTED_SOURCES = FLAG_CRITERIA.keys() +# Assume all quant-aq products share the same database flagging criteria. +for _model in SUPPORTED_MODELS: + FLAG_CRITERIA["database"][_model] = FLAG_CRITERIA["database"]["default"] +# Assume cloudapi uses the same flag criteria as database +# (besides name of column to be flagged) +FLAG_CRITERIA["cloudapi"] = FLAG_CRITERIA["database"] + +SUPPORTED_SOURCES = FLAG_CRITERIA.keys() def get_flag_criteria(source, model): """ source = 'rawsd' or 'database' model = 'default', 'modulair', 'modulair-x' """ + if source == 'rawsd': + logger.debug("{!r} flag criteria are not yet defined", source) + raise NotImplementedError by_model = FLAG_CRITERIA[source] return by_model[model] From e19c02871f96c5918f1234e0e3b14da3cc51860a Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 13:49:13 -0400 Subject: [PATCH 14/23] Convert model-specific FLAG_DEFINITIONS into one generic list --- quantaq_cli/console/commands/flag.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/quantaq_cli/console/commands/flag.py b/quantaq_cli/console/commands/flag.py index 4a5592f..19b0781 100644 --- a/quantaq_cli/console/commands/flag.py +++ b/quantaq_cli/console/commands/flag.py @@ -6,7 +6,8 @@ from loguru import logger from terminaltables import SingleTable -from quantaq_cli.variables import FLAGS, get_flag_criteria, SUPPORTED_MODELS, SUPPORTED_SOURCES +from quantaq_cli.variables import FLAG_DEFINITIONS, FLAG_CRITERIA, get_flag_criteria +from quantaq_cli.variables import SUPPORTED_MODELS, SUPPORTED_SOURCES from quantaq_cli.variables import Range, Gap from quantaq_cli.utilities import determine_timestamp_column, safe_load from quantaq_cli.utilities import infer_data_source, infer_data_model @@ -82,11 +83,11 @@ def flag_dataframe(df): df["flag"] = 0 # get the flag values for each flag name - flag_values = {flag.name: flag.value for flag in FLAGS[model]} + FLAG_VALUES = {flag.name: flag.value for flag in FLAG_DEFINITIONS} # set the flag for each flag_name and their respective crtieria for flag_name, criteria in get_flag_criteria(source, model).items(): - flag_value = flag_values[flag_name] + flag_value = FLAG_VALUES[flag_name] for criterion in criteria: df = add_flag(df, flag_name, flag_value, criterion) return df From fe018ebdb0ccdf7fef6cac3a23ae972fa9d118de Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 13:49:32 -0400 Subject: [PATCH 15/23] Remove tests for rawSD flagging - currently not implemented --- tests/test_flag.py | 40 ++++++++++++++++++++-------------------- 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/tests/test_flag.py b/tests/test_flag.py index 9d97d71..bf8be45 100644 --- a/tests/test_flag.py +++ b/tests/test_flag.py @@ -42,31 +42,31 @@ def test_flag_files_modulair_db(self): # is it a csv? self.assertEqual(p.suffix, ".csv") - def test_flag_files_modulairx_rawsd(self): - runner = CliRunner() - result = runner.invoke(flag, - [ - "-o", - os.path.join(self.test_dir, "output.csv"), - "-v", - os.path.join(self.test_files_dir, "modulair-x/MOD-X-00891-rawsd-file1.csv"), - "modulair-x", - "rawsd" - ], catch_exceptions=False - ) + #def test_flag_files_modulairx_rawsd(self): + # runner = CliRunner() + # result = runner.invoke(flag, + # [ + # "-o", + # os.path.join(self.test_dir, "output.csv"), + # "-v", + # os.path.join(self.test_files_dir, "modulair-x/MOD-X-00891-rawsd-file1.csv"), + # "modulair-x", + # "rawsd" + # ], catch_exceptions=False + # ) # did it succeed? - self.assertEqual(result.exit_code, 0) + # self.assertEqual(result.exit_code, 0)# - # did it output the correct text? - self.assertTrue("File to read" in result.output) + # # did it output the correct text? + # self.assertTrue("File to read" in result.output) - # make sure the file exists - p = Path(self.test_dir + "/output.csv") - self.assertTrue(p.exists()) + # # make sure the file exists + # p = Path(self.test_dir + "/output.csv") + # self.assertTrue(p.exists()) - # is it a csv? - self.assertEqual(p.suffix, ".csv") + # # is it a csv? + # self.assertEqual(p.suffix, ".csv") def test_flag_files_modulairx_cloudapi(self): runner = CliRunner() From 30646f7688759d0dc68da7ec45cd9d8523742ee4 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 13:56:19 -0400 Subject: [PATCH 16/23] Upgrade deps to Click8.x --- poetry.lock | 13 ++++++++----- pyproject.toml | 2 +- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/poetry.lock b/poetry.lock index aa7592e..0ed1418 100644 --- a/poetry.lock +++ b/poetry.lock @@ -197,15 +197,18 @@ files = [ [[package]] name = "click" -version = "7.1.2" +version = "8.1.8" description = "Composable command line interface toolkit" optional = false -python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*" +python-versions = ">=3.7" files = [ - {file = "click-7.1.2-py2.py3-none-any.whl", hash = "sha256:dacca89f4bfadd5de3d7489b7c8a566eee0d3676333fbb50030263894c38c0dc"}, - {file = "click-7.1.2.tar.gz", hash = "sha256:d2b5255c7c6349bc1bd1e59e08cd12acbbd63ce649f2588755783aa94dfb6b1a"}, + {file = "click-8.1.8-py3-none-any.whl", hash = "sha256:63c132bbbed01578a06712a2d1f497bb62d9c1c0d329b7903a866228027263b2"}, + {file = "click-8.1.8.tar.gz", hash = "sha256:ed53c9d8990d83c2a27deae68e4ee337473f6330c040a31d4225c9574d16096a"}, ] +[package.dependencies] +colorama = {version = "*", markers = "platform_system == \"Windows\""} + [[package]] name = "colorama" version = "0.4.6" @@ -1416,4 +1419,4 @@ type = ["pytest-mypy"] [metadata] lock-version = "2.0" python-versions = ">=3.9, <3.14" -content-hash = "35f6c41bcd2f19d5c5086be42b9c8dc8f40d414e1678f01755628a92fc4dddd8" +content-hash = "5a23c1ec6d3b12bb209f2ff5a2432d629111fa656681403be8d24f392a35fc3f" diff --git a/pyproject.toml b/pyproject.toml index 7b5b691..1e240f8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,7 +13,7 @@ classifiers = [] [tool.poetry.dependencies] python = ">=3.9, <3.14" pandas = ">=2.1.4,<2.2" -click = "^7.1.2" +click = "^8.1" pyarrow = ">=14.0.2" terminaltables = "^3.1.0" urllib3 = "^1.26.10" From 89a2ce676c6dde1c409b2dc783103cebd94002d2 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 14:04:21 -0400 Subject: [PATCH 17/23] Port expunge_dataframe() and echo_flag_table() from Isoprene --- quantaq_cli/console/commands/expunge.py | 145 +++++++++++++++--------- 1 file changed, 94 insertions(+), 51 deletions(-) diff --git a/quantaq_cli/console/commands/expunge.py b/quantaq_cli/console/commands/expunge.py index 6b62f7a..ce74f95 100644 --- a/quantaq_cli/console/commands/expunge.py +++ b/quantaq_cli/console/commands/expunge.py @@ -1,19 +1,100 @@ from pathlib import Path -import pandas as pd -import numpy as np + import click -from terminaltables import SingleTable +import numpy as np +import pandas as pd +import rich +from rich.table import Table + +from quantaq_cli.exceptions import InvalidFileExtension, InvalidDeviceModel +from quantaq_cli.utilities import safe_load +from quantaq_cli.variables import FLAG_DEFINITIONS, SUPPORTED_MODELS + + +def expunge_dataframe(df): + + # get the flags (in the future, this will come from the file itself) + list_of_flags = FLAG_DEFINITIONS + + # force the flag column to be an int + df["flag"] = df["flag"].astype(int, errors='ignore') + + # Drop NaNs + df = df.dropna(how='any', subset=["flag"]) + + for label, value, cols in list_of_flags: + mask = df["flag"] & value == value + n_affected = mask.sum() + pct_affected = round((n_affected / df.shape[0]) * 100.0, 2) + + # NaN the necessary columns + if cols is None: + cols = df.columns + elif len(cols) > 0: + cols = [c for c in cols if c in df.columns] + + # set the mask + df.loc[mask, cols] = np.nan + + return df -from ...exceptions import InvalidFileExtension, InvalidDeviceModel -from ...utilities import safe_load -from ...variables import FLAGS, SUPPORTED_MODELS +def flag_summary(df): + """Create a table with a summary of flags. + + Parameters + ---------- + sensor_df + DataFrame with flags to summarize. + + Returns + ------- + A new DataFrame, suitable for human consumption. + + TO DO: move this to flag.py after merging sc-20242 + """ + # force the flag column to be an int + df["flag"] = df["flag"].astype(int, errors='ignore') + + rows = [] + for name, value, cols in FLAG_DEFINITIONS: + mask = df["flag"] & value + num_naffected = mask.astype(bool).sum() + percent_affected = f"{100 * num_naffected / df.shape[0]:.1f}" + rows.append([name, int(value), num_naffected, percent_affected]) + + explained_df = pd.DataFrame( + rows, + columns=["FLAG", "FLAG VALUE", "# OCCURENCES", "% DATA"], + ).set_index("FLAG") + return explained_df + +def echo_flag_table(df): + """Print a table of flag statistics for a DataFrame. + + Parameters + ---------- + sensor_df + DataFrame to describe. + + TO DO: move this to flag.py after merging sc-20242 + """ + explained_df = flag_summary(df) + + table = Table() + for column in ["FLAG", *explained_df.columns]: + table.add_column( + column, + justify="right" if column != "FLAG" else "left", + style="bold", + ) + for row in explained_df.itertuples(): + table.add_row(*map(str, row)) + rich.print(table) def expunge_command(file, output, **kwargs): verbose = kwargs.pop("verbose", False) dry_run = kwargs.pop("dry_run", False) - flagcol = kwargs.pop("flagcol", "flag") - model = kwargs.pop("model", "modulair_pm") table = kwargs.pop("table", False) # make sure the extension is either a csv or feather format @@ -21,10 +102,6 @@ def expunge_command(file, output, **kwargs): if output.suffix not in (".csv", ".feather"): raise InvalidFileExtension("Invalid file extension") - # ensure the model is valid - if model not in SUPPORTED_MODELS: - raise InvalidDeviceModel("Invalid device model. Must be one of {}".format(SUPPORTED_MODELS)) - save_as_csv = True if output.suffix == ".csv" else False # concat everything in filepath @@ -34,49 +111,15 @@ def expunge_command(file, output, **kwargs): # load the file df = safe_load(file) + # expunge if verbose: - click.echo("Expunging data for {}".format(model)) - - # init an array to hold the table data - data = [] - data.append(["FLAG", "FLAG VALUE", "# OCCURENCES", "% DATA"]) - - # get the flags (in the future, this will come from the file itself) - list_of_flags = FLAGS.get(model) - - # force the flag column to be an int - df[flagcol] = df[flagcol].astype(int, errors='ignore') - - # Drop NaNs - df = df.dropna(how='any', subset=[flagcol]) - - for label, value, cols in list_of_flags: - mask = df[flagcol] & value == value - n_affected = mask.sum() - pct_affected = round((n_affected / df.shape[0]) * 100.0, 2) - - # NaN the necessary columns - if cols is None: - cols = df.columns - elif len(cols) > 0: - cols = [c for c in cols if c in df.columns] - - # set the mask - df.loc[mask, cols] = np.nan + click.echo("Expunging data for {}".format(file)) - # add a row to the output table - data.append([label, value, n_affected, pct_affected]) + df = expunge_dataframe(df) if dry_run or verbose: - tbl = SingleTable(data) - tbl.title = "Flag Breakdown".upper() - - if table: - click.echo(tbl.table) - else: - for item in data: - click.echo(item) - + echo_flag_table(df) + # save the file (if not a dry run) if not dry_run: if verbose: From 49c186a29a0827864204cbc6b38b7144339b2f2e Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 14:53:49 -0400 Subject: [PATCH 18/23] Move flag_summary and echo_flag_table to flag.py and fix all_nan_columns definition --- quantaq_cli/console/commands/expunge.py | 71 ++++--------------------- quantaq_cli/console/commands/flag.py | 55 +++++++++++++++++++ quantaq_cli/variables.py | 2 +- 3 files changed, 66 insertions(+), 62 deletions(-) diff --git a/quantaq_cli/console/commands/expunge.py b/quantaq_cli/console/commands/expunge.py index ce74f95..fa6a243 100644 --- a/quantaq_cli/console/commands/expunge.py +++ b/quantaq_cli/console/commands/expunge.py @@ -3,12 +3,13 @@ import click import numpy as np import pandas as pd -import rich -from rich.table import Table from quantaq_cli.exceptions import InvalidFileExtension, InvalidDeviceModel -from quantaq_cli.utilities import safe_load +from quantaq_cli.utilities import safe_load, determine_timestamp_column from quantaq_cli.variables import FLAG_DEFINITIONS, SUPPORTED_MODELS +from quantaq_cli.console.commands.flag import ( + echo_flag_table +) def expunge_dataframe(df): @@ -24,12 +25,14 @@ def expunge_dataframe(df): for label, value, cols in list_of_flags: mask = df["flag"] & value == value - n_affected = mask.sum() - pct_affected = round((n_affected / df.shape[0]) * 100.0, 2) + if not mask.any(): + continue # NaN the necessary columns - if cols is None: - cols = df.columns + if cols == "all_columns": + tscol = determine_timestamp_column(df) + cols_to_keep = {tscol, "sn", "flag"} + cols = [c for c in df.columns if c not in cols_to_keep] elif len(cols) > 0: cols = [c for c in cols if c in df.columns] @@ -38,60 +41,6 @@ def expunge_dataframe(df): return df -def flag_summary(df): - """Create a table with a summary of flags. - - Parameters - ---------- - sensor_df - DataFrame with flags to summarize. - - Returns - ------- - A new DataFrame, suitable for human consumption. - - TO DO: move this to flag.py after merging sc-20242 - """ - # force the flag column to be an int - df["flag"] = df["flag"].astype(int, errors='ignore') - - rows = [] - for name, value, cols in FLAG_DEFINITIONS: - mask = df["flag"] & value - num_naffected = mask.astype(bool).sum() - percent_affected = f"{100 * num_naffected / df.shape[0]:.1f}" - rows.append([name, int(value), num_naffected, percent_affected]) - - explained_df = pd.DataFrame( - rows, - columns=["FLAG", "FLAG VALUE", "# OCCURENCES", "% DATA"], - ).set_index("FLAG") - return explained_df - -def echo_flag_table(df): - """Print a table of flag statistics for a DataFrame. - - Parameters - ---------- - sensor_df - DataFrame to describe. - - TO DO: move this to flag.py after merging sc-20242 - """ - explained_df = flag_summary(df) - - table = Table() - for column in ["FLAG", *explained_df.columns]: - table.add_column( - column, - justify="right" if column != "FLAG" else "left", - style="bold", - ) - for row in explained_df.itertuples(): - table.add_row(*map(str, row)) - rich.print(table) - - def expunge_command(file, output, **kwargs): verbose = kwargs.pop("verbose", False) dry_run = kwargs.pop("dry_run", False) diff --git a/quantaq_cli/console/commands/flag.py b/quantaq_cli/console/commands/flag.py index 19b0781..bb7ee9f 100644 --- a/quantaq_cli/console/commands/flag.py +++ b/quantaq_cli/console/commands/flag.py @@ -4,6 +4,8 @@ import pandas as pd import numpy as np from loguru import logger +import rich +from rich.table import Table from terminaltables import SingleTable from quantaq_cli.variables import FLAG_DEFINITIONS, FLAG_CRITERIA, get_flag_criteria @@ -59,6 +61,59 @@ def add_flag(df, flag_name, flag_value, criterion): return df +def flag_summary(df): + """Create a table with a summary of flags. + + Parameters + ---------- + sensor_df + DataFrame with flags to summarize. + + Returns + ------- + A new DataFrame, suitable for human consumption. + + TO DO: move this to flag.py after merging sc-20242 + """ + # force the flag column to be an int + df["flag"] = df["flag"].astype(int, errors='ignore') + + rows = [] + for name, value, cols in FLAG_DEFINITIONS: + mask = df["flag"] & value + num_naffected = mask.astype(bool).sum() + percent_affected = f"{100 * num_naffected / df.shape[0]:.1f}" + rows.append([name, int(value), num_naffected, percent_affected]) + + explained_df = pd.DataFrame( + rows, + columns=["FLAG", "FLAG VALUE", "# OCCURENCES", "% DATA"], + ).set_index("FLAG") + return explained_df + +def echo_flag_table(df): + """Print a table of flag statistics for a DataFrame. + + Parameters + ---------- + sensor_df + DataFrame to describe. + + TO DO: move this to flag.py after merging sc-20242 + """ + explained_df = flag_summary(df) + + table = Table() + for column in ["FLAG", *explained_df.columns]: + table.add_column( + column, + justify="right" if column != "FLAG" else "left", + style="bold", + ) + for row in explained_df.itertuples(): + table.add_row(*map(str, row)) + rich.print(table) + def flag_dataframe(df): """ df = pandas dataframe to be flagged (or re-flagged) diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py index d0ac23f..34fcae7 100644 --- a/quantaq_cli/variables.py +++ b/quantaq_cli/variables.py @@ -61,7 +61,7 @@ # Flag Name, Flag Bitmask Value, Columns to Nan FLAG_DEFINITIONS = [ - Flag("FLAG_STARTUP", 1, []), + Flag("FLAG_STARTUP", 1, "all_columns"), Flag("FLAG_OPC", 2, _OPC_COLUMNS), Flag("FLAG_NEPH", 4, _NEPH_COLUMNS), Flag("FLAG_RHTP", 8, _RHTP_COLUMNS), From cae63d55d646230430dd3bc331235d1013384dd7 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 14:55:04 -0400 Subject: [PATCH 19/23] Remove arisense from test_expunge() and add some expunge tests from Isoprene --- tests/test_expunge.py | 298 +++++++++++++++++++++++++++++------------- 1 file changed, 206 insertions(+), 92 deletions(-) diff --git a/tests/test_expunge.py b/tests/test_expunge.py index 7e7f454..487d1a5 100644 --- a/tests/test_expunge.py +++ b/tests/test_expunge.py @@ -5,10 +5,16 @@ import os import shutil, tempfile import pandas as pd +from pandas.testing import assert_frame_equal + import numpy as np from quantaq_cli.console import expunge - +from quantaq_cli.console.commands.expunge import expunge_dataframe +from quantaq_cli.console.commands.flag import ( + flag_dataframe, + flag_summary, +) class SetupTestCase(unittest.TestCase): def setUp(self): @@ -18,93 +24,6 @@ def setUp(self): def tearDown(self): shutil.rmtree(self.test_dir) - def test_expunge_csv_arisense_db(self): - runner = CliRunner() - result = runner.invoke(expunge, - [ - "-o", - os.path.join(self.test_dir, "output.csv"), - "-v", - os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - ] - ) - - # did it succeed? - self.assertEqual(result.exit_code, 0) - - # did it output the correct text? - self.assertTrue("Saving file" in result.output) - - # make sure the file exists - p = Path(self.test_dir + "/output.csv") - self.assertTrue(p.exists()) - - # is it a csv? - self.assertEqual(p.suffix, ".csv") - - def test_expunge_dryrun(self): - runner = CliRunner() - result = runner.invoke(expunge, - [ - "-o", - os.path.join(self.test_dir, "output.feather"), - "--dry-run", - "--table", - os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - ] - ) - - # did it succeed? - self.assertEqual(result.exit_code, 0) - - # did it output the correct text? - self.assertTrue("FLAG BREAKDOWN" in result.output) - - # make sure the file does not exist - p = Path(self.test_dir + "/output.feather") - self.assertFalse(p.exists()) - - def test_expunge_dict(self): - runner = CliRunner() - result = runner.invoke(expunge, - [ - "-o", - os.path.join(self.test_dir, "output.feather"), - "--dry-run", - os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - ] - ) - - # did it succeed? - self.assertEqual(result.exit_code, 0) - - # did it output the correct text? - self.assertFalse("FLAG BREAKDOWN" in result.output) - - def test_expunge_feather_arisense_db(self): - runner = CliRunner() - result = runner.invoke(expunge, - [ - "-o", - os.path.join(self.test_dir, "output.feather"), - os.path.join(self.test_files_dir, "arisense/SN000-063-db-file1.csv"), - ] - ) - - # did it succeed? - self.assertEqual(result.exit_code, 0) - - - # did it output the correct text? - self.assertFalse("FLAG BREAKDOWN" in result.output) - - # make sure the file does not exist - p = Path(self.test_dir + "/output.feather") - self.assertTrue(p.exists()) - - # is it a csv? - self.assertEqual(p.suffix, ".feather") - def test_expunge_csv_modulairpm_rawsd(self): runner = CliRunner() result = runner.invoke(expunge, @@ -113,7 +32,7 @@ def test_expunge_csv_modulairpm_rawsd(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair-pm/MOD-PM-00001-rawsd-file1.csv"), - ] + ], catch_exceptions=False ) # did it succeed? @@ -137,7 +56,7 @@ def test_expunge_csv_modulair_db(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair/MOD-00014-db-raw.csv"), - ] + ], catch_exceptions=False ) # did it succeed? @@ -161,7 +80,7 @@ def test_expunge_csv_modulairx_rawsd(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair-x/MOD-X-00891-rawsd-file1.csv"), - ] + ], catch_exceptions=False ) # did it succeed? @@ -185,7 +104,7 @@ def test_expunge_csv_modulairx_cloudapi(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair-x/MOD-X-00993-cloudapi-file1.csv"), - ] + ], catch_exceptions=False ) # did it succeed? @@ -200,3 +119,198 @@ def test_expunge_csv_modulairx_cloudapi(self): # is it a csv? self.assertEqual(p.suffix, ".csv") + +def test_expunge_dataframe_noop(): + """Ensure expunge_dataframe doesn't change data that doesn't meet new flags.""" + sample_df = pd.DataFrame( + [ + { + "timestamp": pd.to_datetime("2023-03-10T17:41:24Z"), + "rh": 77.6, + "temp": -10.4, + "pm1_env": 10.6, + "pm25_env": 25.0, + "pm10_env": 32.4, + "neph_bin0": 1590.375, + "co_we": 758.8, + "co_ae": 656.8, + "co_diff": 102.0, + "flag": 0, + "sn" : 'MOD-00001', + }, + ], + ) + + flagged = expunge_dataframe(sample_df) + + assert_frame_equal(flagged, sample_df) + + +def test_expunge_dataframe_irrelevant_flag(): + """Ensure expunge_dataframe doesn't change data with irrelevant flags. + + In this case, the flags don't apply to the current columns. + """ + sample_df = pd.DataFrame( + [ + { + "timestamp": pd.to_datetime("2023-03-10T17:41:24Z"), + "rh": 77.6, + "temp": -10.4, + "pm1_env": 10.6, + "pm25_env": 25.0, + "pm10_env": 32.4, + "neph_bin0": 1590.375, + "co_we": 758.8, + "co_ae": 656.8, + "co_diff": 102.0, + "flag": 128, # FLAG_O3 -- doesn't change any current values + "sn" : 'MOD-00001', + }, + ], + ) + + flagged = expunge_dataframe(sample_df) + + assert_frame_equal(flagged, sample_df) + + +def test_expunge_dataframe_expunges_rht(): + """Ensure expunge_dataframe expunges values correctly for FLAG_RHT.""" + sample_df = pd.DataFrame( + [ + { + "timestamp": pd.to_datetime("2023-03-10T17:41:24Z"), + "rh": 20000, # too high! + "temp": -10.4, + "pm1_env": 10.6, + "pm25_env": 25.0, + "pm10_env": 32.4, + "neph_bin0": 1590.375, + "co_we": 758.8, + "co_ae": 656.8, + "co_diff": 102.0, + "flag": 8, # FLAG_RHT + "sn" : 'MOD-00001', + }, + ], + ) + + flagged = expunge_dataframe(sample_df) + + # Ensure rh and temp are expunged. + expected = sample_df.copy().assign(rh=np.nan, temp=np.nan) + assert_frame_equal(flagged, expected, check_dtype=False) + + +def test_expunge_dataframe_expunges_startup(): + """Ensure expunge_dataframe expunges all columns correctly for FLAG_STARTUP.""" + sample_df = pd.DataFrame( + [ + { + "timestamp": pd.to_datetime("2023-03-10T17:41:24Z"), + "sn": "MOD-12345", + "rh": 20000., # too high! + "temp": -10.4, + "pm1_env": 10.6, + "pm25_env": 25.0, + "pm10_env": 32.4, + "neph_bin0": 1590.375, + "co_we": 758.8, + "co_ae": 656.8, + "co_diff": 102.0, + "flag": 1, # FLAG_STARTUP + }, + ], + ) + + flagged = expunge_dataframe(sample_df) + + # Ensure most columns are expunged but key metadata is unaffected. + expected = pd.DataFrame( + [ + { + "timestamp": pd.to_datetime("2023-03-10T17:41:24Z"), + "sn": "MOD-12345", + "rh": np.nan, + "temp": np.nan, + "pm1_env": np.nan, + "pm25_env": np.nan, + "pm10_env": np.nan, + "neph_bin0": np.nan, + "co_we": np.nan, + "co_ae": np.nan, + "co_diff": np.nan, + "flag": 1, # FLAG_STARTUP + }, + ], + ) + assert_frame_equal(flagged, expected) + + +def test_flag_summary_basic(): + """Simple check to confirm flag_summary correctness.""" + sample_df = pd.DataFrame( + [ + { + "flag": 1 | 4, # FLAG_STARTUP | FLAG_NEPH + }, + { + "flag": 1 | 32 | 64, # FLAG_STARTUP | FLAG_NO | FLAG_NO2, + }, + { + "flag": 0, + }, + ], + ) + summary = flag_summary(sample_df) + + expected = pd.DataFrame( + { + "FLAG VALUE": { + "FLAG_BAT": 2048, + "FLAG_CO": 16, + "FLAG_CO2": 256, + "FLAG_H2S": 1024, + "FLAG_NEPH": 4, + "FLAG_NO": 32, + "FLAG_NO2": 64, + "FLAG_O3": 128, + "FLAG_OPC": 2, + "FLAG_RHTP": 8, + "FLAG_SO2": 512, + "FLAG_STARTUP": 1, + }, + "# OCCURENCES": { + "FLAG_BAT": 0, + "FLAG_CO": 0, + "FLAG_CO2": 0, + "FLAG_H2S": 0, + "FLAG_NEPH": 1, + "FLAG_NO": 1, + "FLAG_NO2": 1, + "FLAG_O3": 0, + "FLAG_OPC": 0, + "FLAG_RHTP": 0, + "FLAG_SO2": 0, + "FLAG_STARTUP": 2, + }, + "% DATA": { + "FLAG_BAT": "0.0", + "FLAG_CO": "0.0", + "FLAG_CO2": "0.0", + "FLAG_H2S": "0.0", + "FLAG_NEPH": "33.3", + "FLAG_NO": "33.3", + "FLAG_NO2": "33.3", + "FLAG_O3": "0.0", + "FLAG_OPC": "0.0", + "FLAG_RHTP": "0.0", + "FLAG_SO2": "0.0", + "FLAG_STARTUP": "66.7", + }, + }, + ) + expected.index.name = "FLAG" + + assert_frame_equal(summary.sort_index(), expected.sort_index()) From 2e678574055a09cc3753c81b5c9ca1aacdba3444 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Tue, 30 Jun 2026 15:05:02 -0400 Subject: [PATCH 20/23] Remove feather tests and fix --wind unpack error in resample CLI --- quantaq_cli/console/__init__.py | 8 +++++- quantaq_cli/console/commands/resample.py | 6 ++--- tests/test_resample.py | 32 ------------------------ 3 files changed, 10 insertions(+), 36 deletions(-) diff --git a/quantaq_cli/console/__init__.py b/quantaq_cli/console/__init__.py index 6c26008..9733774 100644 --- a/quantaq_cli/console/__init__.py +++ b/quantaq_cli/console/__init__.py @@ -52,7 +52,13 @@ def merge(files, tscol, output, verbose, **kwargs): @click.option("-o", "--output", default="output.csv", help="The filepath where you would like to save the file", type=str) @click.option("--on", default="timestamp", help="Name of the datetime column to resample over.", type=str) @click.option("--by", default=None, help="Optional column(s) to group by first") -@click.option("--wind", default=("wx_u", "wx_v", "wx_ws", "wx_wd"), help="``(u, v, speed, direction)`` column names") +@click.option( + "--wind", + nargs=4, + type=str, + default=("wx_u", "wx_v", "wx_ws", "wx_wd"), + help="(u, v, speed, direction) column names", +) @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("-v", "--verbose", is_flag=True, help="Enable verbose mode (debugging)") diff --git a/quantaq_cli/console/commands/resample.py b/quantaq_cli/console/commands/resample.py index e15c565..228c0fc 100644 --- a/quantaq_cli/console/commands/resample.py +++ b/quantaq_cli/console/commands/resample.py @@ -5,7 +5,7 @@ from loguru import logger import numpy as np import pandas as pd -from pandas.api.types import is_numeric_dtype +from pandas.api.types import is_numeric_dtype, is_datetime64_any_dtype from quantaq_cli.exceptions import InvalidFileExtension from quantaq_cli.utilities import safe_load @@ -89,8 +89,8 @@ def resample_dataframe( Returns: A new frame with ``on`` (and any ``by`` keys) as columns. """ - if type(df[on]) != np.datetime64: - df[on] = df[on].map(pd.to_datetime) + 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 []) diff --git a/tests/test_resample.py b/tests/test_resample.py index 4c84c00..8010c76 100644 --- a/tests/test_resample.py +++ b/tests/test_resample.py @@ -56,38 +56,6 @@ def test_resample_files_csv(self): self.assertEqual((idx[1] - idx[0]) / np.timedelta64(1, 's'), 600.0) - def test_resample_files_feather(self): - runner = CliRunner() - result = runner.invoke(resample, - [ - "-o", - os.path.join(self.test_dir, "output.feather"), - "-v", - os.path.join(self.test_files_dir, "arisense/ref/ref.csv"), - "10min", - ], catch_exceptions=False - ) - - # did it succeed? - self.assertEqual(result.exit_code, 0) - - # did it output the correct text? - self.assertTrue("Saving file" in result.output) - - # make sure the file exists - p = Path(self.test_dir + "/output.feather") - self.assertTrue(p.exists()) - - # is it a csv? - self.assertEqual(p.suffix, ".feather") - - # are the number of lines correct? - df = pd.read_feather(os.path.join(self.test_dir, "output.feather")) - - idx = df.timestamp.values - - self.assertEqual((idx[1] - idx[0]) / np.timedelta64(1, 's'), 600.0) - def test_resample_modulair_db(self): runner = CliRunner() result = runner.invoke(resample, From eb2c043351ea3cf0cf4cf24848be8610aa2d4f21 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Wed, 1 Jul 2026 10:55:14 -0400 Subject: [PATCH 21/23] Remove data model and data source as required args to flag_command --- quantaq_cli/console/__init__.py | 12 ++---------- 1 file changed, 2 insertions(+), 10 deletions(-) diff --git a/quantaq_cli/console/__init__.py b/quantaq_cli/console/__init__.py index 9733774..8612db9 100644 --- a/quantaq_cli/console/__init__.py +++ b/quantaq_cli/console/__init__.py @@ -88,21 +88,13 @@ def expunge(file, dry_run, table, output, flag, verbose, model, **kwargs): @click.command("flag", short_help="flag data based on specific criteria") @click.argument("file", nargs=1, type=click.Path()) -@click.argument("model", type=click.Choice(SUPPORTED_MODELS)) -@click.argument("source", type=click.Choice(SUPPORTED_SOURCES)) @click.option("-o", "--output", default="output.csv", help="The filepath where you would like to save the file", type=str) @click.option("-v", "--verbose", is_flag=True, help="Enable verbose mode (debugging)") def flag(file, output, verbose, model, source, **kwargs): - """Reflag a data file based on data model and data source. - - Three arguments are required: - 1. FILE -> the path to the file of interest - 2. MODEL -> the device model type (one of SUPPORTED_MODELS) - 3. SOURCE -> the data source (one of SUPPORTED_SOURCES) - """ + """Reflag a data file""" from .commands.flag import flag_command - flag_command(file, output, model=model, source=source, verbose=verbose) + flag_command(file, output, verbose=verbose) @click.command("clean") From 04d34eda71a0490aa179178df4c663fe78a6bbd1 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Wed, 1 Jul 2026 11:02:04 -0400 Subject: [PATCH 22/23] Remove data model and data source as required args in cli logic and tests --- quantaq_cli/console/__init__.py | 4 +--- tests/test_flag.py | 4 ---- 2 files changed, 1 insertion(+), 7 deletions(-) diff --git a/quantaq_cli/console/__init__.py b/quantaq_cli/console/__init__.py index 8612db9..fa8b9ec 100644 --- a/quantaq_cli/console/__init__.py +++ b/quantaq_cli/console/__init__.py @@ -75,15 +75,13 @@ def resample(file, rule, output, verbose, **kwargs): @click.option("-d", "--dry-run", is_flag=True, help="Print table to screen and bypass file save") @click.option("-t", "--table", is_flag=True, help="Print as a dict instead of as a table") @click.option("-o", "--output", default="output.csv", help="The filepath where you would like to save the file", type=str) -@click.option("-f", "--flag", default="flag", help="The name of the flag column", type=str) @click.option("-v", "--verbose", is_flag=True, help="Enable verbose mode (debugging)") -@click.option("-m", "--model", default="modulair_pm", help="The device model type. One of {}".format(SUPPORTED_MODELS)) def expunge(file, dry_run, table, output, flag, verbose, model, **kwargs): """Expunge (NaN flagged values) FILE and save to OUTPUT. """ from .commands.expunge import expunge_command - expunge_command(file, output, flagcol=flag, dry_run=dry_run, verbose=verbose, model=model, table=table, **kwargs) + expunge_command(file, output, dry_run=dry_run, verbose=verbose, table=table, **kwargs) @click.command("flag", short_help="flag data based on specific criteria") diff --git a/tests/test_flag.py b/tests/test_flag.py index bf8be45..d49f7f1 100644 --- a/tests/test_flag.py +++ b/tests/test_flag.py @@ -24,8 +24,6 @@ def test_flag_files_modulair_db(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair/MOD-00014-db-raw.csv"), - "modulair", - "database" ], catch_exceptions=False ) @@ -76,8 +74,6 @@ def test_flag_files_modulairx_cloudapi(self): os.path.join(self.test_dir, "output.csv"), "-v", os.path.join(self.test_files_dir, "modulair-x/MOD-X-00993-cloudapi-file1.csv"), - "modulair-x", - "cloudapi" ], catch_exceptions=False ) From 9e729a7b52f67d583fe319183d3065973ca16eb4 Mon Sep 17 00:00:00 2001 From: Emmie Le Roy <47359313+emmleroy@users.noreply.github.com> Date: Wed, 1 Jul 2026 11:07:03 -0400 Subject: [PATCH 23/23] Fix cli args again --- quantaq_cli/console/__init__.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/quantaq_cli/console/__init__.py b/quantaq_cli/console/__init__.py index fa8b9ec..8b2fad6 100644 --- a/quantaq_cli/console/__init__.py +++ b/quantaq_cli/console/__init__.py @@ -76,19 +76,19 @@ def resample(file, rule, output, verbose, **kwargs): @click.option("-t", "--table", is_flag=True, help="Print as a dict instead of as a table") @click.option("-o", "--output", default="output.csv", help="The filepath where you would like to save the file", type=str) @click.option("-v", "--verbose", is_flag=True, help="Enable verbose mode (debugging)") -def expunge(file, dry_run, table, output, flag, verbose, model, **kwargs): +def expunge(file, dry_run, table, output, verbose): """Expunge (NaN flagged values) FILE and save to OUTPUT. """ from .commands.expunge import expunge_command - expunge_command(file, output, dry_run=dry_run, verbose=verbose, table=table, **kwargs) + expunge_command(file, output, dry_run=dry_run, verbose=verbose, table=table) @click.command("flag", short_help="flag data based on specific criteria") @click.argument("file", nargs=1, type=click.Path()) @click.option("-o", "--output", default="output.csv", help="The filepath where you would like to save the file", type=str) @click.option("-v", "--verbose", is_flag=True, help="Enable verbose mode (debugging)") -def flag(file, output, verbose, model, source, **kwargs): +def flag(file, output, verbose): """Reflag a data file""" from .commands.flag import flag_command