magscope.scope ============== .. py:module:: magscope.scope .. autoapi-nested-parse:: 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. .. rubric:: 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 :meth:`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 ---------- .. autoapisummary:: magscope.scope.logger Classes ------- .. autoapisummary:: magscope.scope.MagScope Module Contents --------------- .. py:data:: logger .. py:class:: 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 :meth:`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 :meth:`start` while an instance is already running logs a warning, and invoking :meth:`start` after the instance has quit raises an error. Use :meth:`stop` to request a graceful shutdown; it blocks until all managers acknowledge the quit sequence and exit. .. py:attribute:: beadlock_manager .. py:attribute:: camera_manager .. py:attribute:: video_processor_manager .. py:attribute:: zlut_generation_manager .. py:attribute:: ui_manager .. py:attribute:: script_manager .. py:attribute:: _hardware :type: dict[str, magscope.hardware.HardwareManagerBase] .. py:attribute:: _hardware_buffers :type: dict[str, magscope.datatypes.MatrixBuffer] .. py:attribute:: processes :type: dict[str, magscope.processes.ManagerProcessBase] .. py:attribute:: command_registry :type: magscope.ipc.CommandRegistry .. py:attribute:: locks :type: dict[str, multiprocessing.synchronize.Lock] .. py:attribute:: lock_names :type: list[str] :value: ['BeadRoiBuffer', 'LiveProfileBuffer', 'TracksBuffer', 'VideoBuffer',... .. py:attribute:: pipes :type: dict[str, multiprocessing.connection.Connection] .. py:attribute:: quitting_events :type: dict[str, multiprocessing.synchronize.Event] .. py:attribute:: shared_values :type: magscope.processes.InterprocessValues .. py:attribute:: _quitting :type: multiprocessing.Event .. py:attribute:: _settings .. py:attribute:: _running :type: bool :value: False .. py:attribute:: _log_level :value: 20 .. py:attribute:: _command_registry_initialized :type: bool :value: False .. py:attribute:: _print_ipc_commands :value: False .. py:attribute:: _print_script_commands :value: False .. py:attribute:: _terminated :type: bool :value: False .. py:attribute:: _startup_splash_deadline :type: float | None :value: None .. py:attribute:: _startup_splash_close_event :type: multiprocessing.Event | None :value: None .. py:attribute:: _startup_splash_process :type: multiprocessing.Process | None :value: None .. py:attribute:: _startup_splash_timeout_seconds :type: float :value: 600.0 .. py:attribute:: _startup_splash_waiting_for_ui_ready :type: bool :value: False .. py:attribute:: _camera_health_log_interval_seconds :type: float :value: 60.0 .. py:attribute:: _next_camera_health_log_deadline :type: float | None :value: None .. py:attribute:: _last_camera_health_sample_time :type: float | None :value: None .. py:attribute:: _last_camera_health_frame_count :type: int :value: 0 .. py:attribute:: live_profile_buffer :type: magscope.datatypes.LiveProfileBuffer | None :value: None .. py:attribute:: bead_roi_buffer :type: magscope.datatypes.BeadRoiBuffer | None :value: None .. py:attribute:: tracks_buffer :type: magscope.datatypes.MatrixBuffer | None :value: None .. py:attribute:: video_buffer :type: magscope.datatypes.VideoBuffer | None :value: None .. py:method:: 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 :attr:`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. .. py:method:: 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. .. py:method:: add_hardware(hardware: magscope.hardware.HardwareManagerBase) Register a hardware manager so its process launches with MagScope. .. py:method:: add_control(control_type: type[magscope.ui.controls.ControlPanelBase], column: int) Schedule a GUI control panel to be added when the UI manager starts. .. py:method:: add_timeplot(plot: magscope.ui.plots.TimeSeriesPlotBase) Schedule a time-series plot for inclusion in the GUI at startup. .. py:property:: print_ipc_commands :type: bool Return whether :meth:`start` should print IPC commands and exit early. .. py:property:: print_script_commands :type: bool Return whether :meth:`start` should print script commands and exit early. .. py:method:: _coerce_settings(value: magscope.settings.MagScopeSettings | dict) -> magscope.settings.MagScopeSettings .. py:property:: settings .. py:method:: _reset_singleton_for_testing() -> None :classmethod: Clear the singleton registry so tests can create fresh instances. .. py:method:: set_verbose_logging(enabled: bool = True) -> None Toggle informational console output for MagScope internals. .. py:method:: _mark_running() -> bool Mark the orchestrator as running if it is not already active. .. py:method:: _ensure_not_terminated() -> None Prevent reusing a MagScope instance after it has been stopped. .. py:method:: _apply_logging_preferences() -> None Apply the current verbosity preference to the logging system. .. py:method:: _mark_terminated() -> None Record that this MagScope instance has finished its lifecycle. .. py:method:: _start_startup_splash() -> None Launch a lightweight splash window in a helper process. .. py:method:: _dismiss_startup_splash_if_pending() -> None Dismiss the splash while startup is still waiting on the UI. .. py:method:: _stop_startup_splash() -> None Request the startup splash helper process to exit. .. py:method:: _check_startup_splash_timeout() -> None Dismiss the splash if UI startup has been pending too long. .. py:method:: _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. .. py:method:: _setup_command_registry() -> None Register all command handlers and validate destinations. .. py:method:: print_registered_commands() -> None Print the registered IPC commands without launching managers. .. py:method:: print_registered_script_commands() -> None Print the registered script commands without launching managers. .. py:method:: _initialize_shared_state() -> None Load configuration and prepare shared resources for all managers. .. py:method:: _start_managers() -> None Start each manager process. .. py:method:: _main_ipc_loop() -> None Forward IPC messages until a quit request is observed. .. py:method:: _join_processes() -> None Join every managed process once shutdown has been requested. .. py:method:: receive_ipc() Poll every IPC pipe once and relay any commands that arrive. .. py:method:: _read_command(pipe: multiprocessing.connection.Connection) -> magscope.ipc_commands.Command | object | None Retrieve a command from ``pipe`` if one is waiting. .. py:method:: _process_command(command: magscope.ipc_commands.Command) -> bool Route a valid command and indicate whether the IPC loop should break. .. py:method:: _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. .. py:method:: _dispatch_mag_scope_command(command: magscope.ipc_commands.Command, spec) -> None Handle commands destined for the MagScope orchestrator. .. py:method:: _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). .. py:method:: log_exception(process_name: str, details: str) -> None Surface an exception raised in a managed process. .. py:method:: startup_ready(process_name: str = 'UIManager') -> None Dismiss the startup splash once the UI process is ready. .. py:method:: _sleep_when_idle() -> None Throttle the IPC loop when no messages were processed. .. py:method:: _reset_camera_health_logging_state() -> None Start a fresh sampling window for periodic camera health logging. .. py:method:: _log_camera_health_if_due() -> None Emit a 1-minute camera health summary while verbose logging is enabled. .. py:method:: _drain_child_pipes_after_quit() -> None Drain child pipes until they acknowledge the quit event. .. py:method:: _setup_shared_resources() Create and distribute shared locks, pipes, buffers, and metadata. .. py:method:: _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. .. py:method:: _create_shared_buffers() Instantiate shared memory buffers used throughout the application. .. py:method:: _setup_locks() Instantiate per-buffer locks and make them available to processes. .. py:method:: _setup_pipes() -> dict[str, multiprocessing.connection.Connection] Create duplex pipes that allow processes to exchange messages. .. py:method:: _register_script_methods() Expose manager methods to the scripting subsystem. .. py:method:: update_settings(settings: magscope.settings.MagScopeSettings | dict) -> None Replace the active settings and broadcast them to all managers.