import numpy as np
from .baseMetric import BaseMetric
__all__ = ['TemplateExistsMetric', 'UniformityMetric',
'RapidRevisitUniformityMetric', 'RapidRevisitMetric','NRevisitsMetric', 'IntraNightGapsMetric',
'InterNightGapsMetric', 'AveGapMetric']
[docs]class TemplateExistsMetric(BaseMetric):
"""Calculate the fraction of images with a previous template image of desired quality.
"""
def __init__(self, seeingCol='seeingFwhmGeom', observationStartMJDCol='observationStartMJD',
metricName='TemplateExistsMetric', **kwargs):
cols = [seeingCol, observationStartMJDCol]
super(TemplateExistsMetric, self).__init__(col=cols, metricName=metricName,
units='fraction', **kwargs)
self.seeingCol = seeingCol
self.observationStartMJDCol = observationStartMJDCol
[docs] def run(self, dataSlice, slicePoint=None):
""""Calculate the fraction of images with a previous template image of desired quality.
Parameters
----------
dataSlice : numpy.array
Numpy structured array containing the data related to the visits provided by the slicer.
slicePoint : dict, optional
Dictionary containing information about the slicepoint currently active in the slicer.
Returns
-------
float
The fraction of images with a 'good' previous template image.
"""
# Check that data is sorted in observationStartMJD order
dataSlice.sort(order=self.observationStartMJDCol)
# Find the minimum seeing up to a given time
seeing_mins = np.minimum.accumulate(dataSlice[self.seeingCol])
# Find the difference between the seeing and the minimum seeing at the previous visit
seeing_diff = dataSlice[self.seeingCol] - np.roll(seeing_mins, 1)
# First image never has a template; check how many others do
good = np.where(seeing_diff[1:] >= 0.)[0]
frac = (good.size) / float(dataSlice[self.seeingCol].size)
return frac
[docs]class RapidRevisitMetric(BaseMetric):
def __init__(self, mjdCol='observationStartMJD', metricName='RapidRevisit',
dTmin=40.0 / 60.0 / 60.0 / 24.0, dTpairs = 20.0 / 60.0 / 24.0,
dTmax = 30.0 / 60.0 / 24.0, minN1 = 28, minN2 = 82, **kwargs):
self.mjdCol = mjdCol
self.dTmin = dTmin
self.dTpairs = dTpairs
self.dTmax = dTmax
self.minN1 = minN1
self.minN2 = minN2
super().__init__(col=self.mjdCol, metricName=metricName, **kwargs)
[docs] def run(self, dataSlice, slicePoint=None):
dtimes = np.diff(np.sort(dataSlice[self.mjdCol]))
N1 = len(np.where((dtimes >= self.dTmin) & (dtimes <= self.dTpairs))[0])
N2 = len(np.where((dtimes >= self.dTmin) & (dtimes <= self.dTmax))[0])
if (N1 >= self.minN1) and (N2 >= self.minN2):
val = 1
else:
val = 0
return val
[docs]class NRevisitsMetric(BaseMetric):
"""Calculate the number of consecutive visits with time differences less than dT.
Parameters
----------
dT : float, optional
The time interval to consider (in minutes). Default 30.
normed : bool, optional
Flag to indicate whether to return the total number of consecutive visits with time
differences less than dT (False), or the fraction of overall visits (True).
Note that we would expect (if all visits occur in pairs within dT) this fraction would be 0.5!
"""
def __init__(self, mjdCol='observationStartMJD', dT=30.0, normed=False, metricName=None, **kwargs):
units = ''
if metricName is None:
if normed:
metricName = 'Fraction of revisits faster than %.1f minutes' % (dT)
else:
metricName = 'Number of revisits faster than %.1f minutes' % (dT)
units = '#'
self.mjdCol = mjdCol
self.dT = dT / 60. / 24. # convert to days
self.normed = normed
super(NRevisitsMetric, self).__init__(col=self.mjdCol, units=units, metricName=metricName, **kwargs)
[docs] def run(self, dataSlice, slicePoint=None):
"""Count the number of consecutive visits occuring within time intervals dT.
Parameters
----------
dataSlice : numpy.array
Numpy structured array containing the data related to the visits provided by the slicer.
slicePoint : dict, optional
Dictionary containing information about the slicepoint currently active in the slicer.
Returns
-------
float
Either the total number of consecutive visits within dT or the fraction compared to overall visits.
"""
dtimes = np.diff(np.sort(dataSlice[self.mjdCol]))
nFastRevisits = np.size(np.where(dtimes <= self.dT)[0])
if self.normed:
nFastRevisits = nFastRevisits / float(np.size(dataSlice[self.mjdCol]))
return nFastRevisits
[docs]class IntraNightGapsMetric(BaseMetric):
"""
Calculate the gap between consecutive observations within a night, in hours.
Parameters
----------
reduceFunc : function, optional
Function that can operate on array-like structures. Typically numpy function.
Default np.median.
"""
def __init__(self, mjdCol='observationStartMJD', nightCol='night', reduceFunc=np.median,
metricName='Median Intra-Night Gap', **kwargs):
units = 'hours'
self.mjdCol = mjdCol
self.nightCol = nightCol
self.reduceFunc = reduceFunc
super(IntraNightGapsMetric, self).__init__(col=[self.mjdCol, self.nightCol],
units=units, metricName=metricName, **kwargs)
[docs] def run(self, dataSlice, slicePoint=None):
"""Calculate the (reduceFunc) of the gap between consecutive obervations within a night.
Parameters
----------
dataSlice : numpy.array
Numpy structured array containing the data related to the visits provided by the slicer.
slicePoint : dict, optional
Dictionary containing information about the slicepoint currently active in the slicer.
Returns
-------
float
The (reduceFunc) value of the gap, in hours.
"""
dataSlice.sort(order=self.mjdCol)
dt = np.diff(dataSlice[self.mjdCol])
dn = np.diff(dataSlice[self.nightCol])
good = np.where(dn == 0)
if np.size(good[0]) == 0:
result = self.badval
else:
result = self.reduceFunc(dt[good]) * 24
return result
[docs]class InterNightGapsMetric(BaseMetric):
"""
Calculate the gap between consecutive observations between nights, in days.
Parameters
----------
reduceFunc : function, optional
Function that can operate on array-like structures. Typically numpy function.
Default np.median.
"""
def __init__(self, mjdCol='observationStartMJD', nightCol='night', reduceFunc=np.median,
metricName='Median Inter-Night Gap', **kwargs):
units = 'days'
self.mjdCol = mjdCol
self.nightCol = nightCol
self.reduceFunc = reduceFunc
super(InterNightGapsMetric, self).__init__(col=[self.mjdCol, self.nightCol],
units=units, metricName=metricName, **kwargs)
[docs] def run(self, dataSlice, slicePoint=None):
"""Calculate the (reduceFunc) of the gap between consecutive nights of observations.
Parameters
----------
dataSlice : numpy.array
Numpy structured array containing the data related to the visits provided by the slicer.
slicePoint : dict, optional
Dictionary containing information about the slicepoint currently active in the slicer.
Returns
-------
float
The (reduceFunc) of the gap between consecutive nights of observations, in days.
"""
dataSlice.sort(order=self.mjdCol)
unights = np.unique(dataSlice[self.nightCol])
if np.size(unights) < 2:
result = self.badval
else:
# Find the first and last observation of each night
firstOfNight = np.searchsorted(dataSlice[self.nightCol], unights)
lastOfNight = np.searchsorted(dataSlice[self.nightCol], unights, side='right') - 1
diff = dataSlice[self.mjdCol][firstOfNight[1:]] - dataSlice[self.mjdCol][lastOfNight[:-1]]
result = self.reduceFunc(diff)
return result
[docs]class AveGapMetric(BaseMetric):
"""
Calculate the gap between any consecutive observations, in hours, regardless of night boundaries.
Parameters
----------
reduceFunc : function, optional
Function that can operate on array-like structures. Typically numpy function.
Default np.median.
"""
def __init__(self, mjdCol='observationStartMJD', nightCol='night', reduceFunc=np.median,
metricName='AveGap', **kwargs):
units = 'hours'
self.mjdCol = mjdCol
self.nightCol = nightCol
self.reduceFunc = reduceFunc
super(AveGapMetric, self).__init__(col=[self.mjdCol, self.nightCol],
units=units, metricName=metricName, **kwargs)
[docs] def run(self, dataSlice, slicePoint=None):
"""Calculate the (reduceFunc) of the gap between consecutive observations.
Different from inter-night and intra-night gaps, between this is really just counting
all of the times between consecutive observations (not time between nights or time within a night).
Parameters
----------
dataSlice : numpy.array
Numpy structured array containing the data related to the visits provided by the slicer.
slicePoint : dict, optional
Dictionary containing information about the slicepoint currently active in the slicer.
Returns
-------
float
The (reduceFunc) of the time between consecutive observations, in hours.
"""
dataSlice.sort(order=self.mjdCol)
diff = np.diff(dataSlice[self.mjdCol])
result = self.reduceFunc(diff) * 24.
return result