'''
Created on Jun 26, 2017
@author: jfpiolle
'''
import numpy
from s3analysis.slstr.cloud import WST_MASK, cloud_mask
[docs]def quality_level(
l2p_flags, cloud, sst_algo, uncertainty,
dt_analysis, sst, dt_threshold=5.
):
"""process the quality level
Default settings as provided by G.Corlett
Args:
cloud : a cloud mask or a tuple (nadir cloud mask, oblique cloud mask)
"""
quality = numpy.ma.zeros(l2p_flags.shape, dtype=numpy.byte)
if type(cloud) is tuple:
# combine cloud masks depending on the view combination in the
# SST calculation itself
cloud_in, cloud_io = cloud
cloudm = (
((sst_algo >= 4) & cloud_mask(cloud_io, cloud_mask)) |
((sst_algo < 4) & cloud_mask(cloud_in, cloud_mask))
)
else:
cloudm = cloud
quality[:] = 2
quality[uncertainty <= 0.25 ] = 5
quality[(quality < 5) & (uncertainty <= 0.8)] = 4
quality[(quality < 4) & (uncertainty <= 1.2) ] = 3
quality[numpy.ma.getmaskarray(uncertainty)] = 2
quality[cloud] = 1
quality[
(((dt_analysis >= dt_threshold) & (l2p_flags & 8 == 0)) |
(sst_algo == 0) |
numpy.ma.getmaskarray(sst))
] = 0
return quality
[docs]def granule_quality_level(wst, wct, dt_threshold=5.):
"""Return the quality level field given a pair of WST and WCT granules
provided as cerbere Swath objects.
Args:
wst (:py:class:cerbere.datamodel.swath.Swath)
wct (:py:class:cerbere.datamodel.swath.Swath)
"""
cloud_mask_io = wct.get_field('cloud_io').bitmask_or(WST_MASK)
cloud_mask_in = wct.get_field('cloud_in').bitmask_or(WST_MASK)
algo = wst.get_values('sst_algorithm_type')
cloud = (((algo < 4) & cloud_mask_in) | ((algo >= 4) & cloud_mask_io.filled(False)))
return quality_level(
wst.get_values("l2p_flags"),
cloud,
wst.get_values("sst_algorithm_type"),
wst.get_values("sst_theoretical_uncertainty"),
wst.get_values("dt_analysis"),
wst.get_values("sea_surface_temperature"),
dt_threshold=dt_threshold
)