diff --git a/mvcm/spectral_tests.py b/mvcm/spectral_tests.py
index 183efb591a7d71c685456166ede70c71d430c52d..4c2b32652f905c20e6358133f3d4b338669fb74c 100644
--- a/mvcm/spectral_tests.py
+++ b/mvcm/spectral_tests.py
@@ -2,102 +2,114 @@ import numpy as np
import xarray as xr
from numpy.lib.stride_tricks import sliding_window_view
+from attrs import define, field, validators, Factory
from typing import Dict
import functools
+import logging
# import utils
-import conf
-import conf_xr
-import scene as scn
-import preprocess_thresholds as preproc
-import restoral
+import mvcm.conf as conf
+import dev.conf_xr as conf_xr
+import mvcm.scene as scn
+import mvcm.preprocess_thresholds as preproc
+import mvcm.restoral as restoral
import importlib
_RTD = 180./np.pi
_DTR = np.pi/180
+_scene_list = ['Ocean_Day', 'Ocean_Night', 'Land_Day', 'Land_Night', 'Day_Snow', 'Night_Snow', 'Coast_Day',
+ 'Land_Day_Desert', 'Antarctic_Day', 'Polar_Day_Snow', 'Polar_Day_Desert', 'Polar_Day_Ocean',
+ 'Polar_Day_Desert_Coast', 'Polar_Day_Coast', 'Polar_Day_Land', 'Polar_Night_Snow',
+ 'Polar_Night_Land', 'Polar_Night_Ocean', 'Land_Day_Desert_Coast', 'Land_Day_Coast', 'Desert',
+ 'Australia']
# this is used for testing, eventually we want to remove it
importlib.reload(preproc)
importlib.reload(conf)
importlib.reload(restoral)
+logger = logging.getLogger('__name__')
-class CloudTests(object):
-
- def __init__(self,
- data: xr.Dataset,
- scene_name: str,
- thresholds: Dict) -> None:
- self.data = data
- self.scene_name = scene_name
- self.thresholds = thresholds
- self.scene_idx = tuple(np.nonzero(data[scene_name] == 1))
- if self.scene_idx[0].shape[0] == 0:
- self.pixels_in_scene = False
- else:
- self.pixels_in_scene = True
-
- def run_if_test_exists_for_scene(func):
- @functools.wraps(func)
- def wrapper(self, *args, test_name, **kwargs):
- if test_name not in self.thresholds[self.scene_name]:
- print('Not running test for this scene')
- # returns cmin. This could be changed into a keyworded argument for readability
- test_bit = np.zeros(args[-1].shape)
- return args[-1], test_bit
- return func(self, *args, test_name, **kwargs)
- return wrapper
- @run_if_test_exists_for_scene
+@define(kw_only=True, slots=True)
+class CloudTests(object):
+ """
+ """
+ data: xr.Dataset = field(validator=[validators.instance_of(xr.Dataset), ])
+ scene_name: str = field(validator=[validators.instance_of(str),
+ validators.in_(_scene_list)])
+ thresholds: str = field(validator=[validators.instance_of(Dict), ])
+ scene_idx: tuple = field(init=False,
+ default=Factory(lambda self: tuple(np.nonzero(self.data[self.scene_name] == 1)),
+ takes_self=True),
+ validator=[validators.instance_of(tuple), ])
+ pixels_in_scene: bool = field(init=False,
+ default=Factory(lambda self: self.scene_idx[0].shape[0] != 0,
+ takes_self=True),
+ validator=[validators.instance_of(bool), ])
+
+ def run_if_test_exists_for_scene(test_name):
+ def decorate(func):
+
+ @functools.wraps(func)
+ def wrapper(self, *args, **kwargs):
+ if test_name not in self.thresholds[self.scene_name]:
+ test_bit = np.zeros(args[-1].shape)
+ return args[-1], test_bit
+ else:
+ kwargs['confidence'] = np.ones(self.data.latitude.shape)
+ kwargs['qa_bit'] = np.zeros(self.data.latitude.shape)
+ kwargs['test_bit'] = np.zeros(self.data.latitude.shape)
+ kwargs['thresholds'] = self.thresholds[self.scene_name][test_name]
+ return func(self, *args, **kwargs)
+
+ return wrapper
+ return decorate
+
+ @run_if_test_exists_for_scene('11um_Test')
def test_11um(self,
band: str,
cmin: np.ndarray,
- test_name: str = '11um_Test') -> np.ndarray:
+ **kwargs) -> np.ndarray:
- confidence = np.ones(self.data[band].shape)
- qa_bit = np.zeros(self.data[band].shape)
- test_bit = np.zeros(self.data[band].shape)
- threshold = self.thresholds[self.scene_name][test_name]
+ threshold = kwargs['thresholds']
if (threshold['perform'] is True and self.pixels_in_scene is True):
- qa_bit[self.scene_idx] = 1
+ kwargs['qa_bit'][self.scene_idx] = 1
print(f'Testing "{self.scene_name}"\n')
rad = self.data[band].values[self.scene_idx]
idx = np.nonzero((self.data[band].values >= threshold['thr'][1]) &
(self.data[self.scene_name] == 1))
- test_bit[idx] = 1
- confidence[self.scene_idx] = conf.conf_test_new(rad, threshold['thr'])
+ kwargs['test_bit'][idx] = 1
+ kwargs['confidence'][self.scene_idx] = conf.conf_test_new(rad, threshold['thr'])
- cmin = np.fmin(cmin, confidence)
+ cmin = np.fmin(cmin, kwargs['confidence'])
- return cmin, test_bit
+ return cmin, kwargs['test_bit']
- @run_if_test_exists_for_scene
+ @run_if_test_exists_for_scene('Surface_Temperature_Test')
def surface_temperature_test(self,
band: str,
viirs_data: xr.Dataset,
cmin: np.ndarray,
- test_name: str = 'Surface_Temperature_Test') -> np.ndarray:
+ **kwargs) -> np.ndarray:
- confidence = np.ones(self.data[band].shape)
- qa_bit = np.zeros(self.data[band].shape)
- test_bit = np.zeros(self.data[band].shape)
- threshold = self.thresholds[self.scene_name][test_name]
+ threshold = kwargs['thresholds']
if (threshold['perform'] is True and self.pixels_in_scene is True):
- qa_bit[self.scene_idx] = 1
+ kwargs['qa_bit'][self.scene_idx] = 1
print(f'Testing "{self.scene_name}"\n')
rad = self.data[band].values[self.scene_idx]
sfcdif = viirs_data.geos_sfct.values[self.scene_idx] - rad
# need to write the test_bit here
thr = preproc.thresholds_surface_temperature(viirs_data, threshold, self.scene_idx)
- confidence[self.scene_idx] = conf.conf_test_new(sfcdif, thr)
+ kwargs['confidence'][self.scene_idx] = conf.conf_test_new(sfcdif, thr)
- cmin = np.fmin(cmin, confidence)
+ cmin = np.fmin(cmin, kwargs['confidence'])
- return cmin, test_bit
+ return cmin, kwargs['test_bit']
@run_if_test_exists_for_scene
def sst_test(self,