MeteoScreening from database (influxdb)

Modified

2 August 2026

*****Notebook version: 9 (15 Jul 2026) Author**: Lukas Hörtnagl (holukas@ethz.ch)

ℹ️ About this notebook

Download raw meteo data from the InfluxDB database, quality-screen and correct it on the high-resolution data, resample to 30MIN, and upload the result back to the database. Screening uses StepwiseMeteoScreeningDb from diive; download and upload use diive’s in-house InfluxDB engine (InfluxIO, in diive/core/io/db/influx), needing only uv sync --group db.

Flow: download (InfluxIO) → screen on high-res data (diive) → resample to 30MIN → upload.

Outlier detection is stepwise: run a test, inspect its preview plot, then commit it with mscr.addflag(). Re-run with different parameters as often as you like before committing. Run only the tests a variable actually needs. At the end all committed flags are aggregated into one overall quality flag QCF.

⏱️ Timestamp convention (important)

The database stores every timestamp in UTC and as TIMESTAMP_END (the stamp marks the end of the averaging interval). Getting the timestamp right is the one thing that must not go wrong, because the day/night split during screening — and the value written back to the DB — both depend on it.

The single knob is TIMEZONE_OFFSET_TO_UTC_HOURS (set in User settings). It is applied identically on download and on upload. Here is the full round-trip, all handled for you:

Stage	Timezone	Convention	Done by
Database	UTC	`TIMESTAMP_END`	InfluxDB
After `dbc.download(..., timezone_offset_to_utc_hours=N)`	local (UTC+N)	`TIMESTAMP_END`	InfluxIO
During screening	local	`TIMESTAMP_MIDDLE` (converted internally)	`StepwiseMeteoScreeningDb`
After `mscr.resample()`	local	back to `TIMESTAMP_END`	diive
After `dbc.upload_singlevar(..., timezone_offset_to_utc_hours=N)`	UTC	`TIMESTAMP_END`	InfluxIO

So screening runs on correct local middle-of-period timestamps, and the resampled value lands on the correct UTC end-of-period stamp in the DB. TIMEZONE_OFFSET_TO_UTC_HOURS must match the timezone the raw data was logged in (e.g. 1 for CET winter time) and must be the same value everywhere. This notebook prints the timestamps right after download and right after the verification download so you can confirm they look correct.

✏️ User settings (please adjust)

Adjust these before running. What each setting means:

Site - SITE, SITE_LAT, SITE_LON: site ID and coordinates. The coordinates set the day/night split used during screening.

Variables to screen - FIELDS: variable name(s) exactly as stored in the database (the InfluxDB _field). Multiple allowed, e.g. ['TA_T1_20_1', 'TA_NABEL_T1_35_1']. - MEASUREMENT: exactly one measurement grouping the variables, e.g. TA (air temperature), SW (short-wave radiation), SWC (soil water content).

Time range to screen - START: first timestamp to screen — is included. - STOP: upper bound — is not included.

Data settings - TIMEZONE_OFFSET_TO_UTC_HOURS: the critical timestamp knob — see the Timestamp convention section above. Must match how the raw data was logged (e.g. 1 for CET winter time). - DATA_VERSION: the source data version in the database (raw). - DIRCONF: local folder holding the database connection config.

Resampling - RESAMPLING_FREQ: the screened high-res data is resampled to this frequency. - RESAMPLING_AGG: aggregation used when resampling: 'mean' or 'sum'. ⚠️ For precipitation use 'sum'.

Parameter help - SHOW_PARAM_HELP: set True to print the full docstring (all parameters) of each screening method right before it runs. Leave False for a clean, concise notebook.

# --- Site ---
SITE = 'ch-hon'
SITE_LAT = 47.41887  # CH-HON
SITE_LON = 8.491318  # CH-HON

# --- Variables to screen ---
FIELDS = [
    'TA_T1_4_2',
]
MEASUREMENT = 'TA'

# --- Time range to screen ---
START = '2026-03-01 00:00:01'  # included
STOP = '2026-04-01 00:00:01'  # not included

# --- Data settings ---
DATA_VERSION = 'raw'
TIMEZONE_OFFSET_TO_UTC_HOURS = 1  # UTC+01:00 (CET, winter time). Must match how the raw data was logged.
DIRCONF = r'F:\dev\poet\configs'  # <-- set to your config folder
# DIRCONF = r'P:\Flux\RDS_calculations\_scripts\_configs\configs'

# --- Resampling ---
RESAMPLING_FREQ = '30min'  # screened high-res data is resampled to this frequency
RESAMPLING_AGG = 'mean'  # 'mean' or 'sum' (use 'sum' for precipitation)

# --- Parameter help ---
SHOW_PARAM_HELP = False

🤖 Auto settings

Buckets (do not adjust)

BUCKET_RAW = f'{SITE}_raw'  # source bucket, e.g. 'ch-hon_raw'
BUCKET_PROCESSED = f'{SITE}_processed'  # destination bucket, e.g. 'ch-hon_processed'
print(f'Source bucket (raw data):      {BUCKET_RAW}')
print(f'Destination bucket (processed): {BUCKET_PROCESSED}')

Imports

import importlib.metadata
import warnings
from datetime import datetime

import pandas as pd

import diive as dv
from diive.core.io.db.influx import InfluxIO  # diive's in-house InfluxDB engine (needs: uv sync --group db)

warnings.filterwarnings(action='ignore', category=FutureWarning)
warnings.filterwarnings(action='ignore', category=UserWarning)
pd.set_option('display.max_rows', 30)
pd.set_option('display.max_columns', 30)
pd.set_option('display.width', 1000)
print(f"Last run: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"diive v{importlib.metadata.version('diive')}")

⬇️ Download data from database

Connect to database

dbc = InfluxIO(dirconf=DIRCONF)

Optional — list all fields available in the measurement (does not check the selected time range):

# display(dbc.show_fields_in_measurement(bucket=BUCKET_RAW, measurement=MEASUREMENT))

Download

Returns three objects: - data_simple: high-res time series, one column per variable (nice to look at). - data_detailed: dict {varname: DataFrame} with each variable’s time series and its database tags — this is what the screening consumes. - assigned_measurements: the auto-detected measurement per variable (a sanity check).

%%time
data_simple, data_detailed, assigned_measurements = dbc.download(
    bucket=BUCKET_RAW,
    measurements=[MEASUREMENT],
    fields=FIELDS,
    start=START,
    stop=STOP,
    timezone_offset_to_utc_hours=TIMEZONE_OFFSET_TO_UTC_HOURS,
    data_version=DATA_VERSION,
)

Inspect downloaded data

data_simple

assigned_measurements

Drop any requested variable that has no data in this period:

vars_not_available = [v for v in FIELDS if v not in data_detailed.keys()]
for rem in vars_not_available:
    FIELDS.remove(rem)
    print(f'Removed {rem} from FIELDS (no data in this period).')
print(f'Data available for: {list(data_detailed.keys())}')

Verify download timestamps

Confirm the timestamps look right: they should be in local time (UTC+TIMEZONE_OFFSET_TO_UTC_HOURS) and mark the end of each averaging interval. The DB itself stores UTC — InfluxIO applied the offset on download. Eyeball the first/last stamps against the START/STOP you requested.

for v in data_detailed.keys():
    idx = data_detailed[v].index
    print(f'{v}: index name={idx.name!r}, tz={idx.tz}, freq={idx.freqstr}')
    print(f'   first={idx[0]}   last={idx[-1]}')
print(f'\nApplied UTC offset: +{TIMEZONE_OFFSET_TO_UTC_HOURS}h (timestamps above are local time)')

Optional — save the full-resolution raw data to a file:

# data_detailed[FIELDS[0]].to_csv('rawdata_highres.csv')

Plot downloaded high-res data

for varname, frame in data_detailed.items():
    dv.plotting.TimeSeries(series=frame[varname]).plot()

▶️ Start MeteoScreening with `diive`

mscr = dv.qaqc.StepwiseMeteoScreeningDb(
    site=SITE,
    data_detailed=data_detailed,
    fields=FIELDS,
    site_lat=SITE_LAT,
    site_lon=SITE_LON,
    utc_offset=TIMEZONE_OFFSET_TO_UTC_HOURS,
)
mscr.showplot_orig()

🔍 Outlier detection

Run a test → inspect its preview → commit with mscr.addflag(). Only the committed flag of the most recent test is added. Skip any test a variable does not need.

💡 If the data already look clean, you can skip straight to Corrections or Resampling.

mscr.start_outlier_detection()

Plot the current cleaned data at any point during detection:

for key, val in mscr.outlier_detection.items():
    val.showplot_cleaned(interactive=False)

Manual removal

Flag specific timestamps or time ranges for removal — known sensor failures, maintenance windows, etc. Give [start, stop] pairs and/or single timestamps.

if SHOW_PARAM_HELP:
    help(dv.outliers.ManualRemoval)

REMOVE_DATES = [
    ['2024-07-14 07:15:00', '2024-07-19 00:00:15'],  # remove a time range
    # '2022-08-23 11:45:00',                          # remove a single point
]
mscr.flag_manualremoval_test(remove_dates=REMOVE_DATES, showplot=True, verbose=True)

mscr.addflag()

Hampel filter

Sliding-window spike detection using the median absolute deviation (MAD) — robust to spikes already present in the window. Good general-purpose filter. separate_daytime_nighttime=True applies independent thresholds to day and night (n_sigma_daytime / n_sigma_nighttime).

if SHOW_PARAM_HELP:
    help(dv.outliers.Hampel)

mscr.flag_outliers_hampel_test(
    window_length=60 * 24 * 7,  # 7 days of 1-min data used for the MAD window
    n_sigma_daytime=5.5, n_sigma_nighttime=5.5,
    use_differencing=True, separate_daytime_nighttime=True,
    repeat=True, showplot=True, verbose=True,
)

mscr.addflag()

Z-score

Flags values whose deviation from the mean, in standard deviations, exceeds thres_zscore. separate_daytime_nighttime=True computes separate mean/SD for day and night (day/night boundaries come from the site coordinates given at init).

⚠️ Radiation: this easily removes legitimate below-zero nighttime values — those are corrected later (Remove radiation zero offset). Don’t let it remove them here.
⚠️ Relative humidity: likewise it can remove valid values >100%, which are corrected later.

if SHOW_PARAM_HELP:
    help(dv.outliers.zScore)

mscr.flag_outliers_zscore_test(
    thres_zscore=4.5, separate_daytime_nighttime=True,
    repeat=True, showplot=True, verbose=True,
)

mscr.addflag()

Z-score (rolling window)

Like the z-score test but computed in a moving window of winsize records, so it adapts to slow drifts and flags only local spikes.

if SHOW_PARAM_HELP:
    help(dv.outliers.zScoreRolling)

mscr.flag_outliers_zscore_rolling_test(
    thres_zscore=4.5, winsize=1440 * 7,  # 7 days of 1-min data
    repeat=True, showplot=True, verbose=True,
)

mscr.addflag()

Local standard deviation

Flags values beyond n_sd standard deviations of a local rolling window (winsize). separate_daytime_nighttime=True allows different n_sd for day and night. constant_sd=True uses one global SD instead of a per-window SD.

if SHOW_PARAM_HELP:
    help(dv.outliers.LocalSD)

mscr.flag_outliers_localsd_test(
    separate_daytime_nighttime=True, n_sd=5.5, winsize=60 * 24 * 7,
    constant_sd=False, repeat=False, showplot=True, verbose=True,
)

mscr.addflag()

Increments z-score

Flags unrealistic jumps between consecutive records (z-score of the first difference). Catches spikes and steps that absolute-value tests miss.

if SHOW_PARAM_HELP:
    help(dv.outliers.zScoreIncrements)

mscr.flag_outliers_increments_zcore_test(thres_zscore=40, repeat=True, showplot=True, verbose=True)

mscr.addflag()

Local outlier factor

Density-based detection (k-nearest neighbours): flags points isolated relative to their neighbours.

🐌 Do not run on high-resolution data (1S/10S/1MIN) — extremely slow. Fine on 30MIN or coarser.

if SHOW_PARAM_HELP:
    help(dv.outliers.LocalOutlierFactor)

mscr.flag_outliers_lof_test(
    n_neighbors=30, contamination=0.01, separate_daytime_nighttime=False,
    repeat=False, n_jobs=-1, showplot=True, verbose=True,
)

mscr.addflag()

Absolute limits

Flags values outside a fixed physical range [minval, maxval]. separate_daytime_nighttime=True allows separate ranges via daytime_minmax / nighttime_minmax.

if SHOW_PARAM_HELP:
    help(dv.outliers.AbsoluteLimits)

mscr.flag_outliers_abslim_test(minval=-18, maxval=50, showplot=True)

mscr.addflag()

Trim low

Flags values below lower_limit, then flags an equal number of the highest values (symmetric trim). trim_daytime / trim_nighttime restrict which period is trimmed.

if SHOW_PARAM_HELP:
    help(dv.outliers.TrimLow)

mscr.flag_outliers_trim_low_test(
    trim_daytime=False, trim_nighttime=True, lower_limit=10,
    showplot=True, verbose=True,
)

mscr.addflag()

Missing values

Not an outlier test — flags missing records so they are counted in the overall QCF.

mscr.flag_missingvals_test(verbose=True)

Overall quality flag QCF

Aggregate all committed test flags into one overall flag QCF (0 = good, 1 = marginal, 2 = bad) and filter the series. Required before corrections and resampling.

mscr.finalize_outlier_detection()

Reports

mscr.report_outlier_detection_qcf_evolution()

mscr.report_outlier_detection_qcf_flags()

mscr.report_outlier_detection_qcf_series()

Plots

mscr.showplot_outlier_detection_qcf_heatmaps()
# mscr.showplot_outlier_detection_qcf_timeseries()

🔧 Corrections

Applied to the high-res, QCF-filtered data. Run only what applies to the variable.

⚠️ The correction calls below are commented out on purpose — each one mutates the data with example parameters. Edit the parameters for your variable, then uncomment the call to apply it. This keeps a top-to-bottom Run All from silently corrupting data (e.g. clipping air temperature or deleting valid 0 values).

mscr.showplot_cleaned()

Remove radiation zero offset

For radiation (SW_IN, SW_OUT, PPFD_IN, PPFD_OUT): detect the per-day nighttime offset, set night to zero, and correct daytime values by the interpolated slope.

# Uncomment to apply (radiation variables only):
# mscr.correction_remove_nighttime_zero_offset()

Remove relative humidity offset

For RH: remove the offset so values do not exceed 100%.

# Uncomment to apply (RH only):
# mscr.correction_remove_relativehumidity_offset()

Set to max / min threshold

Clip values above (below) a threshold to the threshold value.

# Edit thresholds, then uncomment to apply:
# mscr.correction_setto_max_threshold(threshold=30)
# mscr.correction_setto_min_threshold(threshold=-5)

Set time range(s) to a value

Set all records within the given date range(s) to a constant value.

# Edit DATES/value, then uncomment to apply:
DATES = [
    ['2022-04-01', '2022-04-05'],
    ['2022-09-05', '2022-09-07'],
]
# mscr.correction_setto_value(dates=DATES, value=3.7, verbose=1)
# mscr.showplot_cleaned(interactive=False)

Set exact values to missing

Set records exactly equal to given value(s) to NaN — e.g. a stuck 0. Inspect the most frequent values first (the inspection below is read-only and safe to run).

for ff in mscr.fields:
    vc = mscr.series_hires_cleaned[ff].value_counts()
    print(f'--- {ff} (top 20 of {mscr.series_hires_cleaned[ff].count()} records) ---')
    print(vc.head(20))

# Edit values, then uncomment to apply:
# mscr.correction_set_exact_value_to_missing(values=[0])
# mscr.showplot_cleaned(interactive=False)

📈 Analyses (optional)

Check for timestamp shifts (radiation only)

For SW_IN, SW_OUT, PPFD_IN, PPFD_OUT: compare the measured radiation to potential (clear-sky) radiation. A consistent offset between the two points to a timestamp shift.

# _ = mscr.analysis_potential_radiation_correlation(utc_offset=TIMEZONE_OFFSET_TO_UTC_HOURS, mincorr=0.7, showplot=True)

🔁 Resampling

Resample to 30MIN

Resample the screened high-res series to RESAMPLING_FREQ. The output timestamp is TIMESTAMP_END again (see Timestamp convention), ready for upload.

mscr.resample(to_freqstr=RESAMPLING_FREQ, agg=RESAMPLING_AGG, mincounts_perc=.25)
mscr.showplot_resampled()

Check the resampled time resolution

for v in mscr.resampled_detailed.keys():
    freq = dv.times.DetectFrequency(index=mscr.resampled_detailed[v].index, verbose=True).get()
    status = 'PASSED' if freq == RESAMPLING_FREQ else '(!) FAILED'
    print(f'{status} - {v}: {freq}')

⬆️ Upload data to database

Re-uploading overwrites the same variant — safe to re-run. With delete_from_db_before_upload=True (below), the upload first deletes, then writes. The delete is scoped to the exact match _measurement + varname + data_version (meteoscreening_diive) over the uploaded time range, so re-screening a period replaces only its previous screened result. It never touches the raw data (different data_version, and a different _raw bucket), other variables, or other data versions. The delete-first step (rather than a plain overwrite) matters because InfluxDB keys a point by its full tag set: if a tag changed between runs (e.g. units, gain, offset), a plain write would leave the old point as a duplicate — the delete removes it regardless of tags.

print(f'Uploading to bucket {BUCKET_PROCESSED}')
for v in mscr.resampled_detailed.keys():
    dbc.upload_singlevar(
        to_bucket=BUCKET_PROCESSED,
        to_measurement=assigned_measurements[v],
        var_df=mscr.resampled_detailed[v],
        timezone_offset_to_utc_hours=TIMEZONE_OFFSET_TO_UTC_HOURS,
        delete_from_db_before_upload=True,
    )

Verify upload

Download the just-uploaded data back and confirm the time resolution and timestamps. The offset is the same TIMEZONE_OFFSET_TO_UTC_HOURS, so the timestamps below should again be local TIMESTAMP_END — matching what you screened.

# Fresh variable names so the screened originals (data_detailed etc.) are not overwritten:
check_simple, check_detailed, check_measurements = dbc.download(
    bucket=BUCKET_PROCESSED,
    measurements=[MEASUREMENT],
    fields=FIELDS,
    start=START,
    stop=STOP,
    timezone_offset_to_utc_hours=TIMEZONE_OFFSET_TO_UTC_HOURS,
    data_version='meteoscreening_diive',
)
check_simple

for v in check_detailed.keys():
    idx = check_detailed[v].index
    freq = dv.times.DetectFrequency(index=idx, verbose=True).get()
    status = 'PASSED' if freq == RESAMPLING_FREQ else '(!) FAILED'
    print(f'{status} - {v}: freq={freq}, first={idx[0]}, last={idx[-1]}')

✅ End of notebook

print(f"Finished: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")