import numpy as np
import math
import os.path
from astropy.time import Time
from collections import namedtuple
from typing import Pattern, List, Dict, Tuple, Any, NamedTuple, Union
#------------------------------------------------------------------------------
# class Level0_Image
#------------------------------------------------------------------------------
[docs]class Level0_Image:
"""
Instance Attributes:
.. py:attribute:: mjd
The UTC time of the image represented as a modified julian date. Float.
.. py:attribute:: exposure_time
The exposure time in micro-seconds seconds. Float.
.. py:attribute:: sensor_temp
The detector temperature in Celsius. Float.
.. py:attribute:: sensor_pcb_temp
The detector printed circuit board temperature in Celsius. Float.
.. py:attribute:: top_intferom_temp
The temperature in Celsius of the top of the interferometer. Float.
.. py:attribute:: bottom_intferom_temp
The temperature in Celsius of the bottom of the interferometer. Float
.. py:attribute:: optobox_temp
The temperature in Celsius of the optics box. Float
.. py:attribute:: Q7_temp
The temperature in Celsius of Q7. Float
.. py:attribute:: high_gain
The detector gain setting. The value is `True` for high gain and `False` for low gain.
.. py:attribute:: comment
An optional comment string for each image. The comment is typically supplied by the operator during data acquisition.
.. py:attribute:: image
The Level 0 interferogram image expressed as a numpy 2-D float array of dimension (H,M) where H is the number of height rows and M is the number
of interferogram columns. Depending upon context the image may represent either a raw 2-D image read directly from disk or some other processed
level 0 image, e.g. sub-windowed or calibrated image.
.. py:attribute:: error
The error on the Level 0 interferogram image. The error field may be None meaning no error value is available. If it is not None then it will be
a numpy 2-D float array of dimension (H,M) where H is the number of height rows and M is the number of interferogram columns.
It will be the the same size as the image attribute.
"""
def __init__(self):
self.mjd = 0.0
self.exposure_time = 0.0
self.sensor_temp = 0.0
self.setpoint_temp = 0.0
self.sensor_pcb_temp = 0.0
self.top_intferom_temp = 0.0
self.bottom_intferom_temp = 0.0
self.optobox_temp = 0.0
self.Q7_temp = 0.0
self.periscope_temp = 0.0
self.wingpod_window_temp = 0.0
self.top_heater_power = 0.0
self.bot_heater_power = 0.0
self.high_gain = False
self.comment = None
self.image = None
self.error = None
# ------------------------------------------------------------------------------
# ImageStats
#------------------------------------------------------------------------------
ImageStats = NamedTuple('ImageStats', [('average', Level0_Image), ('error',np.ndarray), ('stddev', np.ndarray), ('median', np.ndarray)])
""" A simple namedtuple to hold image statistics.
.. py:attribute:: average
The average of all the images on a pixel by pixel basis. The average is an instance of :class:`showapi.level0.level0imagearray.Level0_Image` so you have to access to averaged header information as well as theimage. Access statsobj.average.image if
you want to get at the numpy 2-d image
.. py:attribute:: stddev
The standard deviation of all images expressed as a numpy 2-D array
.. py:attribute:: error
The error of all images expressed as a numpy 2-D array. This is typically the standard deviation devided by the square oot of the number of images.
.. py:attribute:: median
The median value of all images that contributed to the statistics, This can be useful if you think the statistics are being skewed by a bad image.
"""
#------------------------------------------------------------------------------
# class Level0_ImageCollection:
# This represents a collection of Level 0 images
#------------------------------------------------------------------------------
[docs]class Level0_ImageCollection:
"""
A class to support loading and maniplulating a collection of SHOW level 0 images. The general concept is to
* quickly load image headers from directories
* select, modify, delete various headers
* only read image data when specifically requested for a specific record, see :meth:`at`
Users can access header fields either via the attributes o fthe class as (numpy) arrays or via individual records using :meth:`~showapi.level0.level0imagearray.Level0_ImageCollection.at` which returns the image and header
for one record as an instance of :class:`showapi.level0.level0imagearray.Level0_Image`.
Instance Attributes:
.. py:attribute:: mjd
The UTC time of each image represented as a modified julian date. This variable is a **numpy** 1-D array of floating point.
.. py:attribute:: exposure_time
The exposure time in seconds of each image. This variable is a **numpy** 1-D array of floating point.
.. py:attribute:: sensor_temp
The detector temperature in Celsius for each image. This variable is a **numpy** 1-D array of 32 bit floating point. The OWL640 :attr:`sensor_temp`
is normally derived from interpolation of the telemetry stream.
.. py:attribute:: sensor_pcb_temp
The detector printed circuit board temperature in Celsius for each image. This variable is a **numpy** 1-D array of 32 bit floating point. The OWL640 :attr:`sensor_pcb_temp`
is normally derived from interpolation of the telemetry stream.
.. py:attribute:: top_intferom_temp
The temperature in Celsius of the top of the interferometer for each image. This variable is a **numpy** 1-D array of 32 bit floating point. The OWL640 :attr:`top_intferom_temp`
is derived from interpolation of the telemetry stream.
.. py:attribute:: bottom_intferom_temp
The temperature in Celsius of the bottom of the interferometer for each image. This variable is a **numpy** 1-D array of 32 bit floating point. The OWL640 :attr:`bottom_intferom_temp`
is derived from interpolation of the telemetry stream.
.. py:attribute:: optobox_temp
The temperature in Celsius of the optics box for each image. This variable is a **numpy** 1-D array of 32 bit floating point. The OWL640 :attr:`optobox_temp`
is derived from interpolation of the telemetry stream.
.. py:attribute:: Q7_temp
The temperature in Celsius of Q7. This variable is a **numpy** 1-D array of 32 bit floating point. The OWL640 :attr:`Q7_temp`
is derived from interpolation of the telemetry stream.
.. py:attribute:: high_gain
The detector gain setting for each image. The value is `True` for a high gain expsoure and `False` for a low gain exposure. This variable is a **numpy** 1-D array of `bool`.
.. py:attribute:: comment
An optional comment string for each image. The comment is typically supplied by the operator during data acquisition.
"""
def __init__(self):
self.mjd = np.array( [], dtype='f8')
self.exposure_time = np.array( [], dtype='f8')
self.sensor_temp = np.array( [], dtype='f4')
self.setpoint_temp = np.array( [], dtype='f4')
self.sensor_pcb_temp = np.array( [], dtype='f4')
self.top_intferom_temp = np.array( [], dtype='f4')
self.bottom_intferom_temp = np.array( [], dtype='f4')
self.optobox_temp = np.array( [], dtype='f4')
self.Q7_temp = np.array( [], dtype='f4')
self.periscope_temp = np.array( [], dtype='f4')
self.wingpod_window_temp = np.array( [], dtype='f4')
self.top_heater_power = np.array( [], dtype='f4')
self.bot_heater_power = np.array( [], dtype='f4')
# self.tec_on = np.array( [], dtype='bool')
self.high_gain = np.array( [], dtype='bool')
self.comment = []
self._level0_imageobject = []
#------------------------------------------------------------------------------
# Level0_ImageCollection::assign_from_headers
# Assign the headers read from the raw level 0 files into the standard Level 0
# format.
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
# assign_comment
#------------------------------------------------------------------------------
def assign_comment(self, comment : str) :
for i in range(self.numimages()):
self.comment[i] = comment
#------------------------------------------------------------------------------
# Level0_ImageCollection::unique_exposuretimes
#------------------------------------------------------------------------------
[docs] def unique_exposuretimes(self) -> np.ndarray :
"""
Fetch the unique exposure times in this collection
:return: returns the sorted array of unique exposure times
:rtype: np.ndarray
"""
uniquetimes = np.unique( self.exposure_time )
return uniquetimes
#------------------------------------------------------------------------------
# def unique_commentfield(self) -> List[str] :
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
# where_exposuretimes
#------------------------------------------------------------------------------
[docs] def where_exposuretimes(self, exposuretime: float) -> np.ndarray :
"""
Fetch the indices of exposure times in this collection that match the
selected exposure time. This method returns an array that can be passed
to Level0_ImageCollection.select to create a new collection of images with
fixed exposure time.
:return: returns the index of requested exposure times in this collection
:rtype: np.ndarray
"""
indices, = np.where( self.exposure_time == exposuretime )
return indices
#------------------------------------------------------------------------------
# where_comment
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
# split_by_exposuretime
#------------------------------------------------------------------------------
[docs] def split_by_exposuretime(self) -> Dict[float,'Level0_ImageCollection']:
"""
Splits the current collection by exposure time. It creates a dictionary of Level0_ImageCollection
sorted by exposure time.
:return:
"""
unique_expt = self.unique_exposuretimes()
sortedimages = {}
for t in unique_expt:
indices = self.where_exposuretimes(t)
coll = self.slice(indices)
sortedimages[t] = coll
return sortedimages
#------------------------------------------------------------------------------
# def split_by_commentfield(self) -> Dict[str, 'Level0_ImageCollection']:
#------------------------------------------------------------------------------
#------------------------------------------------------------------------------
# Level0_ImageCollection::at
#------------------------------------------------------------------------------
[docs] def at(self, index : int, get_image : bool = True) -> Level0_Image:
"""
Fetches the specified instance of header and image from the collection of images.
The code will only fetch the image itself if get_image is True as this is usually
the slow part of the entire process.
:param index:
The zero based index of the required image. The code will raise a RuntimeError exception
if the index is out of range
:param get_image:
Controls whether the image field is filled or left blank. The image field is properly filled if this
parameter is True. The image field is set to None if this parameter is false. This is useful for fetching
headers without the readimg and storing the images. Default is True
:return: The Level 0 image as an instance of class Level0_Image
"""
image = Level0_Image()
try:
image.mjd = self.mjd[index]
image.exposure_time = self .exposure_time[index]
image.sensor_temp = self.sensor_temp[index]
image.setpoint_temp = self.setpoint_temp[index]
image.sensor_pcb_temp = self.sensor_pcb_temp[index]
image.top_intferom_temp = self.top_intferom_temp[index]
image.bottom_intferom_temp = self.bottom_intferom_temp[index]
image.optobox_temp = self.optobox_temp[index]
image.Q7_temp = self.Q7_temp[index]
image.periscope_temp = self.periscope_temp[index]
image.wingpod_window_temp = self.wingpod_window_temp[index]
image.top_heater_power = self.top_heater_power[index]
image.bot_heater_power = self.bot_heater_power[index]
# image.tec_on = self.tec_on[index]
image.high_gain = self.high_gain[index]
image.comment = self.comment[index]
image.image = self._level0_imageobject[index].Image() if get_image else None
except Exception as inst:
raise RuntimeError('Error reading element %d from image collection' % (index,))
return image
#------------------------------------------------------------------------------
# Level0_ImageCollection::slice
#------------------------------------------------------------------------------
[docs] def slice(self, index: np.ndarray) -> 'Level0_ImageCollection':
"""
Slice the collection of images with the given indices and return the slice as
a new collection.
:param index: an array of indices typically generated by numpy.where or Level0_ImageCollection.where_exposuretime
:return: a new image collection with just the subset of images.
"""
coll = Level0_ImageCollection()
coll .mjd = self.mjd[index]
coll.exposure_time = self.exposure_time[index]
coll.sensor_temp = self.sensor_temp[index]
coll.sensor_pcb_temp = self.sensor_pcb_temp[index]
coll.setpoint_temp = self.setpoint_temp[index]
coll.top_intferom_temp = self.top_intferom_temp[index]
coll.bottom_intferom_temp = self.bottom_intferom_temp[index]
coll.optobox_temp = self.optobox_temp[index]
coll.Q7_temp = self.Q7_temp[index]
coll.periscope_temp = self.periscope_temp[index]
coll.wingpod_window_temp = self.wingpod_window_temp[index]
coll.top_heater_power = self.top_heater_power[index]
coll.bot_heater_power = self.bot_heater_power[index]
# coll.tec_on = self.tec_on[index]
coll.high_gain = self.high_gain[index]
for i in range(len(index)):
idx = index[i]
coll.comment.append( self.comment[idx] ) # Copy over the comment
coll._level0_imageobject.append( self._level0_imageobject[idx] ) # Copy over the file reader object
return coll
#------------------------------------------------------------------------------
# Level0_ImageCollection::sortbytime
#------------------------------------------------------------------------------
[docs] def sortbytime(self) -> 'Level0_ImageCollection':
"""
Slice the collection of images with the given indices and return the slice as
a new collection.
:param index: an array of indices typically generated by numpy.where or Level0_ImageCollection.where_exposuretime
:return: a new image collection with just the subset of images.
"""
index = np.argsort(self.mjd)
return self.slice(index)
#------------------------------------------------------------------------------
# Level0_ImageCollection::append
#------------------------------------------------------------------------------
[docs] def append(self, other: 'Level0_ImageCollection') -> int:
"""
Apppends the image records in the 'other' collection to this collection. Returns the number of images
in this collection
:param other: the image collection to append to this collection
:return: a new image collection with just the subset of images.
"""
self.mjd = np.append(self.mjd, other.mjd)
self.exposure_time = np.append(self.exposure_time, other.exposure_time)
self.sensor_temp = np.append(self.sensor_temp, other.sensor_temp)
self.sensor_pcb_temp = np.append(self.sensor_pcb_temp, other.sensor_pcb_temp)
self.setpoint_temp = np.append(self.setpoint_temp, other.setpoint_temp)
self.top_intferom_temp = np.append(self.top_intferom_temp, other.top_intferom_temp)
self.bottom_intferom_temp = np.append(self.bottom_intferom_temp, other.bottom_intferom_temp)
self.optobox_temp = np.append(self.optobox_temp, other.optobox_temp)
self.Q7_temp = np.append(self.Q7_temp, other.Q7_temp)
self.periscope_temp = np.append(self.periscope_temp, other.periscope_temp)
self.wingpod_window_temp = np.append(self.wingpod_window_temp, other.wingpod_window_temp)
self.top_heater_power = np.append(self.top_heater_power, other.top_heater_power)
self.bot_heater_power = np.append(self.bot_heater_power, other.bot_heater_power)
# self.tec_on = np.append(self.tec_on, other.tec_on)
self.high_gain = np.append(self.high_gain, other.high_gain)
for i in range(len(other)):
self.comment.append(other.comment[i]) # Copy over the comment
self._level0_imageobject.append(other._level0_imageobject[i]) # Copy over the file reader object
return self.numimages()
#------------------------------------------------------------------------------
# Level0_ImageCollection::numimages
#------------------------------------------------------------------------------
[docs] def numimages(self) -> int:
"""
Fetches the number of images in this collection of images
:return: The number of images in the collection
"""
return self.mjd.size
#------------------------------------------------------------------------------
# Level0_ImageCollection:: __getitem__(self, key)
#------------------------------------------------------------------------------
def __getitem__(self, key):
return self.at(key)
# ------------------------------------------------------------------------------
# Level0_ImageCollection __iter__
# Create an iterator object for the collection of images
# ------------------------------------------------------------------------------
def __iter__(self):
class iterobject:
def __init__(self, imagearray):
self.index = -1
self.imagearray = imagearray
self.maxpts = imagearray.numimages()
def __next__(self):
self.index += 1
if (self.index >= self.maxpts): raise StopIteration
return self.imagearray.at(self.index)
return iterobject(self)
#------------------------------------------------------------------------------
# Level0_ImageCollection::__len__
#------------------------------------------------------------------------------
def __len__(self):
return len(self.mjd)
#------------------------------------------------------------------------------
# Level0_ImageCollection::correct_image_ff_and_dc
#------------------------------------------------------------------------------
def correct_image_ff_and_dc(self, index : int ,
darkcurrent : Dict[float,ImageStats] = None,
ff : Dict[float,np.ndarray] = None) -> Level0_Image :
record = self.at(index)
image = np.copy(record.image)
if (darkcurrent is not None):
dcstats = darkcurrent.get(record.exposure_time)
if (dcstats is not None):
image -= dcstats.average.image
else:
print( 'WARNING, Level0_ImageCollection::correct_image_ff_and_dc could not find a dark current for exposure time %8.3f millisecs. Dark current is not subtracted' % (record.exposure_time / 1000.0,))
if (ff is not None):
ffimage = ff.get(record.exposure_time)
if (ffimage is not None):
image *= ffimage
else:
print( 'WARNING, Level0_ImageCollection::correct_image_ff_and_dc could not find a flat-field for exposure time %8.3f millisecs. Dark current is not subtracted' % (record.exposure_time / 1000.0,))
record.image = image
return record
#------------------------------------------------------------------------------
# Level0_ImageCollection::average_images(self)
#------------------------------------------------------------------------------
[docs] def average_images(self,
darkcurrent : Dict[float,ImageStats] = None,
ff : Dict[float,np.ndarray] = None) ->ImageStats:
"""
Averages all of the images in the collection and returns the average, standard deviation and error. Note that
you will probably want to sort data by exposure time first as we do not to divide by exposure time
:param darkcurrent: a dictionary of dark current statistics for differnt exposure times. The dark current will be
found that matches each images exposure time and subtracted form the image befor ethe image is averaged etc. Default is None
which means no dark current correction is applied.
:return: three element tuple[ average image and header, error image, standard deviation of image].
:rtype: Tuple[ Level0_Image, np.ndarray, np.ndarray ]
"""
avg = None
sd = None
err = None
n = 0.0
median = np.zeros( (self.numimages(),) )
for i in range(self.numimages()):
l0 = self.correct_image_ff_and_dc( i, darkcurrent=darkcurrent, ff=ff)
imge = l0.image
if avg is None:
avg = imge
sd = imge*imge
else:
avg = avg + imge
sd = sd + imge*imge
n += 1.0
median[i] = np.median(imge)
if n > 0.0 :
avg = avg/n # get the average signa
q = (sd - n * (avg * avg)) / n # Get the mean square deviation, (I avoid the n-1 form so it still works if we have only 1 image)
q [ np.where( q < 0.0)] = 0.0 # Check for a few bad points that might stray ever so slightly less than 0.0, it throws the sqrt code
sd = np.sqrt(q) # Get the root mean square deviation
ne = math.sqrt(n-1) if (n > 1) else 1
err = sd/ne # Get the error estimate on the average value.
avgrec = Level0_Image()
avgrec.mjd = np.average(self.mjd)
avgrec.exposure_time = np.average(self.exposure_time)
avgrec.sensor_temp = np.average(self.sensor_temp)
avgrec.setpoint_temp = np.average(self.setpoint_temp)
avgrec.sensor_pcb_temp = np.average(self.sensor_pcb_temp)
avgrec.top_intferom_temp = np.average(self.top_intferom_temp)
avgrec.bottom_intferom_temp = np.average(self.bottom_intferom_temp)
avgrec.optobox_temp = np.average(self.optobox_temp)
avgrec.Q7_temp = np.average(self.Q7_temp)
avgrec.periscope_temp = np.average(self.periscope_temp)
avgrec.wingpod_window_temp = np.average(self.wingpod_window_temp)
avgrec.top_heater_power = np.average(self.top_heater_power)
avgrec.bot_heater_power = np.average(self.bot_heater_power)
# avgrec.tec_on = self.tec_on[0]
avgrec.high_gain = self.high_gain[0]
avgrec.comment = ''
avgrec.image = avg
return ImageStats( average=avgrec, error=err, stddev=sd, median=median)
#------------------------------------------------------------------------------
# Read( dirname : str, format : str):
# Reads image headers from a list_of_files and returns the array of headers
# as a L0 collection
#------------------------------------------------------------------------------
def read_l0_onedirectory( dirname : str, format : str = 'OWL640', ignore_bad_headers = False) -> Level0_ImageCollection:
"""
A wrapper interface to load a list_of_files of images and return the result a collection
of level 0 images. The function provides a format parameters so the user can select
the image acquisition system. The code paradigm is to quiclky load the headers from all images
in the list_of_files and store the result in the collection of images. The actual image data is
not read from disk until the user accesses the specific member of the collection
:param dirname:
The list_of_files containing Level 0 images. The will read headers from all images/files in this list_of_files.
Note the code only supports one image format in any given list_of_files.
:param format:
The file format of the SHOW data acquisition system.
- OWL640, used for the SHOW ER2 flight in 2017
- XENICS, used for the SHOW TImmins balloon flight in 2014
- EPIX, used by the Univ of Sasktachewan frame grabber system in December 2016.
:return:
The collection of images in a single class instance of Level0_ImageCollection
"""
fmt = format.upper()
L0collection = None
if (fmt == 'OWL640'):
from . import owl640_xiphosraw
h = owl640_xiphosraw.SHOWRawDataOwl640XiphosHeader()
h.set_ignore_bad_header(ignore_bad_headers)
L0collection = h.read_level0_images(dirname)
elif (fmt == 'XENICS'):
from . import xenixcamera_show2014
h = xenixcamera_show2014.SHOW_RawData_Xenix_ImageHeader()
L0collection = h.read_level0_images(dirname)
elif (fmt == 'EPIX'):
from . import epixframegrabber
h = epixframegrabber.SHOW_RawData_EPIX_FrameGrabber()
L0collection = h.read_level0_images( dirname )
else:
print("Unsupported SHOW Level 0 format", format)
L0collection = Level0_ImageCollection()
return L0collection
#------------------------------------------------------------------------------
# SHOWLevel0::read_level0
#------------------------------------------------------------------------------
def read_level0( format : str,
dirnames : Union[str, List[str]],
append : bool = False,
assign_comment_from_dirname : bool = False,
ignore_bad_headers : bool = False) -> Level0_ImageCollection:
"""
Reads in SHOW level 0 image collections from one list_of_files or a list of directories and sorts all of the images by integration
time. This method deletes any previous images
:param dirnames: One list_of_files or a list of directories to read in. each list_of_files is represented as a string. The list_of_files can be absolute or relative to the current path
:param append: default is False: If False delete all current images and create a new set of images otherwise append the new images to the current images.
:return: True if any records are read in.
"""
dataset = Level0_ImageCollection()
names = [dirnames] if isinstance(dirnames, str) else dirnames # make sure that a string is converted to a list otherwise just assign 'as is' and assume it is Iterable
for dirname in names: # Iterate over all o fthe list_of_files names
d = read_l0_onedirectory( dirname, format, ignore_bad_headers = ignore_bad_headers ) # Read in all the images from this list_of_files
if (assign_comment_from_dirname):
head,comment = os.path.split(dirname)
d.assign_comment( comment )
dataset.append(d)
if (len(dataset) > 0):
newdataset = dataset.sortbytime()
dataset = newdataset
return dataset