magscope.scope

Core orchestration for the MagScope application.

MagScope is the parent process that builds every other manager process, connects them with shared resources, and relays inter-process messages until shutdown. Its responsibilities span the full application lifetime:

• Loading persisted settings from QSettings, with optional YAML import/export available through the user interface.

• Constructing manager processes (camera, bead lock, GUI, scripting, video processing, and optional hardware integrations) and wiring them to shared locks, buffers, and IPC pipes.

• Running the main IPC loop that forwards typed IPC commands between processes and supervises orderly shutdown.

MagScope.start prepares the shared resources, registers scriptable hooks, starts each process, and then loops until a quit command is received.

Example

Run the simulated scope with its default managers:

>>> from magscope.scope import MagScope
>>> scope = MagScope()
>>> scope.start()

For headless automation you can add hardware adapters and GUI panels before invoking MagScope.start():

>>> scope.add_hardware(custom_hardware_manager)
>>> scope.add_control(CustomPanel, column=0)
>>> scope.start()

MagScope constructs the following high-level pipeline:

CameraManager → VideoBuffer → VideoProcessorManager → UIManager and BeadLockManager → MatrixBuffer → UIManager

Every manager receives shared locks, pipes, and configuration from the main process so that real-time video frames, bead tracking data, and scripted events remain synchronized.

Attributes

logger

Classes

MagScope

Coordinate MagScope managers, shared resources, and IPC.

Module Contents

magscope.scope.logger

class magscope.scope.MagScope(*, verbose: bool = False, print_ipc_commands: bool = False, print_script_commands: bool = False)

Coordinate MagScope managers, shared resources, and IPC.

MagScope owns references to every manager process, shared buffer, and IPC primitive used by the application. Instances can be customized by adding hardware managers, GUI controls, or time-series plots before calling start(). Once started, the instance supervises manager lifetimes, relays messages, and coordinates shutdown when a quit signal is broadcast over the IPC bus. The orchestrator is a singleton: attempts to construct a second instance raise TypeError. Each instance is single-use: calling start() while an instance is already running logs a warning, and invoking start() after the instance has quit raises an error. Use stop() to request a graceful shutdown; it blocks until all managers acknowledge the quit sequence and exit.

beadlock_manager

camera_manager

video_processor_manager

zlut_generation_manager

ui_manager

script_manager

_hardware: dict[str, magscope.hardware.HardwareManagerBase]

_hardware_buffers: dict[str, magscope.datatypes.MatrixBuffer]

processes: dict[str, magscope.processes.ManagerProcessBase]

command_registry: magscope.ipc.CommandRegistry

locks: dict[str, multiprocessing.synchronize.Lock]

lock_names: list[str] = ['BeadRoiBuffer', 'LiveProfileBuffer', 'TracksBuffer', 'VideoBuffer',...

pipes: dict[str, multiprocessing.connection.Connection]

quitting_events: dict[str, multiprocessing.synchronize.Event]

shared_values: magscope.processes.InterprocessValues

_quitting: multiprocessing.Event

_settings

_running: bool = False

_log_level = 20

_command_registry_initialized: bool = False

_print_ipc_commands = False

_print_script_commands = False

_terminated: bool = False

_startup_splash_deadline: float | None = None

_startup_splash_close_event: multiprocessing.Event | None = None

_startup_splash_process: multiprocessing.Process | None = None

_startup_splash_timeout_seconds: float = 600.0

_startup_splash_waiting_for_ui_ready: bool = False

_camera_health_log_interval_seconds: float = 60.0

_next_camera_health_log_deadline: float | None = None

_last_camera_health_sample_time: float | None = None

_last_camera_health_frame_count: int = 0

live_profile_buffer: magscope.datatypes.LiveProfileBuffer | None = None

bead_roi_buffer: magscope.datatypes.BeadRoiBuffer | None = None

tracks_buffer: magscope.datatypes.MatrixBuffer | None = None

video_buffer: magscope.datatypes.VideoBuffer | None = None

start()

Launch all managers and enter the main IPC loop.

The startup sequence performs the following steps:

1. Collect every manager (built-in and user-supplied hardware) and assign them a shared processes mapping for bookkeeping.

2. Load configuration values, prepare shared memory buffers, locks, pipes, and register scriptable methods.

3. Spawn each manager process and then forward IPC messages until a quit signal is observed.

When a quit message is received the method joins every process before returning control to the caller.

stop() → None

Request a graceful shutdown and wait for every manager to exit.

stop mirrors a quit request sent from any manager process: it broadcasts a quit message, drains outstanding IPC, and blocks until all managers have joined. After stop completes the instance is permanently terminated and cannot be restarted.

add_hardware(hardware: magscope.hardware.HardwareManagerBase)

Register a hardware manager so its process launches with MagScope.

add_control(control_type: type[magscope.ui.controls.ControlPanelBase], column: int)

Schedule a GUI control panel to be added when the UI manager starts.

add_timeplot(plot: magscope.ui.plots.TimeSeriesPlotBase)

Schedule a time-series plot for inclusion in the GUI at startup.

property print_ipc_commands: bool

Return whether start() should print IPC commands and exit early.

property print_script_commands: bool

Return whether start() should print script commands and exit early.

_coerce_settings(value: magscope.settings.MagScopeSettings | dict) → magscope.settings.MagScopeSettings

property settings

classmethod _reset_singleton_for_testing() → None

Clear the singleton registry so tests can create fresh instances.

set_verbose_logging(enabled: bool = True) → None

Toggle informational console output for MagScope internals.

_mark_running() → bool

Mark the orchestrator as running if it is not already active.

_ensure_not_terminated() → None

Prevent reusing a MagScope instance after it has been stopped.

_apply_logging_preferences() → None

Apply the current verbosity preference to the logging system.

_mark_terminated() → None

Record that this MagScope instance has finished its lifecycle.

_start_startup_splash() → None

Launch a lightweight splash window in a helper process.

_dismiss_startup_splash_if_pending() → None

Dismiss the splash while startup is still waiting on the UI.

_stop_startup_splash() → None

Request the startup splash helper process to exit.

_check_startup_splash_timeout() → None

Dismiss the splash if UI startup has been pending too long.

_collect_processes() → None

Assemble the ordered list of manager processes to supervise.

ScriptManager must remain first so that the @register_script_command decorator binds correctly before other managers start.

_setup_command_registry() → None

Register all command handlers and validate destinations.

print_registered_commands() → None

Print the registered IPC commands without launching managers.

print_registered_script_commands() → None

Print the registered script commands without launching managers.

_initialize_shared_state() → None

Load configuration and prepare shared resources for all managers.

_start_managers() → None

Start each manager process.

_main_ipc_loop() → None

Forward IPC messages until a quit request is observed.

_join_processes() → None

Join every managed process once shutdown has been requested.

receive_ipc()

Poll every IPC pipe once and relay any commands that arrive.

_read_command(pipe: multiprocessing.connection.Connection) → magscope.ipc_commands.Command | object | None

Retrieve a command from pipe if one is waiting.

_process_command(command: magscope.ipc_commands.Command) → bool

Route a valid command and indicate whether the IPC loop should break.

_route_command(command: magscope.ipc_commands.Command) → bool

Dispatch a command based on its destination.

Returns True when the IPC loop should stop iterating over the current set of pipes (for example, immediately after handling a quit broadcast). This mirrors the previous behavior of breaking out of the receive_ipc loop once a quit command has been processed.

_dispatch_mag_scope_command(command: magscope.ipc_commands.Command, spec) → None

Handle commands destined for the MagScope orchestrator.

_handle_broadcast_command(command: magscope.ipc_commands.Command, spec=None) → bool

Broadcast a command to all processes and handle quit semantics.

Returns True when the caller should stop processing the current IPC loop (e.g., after handling a quit command).

log_exception(process_name: str, details: str) → None

Surface an exception raised in a managed process.

startup_ready(process_name: str = 'UIManager') → None

Dismiss the startup splash once the UI process is ready.

_sleep_when_idle() → None

Throttle the IPC loop when no messages were processed.

_reset_camera_health_logging_state() → None

Start a fresh sampling window for periodic camera health logging.

_log_camera_health_if_due() → None

Emit a 1-minute camera health summary while verbose logging is enabled.

_drain_child_pipes_after_quit() → None

Drain child pipes until they acknowledge the quit event.

_setup_shared_resources()

Create and distribute shared locks, pipes, buffers, and metadata.

_configure_processes_with_shared_resources()

Share locks, pipes, and configuration with each process.

This step must occur before any manager processes are started so they can inherit references to shared multiprocessing primitives.

_create_shared_buffers()

Instantiate shared memory buffers used throughout the application.

_setup_locks()

Instantiate per-buffer locks and make them available to processes.

_setup_pipes() → dict[str, multiprocessing.connection.Connection]

Create duplex pipes that allow processes to exchange messages.

_register_script_methods()

Expose manager methods to the scripting subsystem.

update_settings(settings: magscope.settings.MagScopeSettings | dict) → None

Replace the active settings and broadcast them to all managers.