Coverage for runmacs/spec/calibration/tools.py : 41%

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

""" 

tools 

===== 

This module contanins assorted helper functions for the calibration of specMACS data. 

""" 

import logging 

logger = logging.getLogger(__name__) 

import datetime 

try: 

import dateutil.parser as dateparser 

import dateutil.tz as datetz 

except ImportError: 

dateparser = None 

datetz = None 

def parseIsoTimeSlow(time): 

""" 

Parses iso time format 

:param time: Time to parse as string. 

:returns: Time as datetime 

:note: This should cope all valid ISO time formats, but is intended only as fallback for :py:func:`parseIsoTime`. 

""" 

dt = dateparser.parse(time) 

try: 

dt = dt.astimezone(datetz.tzutc()) 

except ValueError: 

dt = dt.replace(tzinfo = datetz.tzutc()) 

return dt 

def parseIsoTime(time): 

""" 

Parses iso time format 

:param time: Time to parse as string. 

:returns: Time as datetime 

""" 

try: 

return datetime.datetime(int(time[0:4]), 

int(time[5:7]), 

int(time[8:10]), 

int(time[11:13]), 

int(time[14:16]), 

int(time[17:19]), 

int(time[20:26]), 

tzinfo=datetz.tzutc()) 

except: 

return parseIsoTimeSlow(time) 

def parseActionLog(loglist): 

""" 

Parses action log from a action log list. 

:param loglist: The action log as list of tuples. 

:returns: The action log as list of tuples with time as datetime objects. 

""" 

66 ↛ 67line 66 didn't jump to line 67, because the condition on line 66 was never true if dateparser is None: 

raise ImportError('Could not import dateutil.parser') 

for action, time in loglist: 

yield action, parseIsoTime(time) 

def normalizeEnviHeader(header): 

""" 

Normalizes an ENVI header across various versions of the specMACS ENVI header formats. 

:param header: ENVI header to be normalized as :py:class:`dict` 

:returns: The (same,) normalized header 

:note: This function operates in-place 

""" 

for name in ('lines', 'samples', 'bands', 'autodarkstartline'): 

try: 

header[name] = int(header[name]) 

except KeyError: 

pass 

86 ↛ 87line 86 didn't jump to line 87, because the condition on line 86 was never true if not header.get('lines', 0) > 0: 

raise ValueError('cannot normalize header with 0 lines') 

for name in ('tint','fps'): 

try: 

header[name] = float(header[name]) 

except KeyError: 

pass 

for name in ('Wavelength',): 

try: 

header[name] = map(float, header[name]) 

except KeyError: 

pass 

try: 

header['wallTimes'] = map(parseIsoTime, header['wallTimes']) 

header['start time'] = header['wallTimes'][0] 

header['stop time'] = header['wallTimes'][-1] 

except KeyError: 

try: 

date = header['acquisition date'].rsplit(':',1)[1].strip() 

start = header['GPS Start Time'].split(':',1)[1].strip() 

stop = header['GPS Stop Time'].split(':',1)[1].strip() 

header['start time'] = datetime.datetime(*map(int, (date[6:10], date[3:5], date[0:2], start[0:2], start[3:5], start[6:8])), microsecond=int(1e6*float('0.'+start[9:])), tzinfo=datetz.tzutc()) 

header['stop time'] = datetime.datetime(*map(int, (date[6:10], date[3:5], date[0:2], stop[0:2], stop[3:5], stop[6:8])), microsecond=int(1e6*float('0.'+stop[9:])), tzinfo=datetz.tzutc()) 

try: 

calcFps = (float(header['lines']) - 1) / (header['stop time'] - header['start time']).total_seconds() 

except ZeroDivisionError: 

pass 

else: 

if 'fps' not in header: 

header['fps'] = calcFps 

header['wallTimes'] = [header['start time'] + datetime.timedelta(seconds = i / calcFps) for i in xrange(header['lines'])] 

except KeyError: 

pass 

pass 

else: 

try: 

header['fps'] = (float(header['lines']) - 1) / (header['stop time'] - header['start time']).total_seconds() 

except ZeroDivisionError: 

pass 

try: 

header['highResTimes'] = map(parseIsoTime, header['highResTimes']) 

except KeyError: 

try: 

header['highResTimes'] = header['wallTimes'] 

except KeyError: 

pass 

pass 

actionLog = {} 

for k,v in header.items(): 

try: 

n, a = k.rsplit(' ', 1) 

except ValueError: 

continue 

if a == 'actionLog': 

actionLog[n] = list(parseActionLog(v)) 

143 ↛ 144line 143 didn't jump to line 144, because the condition on line 143 was never true if 'shutter' not in actionLog: 

actionLog['shutter'] = [('open', header['wallTimes'][0])] 

try: 

actionLog['shutter'].append(('close', header['wallTimes'][header['autodarkstartline']])) 

except (IndexError, KeyError): 

pass 

150 ↛ 151line 150 didn't jump to line 151, because the condition on line 150 was never true if 'fps' not in actionLog: 

actionLog['fps'] = [(header['fps'], header['wallTimes'][0])] 

153 ↛ 154line 153 didn't jump to line 154, because the condition on line 153 was never true if 'exposureTime' not in actionLog: 

actionLog['exposureTime'] = [(header['tint'], header['wallTimes'][0])] 

header['actionLog'] = actionLog 

header['isDarkfile'] = False 

return header 

def shutterStateToFrameTag(state): 

""" 

Converts shutter states into frame classification tags. 

:param state: shutter state 

:returns: frame classification tag 

""" 

if state in ('open', 'opened'): 

return 'good' 

if state in ('close', 'closed'): 

return 'dark' 

raise ValueError('invalid shutter state') 

def classifyFrames(header): 

""" 

Performs simple frame classification. 

This method operates solely by generating slices at the points im time 

as marked by the entries in the action log. 

:param header: ENVI header to be classified as :py:class:`dict` 

:returns: The (same,) header with added frame classificatoin data. 

:note: This function operates in-place 

:note: The :py:mod:`runmacs.spec.classification.frameclassification` module offers a more sophisticated analysis. 

""" 

if not header.get('isDarkfile', False): 

try: 

shutterLog = header['actionLog']['shutter'] 

wallTimes = header['wallTimes'] 

except (KeyError, AttributeError): 

shutterLog = [] 

if len(shutterLog) == 0: 

frameState = [(0, 'good')] 

logger.warning('no shutter information found in "%s", assuming full time exposure!', header['sourceId']) 

else: 

shutterLogIter = iter(shutterLog) 

if shutterLog[0][1] > wallTimes[0]: 

currentState = 'good' 

logger.warning('no shutter information found in start of "%s", assuming shutter open at the beginning!', header['sourceId']) 

else: 

currentState = shutterStateToFrameTag(shutterLog[0][0]) 

shutterLogIter.next() 

frameState = [(0, currentState)] 

try: 

nextState, nextTime = shutterLogIter.next() 

except StopIteration: 

nextState = None 

for i, frameTime in enumerate(wallTimes): 

if nextState is not None and frameTime >= nextTime: 

n = shutterStateToFrameTag(nextState) 

if n != currentState: 

currentState = n 

frameState.append((i, currentState)) 

try: 

nextState, nextTime = shutterLogIter.next() 

except StopIteration: 

nextState = None 

else: 

frameState = [(0, 'dark')] 

frameState.append((header['lines'], 'end')) 

header['frameState'] = frameState 

header['darkranges'] = [(a[0], b[0]) for a,b in zip(frameState[:-1], frameState[1:]) if a[1] == 'dark'] 

header['goodranges'] = [(a[0], b[0]) for a,b in zip(frameState[:-1], frameState[1:]) if a[1] == 'good'] 

header['darkindices'] = [i for s in header['darkranges'] for i in xrange(*s)] 

header['goodindices'] = [i for s in header['goodranges'] for i in xrange(*s)] 

return header