""" 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' 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 format header: >>> len(main_data._format_header_spec) 4 >>> main_data._format_header_spec[0] 'FORMAT-1' Parse the file header after the format header: >>> len(main_data._file_header_body) 26 >>> main_data._file_header_body[1] '# file information' Parse the profile data: >>> len(main_data) 48 Parse the first profile: >>> len(main_data[0]) 39 >>> main_data[0].timestamp '2009-11-17 00:30:00 00:30:00' >>> 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][-1] == {'z':'200', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.07', 'sigW':'99.99', 'bck':'9.99E+37', ... 'error':'0'} True Parse the last profile: >>> len(main_data[-1]) 39 >>> main_data[-1].timestamp '2009-11-18 00:00:00 00:30:00' >>> 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][-1] == {'z':'200', 'speed':'15.05', 'dir':'71.8', ... 'W':'-0.19', 'sigW':'0.53', 'bck':'9.99E+37', ... 'error':'0'} True """ 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(MainData, 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. self._blocks = [self._block.strip() for self._block in mnd.split('\n\n') if self._block.strip()] # The first block is the specification of the format header. # Divide it into lines. self._format_header_spec = [self._line.strip() for self._line in self._blocks[0].split('\n') if self._line.strip()] # The second block is the body of the format header. # Divide it into lines. self._file_header_body = [self._line.strip() for self._line in self._blocks[1].split('\n') if self._line.strip()] # All the remaing blocks are individual profiles # in chronological order. self.extend([Profile([self._line.strip() for self._line in self._block.split('\n') if self._line.strip()]) for self._block in self._blocks[2:]]) # No need to haul around bookkeeping storage used for initialization. del self._blocks, self._block, self._line 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) >>> profile.timestamp '2009-11-17 00:30:00 00:30:00' >>> profile.variables ['z', 'speed', 'dir', 'W', 'sigW', 'bck', 'error'] >>> 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 """ 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) >>> profile.timestamp '2009-11-17 00:30:00 00:30:00' >>> profile.variables ['z', 'speed', 'dir', 'W', 'sigW', 'bck', 'error'] >>> 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 """ super(Profile, self).__init__() # The first line in the profile block is the timestamp. self.timestamp = profile_block[0] # 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. self.extend([Observation(observation.split(),self.variables) for observation in profile_block[2:]]) 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 >>> observation.z '10' >>> observation.speed '99.99' >>> observation.dir '999.9' >>> observation.W '-0.05' >>> observation.sigW '0.40' >>> observation.bck '5.46E+03' >>> observation.error '0' """ 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 labels. 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 >>> extra_variables = good_variables + ['extra'] >>> observation = Observation(good_observation,extra_variables) Traceback (most recent call last): ... AttributeError: Same number of attributes and values required in 'Observation' object >>> extra_observation = good_observation + ['extra'] >>> observation = Observation(extra_observation,good_variables) Traceback (most recent call last): ... AttributeError: Same number of attributes 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 AttributeError, ' '.join(['Same number of attributes', 'and values required in', repr(self.__class__.__name__), 'object',]) def __getattr__(self,name): """ Get a variable value by name. observation_instance.variable_name -> variable_value >>> observation = Observation(good_observation,good_variables) >>> observation.z '10' >>> observation.speed '99.99' >>> observation.dir '999.9' >>> observation.W '-0.05' >>> observation.sigW '0.40' >>> observation.bck '5.46E+03' >>> observation.error '0' >>> observation.y Traceback (most recent call last): ... AttributeError: 'Observation' object has no attribute 'y' >>> observation.y = 'OK' >>> observation.y 'OK' >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ # Attempt attribute access by dictionary key. if name in self: return self[name] else: raise AttributeError, ' '.join([repr(self.__class__.__name__), 'object has no attribute', repr(name),]) def __setattr__(self,name,value): """ Protect variable values from overwriting. observation_instance.variable_name = variable_value But retrieve any other bound attribute values. >>> observation = Observation(good_observation,good_variables) >>> observation.z = 'OK' Traceback (most recent call last): ... AttributeError: Rebinding attribute 'z' disallowed in 'Observation' object >>> observation.y = 'OK' >>> observation.y 'OK' >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ if name in self: # Protect observation variables raise AttributeError, ' '.join(['Rebinding attribute', repr(name), 'disallowed in', repr(self.__class__.__name__), 'object',]) else: # Otherwise, do it the normal way. super(self.__class__, self).__setattr__(name,value) def __delattr__(self,name): """ Protect variable values from deletion. del observation_instance.variable_name But delete any other bound attribute values. >>> observation = Observation(good_observation,good_variables) >>> del observation.z Traceback (most recent call last): ... AttributeError: Unbinding attribute 'z' disallowed in 'Observation' object >>> del observation.y Traceback (most recent call last): ... AttributeError: 'Observation' object has no attribute 'y' >>> observation.y = 'OK' >>> del observation.y >>> del observation.y Traceback (most recent call last): ... AttributeError: y >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ if name in self: # Protect observation variables raise AttributeError, ' '.join(['Unbinding attribute', repr(name), 'disallowed in', repr(self.__class__.__name__), 'object',]) else: # Otherwise, do it the normal way. super(self.__class__, self).__delattr__(name) def __setitem__(self,key,value): """ Protect items. observation_instance[variable_name] = variable_value >>> observation = Observation(good_observation,good_variables) >>> observation['z'] = 'OK' Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation['y'] = 'OK' Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ raise TypeError, ' '.join([repr(self.__class__.__name__), 'object\'s items are read-only',]) def __delitem__(self,key): """ Protect items. observation_instance[variable_name] = variable_value >>> observation = Observation(good_observation,good_variables) >>> del observation['z'] Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> del observation['y'] Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ raise TypeError, ' '.join([repr(self.__class__.__name__), 'object\'s items are read-only',]) def clear(self): """ Protect items. >>> observation = Observation(good_observation,good_variables) >>> observation.clear() Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ raise TypeError, ' '.join([repr(self.__class__.__name__), 'object\'s items are read-only',]) def pop(self,key,*args): """ Protect items. >>> observation = Observation(good_observation,good_variables) >>> observation.pop('z') Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation.pop('y') Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation.pop('y','OK') Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ raise TypeError, ' '.join([repr(self.__class__.__name__), 'object\'s items are read-only',]) def popitem(self): """ Protect items. >>> observation = Observation(good_observation,good_variables) >>> observation.popitem() Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ raise TypeError, ' '.join([repr(self.__class__.__name__), 'object\'s items are read-only',]) def setdefault(self,key,*args): """ Protect items. >>> observation = Observation(good_observation,good_variables) >>> observation.setdefault('z') Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation.setdefault('y') Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation.setdefault('y','OK') Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ raise TypeError, ' '.join([repr(self.__class__.__name__), 'object\'s items are read-only',]) def update(self,*args,**kwargs): """ Protect items. >>> observation = Observation(good_observation,good_variables) >>> observation.update({'UP':'up','DOWN':'down',}) Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> observation.update((('UP','up'),('DOWN','down'),)) Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True >>> observation.update(UP='up',DOWN='down') Traceback (most recent call last): ... TypeError: 'Observation' object's items are read-only >>> observation == {'z':'10', 'speed':'99.99', 'dir':'999.9', ... 'W':'-0.05', 'sigW':'0.40', 'bck':'5.46E+03', ... 'error':'0'} True """ raise TypeError, ' '.join([repr(self.__class__.__name__), 'object\'s items are read-only',]) 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()