""" Module to handle Scintec sodar .mnd files. >>> from sodar.scintec import maindata """ __author__ = 'Chris Calloway' __email__ = 'cbc@chriscalloway.org' __copyright__ = 'Copyright 2009 UNC-CH Department of Marine Science' __license__ = 'GPL2' from datetime import datetime, timedelta class MainData(list): """ Contain data from a Scintec sodar .mnd file. The data is contained as a list of Profile objects along with some attributes which apply to all the Profile objects. Parse a known good .mnd file: >>> main_data = MainData(good_mnd) Parse the profile data: >>> len(main_data) 48 Parse the first profile: >>> len(main_data[0]) 39 >>> main_data[0].start datetime.datetime(2009, 11, 17, 0, 0) >>> main_data[0].stop datetime.datetime(2009, 11, 17, 0, 30) >>> main_data[0].variables ['z', 'speed', 'dir', 'W', 'sigW', 'bck', 'error'] >>> main_data[0][0] == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> main_data[0](10) == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> main_data[0][-1] == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True >>> main_data[0](200) == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True >>> main_data(datetime(2009, 11, 17, 0, 30)).stop datetime.datetime(2009, 11, 17, 0, 30) >>> main_data(datetime(2009, 11, 17, 0, 0),True).start datetime.datetime(2009, 11, 17, 0, 0) Parse the last profile: >>> len(main_data[-1]) 39 >>> main_data[-1].start datetime.datetime(2009, 11, 17, 23, 30) >>> main_data[-1].stop datetime.datetime(2009, 11, 18, 0, 0) >>> main_data[-1].variables ['z', 'speed', 'dir', 'W', 'sigW', 'bck', 'error'] >>> main_data[-1][0] == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.32', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True >>> main_data[-1](10) == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.32', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True >>> main_data[-1][-1] == {'z':'200', 'speed':'15.05', 'dir':'71.8', ... 'W':'-0.19', 'sigW':'0.53', 'bck':'9.99E+37', ... 'error':'0'} True >>> main_data[-1](200) == {'z':'200', 'speed':'15.05', 'dir':'71.8', ... 'W':'-0.19', 'sigW':'0.53', 'bck':'9.99E+37', ... 'error':'0'} True >>> main_data(datetime(2009, 11, 18, 0, 0)).stop datetime.datetime(2009, 11, 18, 0, 0) >>> main_data(datetime(2009, 11, 17, 23, 0),True).start datetime.datetime(2009, 11, 17, 23, 0) """ def __init__(self, mnd, *args): """ Parse main daily Scintec sodar .mnd file. MainData(mnd[,file_name[,file_path]]) -> Where: mnd is a str object containing the complete contents read from a Scintec .mnd daily sodar file including all line endings, file_name is an optional str object representing a file name for a file which contains the referenced .mnd daily sodar file, file_path is an optional str object representing the path to file_name. Parse a known good .mnd file: >>> main_data = MainData(good_mnd) >>> main_data = MainData(good_mnd,good_name) >>> main_data.file_name == good_name True >>> main_data = MainData(good_mnd,good_name,good_path) >>> main_data.file_name == good_name True >>> main_data.file_path == good_path True """ super(self.__class__, self).__init__() self.file_name = '' self.file_path = '' # Optional args: smoke 'em if ya got 'em. try: self.file_name = str(args[0]) self.file_path = str(args[1]) except IndexError: pass # Divide the data into blocks # across boundaries separated by blank lines. blocks = [block.strip() for block in mnd.split('\n\n') if block.strip()] # The first block is the specification of the format header. # Divide it into lines. format_header_spec = [line.strip() for line in blocks[0].split('\n') if line.strip()] # The second block is the body of the format header. # Divide it into lines. file_header_body = [line.strip() for line in blocks[1].split('\n') if line.strip()] # All the remaing blocks are individual profiles # in chronological order. self.extend([Profile([line.strip() for line in block.split('\n') if line.strip()]) for block in blocks[2:]]) def __call__(self,timestamp,start=False): """ Get a profile by timestamp MainData_instance(timestamp[,start]) -> Where: timestamp is a datetime.datetime object representing either the start or stop timestamp of the Profile object returned. start is a bool object. If start is False (default), then timestamp is matched to the stop timestamp of the returned Profile object. Otherwise, if start is True, then timestamp is matched to the start timestamp of the returned Profile object. Access profiles by timestamp: >>> main_data = MainData(good_mnd) >>> main_data(datetime(2009, 11, 17, 0, 30)).stop datetime.datetime(2009, 11, 17, 0, 30) >>> main_data(datetime(2009, 11, 17, 0, 0),True).start datetime.datetime(2009, 11, 17, 0, 0) >>> main_data(datetime(2009, 11, 18, 0, 0)).stop datetime.datetime(2009, 11, 18, 0, 0) >>> main_data(datetime(2009, 11, 17, 23, 0),True).start datetime.datetime(2009, 11, 17, 23, 0) >>> main_data(datetime(2009, 11, 16, 0, 30)) Traceback (most recent call last): ... ValueError: Timestamp datetime.datetime(2009, 11, 16, 0, 30) not found in 'MainData' object >>> main_data('OK') Traceback (most recent call last): ... ValueError: Timestamp 'OK' not found in 'MainData' object """ for profile in self: if start: boundary = profile.start else: boundary = profile.stop if boundary == timestamp: return profile raise ValueError, ' '.join(['Timestamp', repr(timestamp), 'not found in', repr(self.__class__.__name__), 'object']) class Profile(list): """ Contain data for single profile from a Scintec sodar .mnd file. The data is contained as a list of Observation objects along with some attributes which apply to all the Profile objects. Parse a known good profile block: >>> profile = Profile(good_profile) Access the timestamp attributes: >>> profile.start datetime.datetime(2009, 11, 17, 0, 0) >>> profile.stop datetime.datetime(2009, 11, 17, 0, 30) Access the variable list: >>> profile.variables ['z', 'speed', 'dir', 'W', 'sigW', 'bck', 'error'] Access observations by sequence number: >>> profile[0] == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> profile[-1] == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True Access observations by elevation: >>> profile(10) == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> profile(200) == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True """ def __init__(self, profile_block): """ Parse a profile block from a main daily Scintec sodar .mnd file. Profile(profile_data) -> Where: profile_block is a list of str objects containing all the lines from a single profile in a Scintec .mnd daily sodar file. Parse a known good profile block: >>> profile = Profile(good_profile) Access the timestamp attributes: >>> profile.start datetime.datetime(2009, 11, 17, 0, 0) >>> profile.stop datetime.datetime(2009, 11, 17, 0, 30) Access the variable list: >>> profile.variables ['z', 'speed', 'dir', 'W', 'sigW', 'bck', 'error'] Access observations by sequence number: >>> profile[0] == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> profile[-1] == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True >>> profile[100000] Traceback (most recent call last): ... IndexError: list index out of range """ super(self.__class__, self).__init__() # The first line in the profile block is the timestamp. timestamp = profile_block[0] date, stop, duration = timestamp.split() date_year, date_month, date_day = date.split('-') stop_hour, stop_minute, stop_second = stop.split(':') duration_hour, duration_minute, duration_second = duration.split(':') interval = timedelta(0, # days int(duration_second), 0, # microseconds 0, # milliseconds int(duration_minute), int(duration_hour)) # omit optional weeks self.stop = datetime(int(date_year), int(date_month), int(date_day), int(stop_hour), int(stop_minute), # omit optional microseconds int(stop_second)) # and tzinfo self.start = self.stop - interval # The second line in the profile block is # a commented list of variable names self.variables = profile_block[1].split()[1:] # All the remaining lines in the profile block are # individual obervation columns in the same order # as the variable names. super(self.__class__, self).extend([Observation(observation.split(), self.variables) for observation in profile_block[2:]]) def __call__(self,elevation): """ Get an observation by elevation. Profile_instance(elevation) -> Where: elevation is a numeric or str object representing the elevation (value keyed by 'z') of the Observation object. Access observations by elevation: >>> profile = Profile(good_profile) >>> profile(10) == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> profile(200) == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True >>> profile(100000) Traceback (most recent call last): ... ValueError: Elevation 100000 not found in 'Profile' object >>> profile('10') == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> profile('200') == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True >>> profile('100000') Traceback (most recent call last): ... ValueError: Elevation '100000' not found in 'Profile' object """ for observation in self: if observation['z'] == str(int(elevation)): return observation raise ValueError, ' '.join(['Elevation', repr(elevation), 'not found in', repr(self.__class__.__name__), 'object']) class Observation(dict): """ Contain data for single observation from a Scintec sodar .mnd files. The data is contained as a dictionary of variable name/value pairs. The variable values may also be accessed as attributes of the instance. Parse a known good observation: >>> observation = Observation(good_observation,good_variables) >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ def __init__(self,data,variables): """ Parse an observation from a main daily Scintec sodar .mnd file. Observation(data,variables) -> Where: data is a list of str object representing variable values, variables is a list of str objects representing variable names. Parse a known good observation: >>> observation = Observation(good_observation,good_variables) >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True Raise an Exception if more variable names than values: >>> extra_variables = good_variables + ['extra'] >>> observation = Observation(good_observation,extra_variables) Traceback (most recent call last): ... ValueError: Same number of variable names and values required in 'Observation' object Raise an Exception if more variable values than names: >>> extra_observation = good_observation + ['extra'] >>> observation = Observation(extra_observation,good_variables) Traceback (most recent call last): ... ValueError: Same number of variable names and values required in 'Observation' object """ super(self.__class__, self).__init__() if len(variables) == len(data): # And update the dictionary with the variable names/values. super(self.__class__, self).update(dict(zip(variables,data))) else: # Unless there are an unequal number of variable names/values. raise ValueError, ' '.join(['Same number of variable', 'names and values required in', repr(self.__class__.__name__), 'object',]) def _test(): """ Run module tests in script mode. >>> from sodar.scintec.maindata import _test """ import doctest from sodar.tests import suite mnd_path,mnd_file,mnd,profile,variables,observation = \ suite.setUpGoodMndData() doctest.testmod(extraglobs=dict(good_mnd=mnd, good_name=mnd_file, good_path=mnd_path, good_profile=profile, good_variables=variables, good_observation=observation), optionflags=doctest.NORMALIZE_WHITESPACE) if __name__ == "__main__": _test()