L2C_P.py 3.86 KB
Newer Older
Daniel Scheffler's avatar
Daniel Scheffler committed
1
# -*- coding: utf-8 -*-
2
3
"""Level 2C Processor:  Quality layers"""

4
5
6
7
8
9
import numpy as np
from collections import OrderedDict

from geoarray import GeoArray

from ..misc.definition_dicts import bandslist_all_errors, get_outFillZeroSaturated
10
from .L2B_P import L2B_object
11
from ..options.config import GMS_config as CFG
12
13

__author__ = 'Daniel Scheffler'
Daniel Scheffler's avatar
Daniel Scheffler committed
14

Daniel Scheffler's avatar
GEOP:    
Daniel Scheffler committed
15
shared = {}
16
res = {}
Daniel Scheffler's avatar
Daniel Scheffler committed
17

Daniel Scheffler's avatar
GEOP:    
Daniel Scheffler committed
18
19

class L2C_object(L2B_object):
20
    def __init__(self, L2B_obj=None):
21
        super(L2C_object, self).__init__()
22
23
24

        if L2B_obj:
            # populate attributes
25
            [setattr(self, key, value) for key, value in L2B_obj.__dict__.items()]
26

Daniel Scheffler's avatar
GEOP:    
Daniel Scheffler committed
27
        self.proc_level = 'L2C'
28
        self.proc_status = 'initialized'
Daniel Scheffler's avatar
GEOP:    
Daniel Scheffler committed
29
30


31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class AccuracyCube(GeoArray):
    def __init__(self, GMS_obj):
        self._GMS_obj = GMS_obj

        # privates
        self._layers = None

        if self.layers:
            super(AccuracyCube, self).__init__(self.generate_array(),
                                               geotransform=list(self.layers.values())[0].gt,
                                               projection=list(self.layers.values())[0].prj,
                                               bandnames=self.get_bandnames(),
                                               nodata=get_outFillZeroSaturated('int16')[0])

        else:
46
            raise ValueError('The given GMS_object contains no accuracy layers for combination.')
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

    @property
    def layers(self):
        # type: () -> OrderedDict
        if not self._layers:
            errs = OrderedDict((band, getattr(self._GMS_obj, band)) for band in bandslist_all_errors)
            self._layers = \
                OrderedDict((band, err) for band, err in errs.items() if isinstance(err, (np.ndarray, GeoArray)))

        return self._layers

    def get_bandnames(self):
        bandnames = []
        for errArrName, errArr in self.layers.items():
            if errArrName == 'ac_errors':
                if CFG.ac_bandwise_accuracy:
                    bandnames.extend(['AC errors %s' % bN for bN in errArr.bandnames])
                else:
                    bandnames.append('median of AC errors')

            elif errArrName == 'mask_clouds_confidence':
                bandnames.append('confidence of cloud mask')

70
71
72
            elif errArrName == 'spat_homo_errors':
                bandnames.append('shift reliability percentage')

73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
            elif errArrName == 'spec_homo_errors':
                if CFG.spechomo_bandwise_accuracy:
                    bandnames.extend(['error of spectral homogenization %s' % bN for bN in errArr.bandnames])
                else:
                    bandnames.append('median error of spectral homogenization')

            else:
                raise RuntimeError('Error setting bandnames for %s.' % errArrName)

        return bandnames

    def generate_array(self):
        err_layers = self.layers.copy()  # copy OrdDict, otherwise attributes of GMS_object are overwritten

        # handle CFG.ac_bandwise_accuracy (average ac_errors if needed)
        if 'ac_errors' in err_layers and not CFG.ac_bandwise_accuracy and err_layers['ac_errors'].bands > 1:
            err_layers['ac_errors'].arr = \
                np.median(err_layers['ac_errors'], axis=2).astype(err_layers['ac_errors'].dtype)

        # handle CFG.spechomo_bandwise_accuracy (average spec_homo_errors if needed)
        if 'spec_homo_errors' in err_layers and not CFG.spechomo_bandwise_accuracy and \
           err_layers['spec_homo_errors'].bands > 1:

            err_layers['spec_homo_errors'].arr = \
                np.median(err_layers['spec_homo_errors'], axis=2).astype(err_layers['spec_homo_errors'].dtype)

99
100
101
102
103
104
105
        # stack all accuracy layers together
        accArr = np.dstack(err_layers.values()).astype('int16')

        # apply int16 nodata value
        accArr[self._GMS_obj.arr.mask_nodata.astype(np.int8) == 0] = get_outFillZeroSaturated('int16')[0]

        return accArr