magscope.processes

Attributes

logger
ValueTypeUI8

Classes

InterprocessValues
SingletonMeta
SingletonABCMeta	Metaclass for defining Abstract Base Classes (ABCs).
ManagerProcessBase	Abstract base class for processes in the MagScope

Module Contents

magscope.processes.logger

magscope.processes.ValueTypeUI8

class magscope.processes.InterprocessValues

video_process_busy_count: ValueTypeUI8

video_process_reserved_stacks: multiprocessing.Value[ctypes.c_uint32]

video_process_completed_stacks: multiprocessing.Value[ctypes.c_uint64]

live_profile_enabled: ValueTypeUI8

live_profile_bead: multiprocessing.Value[ctypes.c_int]

camera_total_frames: multiprocessing.Value[ctypes.c_uint64]

camera_consecutive_timeouts: multiprocessing.Value[ctypes.c_uint32]

camera_queue_full_events: multiprocessing.Value[ctypes.c_uint64]

camera_last_frame_timestamp: multiprocessing.Value[ctypes.c_double]

class magscope.processes.SingletonMeta

Bases: type

_instances

__call__(*args, **kwargs)

class magscope.processes.SingletonABCMeta

Bases: abc.ABCMeta, SingletonMeta

Metaclass for defining Abstract Base Classes (ABCs).

Use this metaclass to create an ABC. An ABC can be subclassed directly, and then acts as a mix-in class. You can also register unrelated concrete classes (even built-in classes) and unrelated ABCs as ‘virtual subclasses’ – these and their descendants will be considered subclasses of the registering ABC by the built-in issubclass() function, but the registering ABC won’t show up in their MRO (Method Resolution Order) nor will method implementations defined by the registering ABC be callable (not even via super()).

class magscope.processes.ManagerProcessBase

Bases: multiprocessing.Process, abc.ABC

Abstract base class for processes in the MagScope

Subclass requirements: * Each subclass should have a unique name. * There should only be one instance of each subclass (singleton). * The class name is used for consistent inter-process identification.

_acquisition_on: bool = True

_acquisition_dir: str | None = None

_acquisition_dir_on: bool = False

_acquisition_mode: magscope.utils.AcquisitionMode

bead_roi_buffer: magscope.datatypes.BeadRoiBuffer | None = None

_bead_roi_ids: numpy.ndarray

_bead_roi_values: numpy.ndarray

camera_type: type[magscope.camera.CameraBase] | None = None

hardware_types: dict[str, type[magscope.hardware.HardwareManagerBase]]

locks: dict[str, multiprocessing.synchronize.Lock] | None = None

_magscope_quitting: multiprocessing.synchronize.Event | None = None

name: str = 'ManagerProcessBase'

_pipe: multiprocessing.connection.Connection | None = None

live_profile_buffer: magscope.datatypes.LiveProfileBuffer | None = None

_quitting: multiprocessing.synchronize.Event

_quit_requested: bool = False

_running: bool = False

settings = None

tracks_buffer: magscope.datatypes.MatrixBuffer | None = None

video_buffer: magscope.datatypes.VideoBuffer | None = None

shared_values: InterprocessValues | None = None

_command_registry: magscope.ipc.CommandRegistry | None = None

_command_handlers: dict[type[magscope.ipc_commands.Command], str]

property quitting_event: multiprocessing.synchronize.Event

Event set when this process has begun quitting.

property bead_rois: dict[int, tuple[int, int, int, int]]

get_cached_bead_rois() → tuple[numpy.ndarray, numpy.ndarray]

_refresh_bead_roi_cache() → None

configure_shared_resources(*, camera_type: type[magscope.camera.CameraBase] | None, hardware_types: dict[str, type[magscope.hardware.HardwareManagerBase]], quitting_event: multiprocessing.synchronize.Event, settings: magscope.settings.MagScopeSettings, shared_values: InterprocessValues, locks: dict[str, multiprocessing.synchronize.Lock], pipe_end: multiprocessing.connection.Connection, command_registry: magscope.ipc.CommandRegistry) → None

Attach shared references provided by MagScope.

This centralizes initialization so callers do not need to mutate underscored attributes directly when preparing processes before start() is invoked.

run()

Start the process when start() is called.

Subclasses should create a main loop that calls receive_ipc() last:

while self._running:
    # do other stuff
    self.receive_ipc()

abstractmethod setup()

abstractmethod do_main_loop()

quit()

Shutdown the process (and ask the other processes to quit too).

send_ipc(command: magscope.ipc_commands.Command)

receive_ipc()

set_acquisition_dir(value: str | None)

set_acquisition_dir_on(value: bool)

set_acquisition_mode(mode: magscope.utils.AcquisitionMode)

set_acquisition_on(value: bool)

refresh_bead_rois()

set_settings(settings: magscope.settings.MagScopeSettings)

_report_exception(exc: BaseException) → None