User-interface and worker API

Measurement logic communicates through a small adapter interface. This keeps terminal and Qt front ends separate from the measurement environment. Blocking device operations belong in a worker thread; adapter calls transport prompts, progress, interruption, and log information.

class mpylab.env.ui.ui_adapter.MeasurementUIAdapter

Abstract UI adapter contract for measurement flows.

ask(msg: str = 'Are you ready?', buttons: list[str] | None = None, level: str = '', data: dict[Any, Any] | None = None) → int

Display a question and return the selected button index.

Parameters:

  • msg (str, optional) – User-facing prompt text.

  • buttons (list of str or None, optional) – Available answer labels.

  • level (str, optional) – Message severity or presentation level.

  • data (dict or None, optional) – Structured context supplied with the prompt.

check_interrupt() → int | None

Check for an interrupt request using key polling.

Returns:

Pending key code, or None when no interrupt was requested.

Return type:

int or None

emit_log(block, *args) → None

Emit one structured log block.

Parameters:

  • block (object) – Log block or message.

  • *args (object) – Additional arguments forwarded to logger callbacks.

poll_key() → int | None

Poll one pending key event, if any.

post_user_event() → None

Run hook after user-facing event.

pre_user_event() → None

Run hook before user-facing event.

run_interactive(obj: Any, banner: str) → None

Start an interactive shell for an object.

Parameters:

  • obj (object) – Object exposed to the interactive session.

  • banner (str) – Introductory session text.

class mpylab.env.ui.ui_adapter.TUIAdapter(messenger: Callable[[str, list[str] | None, str, dict[Any, Any] | None], int], logger: list[Callable[[...], Any]] | None, interrupt_tester: Callable[[], int | None], pre_user_event: Callable[[], None], post_user_event: Callable[[], None], interactive_runner: Callable[[Any, str], None])

Default terminal adapter using callbacks supplied by Measure.

Parameters:

  • messenger (callable) – Callback that displays prompts and returns an answer index.

  • logger (list of callable or None) – Callbacks receiving structured log blocks.

  • interrupt_tester (callable) – Non-blocking callback returning a pending key code.

  • pre_user_event (callable) – Hook invoked before user interaction.

  • post_user_event (callable) – Hook invoked after user interaction.

  • interactive_runner (callable) – Callback starting an interactive session.

ask(msg: str = 'Are you ready?', buttons: list[str] | None = None, level: str = '', data: dict[Any, Any] | None = None) → int

Forward a prompt to the configured messenger.

Parameters:

  • msg (str, optional) – User-facing prompt text.

  • buttons (list of str or None, optional) – Available answer labels.

  • level (str, optional) – Message severity or presentation level.

  • data (dict or None, optional) – Structured context supplied with the prompt.

Returns:

Index of the selected answer.

Return type:

int

emit_log(block, *args) → None

Forward one log block to all configured loggers.

Parameters:

  • block (object) – Log block or message.

  • *args (object) – Additional arguments forwarded to every logger callback.

poll_key() → int | None

Poll the configured interrupt tester.

Returns:

Pending key code, or None when no key is available.

Return type:

int or None

post_user_event() → None

Call post-user-event hook.

pre_user_event() → None

Call pre-user-event hook.

run_interactive(obj: Any, banner: str) → None

Call the interactive runner with an object and banner.

Parameters:

  • obj (object) – Object exposed to the interactive session.

  • banner (str) – Introductory session text.

set_interactive_runner(runner: Callable[[Any, str], None]) → None

Replace the interactive-session callback.

Parameters:

runner (callable) – Callback accepting the exposed object and session banner.

set_interrupt_tester(tester: Callable[[], int | None]) → None

Replace the interrupt polling callback.

Parameters:

tester (callable) – Non-blocking callback returning a pending key code.

set_logger(logger: list[Callable[[...], Any]] | None) → None

Replace the logger callback list.

Parameters:

logger (list of callable or None) – Callbacks receiving structured log blocks.

set_messenger(messenger: Callable[[str, list[str] | None, str, dict[Any, Any] | None], int]) → None

Replace the messenger callback.

Parameters:

messenger (callable) – Callback that displays prompts and returns an answer index.

set_post_user_event(cb: Callable[[], None]) → None

Replace the post-user-event callback.

Parameters:

cb (callable) – Hook invoked after user interaction.

set_pre_user_event(cb: Callable[[], None]) → None

Replace the pre-user-event callback.

Parameters:

cb (callable) – Hook invoked before user interaction.

mpylab.env.ui.ui_adapter.resolve_poll_key(handler: Any, caller_locals: dict[str, Any] | None = None) → Callable[[], int | None] | None

Resolve a poll-key callable from a handler or caller context.

Parameters:

  • handler (object) – Explicit polling callback, if callable.

  • caller_locals (dict or None, optional) – Caller namespace whose self.PollKey method is used as a fallback.

Returns:

Non-blocking key polling callback, or None when unavailable.

Return type:

callable or None

Qt adapter and runner

class mpylab.env.ui.qt_adapter.QtUIAdapter(interrupt_key: str = 'q')

Qt-backed measurement UI adapter.

The stop button does not access hardware directly. It sets an interrupt flag that is returned once by poll_key(), allowing the existing measurement safety path to call RFOff_Devices from the measurement worker thread.

Parameters:

interrupt_key (str, optional) – Synthetic key returned once after a stop request. The first character is used and defaults to "q".

answer_prompt_with_quit() → bool

Answer an active prompt with Quit when available.

Returns:

True if a matching answer was submitted, otherwise False.

Return type:

bool

ask(msg: str = 'Are you ready?', buttons: list[str] | None = None, level: str = '', data: dict[Any, Any] | None = None) → int

Display a prompt and wait for the selected button index.

This method is called from the measurement worker. Communication with Qt widgets uses queued signals; waiting occurs on a condition and does not block the Qt event loop.

Parameters:

  • msg (str, optional) – User-facing prompt text.

  • buttons (list of str or None, optional) – Available answers. None provides Ok and Quit.

  • level (str, optional) – Message severity or presentation level.

  • data (dict or None, optional) – Structured context supplied with the prompt.

Returns:

Selected answer index, or -1 for an informational message without buttons.

Return type:

int

create_widget(parent=None) → MeasurementControlWidget

Create the control widget for this adapter.

Parameters:

parent (QWidget or None, optional) – Parent widget.

Returns:

Newly created control widget.

Return type:

MeasurementControlWidget

emit_log(block, *args) → None

Emit log data to Qt and configured logger callbacks.

Parameters:

  • block (object) – Log block or message displayed as text in Qt.

  • *args (object) – Additional arguments forwarded to logger callbacks.

poll_key() → int | None

Return a synthetic key code after a Stop / RF Off request.

Returns:

Configured interrupt key once per request, otherwise None.

Return type:

int or None

post_user_event() → None

Mark UI state after a user-facing event.

pre_user_event() → None

Mark UI state before a user-facing event.

request_interrupt() → None

Request measurement interruption at the next poll point.

request_stop() → None

Request stop, answering an active Quit prompt if possible.

run_interactive(obj: Any, banner: str) → None

Report that terminal-style interactive mode is unavailable.

Parameters:

  • obj (object) – Object that would have been exposed to an interactive session. It is not used by the Qt adapter.

  • banner (str) – Session description included in the log message.

set_configuration(config, trace=None) → None

Store a read-only effective configuration for the Qt view.

Parameters:

  • config (dict) – Effective configuration before runtime UI objects are attached.

  • trace (ConfigTrace or None, optional) – Ordered provenance of the configuration values.

Raises:

TypeError – If the configuration or trace cannot be represented.

set_logger(logger: list[Any] | None) → None

Set optional additional logger callbacks.

Parameters:

logger (list or None) – Objects that are callable are retained as log callbacks.

submit_answer(index: int) → None

Submit a prompt answer from the UI thread.

Parameters:

index (int) – Zero-based answer index.

submit_eut_status(status: str) → bool

Submit a manual during-exposure EUT observation from Qt.

Parameters:

status (str) – Functional-state identifier accepted by the manual EUT monitor.

Returns:

True if the observation was accepted, otherwise False.

Return type:

bool

submit_post_exposure_state(state: str, recovery: str) → bool

Submit post-exposure state and recovery observations.

Parameters:

  • state (str) – Functional state observed after exposure.

  • recovery (str) – Recovery category associated with that state.

Returns:

True if the observation was accepted, otherwise False.

Return type:

bool

class mpylab.env.ui.qt_adapter.MeasurementControlWidget(*args: Any, **kwargs: Any)

Display measurement messages, prompts, EUT state, and stop controls.

Parameters:

  • adapter (QtUIAdapter) – Adapter receiving all user actions from the widget.

  • parent (QWidget or None, optional) – Parent widget.

append_log(text: str) → None

Append one log message to the output pane.

clear_prompt() → None

Clear prompt text and dynamic buttons.

show_prompt(msg: str, buttons: list[str], level: str = '', data: object = None) → None

Show a prompt and create one button per answer.

class mpylab.env.qt_runner.MeasurementTask(measurement_module, config)

Run a measurement script in a Qt worker thread.

Parameters:

  • measurement_module (module) – Module providing a run(config) function.

  • config (dict) – Measurement configuration passed to the module.

run()

Execute the configured measurement.

mpylab.env.qt_runner.attach_ui_adapter(config, ui_adapter)

Attach a Qt UI adapter and remove non-interactive callbacks.

Parameters:

  • config (dict) – Measurement configuration modified in place.

  • ui_adapter (QtUIAdapter) – Adapter and optional manual EUT monitor to attach.

mpylab.env.qt_runner.run_qt_script(script_path, argv=None, module_name=None, window_title=None)

Start a Qt UI and run a measurement script in a worker thread.

Parameters:

  • script_path (path-like) – Path to the measurement script.

  • argv (sequence of str or None, optional) – Configuration arguments. None uses command-line arguments.

  • module_name (str or None, optional) – Import name assigned to the loaded script module.

  • window_title (str or None, optional) – Title overriding the adapter widget’s default title.

Returns:

Qt application exit code, or 2 for an invalid configuration.

Return type:

int

Long receiver scans

class mpylab.device.receiver_scan_worker.ReceiverScanWorker(receiver, scan_kwargs=None, on_point=None, on_done=None, on_error=None, on_cancel=None, safety_stop=None)

Run a receiver RunScan call in a background thread.

Callbacks are executed from the worker thread. GUI frontends should bridge them into the UI thread, for example with Qt signals or a thread-safe queue.

is_running()

Return whether the scan thread is currently active.

join(timeout=None)

Wait for the worker thread and return whether it has finished.

request_stop()

Request cancellation and call the optional safety stop callback.

start()

Start the receiver scan in a daemon thread and return self.

wait_finished(timeout=None)

Wait for the worker completion event.

See UI adapters and worker interface for threading, stop, RF-off, and EUT monitoring guidance.