magscope.camera

Camera manager and dummy camera implementations for MagScope.

This module defines the CameraManager process responsible for coordinating camera acquisition with the shared VideoBuffer, along with several simulation-oriented CameraBase implementations used when hardware is not available. The manager exchanges IPC messages with the GUI to keep camera settings synchronized and ensures buffers are properly released as acquisition states change.

Classes

CameraManager	Manager process that feeds frames from a CameraBase into shared buffers.
CameraBase	Abstract base class describing the camera interface used by managers.
DummyCameraNoise	Noise camera that generates random images at a configurable frame rate.
DummyCameraFastNoise	Noise camera that reuses cached random frames for higher throughput.
DummyCameraBeads	Bead simulator producing synthetic frames for testing without hardware.

Module Contents

class magscope.camera.CameraManager

Bases: magscope.processes.ManagerProcessBase

Manager process that feeds frames from a CameraBase into shared buffers.

The manager owns a camera instance (dummy by default), connects it to the shared VideoBuffer, relays camera settings to the GUI, and orchestrates buffer lifecycles based on acquisition and processing state. Its main loop reacts to pool flags, drains buffers to avoid overflow, and triggers camera fetches when connected.

camera: CameraBase

_released_completed_stacks = 0

setup()

Connect to the camera and publish its current settings.

Connection failures are logged as warnings so the rest of the system can continue running in simulation mode. When a connection succeeds, broadcast the initial camera settings to keep the GUI in sync with the camera process.

do_main_loop()

Main process loop handling buffer lifecycle and fetching frames.

The video processor reports consumed stacks through shared completion counters. This manager drains those completions into camera buffer releases, returns unreserved stacks while acquisition is paused, and purges excess buffered data when capacity gets too low.

_release_unattached_buffers()

Return buffers that are no longer tracked by the processing pool.

_purge_buffers()

Drain video buffer contents until at least 30% capacity is free.

_release_completed_pool_buffers()

Release camera buffers for stacks already consumed by workers.

_take_unreserved_stack() → bool

_stack_coordination_lock()

get_camera_setting(name: str)

Send a camera setting value to the GUI via IPC.

set_camera_setting(name: str, value: str)

Apply a setting to the camera and broadcast the full settings set.

set_simulated_focus(offset: float)

Adjust the simulated camera focus when using DummyCameraBeads.

class magscope.camera.CameraBase

Abstract base class describing the camera interface used by managers.

Concrete cameras must expose immutable dimensions and dtype metadata, a minimal settings API (__getitem__/__setitem__), and methods for connecting, fetching frames into a VideoBuffer, and releasing buffers back to the device or simulation pool.

bits: int

dtype: numpy.dtype

height: int

nm_per_px: float

width: int

settings: list[str] = ['framerate']

is_connected = False

shared_values = None

video_buffer: magscope.datatypes.VideoBuffer | None = None

camera_buffers: queue.Queue | None = None

__del__()

abstractmethod connect(video_buffer)

Attempts to connect to the camera.

But does not start an acquisition. This method should set the value of self.is_connected to True if successful or False if not.

abstractmethod fetch()

Checks if the camera has new images.

If the camera has a new image, then it holds the camera’s buffered image in a queue (self.camera_buffers). And stores the image and timestamp in the video buffer (self._video_buffer).

The timestamp should be the seconds since the unix epoch: (January 1, 1970, 00:00:00 UTC)

abstractmethod release()

Gives the buffer back to the camera.

release_all()

abstractmethod get_setting(name: str) → str

Should return the current value of the setting from the camera

abstractmethod set_setting(name: str, value: str)

Should set the value of the setting on the camera

__getitem__(name: str) → str

Used to get settings. Example: my_cam[‘framerate’]

__setitem__(name: str, value: str) → None

Used to set settings. Example: my_cam[‘framerate’] = 100.0

reset_health_counters() → None

report_frame_received(timestamp: float) → None

report_timeout() → None

report_queue_full() → None

class magscope.camera.DummyCameraNoise

Bases: CameraBase

Noise camera that generates random images at a configurable frame rate.

width = 512

height = 256

bits = 8

dtype

nm_per_px = 5000.0

settings = ['framerate', 'exposure', 'gain']

fake_settings

est_fps = 1000.0

est_fps_count = 0

est_fps_time

last_time = 0

connect(video_buffer)

Attempts to connect to the camera.

But does not start an acquisition. This method should set the value of self.is_connected to True if successful or False if not.

fetch()

Checks if the camera has new images.

If the camera has a new image, then it holds the camera’s buffered image in a queue (self.camera_buffers). And stores the image and timestamp in the video buffer (self._video_buffer).

The timestamp should be the seconds since the unix epoch: (January 1, 1970, 00:00:00 UTC)

_fake_image()

release()

Gives the buffer back to the camera.

get_setting(name: str) → str

Should return the current value of the setting from the camera

set_setting(name: str, value: str)

Should set the value of the setting on the camera

class magscope.camera.DummyCameraFastNoise

Bases: CameraBase

Noise camera that reuses cached random frames for higher throughput.

width = 1280

height = 560

bits = 8

dtype

nm_per_px = 5000.0

settings = ['framerate', 'exposure', 'gain']

fake_settings

est_fps = 1000.0

est_fps_count = 0

est_fps_time

last_time = 0

fake_images = None

fake_images_n = 10

fake_image_index = 0

connect(video_buffer)

Attempts to connect to the camera.

But does not start an acquisition. This method should set the value of self.is_connected to True if successful or False if not.

fetch()

Checks if the camera has new images.

If the camera has a new image, then it holds the camera’s buffered image in a queue (self.camera_buffers). And stores the image and timestamp in the video buffer (self._video_buffer).

The timestamp should be the seconds since the unix epoch: (January 1, 1970, 00:00:00 UTC)

get_fake_image()

release()

Gives the buffer back to the camera.

get_setting(name: str) → str

Should return the current value of the setting from the camera

set_setting(name: str, value: str)

Should set the value of the setting on the camera

class magscope.camera.DummyCameraBeads

Bases: CameraBase

Bead simulator producing synthetic frames for testing without hardware.

width = 512

height = 256

bits = 8

dtype

nm_per_px = 200.0

settings = ['framerate', 'fixed_n', 'fixed_z', 'tethered_n', 'tethered_z', 'tethered_z_sigma',...

_settings

_focus_offset = 0.0

_bead_size_px = 50

_min_sep_px = 50.0

_edge_margin_px = 10.0

_background = 0.4

_radius_nm = 1500.0

_theta_xy = 1.5

_theta_z = 2.0

_rng

_centers_fixed

_centers_teth

_delta_fixed = None

_xy

_z

last_time = 0.0

est_fps = 30.0

est_fps_count = 0

est_fps_time

connect(video_buffer)

Attempts to connect to the camera.

But does not start an acquisition. This method should set the value of self.is_connected to True if successful or False if not.

fetch()

Checks if the camera has new images.

If the camera has a new image, then it holds the camera’s buffered image in a queue (self.camera_buffers). And stores the image and timestamp in the video buffer (self._video_buffer).

The timestamp should be the seconds since the unix epoch: (January 1, 1970, 00:00:00 UTC)

release()

Gives the buffer back to the camera.

get_setting(name: str) → str

Should return the current value of the setting from the camera

set_setting(name: str, value: str)

Should set the value of the setting on the camera

set_focus_offset(offset: float) → None

_reinit_centers_and_fixed()

_recompute_fixed_delta()

_init_tether_state()

static _blit_add(dst, src, x, y, w=1.0)

classmethod _accumulate_bilinear(dst, srcHW, cx, cy)

static _border_median(imgHW)

static _tukey_taper(H, W, pad=4)

classmethod _delta_for_crop(cropHW, pad=4)

static _ou_step(x, dt, theta, sigma, mu, rng)

static _sample_points_uniform_minsep(W, H, n, margin_px, min_sep_px, rng, max_tries=100000, relax=0.95)

Dart throwing with optional relaxation. Returns (n,2) float32.