Revamp UI interactions

.gitignore

@@ -29,3 +29,4 @@ venv37
 .vscode/settings.json
 docs/_build/
 .DS_Store
+.env

docs/release-notes.rst

@@ -26,7 +26,9 @@ Improvements
 - Created a new FixateError base class for all exceptions raised by fixate to use. It inherits from Exception instead of BaseExcepetion to improve error handling.
 - DSO Driver function 'waveform_values' now returns a single channels x and y data as two separate lists, without re-acquiring the signal. This function should
  now be called after performing signal acquisition.
-- Invert channel and vtime functions implemented in the DSO driver.
+- UI has been revamped a little. User input methods will no longer return ("Resp", value) but will return just the value - see Issue #177. To use these, from fixate import user_info etc.
+- Fixed a bug where falsy values such as 0 or 0.0 were not accepted as valid user input - see Issue #213
+- Invert channel and vtime funcitons implemented in the DSO driver.
 - Fxconfig fixes for cmd2 4.0.0
 
 *************

src/fixate/__init__.py

@@ -30,6 +30,25 @@
  generate_relay_matrix_pin_list as generate_relay_matrix_pin_list,
 )
 
+from fixate._ui import (
+ Validator as Validator,
+ UiColour as UiColour,
+ user_input as user_input,
+ user_input_float as user_input_float,
+ user_serial as user_serial,
+ user_yes_no as user_yes_no,
+ user_info as user_info,
+ user_info_important as user_info_important,
+ user_ok as user_ok,
+ user_action as user_action,
+ user_image as user_image,
+ user_image_clear as user_image_clear,
+ user_gif as user_gif,
+ user_post_sequence_info_pass as user_post_sequence_info_pass,
+ user_post_sequence_info_fail as user_post_sequence_info_fail,
+ user_post_sequence_info as user_post_sequence_info,
+)
+
 from fixate.main import run_main_program as run
 
 __version__ = "0.6.5"

src/fixate/_ui.py

@@ -0,0 +1,345 @@
+"""
+This module provides the user interface for fixate. It is agnostic of the
+actual implementation of the UI and provides a standard set of functions used
+to obtain or display information from/to the user.
+"""
+
+from typing import Callable, Literal
+from queue import Queue, Empty
+from enum import StrEnum
+import time
+from pubsub import pub
+
+# going to honour the post sequence info display from `ui.py`
+from fixate.config import RESOURCES
+from fixate.core.exceptions import UserInputError
+from collections import OrderedDict
+
+
+class Validator[T]:
+ """
+ Defines a validator object that can be used to validate user input.
+ """
+
+ def __init__(self, func: Callable[[T], bool], error_msg: str = "Invalid input"):
+ """
+ Args:
+ func (function): The function to validate the input
+ error_msg (str): The message to display if the input is invalid
+ """
+ self.func = func
+ self.error_msg = error_msg
+
+ def __call__(self, resp: T) -> bool:
+ """
+ Args:
+ resp (Any): The response to validate
+
+ Returns:
+ bool: True if the response is valid, False otherwise
+ """
+ return self.func(resp)
+
+ def __str__(self) -> str:
+ return self.error_msg
+
+
+class UiColour(StrEnum):
+ RED = "red"
+ GREEN = "green"
+ BLUE = "blue"
+ YELLOW = "yellow"
+ WHITE = "white"
+ BLACK = "black"
+ CYAN = "cyan"
+ MAGENTA = "magenta"
+ GREY = "grey"
+
+
+def _user_request_input(msg: str) -> str:
+ q: Queue[str] = Queue()
+ pub.sendMessage("UI_block_start")
+ pub.sendMessage("UI_req_input", msg=msg, q=q)
+ resp = q.get()
+ pub.sendMessage("UI_block_end")
+ return resp
+
+
+def user_input(msg: str) -> str:
+ """
+ A blocking function that asks the UI to ask the user for raw input.
+
+ Args:
+ msg (str): A message that will be shown to the user
+
+ Returns:
+ resp (str): The user response from the UI
+ """
+ return _user_request_input(msg)
+
+
+def user_input_float(msg: str, attempts: int = 5) -> float:
+ """
+ A blocking function that asks the UI to ask the user for input and converts the response to a float.
+
+ Args:
+ msg (str): A message that will be shown to the user
+ attempts (int): Number of attempts the user has to get the input right
+
+ Returns:
+ resp (float): The converted user response from the UI
+
+ Raises:
+ UserInputError: If the user fails to enter a number after the specified number of attempts
+ """
+ for _ in range(attempts):
+ resp = _user_request_input(msg)
+ try:
+ return float(resp)
+ except ValueError:
+ pub.sendMessage(
+ "UI_display_important", msg="Invalid input, please enter a number"
+ )
+ raise UserInputError("User failed to enter a number")
+
+
+def _ten_digit_int_serial(serial: str) -> bool:
+ return len(serial) == 10 and serial.isdigit()
+
+
+_ten_digit_int_serial_v = Validator(
+ _ten_digit_int_serial, "Please enter a 10 digit serial number"
+)
+
+
+def user_serial(
+ msg: str,
+ validator: Validator = _ten_digit_int_serial_v,
+ return_type: type[int] | type[str] = int,
+ attempts: int = 5,
+) -> int | str:
+ """
+ A blocking function that asks the UI to ask the user for a serial number.
+
+ Args:
+ msg (str): A message that will be shown to the user
+ validator (Validator): An optional function to validate the serial number,
+ defaults to checking for a 10 digit integer. This function shall return
+ True if the serial number is valid, False otherwise.
+ return_type (int | str): The type to return the serial number as, defaults to int
+
+ Returns:
+ resp (str): The user response from the UI
+ """
+ for _ in range(attempts):
+ resp = _user_request_input(msg)
+ if validator(resp):
+ return return_type(resp)
+ pub.sendMessage("UI_display_important", msg=f"Invalid input: {validator}")
+ raise UserInputError("User failed to enter the correct format serial number")
+
+
+def _user_req_choices(msg: str, choices: tuple[str, ...]) -> str:
+ assert len(choices) >= 2, "There must be at least 2 choices"
+ q: Queue[str] = Queue()
+ pub.sendMessage("UI_block_start")
+ pub.sendMessage("UI_req_choices", msg=msg, q=q, choices=choices)
+ resp = q.get()
+ pub.sendMessage("UI_block_end")
+ return resp
+
+
+def _choice_from_response(choices: tuple[str, ...], resp: str) -> str | Literal[False]:
+ for choice in choices:
+ if choice.startswith(resp):
+ return choice
+ return False
+
+
+def _user_choices(msg: str, choices: tuple[str, ...], attempts: int = 5) -> str:
+ for _ in range(attempts):
+ resp = _user_req_choices(msg, choices).upper()
+ choice = _choice_from_response(choices, resp)
+ # because of how _choice_from_response works, choice will only be False if the user provided an invalid response
+ # which is only possible in the command line UI, otherwise choice will be one of the responses.
+ if choice:
+ return choice
+ pub.sendMessage(
+ "UI_display_important",
+ msg="Invalid input, please enter a valid choice; first letter or full word",
+ )
+ raise UserInputError("User failed to enter a valid response")
+
+
+def user_yes_no(msg: str, attempts: int = 1) -> str:
+ """
+ A blocking function that asks the UI to ask the user for a yes or no response.
+
+ Args:
+ msg (str): A message that will be shown to the user
+
+ Returns:
+ resp (str): 'YES' or 'NO'
+ """
+ CHOICES = ("YES", "NO")
+ return _user_choices(msg, CHOICES, attempts)
+
+
+def user_retry_abort_fail(msg: str, attempts: int = 1) -> str:
+ """This should only ever be used by the sequencer, as such is should never be included in the public API via src/fixate/__init__.py"""
+ CHOICES = ("RETRY", "ABORT", "FAIL")
+ return _user_choices(msg, CHOICES, attempts)
+
+
+def user_info(msg: str):
+ pub.sendMessage("UI_display", msg=msg)
+
+
+def user_info_important(
+ msg: str, colour: UiColour = UiColour.RED, bg_colour: UiColour = UiColour.WHITE
+):
+ pub.sendMessage("UI_display_important", msg=msg, colour=colour, bg_colour=bg_colour)
+
+
+def user_ok(msg: str):
+ """
+ A blocking function that asks the UI to display a message and waits for the user to press OK/Enter.
+ """
+ pub.sendMessage("UI_block_start")
+ pub.sendMessage("UI_req", msg=msg)
+ pub.sendMessage("UI_block_end")
+
+
+def user_action(msg: str, action_monitor: Callable[[], bool]) -> bool:
+ """
+ Prompts the user to complete an action.
+ Actively monitors the target infinitely until the event is detected or a user fail event occurs
+
+ Args:
+ msg (str): Message to display to the user
+ action_monitor (function): A function that will be called until the user action is cancelled. The function
+ should return False if it hasn't completed. If the action is finished return True.
+
+ Returns:
+ bool: True if the action is finished, False otherwise
+ """
+
+ # UserActionCallback is used to handle the cancellation of the action either by the user or by the action itself
+ class UserActionCallback:
+ def __init__(self):
+ # The UI implementation must provide queue.Queue object. We
+ # monitor that object. If it is non-empty, we get the message
+ # in the q and cancel the target call.
+ self.user_cancel_queue: Queue | None = None
+
+ # In the case that the target exits the user action instead
+ # of the user, we need to tell the UI to do any clean up that
+ # might be required. (e.g. return GUI buttons to the default state
+ # Does not need to be implemented by the UI.
+ # Function takes no args and should return None.
+ self.target_finished_callback: Callable[[], None] = lambda: None
+
+ def set_user_cancel_queue(self, cancel_queue: Queue):
+ self.user_cancel_queue = cancel_queue
+
+ def set_target_finished_callback(self, callback: Callable[[], None]):
+ self.target_finished_callback = callback
+
+ callback_obj = UserActionCallback()
+ pub.sendMessage("UI_action", msg=msg, callback_obj=callback_obj)
+ # at this point we should have a cancel queue and a target finished callback, if not, the developer has not implemented the UI_action topic correctly.
+ assert (
+ callback_obj.user_cancel_queue is not None
+ and callback_obj.target_finished_callback is not None
+ ), "user_cancel_queue and target_finished_callback must be set in the UI call for UI_action topic"
+ try:
+ while True:
+ try:
+ callback_obj.user_cancel_queue.get_nowait()
+ return False
+ except Empty:
+ pass
+
+ if action_monitor():
+ return True
+
+ # Yield control for other threads but don't slow down target
+ time.sleep(0)
+ finally:
+ # No matter what, if we exit, we want to reset the UI
+ callback_obj.target_finished_callback()
+
+
+def user_image(path: str):
+ """
+ Display an image to the user
+
+ Args:
+ path (str): The path to the image file. The underlying library does not take a pathlib.Path object.
+ """
+ pub.sendMessage("UI_image", path=path)
+
+
+def user_image_clear():
+ """
+ Clear the image canvas
+ """
+ pub.sendMessage("UI_image_clear")
+
+
+def user_gif(path: str):
+ """
+ Display a gif to the user
+
+ Args:
+ path (str): The path to the gif file. The underlying library does not take a pathlib.Path object.
+ """
+ pub.sendMessage("UI_gif", path=path)
+
+
+def _user_post_sequence_info(msg: str, status: str):
+ if "_post_sequence_info" not in RESOURCES["SEQUENCER"].context_data:
+ RESOURCES["SEQUENCER"].context_data["_post_sequence_info"] = OrderedDict()
+ RESOURCES["SEQUENCER"].context_data["_post_sequence_info"][msg] = status
+
+
+def user_post_sequence_info_pass(msg: str):
+ """
+ Adds information to be displayed to the user at the end if the sequence passes
+ This information will be displayed in the order that this function is called.
+ Multiple calls with the same message will result in the previous being overwritten.
+
+ This is useful for providing a summary of the sequence to the user at the end.
+
+ Args:
+ msg (str): The message to display.
+ """
+ _user_post_sequence_info(msg, "PASSED")
+
+
+def user_post_sequence_info_fail(msg: str):
+ """
+ Adds information to be displayed to the user at the end if the sequence fails.
+ This information will be displayed in the order that this function is called.
+ Multiple calls with the same message will result in the previous being overwritten.
+
+ This is useful for providing a summary of the sequence to the user at the end.
+
+ Args:
+ msg (str): The message to display.
+ """
+ _user_post_sequence_info(msg, "FAILED")
+
+
+def user_post_sequence_info(msg: str):
+ """
+ Adds information to be displayed to the user at the end of the sequence.
+ This information will be displayed in the order that this function is called.
+ Multiple calls with the same message will result in the previous being overwritten.
+
+ This is useful for providing a summary of the sequence to the user at the end.
+
+ Args:
+ msg (str): The message to display.
+ """
+ _user_post_sequence_info(msg, "ALL")