Coverage for runmacs/processor/mixins/caldatamixin.py : 58%

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

from runmacs.processor.product import ProductMixin 

from runmacs.processor.sugar import * 

from runmacs.processor.utils import defaultPartDescriptions 

from runmacs.spec.calibration.calibrationdata import CalibrationData 

import matplotlib.colors 

import matplotlib.cm 

import numpy as np 

class CalDataMixin(ProductMixin): 

productType = 'image_calibration_data' 

productDimensions = {'data': ('fieldid', 'spatial', 'spectral'), 

'act': ('spatial', 'spectral'), 

'alt': ('spatial', 'spectral'), 

'wavelength': ('spatial', 'spectral'), 

'previewdata': ('spatial', 'spectral', 'color'), 

} 

@property 

def partDescription(self): 

return {'data': {'long_name': 'Unstructured calibration data', 

'standard_name': 'unstructured_data'}, 

'act': {'long_name': 'Across track angle', 

'standard_name': 'act', 

'units': self.angleUnit}, 

'alt': {'long_name': 'Along track angle', 

'standard_name': 'alt', 

'units': self.angleUnit}, 

'previewdata': defaultPartDescriptions['previewdata'], 

'wavelength': {'long_name': 'Wavelength', 

'standard_name': 'wavelength', 

'units': self.wavelengthUnit} 

} 

computeComplexity = 1 

date = pprop(lambda self: self.creationTime) 

alphaA = pprop(lambda self: float(iter(self['act'].T('spatial').S(0,1)).next().mean())) 

alphaB = pprop(lambda self: float(iter(self['act'].T('spatial').S(self.dataSize['spatial']-1,self.dataSize['spatial'])).next().mean())) 

wvlnsMin = pprop(lambda self: float(np.nanmin(self['wavelength'].get()))) 

wvlnsMax = pprop(lambda self: float(np.nanmax(self['wavelength'].get()))) 

@returns('float32', unit='radians') 

def _getAct(self): 

fieldId = self.fieldnames.index('act') 

return (iter(self['data'].T('fieldid').S(fieldId, fieldId+1)).next() * self.unitConversion(self.angleUnit, 'radians')).astype("float32") 

@returns('float32', unit='radians') 

def _getAlt(self): 

fieldId = self.fieldnames.index('alt') 

return (iter(self['data'].T('fieldid').S(fieldId, fieldId+1)).next() * self.unitConversion(self.angleUnit, 'radians')).astype("float32") 

@returns('float32', unit='nm') 

def _getWavelength(self): 

fieldId = self.fieldnames.index('wvlns') 

return (iter(self['data'].T('fieldid').S(fieldId, fieldId+1)).next() * self.unitConversion(self.wavelengthUnit, 'nm')).astype("float32") 

@cacheable 

@returns('uint8') 

def _getPreviewdata(self): 

normalizer = matplotlib.colors.Normalize() 

mapper = matplotlib.cm.ScalarMappable(norm=normalizer, cmap='jet') 

data = iter(self['data'].T('fieldid').S(0,1)).next() 

mapper.set_clim(*np.percentile(data, (2,98))) 

data = mapper.to_rgba(data)[...,:3] 

lo = np.nanmin(data) 

hi = np.nanmax(data) 

return (255 * (data-lo) / (hi-lo)).astype('uint8') 

def mkCalibrationData(self): 

cal = CalibrationData() 

cal.setFieldNames(*self.fieldnames) 

cal.setSensorShape(self.dataSize['spatial'], self.dataSize['spectral']) 

cal.creationTime = self.creationTime 

cal.validFrom = self.validFrom 

cal.validUntil = self.validUntil 

cal.optics = self.optics 

cal.sensorId = self.sensorId 

cal.symbolicNonlinearityRule = self.symbolicNonlinearityRule 

cal.calibratedUnit = self.calibratedUnit 

cal.angleUnit = self.angleUnit 

cal.wavelengthUnit = self.wavelengthUnit 

cal.badPixels = self.badPixels 

cal.binningFactor = self.specbin * self.spatbin 

cal.data = self['data'].get() 

return cal