awips.dataaccess.ModelSounding._SoundingCube.__len__() - Code Metrics - Unidata/python-awips - Measure and Improve Code Quality continuously with Scrutinizer

_SoundingCube.len() A
last analyzed 2020-10-19 17:29 UTC

↳ Parent: awips.dataaccess.ModelSounding

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	2
Ratio	100 %

Importance

Changes

Metric	Value
eloc	2
dl	2
loc	2
rs	10
c	0
b	0
f	0
cc	1
nop	1

#
# Classes for retrieving soundings based on gridded data from the Data Access
# Framework
#
#
#
#     SOFTWARE HISTORY
#
#    Date            Ticket#       Engineer       Description
#    ------------    ----------    -----------    --------------------------
#    06/24/15         #4480        dgilling       Initial Creation.
#

from awips.dataaccess import DataAccessLayer
from dynamicserialize.dstypes.com.raytheon.uf.common.dataplugin.level import Level
from shapely.geometry import Point


def getSounding(modelName, weatherElements, levels, samplePoint, timeRange=None):

    """
    Performs a series of Data Access Framework requests to retrieve a sounding object
    based on the specified request parameters.

    Args:
        modelName: the grid model datasetid to use as the basis of the sounding.
        weatherElements: a list of parameters to return in the sounding.
        levels: a list of levels to sample the given weather elements at
        samplePoint: a lat/lon pair to perform the sampling of data at.
        timeRange: (optional) a list of times, or a TimeRange to specify
        which forecast hours to use. If not specified, will default to all forecast hours.

    Returns:
        A _SoundingCube instance, which acts a 3-tiered dictionary, keyed
        by DataTime, then by level and finally by weather element. If no
        data is available for the given request parameters, None is returned.

    """

    (locationNames, parameters, levels, envelope, timeRange) = \
        __sanitizeInputs(modelName, weatherElements, levels, samplePoint, timeRange)

    requestArgs = {'datatype': 'grid', 'locationNames': locationNames,
                   'parameters': parameters, 'levels': levels, 'envelope': envelope}

    req = DataAccessLayer.newDataRequest(**requestArgs)
    response = DataAccessLayer.getGeometryData(req, timeRange)
    soundingObject = _SoundingCube(response)

    return soundingObject


def changeEDEXHost(host):
    """
    Changes the EDEX host the Data Access Framework is communicating with.

    Args:
            host: the EDEX host to connect to
    """

    if host:
        DataAccessLayer.changeEDEXHost(str(host))


def __sanitizeInputs(modelName, weatherElements, levels, samplePoint, timeRange):
    locationNames = [str(modelName)]
    parameters = __buildStringList(weatherElements)
    levels = __buildStringList(levels)
    envelope = Point(samplePoint)
    return locationNames, parameters, levels, envelope, timeRange


def __buildStringList(param):
    if __notStringIter(param):
        return [str(item) for item in param]
    else:
        return [str(param)]


def __notStringIter(iterable):
    if not isinstance(iterable, str):
        try:
            iter(iterable)
            return True
        except TypeError:
            return False


class _SoundingCube(object):

    """
    The top-level sounding object returned when calling ModelSounding.getSounding.

    This object acts as a 3-tiered dict which is keyed by time then level
    then parameter name. Calling times() will return all valid keys into this
    object.
    """

    def __init__(self, geometryDataObjects):
        self._dataDict = {}
        self._sortedTimes = []
        if geometryDataObjects:
            for geometryData in geometryDataObjects:
                dataTime = geometryData.getDataTime()
                level = geometryData.getLevel()
                for parameter in geometryData.getParameters():
                    self.__addItem(parameter, dataTime, level, geometryData.getNumber(parameter))

    def __addItem(self, parameter, dataTime, level, value):
        timeLayer = self._dataDict.get(dataTime, _SoundingTimeLayer(dataTime))
        self._dataDict[dataTime] = timeLayer
        timeLayer._addItem(parameter, level, value)
        if dataTime not in self._sortedTimes:
            self._sortedTimes.append(dataTime)
            self._sortedTimes.sort()

    def __getitem__(self, key):
        return self._dataDict[key]

    def __len__(self):
        return len(self._dataDict)

    def times(self):
        """
        Returns the valid times for this sounding.

        Returns:
            A list containing the valid DataTimes for this sounding in order.
        """
        return self._sortedTimes


class _SoundingTimeLayer(object):

    """
    The second-level sounding object returned when calling ModelSounding.getSounding.

    This object acts as a 2-tiered dict which is keyed by level then parameter
    name. Calling levels() will return all valid keys into this
    object. Calling time() will return the DataTime for this particular layer.
    """

    def __init__(self, dataTime):
        self._dataTime = dataTime
        self._dataDict = {}

    def _addItem(self, parameter, level, value):
        asString = str(level)
        levelLayer = self._dataDict.get(asString, _SoundingTimeAndLevelLayer(self._dataTime, asString))
        levelLayer._addItem(parameter, value)
        self._dataDict[asString] = levelLayer

    def __getitem__(self, key):
        asString = str(key)
        if asString in self._dataDict:
            return self._dataDict[asString]
        else:
            raise KeyError("Level " + str(key) + " is not a valid level for this sounding.")

    def __len__(self):
        return len(self._dataDict)

    def time(self):
        """
        Returns the DataTime for this sounding cube layer.

        Returns:
            The DataTime for this sounding layer.
        """
        return self._dataTime

    def levels(self):
        """
        Returns the valid levels for this sounding.

        Returns:
            A list containing the valid levels for this sounding in order of
            closest to surface to highest from surface.
        """
        sortedLevels = [Level(level) for level in list(self._dataDict.keys())]
        sortedLevels.sort()
        return [str(level) for level in sortedLevels]


class _SoundingTimeAndLevelLayer(object):

    """
    The bottom-level sounding object returned when calling ModelSounding.getSounding.

    This object acts as a dict which is keyed by parameter name. Calling
    parameters() will return all valid keys into this object. Calling time()
    will return the DataTime for this particular layer. Calling level() will
    return the level for this layer.
    """

    def __init__(self, time, level):
        self._time = time
        self._level = level
        self._parameters = {}

    def _addItem(self, parameter, value):
        self._parameters[parameter] = value

    def __getitem__(self, key):
        return self._parameters[key]

    def __len__(self):
        return len(self._parameters)

    def level(self):
        """
        Returns the level for this sounding cube layer.

        Returns:
            The level for this sounding layer.
        """
        return self._level

    def parameters(self):
        """
        Returns the valid parameters for this sounding.

        Returns:
            A list containing the valid parameter names.
        """
        return list(self._parameters.keys())

    def time(self):
        """
        Returns the DataTime for this sounding cube layer.

        Returns:
            The DataTime for this sounding layer.
        """
        return self._time


1		#
2		# Classes for retrieving soundings based on gridded data from the Data Access
3		# Framework
4		#
5		#
6		#
7		# SOFTWARE HISTORY
8		#
9		# Date Ticket# Engineer Description
10		# ------------ ---------- ----------- --------------------------
11		# 06/24/15 #4480 dgilling Initial Creation.
12		#
13
14		from awips.dataaccess import DataAccessLayer
15		from dynamicserialize.dstypes.com.raytheon.uf.common.dataplugin.level import Level
16		from shapely.geometry import Point
17
18
19	View Code Duplication	def getSounding(modelName, weatherElements, levels, samplePoint, timeRange=None):
		0 ignored issues – show Duplication introduced 2018-10-14 19:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
20		"""
21		Performs a series of Data Access Framework requests to retrieve a sounding object
22		based on the specified request parameters.
23
24		Args:
25		modelName: the grid model datasetid to use as the basis of the sounding.
26		weatherElements: a list of parameters to return in the sounding.
27		levels: a list of levels to sample the given weather elements at
28		samplePoint: a lat/lon pair to perform the sampling of data at.
29		timeRange: (optional) a list of times, or a TimeRange to specify
30		which forecast hours to use. If not specified, will default to all forecast hours.
31
32		Returns:
33		A _SoundingCube instance, which acts a 3-tiered dictionary, keyed
34		by DataTime, then by level and finally by weather element. If no
35		data is available for the given request parameters, None is returned.
36
37		"""
38
39		(locationNames, parameters, levels, envelope, timeRange) = \
40		__sanitizeInputs(modelName, weatherElements, levels, samplePoint, timeRange)
41
42		requestArgs = {'datatype': 'grid', 'locationNames': locationNames,
43		'parameters': parameters, 'levels': levels, 'envelope': envelope}
44
45		req = DataAccessLayer.newDataRequest(**requestArgs)
46		response = DataAccessLayer.getGeometryData(req, timeRange)
47		soundingObject = _SoundingCube(response)
48
49		return soundingObject
50
51
52		def changeEDEXHost(host):
53		"""
54		Changes the EDEX host the Data Access Framework is communicating with.
55
56		Args:
57		host: the EDEX host to connect to
58		"""
59
60		if host:
61		DataAccessLayer.changeEDEXHost(str(host))
62
63
64		def __sanitizeInputs(modelName, weatherElements, levels, samplePoint, timeRange):
65		locationNames = [str(modelName)]
66		parameters = __buildStringList(weatherElements)
67		levels = __buildStringList(levels)
68		envelope = Point(samplePoint)
69		return locationNames, parameters, levels, envelope, timeRange
70
71
72		def __buildStringList(param):
73		if __notStringIter(param):
74		return [str(item) for item in param]
75		else:
76		return [str(param)]
77
78
79		def __notStringIter(iterable):
80		if not isinstance(iterable, str):
81		try:
82		iter(iterable)
83		return True
84		except TypeError:
85		return False
86
87
88	View Code Duplication	class _SoundingCube(object):
		0 ignored issues – show Duplication introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
89		"""
90		The top-level sounding object returned when calling ModelSounding.getSounding.
91
92		This object acts as a 3-tiered dict which is keyed by time then level
93		then parameter name. Calling times() will return all valid keys into this
94		object.
95		"""
96
97		def __init__(self, geometryDataObjects):
98		self._dataDict = {}
99		self._sortedTimes = []
100		if geometryDataObjects:
101		for geometryData in geometryDataObjects:
102		dataTime = geometryData.getDataTime()
103		level = geometryData.getLevel()
104		for parameter in geometryData.getParameters():
105		self.__addItem(parameter, dataTime, level, geometryData.getNumber(parameter))
106
107		def __addItem(self, parameter, dataTime, level, value):
108		timeLayer = self._dataDict.get(dataTime, _SoundingTimeLayer(dataTime))
109		self._dataDict[dataTime] = timeLayer
110		timeLayer._addItem(parameter, level, value)
111		if dataTime not in self._sortedTimes:
112		self._sortedTimes.append(dataTime)
113		self._sortedTimes.sort()
114
115		def __getitem__(self, key):
116		return self._dataDict[key]
117
118		def __len__(self):
119		return len(self._dataDict)
120
121		def times(self):
122		"""
123		Returns the valid times for this sounding.
124
125		Returns:
126		A list containing the valid DataTimes for this sounding in order.
127		"""
128		return self._sortedTimes
129
130
131	View Code Duplication	class _SoundingTimeLayer(object):
		0 ignored issues – show Duplication introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
132		"""
133		The second-level sounding object returned when calling ModelSounding.getSounding.
134
135		This object acts as a 2-tiered dict which is keyed by level then parameter
136		name. Calling levels() will return all valid keys into this
137		object. Calling time() will return the DataTime for this particular layer.
138		"""
139
140		def __init__(self, dataTime):
141		self._dataTime = dataTime
142		self._dataDict = {}
143
144		def _addItem(self, parameter, level, value):
145		asString = str(level)
146		levelLayer = self._dataDict.get(asString, _SoundingTimeAndLevelLayer(self._dataTime, asString))
147		levelLayer._addItem(parameter, value)
148		self._dataDict[asString] = levelLayer
149
150		def __getitem__(self, key):
151		asString = str(key)
152		if asString in self._dataDict:
153		return self._dataDict[asString]
154		else:
155		raise KeyError("Level " + str(key) + " is not a valid level for this sounding.")
156
157		def __len__(self):
158		return len(self._dataDict)
159
160		def time(self):
161		"""
162		Returns the DataTime for this sounding cube layer.
163
164		Returns:
165		The DataTime for this sounding layer.
166		"""
167		return self._dataTime
168
169		def levels(self):
170		"""
171		Returns the valid levels for this sounding.
172
173		Returns:
174		A list containing the valid levels for this sounding in order of
175		closest to surface to highest from surface.
176		"""
177		sortedLevels = [Level(level) for level in list(self._dataDict.keys())]
178		sortedLevels.sort()
179		return [str(level) for level in sortedLevels]
180
181
182	View Code Duplication	class _SoundingTimeAndLevelLayer(object):
		0 ignored issues – show Duplication introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
183		"""
184		The bottom-level sounding object returned when calling ModelSounding.getSounding.
185
186		This object acts as a dict which is keyed by parameter name. Calling
187		parameters() will return all valid keys into this object. Calling time()
188		will return the DataTime for this particular layer. Calling level() will
189		return the level for this layer.
190		"""
191
192		def __init__(self, time, level):
193		self._time = time
194		self._level = level
195		self._parameters = {}
196
197		def _addItem(self, parameter, value):
198		self._parameters[parameter] = value
199
200		def __getitem__(self, key):
201		return self._parameters[key]
202
203		def __len__(self):
204		return len(self._parameters)
205
206		def level(self):
207		"""
208		Returns the level for this sounding cube layer.
209
210		Returns:
211		The level for this sounding layer.
212		"""
213		return self._level
214
215		def parameters(self):
216		"""
217		Returns the valid parameters for this sounding.
218
219		Returns:
220		A list containing the valid parameter names.
221		"""
222		return list(self._parameters.keys())
223
224		def time(self):
225		"""
226		Returns the DataTime for this sounding cube layer.
227
228		Returns:
229		The DataTime for this sounding layer.
230		"""
231		return self._time
232

Unidata / python-awips

_SoundingCube.__len__() A last analyzed 2020-10-19 17:29 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

_SoundingCube.len() A
last analyzed 2020-10-19 17:29 UTC