QC Configuration¶

Here we define the generic config object for multiple QARTOD tests, plus the aggregate/rollup flag.

The key “pco2_seawater” indicates which variable in the netcdf file this config should run against.

[6]:

from ioos_qc.config import Config


config = {
    "pco2_seawater": {
        "qartod": {
            "gross_range_test": {"suspect_span": [200, 2400], "fail_span": [0, 3000]},
            "spike_test": {"suspect_threshold": 500, "fail_threshold": 1000},
            "location_test": {"bbox": [-124.5, 44, -123.5, 45]},
            "flat_line_test": {
                "tolerance": 1,
                "suspect_threshold": 3600,
                "fail_threshold": 86400,
            },
            "aggregate": {},
        }
    }
}


config = """
  contexts:
    - streams:
        pco2_seawater:
          qartod:
              gross_range_test:
                suspect_span: [200, 2400]
                fail_span: [0, 3000]
              spike_test:
                suspect_threshold: 500
                fail_threshold: 1000
              location_test:
                bbox: [-124.5, 44, -123.5, 45]
              flat_line_test:
                tolerance: 1
                suspect_threshold: 3600
                fail_threshold: 86400
"""

c = Config(config)

Option A: Manually run tests and store results¶

Store test configurations externally, pass your configuration and netcdf file to ioos_qc, and manually update netcdf variables with results of the test.

Note: For tests that need tinp, zinp, etc, use args to define the t, x, y, z dimensions. In this case, we need latitude and longitude for the location test.

[7]:

from ioos_qc.qartod import aggregate
from ioos_qc.streams import XarrayStream
from ioos_qc.results import collect_results, CollectedResult

qc = XarrayStream(pco2, lon="lon", lat="lat")

# Store as a list to run QC now
runner = list(qc.run(c))

results = collect_results(runner, how='list')

agg = CollectedResult(
    stream_id='',
    package='qartod',
    test='qc_rollup',
    function=aggregate,
    results=aggregate(results),
    tinp=qc.time(),
    data=data
)
results.append(agg)
results

[7]:

[<CollectedResult stream_id=pco2_seawater package=qartod test=gross_range_test>,
 <CollectedResult stream_id=pco2_seawater package=qartod test=spike_test>,
 <CollectedResult stream_id=pco2_seawater package=qartod test=location_test>,
 <CollectedResult stream_id=pco2_seawater package=qartod test=flat_line_test>,
 <CollectedResult stream_id= package=qartod test=qc_rollup>]

[8]:

plot_ncresults(pco2, "pco2_seawater", results, "pCO2 seawater", "gross_range_test")

[9]:

plot_ncresults(pco2, "pco2_seawater", results, "pCO2 seawater", "spike_test")

[10]:

plot_ncresults(pco2, "pco2_seawater", results, "pCO2 seawater", "flat_line_test")

[11]:

plot_ncresults(pco2, "pco2_seawater", results, "pCO2 seawater", "location_test")

[12]:

# To see overall results, use the aggregate test
plot_ncresults(pco2, "", results, "pCO2 seawater", "qc_rollup")

[13]:

# Store results manually
# This is just a simple example and stores the aggregate test flag as a variable.
# You can expand upon this, or use the ioos_qc library to store the results for you (see subsequent examples)
# and use xarray's to_xxx methods to serialize results to whichever format you prefer

agg_da = xr.DataArray(agg.results, {}, ('time',))
output_xds = pco2.assign(
    qartod_aggregate=agg_da
)

output_xds.qartod_aggregate

QARTOD - NetCDF Examples¶

Setup¶

Load the netCDF dataset¶

QC Configuration¶

Option A: Manually run tests and store results¶

Option B¶

Option C¶

ioos_qc

Navigation

Related Topics