electronic_checks.py

import numpy as np
import pandas as pd
import itertools
from aeri_qc.all_checks import checklist

######### CHECKS ###########
@checklist.add_check(depends=['datetime','HBBapexTemp','HBBbottomTemp','HBBtopTemp' ], affects_calibration=True, hides=['missing_data_flag_check','hbb_stable_check','hbb_thermistor_check'], description='outlier in HBB temperature discovered')
def hbb_temp_outlier_check(frame, parameters):
    return find_bb_outliers(frame, parameters, 'HBB')

@checklist.add_check(depends=['datetime','ABBapexTemp','ABBbottomTemp','ABBtopTemp' ], affects_calibration=True, hides=['missing_data_flag_check'], description='outlier in ABB temperature discovered')
def abb_temp_outlier_check(frame, parameters):
    return find_bb_outliers(frame, parameters, 'ABB')
    

###### UTILTIY #############

def find_bb_outliers(frame, parameters, bb):
    """
    Find outliers in blackbody temperatures.
    Annotate any problems, and set quality flags for each thermistor
    """
    window_length = parameters.get('window_length', 50)

    if bb == 'HBB':
        delta_thresh = .001
    else:
        delta_thresh = .01
    bbb_outliers = np.array(list(_scan_for_outliers(frame, '{}bottomTemp'.format(bb), delta_thresh)))
    bba_outliers = np.array(list(_scan_for_outliers(frame, '{}apexTemp'.format(bb), delta_thresh)))
    bbt_outliers = np.array(list(_scan_for_outliers(frame, '{}topTemp'.format(bb), delta_thresh)))

    return ( bbb_outliers | bba_outliers | bbt_outliers )

def _scan_for_outliers(frame, variable, delta_thresh):
    """
    Check for temperature changes exceeding the delta_thresh rate in degK / sec
    """
    last = None
    for i, row in frame.iterrows():
        if pd.isnull(row).any():
            yield False
        # First row cannot be an outlier
        elif last is None:
            yield False
            last = row
        else:
            # Compare each row to previous non-outlier row
            time_diff = (row.datetime - last.datetime).total_seconds()
            variable_diff = row[variable] - last[variable]
            if abs(variable_diff / time_diff) < delta_thresh:
                yield False
                last = row
            else:
                yield True