from util import BaseCheckList, invalidate_records, annotate_all import pandas as pd import numpy as np scene_checks = BaseCheckList() @scene_checks.check(depends=['hatchOpen','sceneMirrorPosition']) def hatch_check(frame, parameters): """ Check that the hatch is open on sky views """ hatch_closed_during_viewing = ((frame.hatchOpen != 1) & (~frame.sceneMirrorPosition.isin([ord('H'), ord('A')]))) frame['hatch_check'] = hatch_closed_during_viewing * 1 annotate_all(frame, hatch_closed_during_viewing, 'hatch closed') return frame @scene_checks.check(depends=['hatchOpen','sceneMirrorPosition']) def safing_check(frame, parameters): """ Check that the mirror doesn't safe during a calibration view and contaminate other records """ hatch_closing = (frame.hatchOpen == -3) mirror_safing = (hatch_closing & frame.sceneMirrorPosition.isin([ord('H'), ord('A')])) frame['safing_check'] = mirror_safing * 1 annotate_all(frame, mirror_safing, 'mirror likely safed during view') frame = invalidate_records(frame, 'safing_check') return frame def encoder_check(frame, parameters): return frame #### TESTS ##### def test_hatch_check(): frame = pd.DataFrame({ 'hatchOpen':[1,1,0], 'sceneMirrorPosition':[ord('H'), ord('A'), ord('S')], 'qc_notes':'' }) assert hatch_check(frame, {})['hatch_check'].values.tolist() == [0,0,1] frame = pd.DataFrame({ 'hatchOpen':[1,0,1], 'sceneMirrorPosition':[ord('H'), ord('A'), ord('S')], 'qc_notes':'' }) assert hatch_check(frame, {})['hatch_check'].values.tolist() == [0,0,0] def test_safing_check(): frame = pd.DataFrame({ 'hatchOpen':[0,0,0], 'sceneMirrorPosition':[ord('H'), ord('A'), ord('S')], 'qc_notes':'' }) assert safing_check(frame, {})['safing_check'].values.tolist() == [0,0,0] frame = pd.DataFrame({ 'hatchOpen':[1,-3,0,0], 'sceneMirrorPosition':[ord('S'), ord('H'), ord('A'), ord('S')], 'qc_notes':'' }) assert safing_check(frame, {})['safing_check'].values.tolist() == [1,1,0,1]