(self)
Main loop to do all the realtime processing.
Main loop to do all the realtime processing.
Main loop to do all the realtime processing .
def run(self):
        """Main loop to do all the realtime processing."""
        check_monitors_interval = 60
        check_monitors_override = False
        check_gamepad_interval = 60
        check_gamepad_override = False

        mouse_distance = 0.0
        old_mouse_pos = old_monitors = None
        old_monitor_index = monitor_index = None
        connected_gamepads = old_gamepads = [False] * 4
        keys = {item: [0] * 255 for item in ('tick', 'prev', 'count', 'held')}
        gamepads = {item: [(0, 0)] * 4 for item in ('thumb_l', 'thumb_r')}
        gamepads.update({item: [0] * 4 for item in ('trig_l', 'trig_r')})
        gamepads['buttons'] = {item: [defaultdict(int)] * 4 for item in ('tick', 'prev', 'count', 'held')}
        ticks = 0
        start = time.time()
        while self.state != ThreadState.Stopped:
            self.process_gui()

            # Receive data from the processing thread
            mouse_speed = 0.0
            while not self.process_receiver.empty():
                event, data = self.process_receiver.get()
                if event == ProcessEvent.Error:
                    return self.stop()
                elif event == ProcessEvent.MouseDistance:
                    mouse_speed = data[0]
                    mouse_distance += mouse_speed
                    self.send_gui_event(ThreadEvent.MouseDistance, mouse_distance)
            self.send_gui_event(ThreadEvent.MouseSpeed, mouse_speed)

            # Get monitor data
            if not ticks % check_monitors_interval or check_monitors_override:
                check_monitors_override = False
                monitors = get_monitor_locations()
            monitors_changed = monitors != old_monitors
            if monitors_changed:
                self.send_process_event(ProcessCommand.MonitorChanged, monitors)

            # Handle realtime data
            if self.state == ThreadState.Running:
                self.send_process_event(ProcessCommand.Tick, ticks)

                # Get mouse data
                mouse_pos = cursor_position()
                mouse_moved = mouse_pos != old_mouse_pos

                # Get keyboard/mouse clicks
                # Note this will not see anything faster than 1/60th of a second
                for key, val in enumerate(keys['tick']):
                    previously_pressed = bool(val)
                    currently_pressed = bool(check_key_press(key))

                    # Detect individual key releases
                    if previously_pressed and not currently_pressed:
                        keys['tick'][key] = keys['held'][key] = 0
                        self.send_process_event(ProcessCommand.KeyReleased, key)

                    # Detect when a key is being held down
                    elif previously_pressed and currently_pressed:
                        keys['held'][key] += 1
                        self.send_process_event(ProcessCommand.KeyHeld, key)

                    # Detect when a new key is pressed
                    elif not previously_pressed and currently_pressed:
                        if ticks - keys['prev'][key] < self.double_click_ticks:
                            keys['count'][key] += 1
                        else:
                            keys['count'][key] = 1
                        keys['tick'][key] = keys['prev'][key] = ticks
                        self.send_process_event(ProcessCommand.KeyPressed, key, keys['count'][key])

                # Check mouse data against monitors
                if mouse_pos is None:  # Cancel if there is no mouse data
                    refresh_monitor_index = False
                elif monitor_index is None:  # If index hasn't been set yet
                    refresh_monitor_index = True
                elif mouse_moved or monitors_changed:  # If there's been any changes
                    refresh_monitor_index = True
                else:
                    refresh_monitor_index = False
                if refresh_monitor_index:
                    mx, my = mouse_pos
                    for i, (x1, y1, x2, y2) in enumerate(monitors):
                        if x1 <= mouse_pos[0] < x2 and y1 <= mouse_pos[1] < y2:
                            # Keep track of which monitor the mouse is on
                            monitor_index = i

                            # Remap mouse position to be within 0 and 1
                            # This is for the GUI "live" preview only
                            remapped = (
                                (mouse_pos[0] - x1) / (x2 - x1),
                                (mouse_pos[1] - y1) / (y2 - y1),
                            )
                            self.send_gui_event(ThreadEvent.MouseMove, remapped)
                            break

                    # If this part fails, monitors may have changed since being recorded
                    # Skip this loop and try again, but this time refresh the data
                    else:
                        print(
                          f'Unknown monitor detected (mouse: {mouse_pos}, saved: '
                          f'{monitors}, actual: {get_monitor_locations()}'
                        )
                        check_monitors_override = True
                        monitor_index = None
                        continue
                mouse_monitor_changed = monitor_index != old_monitor_index

                if mouse_monitor_changed:
                    vres = monitors[monitor_index][3] - monitors[monitor_index][1]
                    print(f'Mouse moved to {vres}p monitor')
                if mouse_moved:
                    self.send_process_event(ProcessCommand.MouseMove, mouse_pos)

                # Get gamepad information
                if XInput is not None:
                    # Determine which gamepads are connected
                    if not ticks % check_gamepad_interval or check_gamepad_override:
                        check_gamepad_override = False
                        connected_gamepads = XInput.get_connected()
                    gamepads_changed = connected_gamepads != old_gamepads

                    # Print message saying when a change is detected
                    if gamepads_changed:
                        diff = sum(filter(bool, connected_gamepads))
                        if old_gamepads is not None:
                            diff -= sum(filter(bool, old_gamepads))
                        print('Gamepad ' + ('removed', 'detected')[diff > 0])

                    for gamepad in (i for i, active in enumerate(connected_gamepads) if active):
                        # Get a snapshot of the current gamepad state
                        try:
                            state = XInput.get_state(gamepad)
                        except XInput.XInputNotConnectedError:
                            check_gamepad_override = True
                            continue
                        thumb_l, thumb_r = XInput.get_thumb_values(state)
                        trig_l, trig_r = XInput.get_trigger_values(state)
                        buttons = XInput.get_button_values(state)

                        if thumb_l != gamepads['thumb_l'][gamepad]:
                            self.send_processEvent(ProcessCommand.GamepadThumbL, gamepad, thumb_l)
                        if thumb_r != gamepads['thumb_r'][gamepad]:
                            self.send_processEvent(ProcessCommand.GamepadThumbR, gamepad, thumb_r)
                        if trig_l != gamepads['trig_l'][gamepad]:
                            self.send_processEvent(ProcessCommand.GamepadTriggerL, gamepad, trig_l)
                        if trig_r != gamepads['trig_r'][gamepad]:
                            self.send_processEvent(ProcessCommand.GamepadTriggerR, gamepad, trig_r)

                        gamepads['thumb_l'][gamepad] = thumb_l
                        gamepads['thumb_r'][gamepad] = thumb_r
                        gamepads['trig_l'][gamepad] = trig_l
                        gamepads['trig_r'][gamepad] = trig_r

                        for button, val in buttons.items():
                            previously_pressed = bool(gamepads['buttons']['tick'][gamepad][button])
                            currently_pressed = val

                            # Detect individual button releases
                            if previously_pressed and not currently_pressed:
                                gamepads['buttons']['tick'][gamepad][button] = gamepads['buttons']['held'][gamepad][button] = 0
                                self.send_process_event(ProcessCommand.GamepadButtonReleased, gamepad, button)

                            # Detect when a button is being held down
                            elif previously_pressed and currently_pressed:
                                gamepads['buttons']['held'][gamepad][button] += 1
                                self.send_process_event(ProcessCommand.GamepadButtonHeld, gamepad, button)

                            # Detect when a new button is pressed
                            elif not previously_pressed and currently_pressed:
                                count = gamepads['buttons']['count'][gamepad]
                                prev = gamepads['buttons']['prev'][gamepad]
                                if ticks - prev[button] < self.double_click_ticks:
                                    count[button] += 1
                                else:
                                    count[button] = 1
                                gamepads['buttons']['tick'][gamepad][button] = prev[button] = ticks
                                self.send_process_event(ProcessCommand.GamepadButtonPressed, gamepad, button, count[button])

                # Store old values
                old_mouse_pos = mouse_pos
                old_monitors = monitors
                old_monitor_index = monitor_index
                old_gamepads = connected_gamepads

            # Ensure loop is running at the correct UPS
            ticks += 1
            time_expected = ticks / self.ups
            time_diff = time.time() - start
            if time_expected > time_diff:
                time.sleep(time_expected - time_diff)
def run ( self ) : check_monitors_interval = 60 check_monitors_override = False check_gamepad_interval = 60 check_gamepad_override = False mouse_distance = 0.0 old_mouse_pos = old_monitors = None old_monitor_index = monitor_index = None connected_gamepads = old_gamepads = [ False ] * 4 keys = { item : [ 0 ] * 255 for item in ( 'tick' , 'prev' , 'count' , 'held' ) } gamepads = { item : [ ( 0 , 0 ) ] * 4 for item in ( 'thumb_l' , 'thumb_r' ) } gamepads . update ( { item : [ 0 ] * 4 for item in ( 'trig_l' , 'trig_r' ) } ) gamepads [ 'buttons' ] = { item : [ defaultdict ( int ) ] * 4 for item in ( 'tick' , 'prev' , 'count' , 'held' ) } ticks = 0 start = time . time ( ) while self . state != ThreadState . Stopped : self . process_gui ( ) # Receive data from the processing thread mouse_speed = 0.0 while not self . process_receiver . empty ( ) : event , data = self . process_receiver . get ( ) if event == ProcessEvent . Error : return self . stop ( ) elif event == ProcessEvent . MouseDistance : mouse_speed = data [ 0 ] mouse_distance += mouse_speed self . send_gui_event ( ThreadEvent . MouseDistance , mouse_distance ) self . send_gui_event ( ThreadEvent . MouseSpeed , mouse_speed ) # Get monitor data if not ticks % check_monitors_interval or check_monitors_override : check_monitors_override = False monitors = get_monitor_locations ( ) monitors_changed = monitors != old_monitors if monitors_changed : self . send_process_event ( ProcessCommand . MonitorChanged , monitors ) # Handle realtime data if self . state == ThreadState . Running : self . send_process_event ( ProcessCommand . Tick , ticks ) # Get mouse data mouse_pos = cursor_position ( ) mouse_moved = mouse_pos != old_mouse_pos # Get keyboard/mouse clicks # Note this will not see anything faster than 1/60th of a second for key , val in enumerate ( keys [ 'tick' ] ) : previously_pressed = bool ( val ) currently_pressed = bool ( check_key_press ( key ) ) # Detect individual key releases if previously_pressed and not currently_pressed : keys [ 'tick' ] [ key ] = keys [ 'held' ] [ key ] = 0 self . send_process_event ( ProcessCommand . KeyReleased , key ) # Detect when a key is being held down elif previously_pressed and currently_pressed : keys [ 'held' ] [ key ] += 1 self . send_process_event ( ProcessCommand . KeyHeld , key ) # Detect when a new key is pressed elif not previously_pressed and currently_pressed : if ticks - keys [ 'prev' ] [ key ] < self . double_click_ticks : keys [ 'count' ] [ key ] += 1 else : keys [ 'count' ] [ key ] = 1 keys [ 'tick' ] [ key ] = keys [ 'prev' ] [ key ] = ticks self . send_process_event ( ProcessCommand . KeyPressed , key , keys [ 'count' ] [ key ] ) # Check mouse data against monitors if mouse_pos is None : # Cancel if there is no mouse data refresh_monitor_index = False elif monitor_index is None : # If index hasn't been set yet refresh_monitor_index = True elif mouse_moved or monitors_changed : # If there's been any changes refresh_monitor_index = True else : refresh_monitor_index = False if refresh_monitor_index : mx , my = mouse_pos for i , ( x1 , y1 , x2 , y2 ) in enumerate ( monitors ) : if x1 <= mouse_pos [ 0 ] < x2 and y1 <= mouse_pos [ 1 ] < y2 : # Keep track of which monitor the mouse is on monitor_index = i # Remap mouse position to be within 0 and 1 # This is for the GUI "live" preview only remapped = ( ( mouse_pos [ 0 ] - x1 ) / ( x2 - x1 ) , ( mouse_pos [ 1 ] - y1 ) / ( y2 - y1 ) , ) self . send_gui_event ( ThreadEvent . MouseMove , remapped ) break # If this part fails, monitors may have changed since being recorded # Skip this loop and try again, but this time refresh the data else : print ( f'Unknown monitor detected (mouse: {mouse_pos}, saved: ' f'{monitors}, actual: {get_monitor_locations()}' ) check_monitors_override = True monitor_index = None continue mouse_monitor_changed = monitor_index != old_monitor_index if mouse_monitor_changed : vres = monitors [ monitor_index ] [ 3 ] - monitors [ monitor_index ] [ 1 ] print ( f'Mouse moved to {vres}p monitor' ) if mouse_moved : self . send_process_event ( ProcessCommand . MouseMove , mouse_pos ) # Get gamepad information if XInput is not None : # Determine which gamepads are connected if not ticks % check_gamepad_interval or check_gamepad_override : check_gamepad_override = False connected_gamepads = XInput . get_connected ( ) gamepads_changed = connected_gamepads != old_gamepads # Print message saying when a change is detected if gamepads_changed : diff = sum ( filter ( bool , connected_gamepads ) ) if old_gamepads is not None : diff -= sum ( filter ( bool , old_gamepads ) ) print ( 'Gamepad ' + ( 'removed' , 'detected' ) [ diff > 0 ] ) for gamepad in ( i for i , active in enumerate ( connected_gamepads ) if active ) : # Get a snapshot of the current gamepad state try : state = XInput . get_state ( gamepad ) except XInput . XInputNotConnectedError : check_gamepad_override = True continue thumb_l , thumb_r = XInput . get_thumb_values ( state ) trig_l , trig_r = XInput . get_trigger_values ( state ) buttons = XInput . get_button_values ( state ) if thumb_l != gamepads [ 'thumb_l' ] [ gamepad ] : self . send_processEvent ( ProcessCommand . GamepadThumbL , gamepad , thumb_l ) if thumb_r != gamepads [ 'thumb_r' ] [ gamepad ] : self . send_processEvent ( ProcessCommand . GamepadThumbR , gamepad , thumb_r ) if trig_l != gamepads [ 'trig_l' ] [ gamepad ] : self . send_processEvent ( ProcessCommand . GamepadTriggerL , gamepad , trig_l ) if trig_r != gamepads [ 'trig_r' ] [ gamepad ] : self . send_processEvent ( ProcessCommand . GamepadTriggerR , gamepad , trig_r ) gamepads [ 'thumb_l' ] [ gamepad ] = thumb_l gamepads [ 'thumb_r' ] [ gamepad ] = thumb_r gamepads [ 'trig_l' ] [ gamepad ] = trig_l gamepads [ 'trig_r' ] [ gamepad ] = trig_r for button , val in buttons . items ( ) : previously_pressed = bool ( gamepads [ 'buttons' ] [ 'tick' ] [ gamepad ] [ button ] ) currently_pressed = val # Detect individual button releases if previously_pressed and not currently_pressed : gamepads [ 'buttons' ] [ 'tick' ] [ gamepad ] [ button ] = gamepads [ 'buttons' ] [ 'held' ] [ gamepad ] [ button ] = 0 self . send_process_event ( ProcessCommand . GamepadButtonReleased , gamepad , button ) # Detect when a button is being held down elif previously_pressed and currently_pressed : gamepads [ 'buttons' ] [ 'held' ] [ gamepad ] [ button ] += 1 self . send_process_event ( ProcessCommand . GamepadButtonHeld , gamepad , button ) # Detect when a new button is pressed elif not previously_pressed and currently_pressed : count = gamepads [ 'buttons' ] [ 'count' ] [ gamepad ] prev = gamepads [ 'buttons' ] [ 'prev' ] [ gamepad ] if ticks - prev [ button ] < self . double_click_ticks : count [ button ] += 1 else : count [ button ] = 1 gamepads [ 'buttons' ] [ 'tick' ] [ gamepad ] [ button ] = prev [ button ] = ticks self . send_process_event ( ProcessCommand . GamepadButtonPressed , gamepad , button , count [ button ] ) # Store old values old_mouse_pos = mouse_pos old_monitors = monitors old_monitor_index = monitor_index old_gamepads = connected_gamepads # Ensure loop is running at the correct UPS ticks += 1 time_expected = ticks / self . ups time_diff = time . time ( ) - start if time_expected > time_diff : time . sleep ( time_expected - time_diff )