started refactoring spectral_tests

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,