Coverage for runmacs/spec/io/ames.py : 55%

# -*- coding: utf-8 -*- 

""" 

ames 

==== 

NASA AMES file format reader 

""" 

# Filename: ames.py 

######################################################################### 

# 

# ames.py - This file is part of the Munich Aerosol Cloud Scanner package. 

# 

# Copyright (C) 2012-2018 Florian Ewald 

# 

# runMACS is free software; you can redistribute it and/ 

# or modify it under the terms of the GNU General Public License 

# as published by the Free Software Foundation; either version 2 

# of the License, or (at your option) any later version. 

# 

# Spectral Python is distributed in the hope that it will be useful, 

# but WITHOUT ANY WARRANTY; without even the implied warranty of 

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

# GNU General Public License for more details. 

# 

# You should have received a copy of the GNU General Public License 

# along with this software; if not, write to 

# 

# Free Software Foundation, Inc. 

# 59 Temple Place, Suite 330 

# Boston, MA 02111-1307 

# USA 

# 

######################################################################### 

# 

# Send comments to: 

# Tobias Koelling, tobias.koelling@physik.uni-muenchen.de 

# 

import nappy 

import numpy as np 

import datetime 

import dateutil.tz as datetz 

import platform 

47 ↛ 48line 47 didn't jump to line 48, because the condition on line 47 was never trueif platform.system() == 'SunOS': #Solaris seems to have problems with pandas, TODO: better solution 

def read_csv(filename, skiprows, *_, **_2): 

""" 

Read csv data using numpy 

""" 

return np.loadtxt(filename, skiprows=skiprows+1) 

else: 

import pandas as pd 

def read_csv(*args, **kwargs): 

""" 

Read csv data using pandas 

""" 

return pd.read_csv(*args, **kwargs).values # pylint: disable=no-member 

class AMESReader(object): 

""" 

Reader for the NASA AMES file format 

""" 

def __init__(self, filename): 

self._header = nappy.openNAFile(filename) 

self._data = read_csv(filename, 

skiprows=self._header.getNumHeaderLines()-1, 

delim_whitespace=True, 

engine='c', 

decimal='.') 

for i in xrange(len(self._header.getVariables())): 

self._data[self._data[..., i] == self._header.getMissingValue(i), i] = float('nan') 

self._basedate = datetime.datetime(*self._header.DATE, tzinfo=datetz.tzutc()) 

self.time = np.array([self._basedate + datetime.timedelta(seconds=x) 

for x in self._data[..., 0]]) 

def getIndex(self, name): 

""" 

Finds the index of the variable unambigously defined by ``name``. 

The ``name`` may be a substring of the full name. 

""" 

possible_indices = [i+1 

for i, v in enumerate(self._header.getVariables()) 

if name in v[0]] 

85 ↛ 86line 85 didn't jump to line 86, because the condition on line 85 was never true if len(possible_indices) > 1: 

raise ValueError('name "%s" is ambigous'%name) 

87 ↛ 88line 87 didn't jump to line 88, because the condition on line 87 was never true elif len(possible_indices) == 0: 

raise ValueError('name "%s" not found'%name) 

else: 

return possible_indices[0] 

def variable(self, i_or_name): 

""" 

Returns the contents of the variable given by one of: 

* The index of the variable 

* The name of the variable 

* An unambigous part of the name of the variable 

""" 

99 ↛ 100line 99 didn't jump to line 100, because the condition on line 99 was never true if isinstance(i_or_name, int): 

return self._data[..., i_or_name] 

else: 

return self._data[..., self.getIndex(i_or_name)] 

def __getitem__(self, key): 

return self._data[key] 

def __getattr__(self, name): 

return getattr(self._header, name) 

@property 

def source(self): 

""" 

The measurement source 

""" 

return self._header.getSource() 

@property 

def originator(self): 

""" 

The measureing person 

""" 

return self._header.getOriginator() 

@property 

def organisation(self): 

""" 

The measureing organisation 

""" 

return self._header.getOrg() 

@property 

def mission(self): 

""" 

The mission during which the data has been acquired 

""" 

return self._header.getMission() 

@property 

def normalComments(self): 

""" 

Comment for the whole dataset 

""" 

return self._header.getNormalComments() 

@property 

def specialComments(self): 

""" 

Comment special for this file 

""" 

return self._header.getSpecialComments() 

class CASReader(object): 

""" 

Reader for AMES files of the CAS sonde 

""" 

variables = map(str.strip, """Ambient total number concentration (cm^-3) 

Ambient number concentration > 3um (cm^-3) 

Ambient number concentration < 3um (cm^-3) 

Effective diameter (um) 

Mean Volume Diameter (um) 

Ice Water Content (g m^-3) 

Bin_0 

Bin_1 

Bin_2 

Bin_3 

Bin_4 

Bin_5 

Bin_6 

Bin_7 

Bin_8 

Bin_9 

Bin_10 

Bin_11 

Bin_12 

Bin_13 

Bin_14 

Bin_15 

Bin_16 

Particle air speed (PAS) measured at the instrument (m/s) 

""".split('\n')) 

def __init__(self, filename, basedate): 

self._skip = int(open(filename).next().split()[0]) 

self._data = np.loadtxt(filename, skiprows=self._skip) 

self._basedate = basedate 

self.time = np.array([self._basedate + datetime.timedelta(seconds=x) 

for x in self._data[..., 0]]) 

def getIndex(self, name): 

""" 

Finds the index of the variable unambigously defined by ``name``. 

The ``name`` may be a substring of the full name. 

""" 

possible_indices = [i+1 

for i, v in enumerate(self.variables) 

if name in v] 

if len(possible_indices) > 1: 

raise ValueError('name "%s" is ambigous' % name) 

elif len(possible_indices) == 0: 

raise ValueError('name "%s" not found' % name) 

else: 

return possible_indices[0] 

def variable(self, i_or_name): 

""" 

Returns the contents of the variable given by one of: 

* The index of the variable 

* The name of the variable 

* An unambigous part of the name of the variable 

""" 

if isinstance(i_or_name, int): 

return self._data[..., i_or_name] 

else: 

return self._data[..., self.getIndex(i_or_name)] 

def __getitem__(self, key): 

return self._data[key]