Merge pull request #655 from kdmukai/hardwarebutton_refactor

[Refactor / bugfix] Simplify `HardwareButtons`
This commit is contained in:
Nick Klockenga
2025-01-13 14:52:03 -05:00
committed by GitHub
5 changed files with 135 additions and 157 deletions
+16 -17
View File
@@ -1,3 +1,4 @@
import logging
import time
from dataclasses import dataclass, field
@@ -15,6 +16,8 @@ from seedsigner.models.encode_qr import BaseQrEncoder
from seedsigner.models.settings import SettingsConstants
from seedsigner.models.threads import BaseThread, ThreadsafeCounter
logger = logging.getLogger(__name__)
# Must be huge numbers to avoid conflicting with the selected_button returned by the
# screens with buttons.
@@ -71,6 +74,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 +236,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 +453,7 @@ class ButtonListScreen(BaseTopNavScreen):
while True:
ret = self._run_callback()
if ret is not None:
logging.info("Exiting ButtonListScreen due to _run_callback")
return ret
user_input = self.hw_inputs.wait_for(
@@ -455,9 +462,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 +646,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 +861,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
@@ -1192,9 +1193,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:
+15 -20
View File
@@ -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
@@ -1466,13 +1458,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
@@ -1489,10 +1481,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(
@@ -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
+49 -68
View File
@@ -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,8 +54,6 @@ 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])
# 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
cls._instance.cur_input_started = None # Track when that input began
@@ -65,7 +64,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 +71,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 +103,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 +147,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,15 +163,16 @@ 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.
class HardwareButtonsConstants:
if GPIO.RPI_INFO['P1_REVISION'] == 3: #This indicates that we have revision 3 GPIO
KEY_UP = 31
@@ -227,5 +210,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
+52 -49
View File
@@ -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)