From c784918531b67744126dfc7bb7e5425d054728df Mon Sep 17 00:00:00 2001 From: Prohurtz <48768484+RedHawk989@users.noreply.github.com> Date: Sat, 17 Jun 2023 12:33:26 -0500 Subject: [PATCH] black format ep --- EyeTrackApp/eye_processor.py | 211 +++++++++++++++++++++-------------- 1 file changed, 128 insertions(+), 83 deletions(-) diff --git a/EyeTrackApp/eye_processor.py b/EyeTrackApp/eye_processor.py index 6fe1c63..6d083d7 100644 --- a/EyeTrackApp/eye_processor.py +++ b/EyeTrackApp/eye_processor.py @@ -1,4 +1,4 @@ -''' +""" ------------------------------------------------------------------------------------------------------ ,@@@@@@ @@ -28,7 +28,7 @@ Additional Contributors: [Assassin], Summer404NotFound, lorow, ZanzyTHEbar Copyright (c) 2023 EyeTrackVR <3 ------------------------------------------------------------------------------------------------------ -''' +""" from operator import truth from dataclasses import dataclass @@ -36,7 +36,6 @@ import sys import asyncio - sys.path.append(".") from config import EyeTrackCameraConfig from config import EyeTrackSettingsConfig @@ -61,6 +60,7 @@ from utils.img_utils import circle_crop from eye import EyeInfo, EyeInfoOrigin from intensity_based_openness import * + def run_once(f): def wrapper(*args, **kwargs): if not wrapper.has_run: @@ -74,8 +74,7 @@ def run_once(f): async def delayed_setting_change(setting, value): await asyncio.sleep(5) setting = value - PlaySound('Audio/completed.wav', SND_FILENAME | SND_ASYNC) - + PlaySound("Audio/completed.wav", SND_FILENAME | SND_ASYNC) class EyeProcessor: @@ -138,7 +137,7 @@ class EyeProcessor: self.er_daddy = None self.ibo = IntensityBasedOpeness(self.eye_id) self.roi_include_set = {"rotation_angle", "roi_window_x", "roi_window_y"} - + self.failed = 0 self.skip_blink_detect = False @@ -148,33 +147,30 @@ class EyeProcessor: self.rawx = 0.0 self.rawy = 0.0 self.eyeopen = 0.7 - #blink + # blink self.max_ints = [] self.max_int = 0 self.min_int = 4000000000000 - self.frames = 0 + self.frames = 0 self.blinkvalue = False self.hasrac_en = False self.radius = 10 - + self.prev_x = None self.prev_y = None self.bd_blink = False self.current_algo = EyeInfoOrigin.HSRAC - try: min_cutoff = float(self.settings.gui_min_cutoff) # 0.0004 beta = float(self.settings.gui_speed_coefficient) # 0.9 except: - print('\033[93m[WARN] OneEuroFilter values must be a legal number.\033[0m') + print("\033[93m[WARN] OneEuroFilter values must be a legal number.\033[0m") min_cutoff = 0.0004 beta = 0.9 noisy_point = np.array([1, 1]) self.one_euro_filter = OneEuroFilter( - noisy_point, - min_cutoff=min_cutoff, - beta=beta + noisy_point, min_cutoff=min_cutoff, beta=beta ) def output_images_and_update(self, threshold_image, output_information: EyeInfo): @@ -189,25 +185,28 @@ class EyeProcessor: self.image_queue_outgoing.put((image_stack, output_information)) self.previous_image = self.current_image self.previous_rotation = self.config.rotation_angle - except: # If this fails it likely means that the images are not the same size for some reason. - print('\033[91m[ERROR] Size of frames to display are of unequal sizes.\033[0m') + except: # If this fails it likely means that the images are not the same size for some reason. + print( + "\033[91m[ERROR] Size of frames to display are of unequal sizes.\033[0m" + ) pass + def capture_crop_rotate_image(self): # Get our current frame - + try: # Get frame from capture source, crop to ROI self.current_image = self.current_image[ - int(self.config.roi_window_y): int( + int(self.config.roi_window_y) : int( self.config.roi_window_y + self.config.roi_window_h ), - int(self.config.roi_window_x): int( + int(self.config.roi_window_x) : int( self.config.roi_window_x + self.config.roi_window_w ), ] self.ibo.change_roi(self.config.dict(include=self.roi_include_set)) - + except: # Failure to process frame, reuse previous frame. self.current_image = self.previous_image @@ -232,7 +231,7 @@ class EyeProcessor: rotation_matrix, (cols, rows), borderMode=cv2.BORDER_CONSTANT, - borderValue=(ar + 10, ag + 10, ab + 10),#(255, 255, 255), + borderValue=(ar + 10, ag + 10, ab + 10), # (255, 255, 255), ) self.current_image_white = cv2.warpAffine( self.current_image, @@ -245,70 +244,96 @@ class EyeProcessor: except: pass - def UPDATE(self): if self.settings.gui_BLINK: self.eyeopen = BLINK(self) if self.settings.gui_IBO: - self.eyeopen = self.ibo.intense(self.rawx, self.rawy, self.current_image_white,self.settings.ibo_filter_samples,self.settings.ibo_average_output_samples) - if self.eyeopen < self.settings.ibo_fully_close_eye_threshold: #threshold so the eye fully closes + self.eyeopen = self.ibo.intense( + self.rawx, + self.rawy, + self.current_image_white, + self.settings.ibo_filter_samples, + self.settings.ibo_average_output_samples, + ) + if ( + self.eyeopen < self.settings.ibo_fully_close_eye_threshold + ): # threshold so the eye fully closes self.eyeopen = 0.0 if self.bd_blink == True: pass if self.settings.gui_IBO and self.settings.gui_BLINK: - ibo = self.ibo.intense(self.rawx, self.rawy, self.current_image_white,self.settings.ibo_filter_samples,self.settings.ibo_average_output_samples) - + ibo = self.ibo.intense( + self.rawx, + self.rawy, + self.current_image_white, + self.settings.ibo_filter_samples, + self.settings.ibo_average_output_samples, + ) + blink = BLINK(self) if blink == 0.0: self.eyeopen = 0.0 else: self.eyeopen = ibo - self.output_images_and_update(self.thresh, EyeInfo(self.current_algo, self.out_x, self.out_y, 0, self.eyeopen)) - + self.output_images_and_update( + self.thresh, + EyeInfo(self.current_algo, self.out_x, self.out_y, 0, self.eyeopen), + ) def BLINKM(self): self.eyeopen = BLINK(self) - + def DADDYM(self): # todo: We should have a proper variable for drawing. self.thresh = self.current_image_gray.copy() - self.rawx, self.rawy, self.eyeopen, self.radius = self.er_daddy.run(self.current_image_gray) + self.rawx, self.rawy, self.eyeopen, self.radius = self.er_daddy.run( + self.current_image_gray + ) # Daddy also uses a one euro filter, so I'll have to use it twice, but I'm not going to think too much about it. self.out_x, self.out_y = cal.cal_osc(self, self.rawx, self.rawy) self.current_algorithm = EyeInfoOrigin.DADDY def HSRACM(self): if self.eye_id in [EyeId.LEFT] and self.settings.gui_circular_crop_left: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass if self.eye_id in [EyeId.RIGHT] and self.settings.gui_circular_crop_right: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass self.hasrac_en = True # todo: add process to initialise er_hsrac when resolution changes - self.rawx, self.rawy, self.thresh, self.radius = self.er_hsf.run(self.current_image_gray) + self.rawx, self.rawy, self.thresh, self.radius = self.er_hsf.run( + self.current_image_gray + ) self.rawx, self.rawy, self.thresh = RANSAC3D(self, True) - #print(self.radius) - #if self.prev_x is None: - # self.prev_x = self.rawx - # self.prev_y = self.rawy + # print(self.radius) + # if self.prev_x is None: + # self.prev_x = self.rawx + # self.prev_y = self.rawy self.out_x, self.out_y = cal.cal_osc(self, self.rawx, self.rawy) self.current_algorithm = EyeInfoOrigin.HSRAC - def HSFM(self): if self.eye_id in [EyeId.LEFT] and self.settings.gui_circular_crop_left: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass if self.eye_id in [EyeId.RIGHT] and self.settings.gui_circular_crop_right: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass # todo: add process to initialise er_hsf when resolution changes @@ -318,26 +343,36 @@ class EyeProcessor: def RANSAC3DM(self): if self.eye_id in [EyeId.LEFT] and self.settings.gui_circular_crop_left: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass if self.eye_id in [EyeId.RIGHT] and self.settings.gui_circular_crop_right: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass self.hasrac_en = False - current_image_gray_copy = self.current_image_gray.copy() # Duplicate before overwriting in RANSAC3D. + current_image_gray_copy = ( + self.current_image_gray.copy() + ) # Duplicate before overwriting in RANSAC3D. self.rawx, self.rawy, self.thresh = RANSAC3D(self, False) self.out_x, self.out_y = cal.cal_osc(self, self.rawx, self.rawy) self.current_algorithm = EyeInfoOrigin.RANSAC def BLOBM(self): if self.eye_id in [EyeId.LEFT] and self.settings.gui_circular_crop_left: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass if self.eye_id in [EyeId.RIGHT] and self.settings.gui_circular_crop_right: - self.current_image_gray, self.cct = circle_crop(self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct) + self.current_image_gray, self.cct = circle_crop( + self.current_image_gray, self.xc, self.yc, self.cc_radius, self.cct + ) else: pass self.rawx, self.rawy, self.thresh = BLOB(self) @@ -345,16 +380,16 @@ class EyeProcessor: self.out_x, self.out_y = cal.cal_osc(self, self.rawx, self.rawy) self.current_algorithm = EyeInfoOrigin.BLOB - - def ALGOSELECT(self): # self.DADDYM() - if self.failed == 0 and self.firstalgo != None: + if self.failed == 0 and self.firstalgo != None: self.firstalgo() else: self.failed = self.failed + 1 - if self.failed == 1 and self.secondalgo != None: #send the tracking algos previous fail number, in algo if we pass set to 0, if fail, + 1 + if ( + self.failed == 1 and self.secondalgo != None + ): # send the tracking algos previous fail number, in algo if we pass set to 0, if fail, + 1 self.secondalgo() else: self.failed = self.failed + 1 @@ -373,17 +408,12 @@ class EyeProcessor: self.fithalgo() else: - self.failed = 0 # we have reached last possible algo and it is disabled, move to first algo - - - - + self.failed = 0 # we have reached last possible algo and it is disabled, move to first algo def run(self): # Run the following somewhere # self.daddy = External_Run_DADDY() - self.firstalgo = None self.secondalgo = None self.thirdalgo = None @@ -391,44 +421,54 @@ class EyeProcessor: self.fithalgo = None algolist = [None, None, None, None, None, None] - #clear HSF values when page is opened to correctly reflect setting changes + # clear HSF values when page is opened to correctly reflect setting changes self.er_hsf = None - - #set algo priorities + + # set algo priorities if self.settings.gui_HSF: if self.er_hsf is None: if self.eye_id in [EyeId.LEFT]: - self.er_hsf = External_Run_HSF(self.settings.gui_skip_autoradius, self.settings.gui_HSF_radius_left) + self.er_hsf = External_Run_HSF( + self.settings.gui_skip_autoradius, + self.settings.gui_HSF_radius_left, + ) else: pass if self.eye_id in [EyeId.RIGHT]: - self.er_hsf = External_Run_HSF(self.settings.gui_skip_autoradius, - self.settings.gui_HSF_radius_right) + self.er_hsf = External_Run_HSF( + self.settings.gui_skip_autoradius, + self.settings.gui_HSF_radius_right, + ) else: pass algolist[self.settings.gui_HSFP] = self.HSFM else: if self.er_hsf is not None: self.er_hsf = None - + if self.settings.gui_HSRAC: if self.er_hsf is None: - + if self.eye_id in [EyeId.LEFT]: - self.er_hsf = External_Run_HSF(self.settings.gui_skip_autoradius, self.settings.gui_HSF_radius_left) + self.er_hsf = External_Run_HSF( + self.settings.gui_skip_autoradius, + self.settings.gui_HSF_radius_left, + ) else: pass if self.eye_id in [EyeId.RIGHT]: - self.er_hsf = External_Run_HSF(self.settings.gui_skip_autoradius, - self.settings.gui_HSF_radius_right) + self.er_hsf = External_Run_HSF( + self.settings.gui_skip_autoradius, + self.settings.gui_HSF_radius_right, + ) else: pass algolist[self.settings.gui_HSRACP] = self.HSRACM else: if self.er_hsf is not None: self.er_hsf = None - + if self.settings.gui_DADDY: if self.er_daddy is None: self.er_daddy = External_Run_DADDY() @@ -443,19 +483,23 @@ class EyeProcessor: if self.settings.gui_BLOB: algolist[self.settings.gui_BLOBP] = self.BLOBM - _, self.firstalgo, self.secondalgo, self.thirdalgo, self.fourthalgo, self.fithalgo = algolist - - + ( + _, + self.firstalgo, + self.secondalgo, + self.thirdalgo, + self.fourthalgo, + self.fithalgo, + ) = algolist f = True while True: - # f = True - # Check to make sure we haven't been requested to close + # f = True + # Check to make sure we haven't been requested to close if self.cancellation_event.is_set(): print("\033[94m[INFO] Exiting Tracking thread\033[0m") return - if self.config.roi_window_w <= 0 or self.config.roi_window_h <= 0: # At this point, we're waiting for the user to set up the ROI window in the GUI. # Sleep a bit while we wait. @@ -463,11 +507,12 @@ class EyeProcessor: return continue - # If our ROI configuration has changed, reset our model and detector - if (self.camera_model is None + if ( + self.camera_model is None or self.detector_3d is None - or self.camera_model.resolution != ( + or self.camera_model.resolution + != ( self.config.roi_window_w, self.config.roi_window_h, ) @@ -492,16 +537,16 @@ class EyeProcessor: except queue.Empty: # print("No image available") continue - + if not self.capture_crop_rotate_image(): continue - self.current_image_gray = cv2.cvtColor( - self.current_image, cv2.COLOR_BGR2GRAY + self.current_image, cv2.COLOR_BGR2GRAY ) - self.current_image_gray_clean = self.current_image_gray.copy() #copy this frame to have a clean image for blink algo - - self.ALGOSELECT() #run our algos in priority order set in settings - self.UPDATE() + self.current_image_gray_clean = ( + self.current_image_gray.copy() + ) # copy this frame to have a clean image for blink algo + self.ALGOSELECT() # run our algos in priority order set in settings + self.UPDATE()