Source code for rainfallqc.qc_frameworks.inbuilt_qc_frameworks

# -*- coding: utf-8 -*-
"""In-built QC frameworks to apply to rainfall data to create quality controlled data."""

from rainfallqc.checks import comparison_checks, gauge_checks, neighbourhood_checks, pypwsqc_filters, timeseries_checks

INTENSE_QC = {
    "QC1": {"function": gauge_checks.check_years_where_nth_percentile_is_zero},
    "QC2": {"function": gauge_checks.check_years_where_annual_mean_k_top_rows_are_zero},
    "QC3": {"function": gauge_checks.check_temporal_bias},
    "QC4": {"function": gauge_checks.check_temporal_bias},
    "QC5": {"function": gauge_checks.check_intermittency},
    "QC6": {"function": gauge_checks.check_breakpoints},
    "QC7": {"function": gauge_checks.check_min_val_change},
    "QC8": {"function": comparison_checks.check_annual_exceedance_etccdi_r99p},
    "QC9": {"function": comparison_checks.check_annual_exceedance_etccdi_prcptot},
    "QC10": {"function": comparison_checks.check_exceedance_of_rainfall_world_record},
    "QC11": {"function": comparison_checks.check_hourly_exceedance_etccdi_rx1day},
    "QC12": {"function": timeseries_checks.check_dry_period_cdd},
    "QC13": {"function": timeseries_checks.check_daily_accumulations},
    "QC14": {"function": timeseries_checks.check_monthly_accumulations},
    "QC15": {"function": timeseries_checks.check_streaks},
    "QC16": {"function": neighbourhood_checks.check_wet_neighbours},
    "QC17": {"function": neighbourhood_checks.check_wet_neighbours},
    "QC18": {"function": neighbourhood_checks.check_dry_neighbours},
    "QC19": {"function": neighbourhood_checks.check_dry_neighbours},
    "QC20": {"function": neighbourhood_checks.check_monthly_neighbours},
    "QC21": {"function": neighbourhood_checks.check_timing_offset},
    "QC22": {"function": neighbourhood_checks.check_neighbour_affinity_index},
    "QC23": {"function": neighbourhood_checks.check_neighbour_correlation},
    "QC24": {"function": neighbourhood_checks.check_daily_factor},
    "QC25": {"function": neighbourhood_checks.check_monthly_factor},
}

INTENSE_RULEBASE_QC = {
    "QC2": {
        "function": gauge_checks.check_years_where_annual_mean_k_top_rows_are_zero,
    },
    "QC10": {
        "function": comparison_checks.check_exceedance_of_rainfall_world_record,
    },
    "QC11": {
        "function": comparison_checks.check_hourly_exceedance_etccdi_rx1day,
    },
    "QC12": {
        "function": timeseries_checks.check_dry_period_cdd,
    },
    "QC13": {
        "function": timeseries_checks.check_daily_accumulations,
    },
    "QC14": {
        "function": timeseries_checks.check_monthly_accumulations,
    },
    "QC15": {
        "function": timeseries_checks.check_streaks,
    },
    "QC17": {
        "function": neighbourhood_checks.check_wet_neighbours,
    },
    "QC19": {
        "function": neighbourhood_checks.check_dry_neighbours,
    },
    "QC20": {
        "function": neighbourhood_checks.check_monthly_neighbours,
    },
}

PYPWSQC = {
    "BC": {"function": pypwsqc_filters.run_bias_correction},
    "EBF": {"function": pypwsqc_filters.run_event_based_filter},
    "IC": {"function": pypwsqc_filters.run_indicator_correlation},
    "FZ": {"function": pypwsqc_filters.check_faulty_zeros},
    "HI": {"function": pypwsqc_filters.check_high_influx_filter},
    "PRF": {"function": pypwsqc_filters.run_peak_removal},
    "SO": {"function": pypwsqc_filters.check_station_outlier},
}

INBUILT_QC_FRAMEWORKS = {"intenseqc": INTENSE_QC, "intenseqc_rulebase_only": INTENSE_RULEBASE_QC, "pypwsqc": PYPWSQC}


# all checks that are computed as summary of overall data or once per year values
NON_ROWWISE_QC_CHECKS = ["QC1", "QC2", "QC3", "QC4", "QC5", "QC7", "QC8", "QC9", "QC21", "QC22", "QC23"]
NON_ROWWISE_QC_CHECKS_NAMES = [
    "percentiles_zero",
    "k-largest_zero",
    "days_of_week",
    "hours_of_day",
    "intermittency",
    "min_val_change",
    "r99p",
    "prcptot",
    "timings_offset",
    "affinity_index",
    "pearson_corr",
]
NON_ROWWISE_QC_CONVERTER = dict(zip(NON_ROWWISE_QC_CHECKS, NON_ROWWISE_QC_CHECKS_NAMES, strict=True))


[docs] def get_qc_name_from_qc_key(qc_key: str) -> str: """ Return the name of a QC check. Parameters ---------- qc_key: QC check ID Returns ------- qc_name: QC check name """ if qc_key in NON_ROWWISE_QC_CONVERTER.keys(): return NON_ROWWISE_QC_CONVERTER[qc_key]