diff --git a/quantaq_cli/cli.py b/quantaq_cli/cli.py index 1177fc2..5fb37a6 100644 --- a/quantaq_cli/cli.py +++ b/quantaq_cli/cli.py @@ -1,5 +1,6 @@ from pathlib import Path -import pkg_resources +from pathlib import Path +from importlib.metadata import version import rich_click as click from loguru import logger @@ -14,8 +15,7 @@ CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help']) -__version__ = pkg_resources.get_distribution('quantaq_cli').version - +__version__ = version("quantaq_cli") @click.group(context_settings=CONTEXT_SETTINGS) @click.version_option(__version__) diff --git a/quantaq_cli/schema.py b/quantaq_cli/schema.py index 9bf0e51..06c6ce6 100644 --- a/quantaq_cli/schema.py +++ b/quantaq_cli/schema.py @@ -1,41 +1,43 @@ import contextlib +import json from loguru import logger import numpy as np +import pandas as pd import pandera.pandas as pa # Default dtypes COLUMN_DEFINITIONS = [ # --- OPC colunms --- - ('opc_bin0', np.float64), - ('opc_bin1', np.float64), - ('opc_bin2', np.float64), - ('opc_bin3', np.float64), - ('opc_bin4', np.float64), - ('opc_bin5', np.float64), - ('opc_bin6', np.float64), - ('opc_bin7', np.float64), - ('opc_bin8', np.float64), - ('opc_bin9', np.float64), - ('opc_bin10', np.float64), - ('opc_bin11', np.float64), - ('opc_bin12', np.float64), - ('opc_bin13', np.float64), - ('opc_bin14', np.float64), - ('opc_bin15', np.float64), - ('opc_bin16', np.float64), - ('opc_bin17', np.float64), - ('opc_bin18', np.float64), - ('opc_bin19', np.float64), - ('opc_bin20', np.float64), - ('opc_bin21', np.float64), - ('opc_bin22', np.float64), - ('opc_bin23', np.float64), - ('opc_bin1MToF', np.float64), - ('opc_bin3MToF', np.float64), - ('opc_bin5MToF', np.float64), - ('opc_bin7MToF', np.float64), + ('bin0', np.float64), + ('bin1', np.float64), + ('bin2', np.float64), + ('bin3', np.float64), + ('bin4', np.float64), + ('bin5', np.float64), + ('bin6', np.float64), + ('bin7', np.float64), + ('bin8', np.float64), + ('bin9', np.float64), + ('bin10', np.float64), + ('bin11', np.float64), + ('bin12', np.float64), + ('bin13', np.float64), + ('bin14', np.float64), + ('bin15', np.float64), + ('bin16', np.float64), + ('bin17', np.float64), + ('bin18', np.float64), + ('bin19', np.float64), + ('bin20', np.float64), + ('bin21', np.float64), + ('bin22', np.float64), + ('bin23', np.float64), + ('bin1MToF', np.float64), + ('bin3MToF', np.float64), + ('bin5MToF', np.float64), + ('bin7MToF', np.float64), ('opc_temp', np.float64), ('opc_rh', np.float64), ('opc_pm1', np.float64), @@ -116,6 +118,14 @@ ('iteration', np.int16), ('dd_measurement_state', np.float64), # needs to be nullable for older data ('dd_operating_state', np.float64), # needs to be nullable for older data + + # -- Wind columns -- + ('wx_u', np.float64), + ('wx_u', np.float64), + ('wx_wd', np.float64), + ('wx_ws', np.float64), + ('wx_ws_scalar', np.float64), + ] STATIC_COLUMN_RENAMES = { @@ -155,11 +165,20 @@ "rh": "sample_rh", # --- Device / metadata columns --- - "operating_state": "dd_operating_state" + "operating_state": "dd_operating_state", + + # -- Wind columns --- + "w_u": "wx_u", + "w_v": "wx_v", + "u": "wx_u", + "v": "wx_v", + "wd": "wx_wd", + "ws_vector": "wx_ws", + "ws" : "wx_ws_scalar" } # Prefixes for unstandardized column names -COLUMN_RENAME_PREFIXES = ("bin", "opc.bin", "met.", "gases.", "geo.") +COLUMN_RENAME_PREFIXES = ("opc.bin", "met.", "gases.", "geo.") def validate_schema(df, nullable=True, required=False, coerce_dtypes=True, coerce_rename=True): @@ -179,16 +198,23 @@ def validate_schema(df, nullable=True, required=False, coerce_dtypes=True, coerc Returns: df (pd.DataFrame): the DataFrame with validated dtypes and column names """ - df = df.copy() columns = { - name: pa.Column(dtype, nullable=nullable, required=required) + name: pa.Column(dtype, nullable=nullable, required=required, coerce=coerce_dtypes) for name, dtype in COLUMN_DEFINITIONS } + expected_dtypes = dict(COLUMN_DEFINITIONS) legacy_names = set(STATIC_COLUMN_RENAMES.keys()) prefix_patterns = COLUMN_RENAME_PREFIXES + def _wrong_dtype_columns(df): + """Return columns whose dtype doesn't match their expected dtype.""" + return sorted( + col for col in df.columns + if col in expected_dtypes and df[col].dtype != np.dtype(expected_dtypes[col]) + ) + def _has_legacy_column_names(df): """Return True if any legacy names/prefixes remain.""" has_legacy_name = bool(legacy_names.intersection(df.columns)) @@ -201,6 +227,40 @@ def _no_legacy_column_names(df): """Pandera check: True if no legacy names/prefixes remain.""" return not _has_legacy_column_names(df) + def _expected_rename(col): + """Predict what standardize_columns() would rename this column to.""" + if col in STATIC_COLUMN_RENAMES: + return STATIC_COLUMN_RENAMES[col] + if col.startswith("opc.bin"): + return col.replace("opc.", "") + if col.startswith("met."): + return col.replace("met.", "") + if col.startswith("gases."): + return "ox_diff" if col == "gases.o3.diff" else col.removeprefix("gases.").replace(".", "_") + if col.startswith("geo."): + return col.removeprefix("geo.") + return "unknown rename rule" + + def _log_failures(df): + """Log one line per failure: dtype mismatches and legacy column names.""" + wrong_dtype_cols = _wrong_dtype_columns(df) + for col in wrong_dtype_cols: + logger.info( + "Schema validation failed - wrong dtype: '{}' should be {}, got {}", + col, np.dtype(expected_dtypes[col]), df[col].dtype, + ) + + if _has_legacy_column_names(df): + legacy_cols = ( + sorted(legacy_names.intersection(df.columns)) + + sorted(col for col in df.columns for p in prefix_patterns if col.startswith(p)) + ) + for col in legacy_cols: + logger.info( + "Schema validation failed - unstandardized column name: should be '{}', got '{}'", + _expected_rename(col), col, + ) + # index = None means no index is specified # strict = False allows missing and extra columns in the DataFrame schema = pa.DataFrameSchema( @@ -214,34 +274,40 @@ def _no_legacy_column_names(df): ) try: - # print all schema errors instead of raising on the first error schema.validate(df, lazy=True) except pa.errors.SchemaErrors as err: - logger.error("Schema validation failed.") - logger.error("{}", err.failure_cases.to_string()) + logger.bind(schema_errors=err.message).error("Schema validation failed") + _log_failures(df) - # Check the actual condition directly rather than parsing - # failure_cases["check"], since pandera's naming of anonymous - # check functions in that column isn't a stable contract. if coerce_rename and _has_legacy_column_names(df): logger.warning("Standardizing unstandardized column names.") df = standardize_columns(df) - if coerce_dtypes: + if coerce_dtypes and _wrong_dtype_columns(df): logger.warning("Coercing dtypes to expected types.") - dtype_map = { - col: dtype for col, dtype in COLUMN_DEFINITIONS - if col in df.columns - } - df = df.astype(dtype_map) + for col, dtype in COLUMN_DEFINITIONS: + if col not in df.columns: + continue + if np.issubdtype(np.dtype(dtype), np.number): + before_na = df[col].isna().sum() + df[col] = pd.to_numeric(df[col], errors="coerce").astype(dtype) + new_na = df[col].isna().sum() - before_na + if new_na > 0: + logger.warning( + "Coerced {} unparseable value(s) in '{}' to NaN.", + new_na, col, + ) + else: + df[col] = df[col].astype(dtype) if coerce_rename or coerce_dtypes: try: schema.validate(df, lazy=True) logger.info("Schema validation passed after coercion.") except pa.errors.SchemaErrors as second_err: - logger.error("Schema validation still failing after coercion.") - logger.error("{}", second_err.failure_cases.to_string()) + logger.bind(schema_errors=second_err.message).error("Schema validation still failing after coercion.") + _log_failures(df) + raise ValueError(f"Schema validation failed: {json.dumps(second_err.message)}") from second_err return df @@ -273,11 +339,6 @@ def standardize_columns(df): # STATIC_COLUMN_RENAMES dict used by validate_schema column_renames = dict(STATIC_COLUMN_RENAMES) - # bin0 --> opc_bin0, etc.. - for column in df.columns: - if column.startswith("bin"): # opc_bin0, opc_bin23 - column_renames[column] = column.replace("bin", "opc_bin") - # Add diff columns (we minus ae) if they don't already exist if not any('diff' in col for col in df.columns): for pollutant in ("co", "no", "no2"): @@ -291,8 +352,8 @@ def standardize_columns(df): # CloudAPI schema to database schema for column in df.columns: - if column.startswith("opc.bin"): # opc_bin0, opc_bin23 - column_renames[column] = column.replace(".", "_") + if column.startswith("opc.bin"): + column_renames[column] = column.replace("opc.", "") elif column.startswith("met."): column_renames[column] = column.replace("met.", "") elif column.startswith("gases."): diff --git a/quantaq_cli/toolkit/flag.py b/quantaq_cli/toolkit/flag.py index ca6cd29..cd026c5 100644 --- a/quantaq_cli/toolkit/flag.py +++ b/quantaq_cli/toolkit/flag.py @@ -176,21 +176,22 @@ def echo_flag_table(df): table.add_row(*map(str, row)) rich.print(table) -def flag_dataframe(df): +def flag_dataframe(df, reset_flags=True): """Re-flags a DataFrame by iterating through the FLAG_DEFINITIONS and calling the _add_flag() function one-by-one. Args: df (pd.DataFrame): DataFrame to be flagged (or re-flagged). + reset_flags (bool): If True, the 'flag' column is reset to zero. Default is True. Returns: pd.DataFrame: The flagged DataFrame. """ df = df.copy() - # Drop nan flags (could happen after a merge) - if "flag" not in df.columns: - df["flag"] = 0 + # set flags to zero before re-flagging + if reset_flags or 'flag' not in df.columns: + df["flag"] = np.int64(0) elif df["flag"].isna().any(): logger.warning("Dropping {} rows with NaN flags", df["flag"].isna().sum()) df = df.dropna(how='any', subset=["flag"]) @@ -205,11 +206,6 @@ def flag_dataframe(df): # get flag criteria (this also checks if the data source is valid) name_to_criteria = flag_name_to_criteria(source).items() - # create flag column if it doesn't exist - if "flag" not in df.columns: - df["flag"] = np.int64(0) - - # sort the dataframe once before adding flags df = fix_timestamps(df, sort_values=True) diff --git a/quantaq_cli/toolkit/munge.py b/quantaq_cli/toolkit/munge.py index 4a3c4be..c712bf5 100644 --- a/quantaq_cli/toolkit/munge.py +++ b/quantaq_cli/toolkit/munge.py @@ -24,6 +24,6 @@ def clean_dataframe(df, coerce_dtypes=True, coerce_rename=True): # Validate the schema # by default this also coerces dtypes and column names, but that can be overrided - df = validate_schema(df, coerce_dtypes=True, coerce_rename=True) + df = validate_schema(df, coerce_dtypes=coerce_dtypes, coerce_rename=coerce_rename) return df diff --git a/quantaq_cli/toolkit/resample.py b/quantaq_cli/toolkit/resample.py index 75fddf3..94b56b0 100644 --- a/quantaq_cli/toolkit/resample.py +++ b/quantaq_cli/toolkit/resample.py @@ -118,6 +118,14 @@ def _resampler(frame): out = clean_agg.where(has_clean, all_agg) out["flag"] = np.where(has_clean, 0, flag_or) + + # drop empty bins + is_empty_bin = flag_or.isna() + if is_empty_bin.any(): + logger.debug("Dropping {} empty resample bin(s)", int(is_empty_bin.sum())) + out = out.loc[~is_empty_bin] + + out["flag"] = out["flag"].astype("int64") return out.reset_index() @@ -205,7 +213,7 @@ def resample_dataframe( if have_uv and (df[u_col].isna().all() or df[v_col].isna().all()): logger.debug( "All wind components contain NaNs ({}: {}, {}: {}); " - "Deriving them from the averaged u/v components", + "cannot vector-average from u/v - will derive from speed/direction instead", u_col, int(df[u_col].isna().sum()), v_col, int(df[v_col].isna().sum()), ) diff --git a/quantaq_cli/variables.py b/quantaq_cli/variables.py index aac610f..4ff0d9e 100644 --- a/quantaq_cli/variables.py +++ b/quantaq_cli/variables.py @@ -8,10 +8,10 @@ # Columns to Nan -- only for standardized schema types _OPC_COLUMNS = [ - "opc_bin0", "opc_bin1", "opc_bin2", "opc_bin3", "opc_bin4", "opc_bin5", "opc_bin6", "opc_bin7", "opc_bin8", "opc_bin9", - "opc_bin10", "opc_bin11", "opc_bin12", "opc_bin13", "opc_bin14", "opc_bin15", "opc_bin16", "opc_bin17", "opc_bin18", - "opc_bin19", "opc_bin20", "opc_bin21", "opc_bin22", "opc_bin23", - "opc_bin1MToF", "opc_bin3MToF", "opc_bin5MToF", "opc_bin7MToF", + "bin0", "bin1", "bin2", "bin3", "bin4", "bin5", "bin6", "bin7", "bin8", "bin9", + "bin10", "bin11", "bin12", "bin13", "bin14", "bin15", "bin16", "bin17", "bin18", + "bin19", "bin20", "bin21", "bin22", "bin23", + "bin1MToF", "bin3MToF", "bin5MToF", "bin7MToF", "opc_sample_period", "opc_sample_flow", "opc_pm1", "opc_pm25", "opc_pm10", "opc_laser_status", "opc_temp", "opc_rh", @@ -135,19 +135,29 @@ # Check: ratio between the OPC and nephelometer is within spec Multiple( criteria=( - Single(column="opc_bin0", op=">=", value=10.0), + Single(column="bin0", op=">=", value=10.0), Ratio(column_numerator="neph_bin0", - column_denominator="opc_bin0", + column_denominator="bin0", op=">", value=2000.0), ), logical_operator="AND", ), + # Check: ensure that the neph and opc are not both reading 0s + # (this happens when an entire 1-min window has FLAG_OPC and FLAG_NEPH + # set by the firmware at 5-s intervals) + Multiple( + criteria=( + Single(column="bin0", op="==", value=0.0), + Single(column="neph_bin0", op="==", value=0.0) + ), + logical_operator="AND", + ), ], "FLAG_NEPH": [ # Check 1: ensure that the neph isn't reading 0's when it shouldn't be Multiple( criteria=( - Single(column="opc_bin0", op=">=", value=10.0), + Single(column="bin0", op=">=", value=10.0), Single(column="neph_bin0", op="==", value=0.0) ), logical_operator="AND", @@ -157,11 +167,21 @@ criteria=( Single(column="neph_bin0", op=">=", value=10.0), Ratio(column_numerator="neph_bin0", - column_denominator="opc_bin0", + column_denominator="bin0", op=">", value=2000.0), ), logical_operator="AND", ), + # Check 3: ensure that the neph and opc are not both reading 0s + # (this happens when an entire 1-min window has FLAG_OPC and FLAG_NEPH + # set by the firmware at 5-s intervals) + Multiple( + criteria=( + Single(column="bin0", op="==", value=0.0), + Single(column="neph_bin0", op="==", value=0.0) + ), + logical_operator="AND", + ), ], } @@ -185,7 +205,7 @@ criteria=( Single(column="neph_bin0", op=">=", value=10.0), Ratio(column_numerator="neph_bin0", - column_denominator="opc_bin0", + column_denominator="bin0", op=">", value=2000.0), ), logical_operator="AND", diff --git a/tests/test_flag.py b/tests/test_flag.py index c9c0ac0..c6ecf41 100644 --- a/tests/test_flag.py +++ b/tests/test_flag.py @@ -144,57 +144,3 @@ def test_flag_files_modulair_db_parquet(self): # is it a parquet? self.assertEqual(p.suffix, ".parquet") - -def test_flag_dataframe_nanflag(): - sample_df = pd.DataFrame( - [ - { - "timestamp": pd.to_datetime("2023-03-10T17:41:24Z"), - "sample_rh": 77.6, - "sample_temp": -10.4, - "neph_pm1_env": 10.6, - "neph_pm25_env": 25.0, - "neph_pm10_env": 32.4, - "neph_bin0": 1590.375, - "co_we": 758.8, - "co_ae": 656.8, - "co_diff": 102.0, - "flag": np.nan, # this row should be dropped - "sn": "MOD-00001", - }, - { - "timestamp": pd.to_datetime("2023-03-10T17:42:24Z"), - "sample_rh": 77.9, - "sample_temp": -10.3, - "neph_pm1_env": 10.4, - "neph_pm25_env": 24.8, - "neph_pm10_env": 32.1, - "neph_bin0": 1588.0, - "co_we": 759.1, - "co_ae": 657.0, - "co_diff": 102.1, - "flag": 0., # this row should survive - "sn": "MOD-00001", - }, - { - "timestamp": pd.to_datetime("2023-03-10T17:43:24Z"), - "sample_rh": 78.1, - "sample_temp": -10.2, - "neph_pm1_env": 10.5, - "neph_pm25_env": 24.9, - "neph_pm10_env": 32.0, - "neph_bin0": 1589.0, - "co_we": 759.3, - "co_ae": 657.2, - "co_diff": 102.0, - "flag": 0., # this row should survive - "sn": "MOD-00001", - }, - ], - ) - - expected = sample_df.dropna(how="any") - - flagged = flag_dataframe(sample_df) - - assert_frame_equal(flagged, expected, check_dtype=False) diff --git a/tests/test_schema.py b/tests/test_schema.py index 121be84..b84db7a 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -85,34 +85,34 @@ def test_schema_migration_modpm_rawsd(self): 'sample_rh', 'sample_temp', 'sample_pres', - 'opc_bin0', - 'opc_bin1', - 'opc_bin2', - 'opc_bin3', - 'opc_bin4', - 'opc_bin5', - 'opc_bin6', - 'opc_bin7', - 'opc_bin8', - 'opc_bin9', - 'opc_bin10', - 'opc_bin11', - 'opc_bin12', - 'opc_bin13', - 'opc_bin14', - 'opc_bin15', - 'opc_bin16', - 'opc_bin17', - 'opc_bin18', - 'opc_bin19', - 'opc_bin20', - 'opc_bin21', - 'opc_bin22', - 'opc_bin23', - 'opc_bin1MToF', - 'opc_bin3MToF', - 'opc_bin5MToF', - 'opc_bin7MToF', + 'bin0', + 'bin1', + 'bin2', + 'bin3', + 'bin4', + 'bin5', + 'bin6', + 'bin7', + 'bin8', + 'bin9', + 'bin10', + 'bin11', + 'bin12', + 'bin13', + 'bin14', + 'bin15', + 'bin16', + 'bin17', + 'bin18', + 'bin19', + 'bin20', + 'bin21', + 'bin22', + 'bin23', + 'bin1MToF', + 'bin3MToF', + 'bin5MToF', + 'bin7MToF', 'opc_sample_period', 'opc_sample_flow', 'opc_temp', @@ -236,30 +236,30 @@ def test_schema_migration_modx_api(self): 'neph_pm1_env', 'neph_pm10_env', 'neph_pm25_env', - 'opc_bin0', - 'opc_bin1', - 'opc_bin10', - 'opc_bin11', - 'opc_bin12', - 'opc_bin13', - 'opc_bin14', - 'opc_bin15', - 'opc_bin16', - 'opc_bin17', - 'opc_bin18', - 'opc_bin19', - 'opc_bin2', - 'opc_bin20', - 'opc_bin21', - 'opc_bin22', - 'opc_bin23', - 'opc_bin3', - 'opc_bin4', - 'opc_bin5', - 'opc_bin6', - 'opc_bin7', - 'opc_bin8', - 'opc_bin9', + 'bin0', + 'bin1', + 'bin10', + 'bin11', + 'bin12', + 'bin13', + 'bin14', + 'bin15', + 'bin16', + 'bin17', + 'bin18', + 'bin19', + 'bin2', + 'bin20', + 'bin21', + 'bin22', + 'bin23', + 'bin3', + 'bin4', + 'bin5', + 'bin6', + 'bin7', + 'bin8', + 'bin9', 'opc_pm1', 'opc_pm10', 'opc_pm25'