From 8c902aa6fb182e8655558b0d84e36bdc90201b62 Mon Sep 17 00:00:00 2001 From: kdmukai Date: Wed, 1 Jan 2025 16:20:13 -0600 Subject: [PATCH] Refactor / simplify `HardwareButtons` --- src/seedsigner/gui/screens/screen.py | 30 ++--- src/seedsigner/gui/screens/seed_screens.py | 35 +++--- .../gui/screens/settings_screens.py | 2 +- src/seedsigner/hardware/buttons.py | 115 ++++++++---------- src/seedsigner/views/seed_views.py | 105 ++++++++-------- 5 files changed, 133 insertions(+), 154 deletions(-) diff --git a/src/seedsigner/gui/screens/screen.py b/src/seedsigner/gui/screens/screen.py index 821424d1..fa2865c4 100644 --- a/src/seedsigner/gui/screens/screen.py +++ b/src/seedsigner/gui/screens/screen.py @@ -71,6 +71,13 @@ class BaseScreen(BaseComponent): for t in self.get_threads(): t.stop() + for t in self.get_threads(): + # Wait for each thread to stop; equivalent to `join()` but gracefully + # handles threads that were never run (necessary for screenshot generator + # compatibility, perhaps other edge cases). + while t.is_alive(): + time.sleep(0.01) + def clear_screen(self): # Clear the whole canvas @@ -226,11 +233,7 @@ class BaseTopNavScreen(BaseScreen): time.sleep(0.1) continue - user_input = self.hw_inputs.wait_for( - HardwareButtonsConstants.ALL_KEYS, - check_release=True, - release_keys=HardwareButtonsConstants.KEYS__ANYCLICK - ) + user_input = self.hw_inputs.wait_for(HardwareButtonsConstants.ALL_KEYS) with self.renderer.lock: if not self.top_nav.is_selected and user_input in [ @@ -447,6 +450,7 @@ class ButtonListScreen(BaseTopNavScreen): while True: ret = self._run_callback() if ret is not None: + print("Exiting ButtonListScreen due to _run_callback") return ret user_input = self.hw_inputs.wait_for( @@ -455,9 +459,7 @@ class ButtonListScreen(BaseTopNavScreen): HardwareButtonsConstants.KEY_DOWN, HardwareButtonsConstants.KEY_LEFT, HardwareButtonsConstants.KEY_RIGHT, - ] + HardwareButtonsConstants.KEYS__ANYCLICK, - check_release=True, - release_keys=HardwareButtonsConstants.KEYS__ANYCLICK + ] + HardwareButtonsConstants.KEYS__ANYCLICK ) with self.renderer.lock: @@ -641,9 +643,7 @@ class LargeButtonScreen(BaseTopNavScreen): HardwareButtonsConstants.KEY_DOWN, HardwareButtonsConstants.KEY_LEFT, HardwareButtonsConstants.KEY_RIGHT - ] + HardwareButtonsConstants.KEYS__ANYCLICK, - check_release=True, - release_keys=HardwareButtonsConstants.KEYS__ANYCLICK + ] + HardwareButtonsConstants.KEYS__ANYCLICK ) with self.renderer.lock: @@ -858,9 +858,7 @@ class QRDisplayScreen(BaseScreen): HardwareButtonsConstants.KEY_DOWN, HardwareButtonsConstants.KEY_LEFT, HardwareButtonsConstants.KEY_RIGHT, - ] + HardwareButtonsConstants.KEYS__ANYCLICK, - check_release=True, - release_keys=HardwareButtonsConstants.KEYS__ANYCLICK + ] + HardwareButtonsConstants.KEYS__ANYCLICK ) if user_input == HardwareButtonsConstants.KEY_DOWN: # Reduce QR code background brightness @@ -1191,9 +1189,7 @@ class KeyboardScreen(BaseTopNavScreen): # Start the interactive update loop while True: input = self.hw_inputs.wait_for( - HardwareButtonsConstants.KEYS__LEFT_RIGHT_UP_DOWN + [HardwareButtonsConstants.KEY_PRESS, HardwareButtonsConstants.KEY3], - check_release=True, - release_keys=[HardwareButtonsConstants.KEY_PRESS, HardwareButtonsConstants.KEY3] + HardwareButtonsConstants.KEYS__LEFT_RIGHT_UP_DOWN + [HardwareButtonsConstants.KEY_PRESS, HardwareButtonsConstants.KEY3] ) with self.renderer.lock: diff --git a/src/seedsigner/gui/screens/seed_screens.py b/src/seedsigner/gui/screens/seed_screens.py index 6856f578..ae5e137a 100644 --- a/src/seedsigner/gui/screens/seed_screens.py +++ b/src/seedsigner/gui/screens/seed_screens.py @@ -243,11 +243,7 @@ class SeedMnemonicEntryScreen(BaseTopNavScreen): def _run(self): while True: - input = self.hw_inputs.wait_for( - HardwareButtonsConstants.ALL_KEYS, - check_release=True, - release_keys=[HardwareButtonsConstants.KEY_PRESS, HardwareButtonsConstants.KEY2] - ) + input = self.hw_inputs.wait_for(HardwareButtonsConstants.ALL_KEYS) with self.renderer.lock: if self.is_input_in_top_nav: @@ -885,11 +881,7 @@ class SeedAddPassphraseScreen(BaseTopNavScreen): # Start the interactive update loop while True: - input = self.hw_inputs.wait_for( - HardwareButtonsConstants.ALL_KEYS, - check_release=True, - release_keys=[HardwareButtonsConstants.KEY_PRESS, HardwareButtonsConstants.KEY1, HardwareButtonsConstants.KEY2, HardwareButtonsConstants.KEY3] - ) + input = self.hw_inputs.wait_for(HardwareButtonsConstants.ALL_KEYS) keyboard_swap = False @@ -1467,13 +1459,13 @@ class SeedAddressVerificationScreen(ButtonListScreen): def _run_callback(self): - # Exit the screen on success via a non-None value - logger.info(f"verified_index: {self.verified_index.cur_count}") + # Exit the screen on success via a non-None value. + # see: ButtonListScreen._run() if self.verified_index.cur_count is not None: - logger.info("Screen callback returning success!") - self.threads[-1].stop() - while self.threads[-1].is_alive(): - time.sleep(0.01) + # Note that the ProgressThread will have already exited on its own. + + # Return a success value (anything other than None) to end the + # ButtonListScreen._run() loop. return 1 @@ -1490,10 +1482,13 @@ class SeedAddressVerificationScreen(ButtonListScreen): while self.keep_running: if self.verified_index.cur_count is not None: # This thread will detect the success state while its parent Screen - # holds in its `wait_for`. Have to trigger a hw_input event to break - # the Screen._run out of the `wait_for` state. The Screen will then - # call its `_run_callback` and detect the success state and exit. - HardwareButtons.get_instance().trigger_override(force_release=True) + # blocks in its `wait_for`. Have to trigger a hw_input override event + # to break the Screen._run out of the `wait_for` state. The Screen + # will then call its `_run_callback` and detect the success state and + # exit. + HardwareButtons.get_instance().trigger_override() + + # Exit the loop and thereby end this thread return textarea = TextArea( diff --git a/src/seedsigner/gui/screens/settings_screens.py b/src/seedsigner/gui/screens/settings_screens.py index 082ad497..794e5b2e 100644 --- a/src/seedsigner/gui/screens/settings_screens.py +++ b/src/seedsigner/gui/screens/settings_screens.py @@ -183,7 +183,7 @@ class IOTestScreen(BaseTopNavScreen): screen_y=int((self.canvas_height - msg_height)/ 2), ) while True: - input = self.hw_inputs.wait_for(keys=HardwareButtonsConstants.ALL_KEYS, check_release=False) + input = self.hw_inputs.wait_for(keys=HardwareButtonsConstants.ALL_KEYS) if input == HardwareButtonsConstants.KEY1: # Note that there are three distinct screen updates that happen at diff --git a/src/seedsigner/hardware/buttons.py b/src/seedsigner/hardware/buttons.py index f37dfa80..4261dd31 100644 --- a/src/seedsigner/hardware/buttons.py +++ b/src/seedsigner/hardware/buttons.py @@ -33,6 +33,7 @@ class HardwareButtons(Singleton): KEY2_PIN = 12 KEY3_PIN = 8 + @classmethod def get_instance(cls): # This is the only way to access the one and only instance @@ -53,7 +54,7 @@ class HardwareButtons(Singleton): cls._instance.GPIO = GPIO cls._instance.override_ind = False - cls._instance.add_events([HardwareButtonsConstants.KEY_UP, HardwareButtonsConstants.KEY_DOWN, HardwareButtonsConstants.KEY_PRESS, HardwareButtonsConstants.KEY_LEFT, HardwareButtonsConstants.KEY_RIGHT, HardwareButtonsConstants.KEY1, HardwareButtonsConstants.KEY2, HardwareButtonsConstants.KEY3]) + # cls._instance.add_events([HardwareButtonsConstants.KEY_UP, HardwareButtonsConstants.KEY_DOWN, HardwareButtonsConstants.KEY_PRESS, HardwareButtonsConstants.KEY_LEFT, HardwareButtonsConstants.KEY_RIGHT, HardwareButtonsConstants.KEY1, HardwareButtonsConstants.KEY2, HardwareButtonsConstants.KEY3]) # Track state over time so we can apply input delays/ignores as needed cls._instance.cur_input = None # Track which direction or button was last pressed @@ -65,7 +66,6 @@ class HardwareButtons(Singleton): return cls._instance - @classmethod def get_instance_no_hardware(cls): # This is the only way to access the one and only instance @@ -73,17 +73,23 @@ class HardwareButtons(Singleton): cls._instance = cls.__new__(cls) + def wait_for(self, keys=[]) -> int: + """ + Block execution until one of the target keys is pressed. - def wait_for(self, keys=[], check_release=True, release_keys=[]) -> int: + Optionally override the wait by calling `trigger_override()`. + """ # TODO: Refactor to keep control in the Controller and not here from seedsigner.controller import Controller controller = Controller.get_instance() - - if not release_keys: - release_keys = keys self.override_ind = False while True: + if self.override_ind: + # Break out of the wait_for without waiting for user input + self.override_ind = False + return HardwareButtonsConstants.OVERRIDE + cur_time = int(time.time() * 1000) if cur_time - self.last_input_time > controller.screensaver_activation_ms and not controller.is_screensaver_running: # Start the screensaver. Will block execution until input detected. @@ -99,48 +105,42 @@ class HardwareButtons(Singleton): # Resume from a fresh loop continue + # Check each candidate key to see if it was pressed for key in keys: - if not check_release or ((check_release and key in release_keys and HardwareButtonsConstants.release_lock) or check_release and key not in release_keys): - # when check release is False or the release lock is released (True) - if self.GPIO.input(key) == GPIO.LOW or self.override_ind: - HardwareButtonsConstants.release_lock = False - if self.override_ind: - self.override_ind = False - return HardwareButtonsConstants.OVERRIDE + if self.GPIO.input(key) == GPIO.LOW: + if self.cur_input != key: + self.cur_input = key + self.cur_input_started = int(time.time() * 1000) # in milliseconds + self.last_input_time = self.cur_input_started + return key - if self.cur_input != key: - self.cur_input = key - self.cur_input_started = int(time.time() * 1000) # in milliseconds - self.last_input_time = self.cur_input_started + else: + # Still pressing the same input + if cur_time - self.last_input_time > self.next_repeat_threshold: + # Too much time has elapsed to consider this the same + # continuous input. Treat as a new separate press. + self.cur_input_started = cur_time + self.last_input_time = cur_time + return key + + elif cur_time - self.cur_input_started > self.first_repeat_threshold: + # We're good to relay this immediately as continuous + # input. + self.last_input_time = cur_time return key else: - # Still pressing the same input - if cur_time - self.last_input_time > self.next_repeat_threshold: - # Too much time has elapsed to consider this the same - # continuous input. Treat as a new separate press. - self.cur_input_started = cur_time - self.last_input_time = cur_time - return key - - elif cur_time - self.cur_input_started > self.first_repeat_threshold: - # We're good to relay this immediately as continuous - # input. - self.last_input_time = cur_time - return key - - else: - # We're not yet at the first repeat threshold; triggering - # a key now would be too soon and yields a bad user - # experience when only a single click was intended but - # a second input is processed because of race condition - # against human response time to release the button. - # So there has to be a delay before we allow the first - # continuous repeat to register. So we'll ignore this - # round's input and **won't update any of our - # timekeeping vars**. But once we cross the threshold, - # we let the repeats fly. - pass + # We're not yet at the first repeat threshold; triggering + # a key now would be too soon and yields a bad user + # experience when only a single click was intended but + # a second input is processed because of race condition + # against human response time to release the button. + # So there has to be a delay before we allow the first + # continuous repeat to register. So we'll ignore this + # round's input and **won't update any of our + # timekeeping vars**. But once we cross the threshold, + # we let the repeats fly. + pass time.sleep(0.01) # wait 10 ms to give CPU chance to do other things @@ -149,29 +149,13 @@ class HardwareButtons(Singleton): self.last_input_time = int(time.time() * 1000) - def add_events(self, keys=[]): - for key in keys: - GPIO.add_event_detect(key, self.GPIO.RISING, callback=HardwareButtons.rising_callback) + def trigger_override(self) -> bool: + """ Set the override flag to break out of the current `wait_for` loop """ + self.override_ind = True - def rising_callback(channel): - HardwareButtonsConstants.release_lock = True - - - def trigger_override(self, force_release = False) -> bool: - if force_release: - HardwareButtonsConstants.release_lock = True - - if not self.override_ind: - self.override_ind = True - return True - return False - - def force_release(self) -> bool: - HardwareButtonsConstants.release_lock = True - return True - def check_for_low(self, key: int = None, keys: List[int] = None) -> bool: + """ Returns True if one of the target keys/key is pressed """ if key: keys = [key] for key in keys: @@ -181,12 +165,15 @@ class HardwareButtons(Singleton): else: return False + def has_any_input(self) -> bool: + """ Returns True if any of the keys are pressed """ for key in HardwareButtonsConstants.ALL_KEYS: if self.GPIO.input(key) == GPIO.LOW: return True return False + # class used as short hand for static button/channel lookup values # TODO: Implement `release_lock` functionality as a global somewhere. Mixes up design # patterns to have a static constants class plus a settable global value. @@ -227,5 +214,3 @@ class HardwareButtonsConstants: KEYS__LEFT_RIGHT_UP_DOWN = [KEY_LEFT, KEY_RIGHT, KEY_UP, KEY_DOWN] KEYS__ANYCLICK = [KEY_PRESS, KEY1, KEY2, KEY3] - - release_lock = True # released when True, locked when False diff --git a/src/seedsigner/views/seed_views.py b/src/seedsigner/views/seed_views.py index d478a4d0..35587630 100644 --- a/src/seedsigner/views/seed_views.py +++ b/src/seedsigner/views/seed_views.py @@ -1848,66 +1848,69 @@ class SeedAddressVerificationView(View): def run(self): # Start brute-force calculations from the zero-th index - self.addr_verification_thread.start() + try: + self.addr_verification_thread.start() - button_data = [self.SKIP_10, self.CANCEL] + button_data = [self.SKIP_10, self.CANCEL] - script_type_settings_entry = SettingsDefinition.get_settings_entry(SettingsConstants.SETTING__SCRIPT_TYPES) - script_type_display = script_type_settings_entry.get_selection_option_display_name_by_value(self.script_type) + script_type_settings_entry = SettingsDefinition.get_settings_entry(SettingsConstants.SETTING__SCRIPT_TYPES) + script_type_display = script_type_settings_entry.get_selection_option_display_name_by_value(self.script_type) - sig_type_settings_entry = SettingsDefinition.get_settings_entry(SettingsConstants.SETTING__SIG_TYPES) - sig_type_display = sig_type_settings_entry.get_selection_option_display_name_by_value(self.sig_type) + sig_type_settings_entry = SettingsDefinition.get_settings_entry(SettingsConstants.SETTING__SIG_TYPES) + sig_type_display = sig_type_settings_entry.get_selection_option_display_name_by_value(self.sig_type) - network_settings_entry = SettingsDefinition.get_settings_entry(SettingsConstants.SETTING__NETWORK) - network_display = network_settings_entry.get_selection_option_display_name_by_value(self.network) - mainnet = network_settings_entry.get_selection_option_display_name_by_value(SettingsConstants.MAINNET) + network_settings_entry = SettingsDefinition.get_settings_entry(SettingsConstants.SETTING__NETWORK) + network_display = network_settings_entry.get_selection_option_display_name_by_value(self.network) + mainnet = network_settings_entry.get_selection_option_display_name_by_value(SettingsConstants.MAINNET) - # Display the Screen to show the brute-forcing progress. - # Using a loop here to handle the SKIP_10 button presses to increment the counter - # and resume displaying the screen. User won't even notice that the Screen is - # being re-constructed. - while True: - selected_menu_num = self.run_screen( - seed_screens.SeedAddressVerificationScreen, - address=self.address, - derivation_path=self.derivation_path, - script_type=script_type_display, - sig_type=sig_type_display, - network=network_display, - is_mainnet=network_display == mainnet, - threadsafe_counter=self.threadsafe_counter, - verified_index=self.verified_index, - button_data=button_data, - ) + # Display the Screen to show the brute-forcing progress. + # Using a loop here to handle the SKIP_10 button presses to increment the counter + # and resume displaying the screen. User won't even notice that the Screen is + # being re-constructed. + while True: + selected_menu_num = self.run_screen( + seed_screens.SeedAddressVerificationScreen, + address=self.address, + derivation_path=self.derivation_path, + script_type=script_type_display, + sig_type=sig_type_display, + network=network_display, + is_mainnet=network_display == mainnet, + threadsafe_counter=self.threadsafe_counter, + verified_index=self.verified_index, + button_data=button_data, + ) + + if self.verified_index.cur_count is not None: + break + + if selected_menu_num == RET_CODE__BACK_BUTTON: + break + + if selected_menu_num is None: + # Only happens in the test suite; the screen isn't actually executed so + # it returns before the brute force thread has completed. + time.sleep(0.1) + continue + + if button_data[selected_menu_num] == self.SKIP_10: + self.threadsafe_counter.increment(10) + + elif button_data[selected_menu_num] == self.CANCEL: + break if self.verified_index.cur_count is not None: - break + # Successfully verified the addr; update the data + self.controller.unverified_address["verified_index"] = self.verified_index.cur_count + self.controller.unverified_address["verified_index_is_change"] = self.verified_index_is_change.cur_count == 1 + return Destination(SeedAddressVerificationSuccessView, view_args=dict(seed_num=self.seed_num)) - if selected_menu_num == RET_CODE__BACK_BUTTON: - break - - if selected_menu_num is None: - # Only happens in the test suite; the screen isn't actually executed so - # it returns before the brute force thread has completed. - time.sleep(0.1) - continue - - if button_data[selected_menu_num] == self.SKIP_10: - self.threadsafe_counter.increment(10) - - elif button_data[selected_menu_num] == self.CANCEL: - break - - if self.verified_index.cur_count is not None: - # Successfully verified the addr; update the data - self.controller.unverified_address["verified_index"] = self.verified_index.cur_count - self.controller.unverified_address["verified_index_is_change"] = self.verified_index_is_change.cur_count == 1 - return Destination(SeedAddressVerificationSuccessView, view_args=dict(seed_num=self.seed_num)) - - else: + finally: # Halt the thread if the user gave up (will already be stopped if it verified the # target addr). self.addr_verification_thread.stop() + + # Block until the thread has stopped while self.addr_verification_thread.is_alive(): time.sleep(0.01) @@ -1933,7 +1936,7 @@ class SeedAddressVerificationView(View): if self.seed: self.xpub = self.seed.get_xpub(wallet_path=self.derivation_path, network=Settings.get_instance().get_value(SettingsConstants.SETTING__NETWORK)) - + def run(self): from seedsigner.helpers import embit_utils @@ -1965,7 +1968,7 @@ class SeedAddressVerificationView(View): # Increment our index counter self.threadsafe_counter.increment() - + class SeedAddressVerificationSuccessView(View):