magscope.camera =============== .. py:module:: magscope.camera .. autoapi-nested-parse:: 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 ------- .. autoapisummary:: magscope.camera.CameraManager magscope.camera.CameraBase magscope.camera.DummyCameraNoise magscope.camera.DummyCameraFastNoise magscope.camera.DummyCameraBeads Module Contents --------------- .. py:class:: CameraManager Bases: :py:obj:`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. .. py:attribute:: camera :type: CameraBase .. py:attribute:: _released_completed_stacks :value: 0 .. py:method:: 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. .. py:method:: 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. .. py:method:: _release_unattached_buffers() Return buffers that are no longer tracked by the processing pool. .. py:method:: _purge_buffers() Drain video buffer contents until at least 30% capacity is free. .. py:method:: _release_completed_pool_buffers() Release camera buffers for stacks already consumed by workers. .. py:method:: _take_unreserved_stack() -> bool .. py:method:: _stack_coordination_lock() .. py:method:: get_camera_setting(name: str) Send a camera setting value to the GUI via IPC. .. py:method:: set_camera_setting(name: str, value: str) Apply a setting to the camera and broadcast the full settings set. .. py:method:: set_simulated_focus(offset: float) Adjust the simulated camera focus when using :class:`DummyCameraBeads`. .. py:class:: 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. .. py:attribute:: bits :type: int .. py:attribute:: dtype :type: numpy.dtype .. py:attribute:: height :type: int .. py:attribute:: nm_per_px :type: float .. py:attribute:: width :type: int .. py:attribute:: settings :type: list[str] :value: ['framerate'] .. py:attribute:: is_connected :value: False .. py:attribute:: shared_values :value: None .. py:attribute:: video_buffer :type: magscope.datatypes.VideoBuffer | None :value: None .. py:attribute:: camera_buffers :type: queue.Queue | None :value: None .. py:method:: __del__() .. py:method:: connect(video_buffer) :abstractmethod: 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. .. py:method:: fetch() :abstractmethod: 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) .. py:method:: release() :abstractmethod: Gives the buffer back to the camera. .. py:method:: release_all() .. py:method:: get_setting(name: str) -> str :abstractmethod: Should return the current value of the setting from the camera .. py:method:: set_setting(name: str, value: str) :abstractmethod: Should set the value of the setting on the camera .. py:method:: __getitem__(name: str) -> str Used to get settings. Example: my_cam['framerate'] .. py:method:: __setitem__(name: str, value: str) -> None Used to set settings. Example: my_cam['framerate'] = 100.0 .. py:method:: reset_health_counters() -> None .. py:method:: report_frame_received(timestamp: float) -> None .. py:method:: report_timeout() -> None .. py:method:: report_queue_full() -> None .. py:class:: DummyCameraNoise Bases: :py:obj:`CameraBase` Noise camera that generates random images at a configurable frame rate. .. py:attribute:: width :value: 512 .. py:attribute:: height :value: 256 .. py:attribute:: bits :value: 8 .. py:attribute:: dtype .. py:attribute:: nm_per_px :value: 5000.0 .. py:attribute:: settings :value: ['framerate', 'exposure', 'gain'] .. py:attribute:: fake_settings .. py:attribute:: est_fps :value: 1000.0 .. py:attribute:: est_fps_count :value: 0 .. py:attribute:: est_fps_time .. py:attribute:: last_time :value: 0 .. py:method:: 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. .. py:method:: 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) .. py:method:: _fake_image() .. py:method:: release() Gives the buffer back to the camera. .. py:method:: get_setting(name: str) -> str Should return the current value of the setting from the camera .. py:method:: set_setting(name: str, value: str) Should set the value of the setting on the camera .. py:class:: DummyCameraFastNoise Bases: :py:obj:`CameraBase` Noise camera that reuses cached random frames for higher throughput. .. py:attribute:: width :value: 1280 .. py:attribute:: height :value: 560 .. py:attribute:: bits :value: 8 .. py:attribute:: dtype .. py:attribute:: nm_per_px :value: 5000.0 .. py:attribute:: settings :value: ['framerate', 'exposure', 'gain'] .. py:attribute:: fake_settings .. py:attribute:: est_fps :value: 1000.0 .. py:attribute:: est_fps_count :value: 0 .. py:attribute:: est_fps_time .. py:attribute:: last_time :value: 0 .. py:attribute:: fake_images :value: None .. py:attribute:: fake_images_n :value: 10 .. py:attribute:: fake_image_index :value: 0 .. py:method:: 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. .. py:method:: 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) .. py:method:: get_fake_image() .. py:method:: release() Gives the buffer back to the camera. .. py:method:: get_setting(name: str) -> str Should return the current value of the setting from the camera .. py:method:: set_setting(name: str, value: str) Should set the value of the setting on the camera .. py:class:: DummyCameraBeads Bases: :py:obj:`CameraBase` Bead simulator producing synthetic frames for testing without hardware. .. py:attribute:: width :value: 512 .. py:attribute:: height :value: 256 .. py:attribute:: bits :value: 8 .. py:attribute:: dtype .. py:attribute:: nm_per_px :value: 200.0 .. py:attribute:: settings :value: ['framerate', 'fixed_n', 'fixed_z', 'tethered_n', 'tethered_z', 'tethered_z_sigma',... .. py:attribute:: _settings .. py:attribute:: _focus_offset :value: 0.0 .. py:attribute:: _bead_size_px :value: 50 .. py:attribute:: _min_sep_px :value: 50.0 .. py:attribute:: _edge_margin_px :value: 10.0 .. py:attribute:: _background :value: 0.4 .. py:attribute:: _radius_nm :value: 1500.0 .. py:attribute:: _theta_xy :value: 1.5 .. py:attribute:: _theta_z :value: 2.0 .. py:attribute:: _rng .. py:attribute:: _centers_fixed .. py:attribute:: _centers_teth .. py:attribute:: _delta_fixed :value: None .. py:attribute:: _xy .. py:attribute:: _z .. py:attribute:: last_time :value: 0.0 .. py:attribute:: est_fps :value: 30.0 .. py:attribute:: est_fps_count :value: 0 .. py:attribute:: est_fps_time .. py:method:: 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. .. py:method:: 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) .. py:method:: release() Gives the buffer back to the camera. .. py:method:: get_setting(name: str) -> str Should return the current value of the setting from the camera .. py:method:: set_setting(name: str, value: str) Should set the value of the setting on the camera .. py:method:: set_focus_offset(offset: float) -> None .. py:method:: _reinit_centers_and_fixed() .. py:method:: _recompute_fixed_delta() .. py:method:: _init_tether_state() .. py:method:: _blit_add(dst, src, x, y, w=1.0) :staticmethod: .. py:method:: _accumulate_bilinear(dst, srcHW, cx, cy) :classmethod: .. py:method:: _border_median(imgHW) :staticmethod: .. py:method:: _tukey_taper(H, W, pad=4) :staticmethod: .. py:method:: _delta_for_crop(cropHW, pad=4) :classmethod: .. py:method:: _ou_step(x, dt, theta, sigma, mu, rng) :staticmethod: .. py:method:: _sample_points_uniform_minsep(W, H, n, margin_px, min_sep_px, rng, max_tries=100000, relax=0.95) :staticmethod: Dart throwing with optional relaxation. Returns (n,2) float32.