awips.dataaccess.SoundingsSupport._SoundingCube.__getitem__() - Code Metrics - Inspection of "SoundingsSupport change setEDEXHost->changeEDEXHos..." - Unidata/python-awips - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9cadfa...6b5914 )

by Michael

created 2018-10-12 02:36 UTC

_SoundingCube.getitem() A

↳ Parent: awips.dataaccess.SoundingsSupport

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	2

# #
# #

#
# 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 collections import defaultdict
from shapely.geometry import Point

from awips import DateTimeConverter
from awips.dataaccess import DataAccessLayer

from dynamicserialize.dstypes.com.raytheon.uf.common.time import DataTime
from dynamicserialize.dstypes.com.raytheon.uf.common.dataplugin.level import Level


def getSounding(modelName, weatherElements, levels, samplePoint, refTime=None, 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.
        refTime: (optional) the grid model reference time to use for the sounding.
        If not specified, the latest ref time in the system will be used.

        timeRange: (optional) 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, refTime, timeRange) = \
        __sanitizeInputs(modelName, weatherElements, levels, samplePoint, refTime, timeRange)

    requestArgs = { 'datatype'      : 'grid',

                    'locationNames' : locationNames,
                    'parameters'    : parameters,
                    'levels'        : levels,
                    'envelope'      : envelope,
                   }

    req = DataAccessLayer.newDataRequest(**requestArgs)

    forecastHours = __determineForecastHours(req, refTime, timeRange)
    if not forecastHours:
        return None
    response = DataAccessLayer.getGeometryData(req, forecastHours)
    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, refTime, timeRange):

    locationNames = [str(modelName)]
    parameters = __buildStringList(weatherElements)
    levels = __buildStringList(levels)
    envelope = Point(samplePoint)
    if refTime is not None:
        refTime = DataTime(refTime=DateTimeConverter.convertToDateTime(refTime))
    if timeRange is not None:
        timeRange = DateTimeConverter.constructTimeRange(*timeRange)
    return (locationNames, parameters, levels, envelope, refTime, timeRange)

def __determineForecastHours(request, refTime, timeRange):

    dataTimes = DataAccessLayer.getAvailableTimes(request, False)
    timesGen = [(DataTime(refTime=dataTime.getRefTime()), dataTime) for dataTime in dataTimes]
    dataTimesMap = defaultdict(list)
    for baseTime, dataTime in timesGen:
        dataTimesMap[baseTime].append(dataTime)

    if refTime is None:
        refTime = max(dataTimesMap.keys())

    forecastHours = dataTimesMap[refTime]
    if timeRange is None:

        return forecastHours
    else:
        return [forecastHour for forecastHour in forecastHours if timeRange.contains(forecastHour.getValidPeriod())]


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 SoundingsSupport.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)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
        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 SoundingsSupport.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)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
        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 SoundingsSupport.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		# #
3
4		#
5		# Classes for retrieving soundings based on gridded data from the Data Access
6		# Framework
7		#
8		#
9		#
10		# SOFTWARE HISTORY
11		#
12		# Date Ticket# Engineer Description
13		# ------------ ---------- ----------- --------------------------
14		# 06/24/15 #4480 dgilling Initial Creation.
15		#
16
17		from collections import defaultdict
18		from shapely.geometry import Point
19
20		from awips import DateTimeConverter
21		from awips.dataaccess import DataAccessLayer
22
23		from dynamicserialize.dstypes.com.raytheon.uf.common.time import DataTime
24		from dynamicserialize.dstypes.com.raytheon.uf.common.dataplugin.level import Level
25
26
27	View Code Duplication	def getSounding(modelName, weatherElements, levels, samplePoint, refTime=None, timeRange=None):
		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... best-practice introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report Too many arguments (6/5) Loading history...
28		"""
29		Performs a series of Data Access Framework requests to retrieve a sounding object
30		based on the specified request parameters.
31
32		Args:
33		modelName: the grid model datasetid to use as the basis of the sounding.
34		weatherElements: a list of parameters to return in the sounding.
35		levels: a list of levels to sample the given weather elements at
36		samplePoint: a lat/lon pair to perform the sampling of data at.
37		refTime: (optional) the grid model reference time to use for the sounding.
38		If not specified, the latest ref time in the system will be used.
39
40		timeRange: (optional) a TimeRange to specify which forecast hours to use.
41		If not specified, will default to all forecast hours.
42
43		Returns:
44		A _SoundingCube instance, which acts a 3-tiered dictionary, keyed
45		by DataTime, then by level and finally by weather element. If no
46		data is available for the given request parameters, None is returned.
47
48		"""
49
50		(locationNames, parameters, levels, envelope, refTime, timeRange) = \
51		__sanitizeInputs(modelName, weatherElements, levels, samplePoint, refTime, timeRange)
52
53		requestArgs = { 'datatype' : 'grid',
		0 ignored issues – show Coding Style introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report No space allowed after bracket Loading history...
54		'locationNames' : locationNames,
55		'parameters' : parameters,
56		'levels' : levels,
57		'envelope' : envelope,
58		}
59
60		req = DataAccessLayer.newDataRequest(**requestArgs)
61
62		forecastHours = __determineForecastHours(req, refTime, timeRange)
63		if not forecastHours:
64		return None
65		response = DataAccessLayer.getGeometryData(req, forecastHours)
66		soundingObject = _SoundingCube(response)
67
68		return soundingObject
69
70		def changeEDEXHost(host):
71		"""
72		Changes the EDEX host the Data Access Framework is communicating with.
73
74		Args:
75		host: the EDEX host to connect to
76		"""
77
78		if host:
79		DataAccessLayer.changeEDEXHost(str(host))
80
81	View Code Duplication	def __sanitizeInputs(modelName, weatherElements, levels, samplePoint, refTime, timeRange):
		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... best-practice introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report Too many arguments (6/5) Loading history...
82		locationNames = [str(modelName)]
83		parameters = __buildStringList(weatherElements)
84		levels = __buildStringList(levels)
85		envelope = Point(samplePoint)
86		if refTime is not None:
87		refTime = DataTime(refTime=DateTimeConverter.convertToDateTime(refTime))
88		if timeRange is not None:
89		timeRange = DateTimeConverter.constructTimeRange(*timeRange)
90		return (locationNames, parameters, levels, envelope, refTime, timeRange)
91
92	View Code Duplication	def __determineForecastHours(request, refTime, timeRange):
		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...
93		dataTimes = DataAccessLayer.getAvailableTimes(request, False)
94		timesGen = [(DataTime(refTime=dataTime.getRefTime()), dataTime) for dataTime in dataTimes]
95		dataTimesMap = defaultdict(list)
96		for baseTime, dataTime in timesGen:
97		dataTimesMap[baseTime].append(dataTime)
98
99		if refTime is None:
100		refTime = max(dataTimesMap.keys())
101
102		forecastHours = dataTimesMap[refTime]
103		if timeRange is None:
		0 ignored issues – show unused-code introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
104		return forecastHours
105		else:
106		return [forecastHour for forecastHour in forecastHours if timeRange.contains(forecastHour.getValidPeriod())]
		0 ignored issues – show Coding Style introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (116/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
107
108		def __buildStringList(param):
109		if __notStringIter(param):
		0 ignored issues – show unused-code introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
110		return [str(item) for item in param]
111		else:
112		return [str(param)]
113
114		def __notStringIter(iterable):
115		if not isinstance(iterable, str):
116		try:
117		iter(iterable)
118		return True
119		except TypeError:
120		return False
121
122
123
124	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... Unused Code introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report The variable `__class__` seems to be unused. Loading history...
125		"""
126		The top-level sounding object returned when calling SoundingsSupport.getSounding.
127
128		This object acts as a 3-tiered dict which is keyed by time then level
129		then parameter name. Calling times() will return all valid keys into this
130		object.
131		"""
132
133		def __init__(self, geometryDataObjects):
134		self._dataDict = {}
135		self._sortedTimes = []
136		if geometryDataObjects:
137		for geometryData in geometryDataObjects:
138		dataTime = geometryData.getDataTime()
139		level = geometryData.getLevel()
140		for parameter in geometryData.getParameters():
141		self.__addItem(parameter, dataTime, level, geometryData.getNumber(parameter))
142
143		def __addItem(self, parameter, dataTime, level, value):
144		timeLayer = self._dataDict.get(dataTime, _SoundingTimeLayer(dataTime))
145		self._dataDict[dataTime] = timeLayer
146		timeLayer._addItem(parameter, level, value)
		0 ignored issues – show Coding Style Best Practice introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report It seems like `_addItem` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
147		if dataTime not in self._sortedTimes:
148		self._sortedTimes.append(dataTime)
149		self._sortedTimes.sort()
150
151		def __getitem__(self, key):
152		return self._dataDict[key]
153
154		def __len__(self):
155		return len(self._dataDict)
156
157		def times(self):
158		"""
159		Returns the valid times for this sounding.
160
161		Returns:
162		A list containing the valid DataTimes for this sounding in order.
163		"""
164		return self._sortedTimes
165
166
167	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... Unused Code introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report The variable `__class__` seems to be unused. Loading history...
168		"""
169		The second-level sounding object returned when calling SoundingsSupport.getSounding.
170
171		This object acts as a 2-tiered dict which is keyed by level then parameter
172		name. Calling levels() will return all valid keys into this
173		object. Calling time() will return the DataTime for this particular layer.
174		"""
175
176		def __init__(self, dataTime):
177		self._dataTime = dataTime
178		self._dataDict = {}
179
180		def _addItem(self, parameter, level, value):
181		asString = str(level)
182		levelLayer = self._dataDict.get(asString, _SoundingTimeAndLevelLayer(self._dataTime, asString))
		0 ignored issues – show Coding Style introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (103/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
183		levelLayer._addItem(parameter, value)
		0 ignored issues – show Coding Style Best Practice introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report It seems like `_addItem` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
184		self._dataDict[asString] = levelLayer
185
186		def __getitem__(self, key):
187		asString = str(key)
188		if asString in self._dataDict:
189		return self._dataDict[asString]
190		else:
191		raise KeyError("Level " + str(key) + " is not a valid level for this sounding.")
192
193		def __len__(self):
194		return len(self._dataDict)
195
196		def time(self):
197		"""
198		Returns the DataTime for this sounding cube layer.
199
200		Returns:
201		The DataTime for this sounding layer.
202		"""
203		return self._dataTime
204
205		def levels(self):
206		"""
207		Returns the valid levels for this sounding.
208
209		Returns:
210		A list containing the valid levels for this sounding in order of
211		closest to surface to highest from surface.
212		"""
213		sortedLevels = [Level(level) for level in list(self._dataDict.keys())]
214		sortedLevels.sort()
215		return [str(level) for level in sortedLevels]
216
217
218	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... Unused Code introduced 2018-10-02 00:51 UTC by Report Bug Copy Issue Report The variable `__class__` seems to be unused. Loading history...
219		"""
220		The bottom-level sounding object returned when calling SoundingsSupport.getSounding.
221
222		This object acts as a dict which is keyed by parameter name. Calling
223		parameters() will return all valid keys into this object. Calling time()
224		will return the DataTime for this particular layer. Calling level() will
225		return the level for this layer.
226		"""
227
228		def __init__(self, time, level):
229		self._time = time
230		self._level = level
231		self._parameters = {}
232
233		def _addItem(self, parameter, value):
234		self._parameters[parameter] = value
235
236		def __getitem__(self, key):
237		return self._parameters[key]
238
239		def __len__(self):
240		return len(self._parameters)
241
242		def level(self):
243		"""
244		Returns the level for this sounding cube layer.
245
246		Returns:
247		The level for this sounding layer.
248		"""
249		return self._level
250
251		def parameters(self):
252		"""
253		Returns the valid parameters for this sounding.
254
255		Returns:
256		A list containing the valid parameter names.
257		"""
258		return list(self._parameters.keys())
259
260		def time(self):
261		"""
262		Returns the DataTime for this sounding cube layer.
263
264		Returns:
265		The DataTime for this sounding layer.
266		"""
267		return self._time
268

Unidata / python-awips

Push — master ( 9cadfa...6b5914 )

_SoundingCube.__getitem__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

_SoundingCube.getitem() A