# While using the garbage collector is frowned upon, it may help during
# large memory allocations
import concurrent.futures
import gc
import inspect
import os
import sys
from collections import OrderedDict
from collections.abc import Iterable
from contextlib import contextmanager
from copy import deepcopy
from datetime import UTC, datetime
from functools import partial
from pathlib import Path
from pprint import pformat
from queue import Empty as EmptyException
from queue import Queue
from threading import Event, Semaphore, Thread
from time import perf_counter
from warnings import warn
import numpy as np
import pandas as pd
import psutil
import xarray as xr
from owl.calibration._keys import brightfield_illumination_modes
from owl.mcam_data._new import add_ml_index_array
from owl.memory_allocator import empty_aligned
from owl.schemas.acquisition_properties import AcquisitionModeProperties
from owl.util import get_localzone, intel
from owl.util.util import RAMONA_MCAM_ADVANCED_MODE, RAMONA_MCAM_DISABLE_MCAM_DATARATE_CHECK
from ..calibration import supported_illumination_modes
from ..calibration.calibration import (
download_factory_calibration,
load_calibration,
save_calibration,
)
from ..calibration.color_correction import define_white_light
from ..calibration.dataset import replace_calibration_data
from ..qdma.boards.falcon import Falcon
from ..schemas.well_plate_keypoints import WellPlateKeypointState
from ..util import user_config_dir
from ..util.array import selection_slice_to_selection
from ..util.memory import fault_pages, try_huge_alloc
from ..util.util import sleep_from
from ._fpga_registry import fpga_registry
from ._mcam_registry import MECHANICAL_STATE_SOLENOID_AXIS, get_registry_entry, mcam_registry
from ._well_positions import (
create_full_coverage_acquisition_mode_from_well_properties,
labware_well_plate_properties,
)
from .fluorescence import FakeFluorescence
from .fluorescence import Fluorescence as RealFluorescence
from .illuminate import FakeIlluminate
from .illuminate import Illuminate as RealIlluminate
from .indicator_board import FakeIndicatorBoard
from .indicator_board import IndicatorBoard as RealIndicatorBoard
from .magellan import SmartAxis
from .motorized_illumination import MotorizedIllumination, motorized_illumination_serial_numbers
from .plate_tapper import FakePlateTapper
from .plate_tapper import PlateTapper as RealPlateTapper
from .solenoid_driver import FakeSolenoidDriver
from .solenoid_driver import SolenoidDriver as RealSolenoidDriver
from .temperature_monitor import FakeTemperatureMonitor
from .temperature_monitor import TemperatureMonitor as RealTemperatureMonitor
from .zaber import X_LSM_E
# AMD CPUs will return None with this call
# Technically we should also check the GPU and the GPU encoder
if intel.cpu_model in [
'Intel(R) Core(TM) Ultra 7 265K',
'Intel(R) Core(TM) Ultra 7 265',
]:
_data_consumption_rate = {
# 2026/02 -- I Mark mostly tested a zebrafish acquisition
# with a
# * 96 well plate
# * Bin x4 (high frame rate)
# * 160 fps
# * 15 mins
# The measured datarate was about 800MB/s, but some overhead
# needs to be considered for debayering and re-tling the
# I tested this with an
# * Intel Ultra-7-265K
# * NVIDIA RTX 4500 Ada Generation
# * 256 GB of RAM
# * h264_nvenc
# I wanted to mostly retain the same functionality that we have with
# systems that have 512GB of RAM which have been capturing
# this amount of data but need a much larger RAM buffer
'multi_mp4': 750 * 1024 ** 2,
'tiledmp4': 750 * 1024 ** 2,
'to_file': (1024 + 512) * 1024 ** 2,
# 32GB/s seems like a large number we will never hit in realistic acquisitions.
'to_null': 32 * 1024 ** 3,
}
else:
_data_consumption_rate = {
# I (Mark) haven't really validated all the different encoding resolutions
# but this limit "feels" right for now given the amount of testing that has
# gone through encoding at high datarates
# maximum_encoder_bandwidth_MBps = 3072 * 4096 * 30 / 1024 ** 2
# https://gitlab.com/ramonaoptics/mcam/python-owl/-/merge_requests/3326
'multi_mp4': 620 * 1024 ** 2,
'tiledmp4': 500 * 1024 ** 2,
'to_file': (1024 + 512) * 1024 ** 2,
# 32GB/s seems like a large number we will never hit in realistic acquisitions.
'to_null': 32 * 1024 ** 3,
}
# Debugging tool for Mark on June 10, 2025
_forced_data_consumption_rate = os.environ.get(
"RAMONA_MCAM_FORCED_DATA_CONSUMPTION_RATE_MBPS", None
)
if _forced_data_consumption_rate is not None:
consumption_rate = int(_forced_data_consumption_rate) * 1024 ** 2
_data_consumption_rate['multi_mp4'] = consumption_rate
_data_consumption_rate['tiledmp4'] = consumption_rate
_data_consumption_rate['to_file'] = consumption_rate
def tqdm(x, *args, **kwargs):
return x
def _select_dataset_selection(dataset, selection):
dataset = dataset
if selection != Ellipsis:
if not isinstance(selection, Iterable):
selection = (selection,)
dims = dataset['images'].dims[:len(selection)]
selector = {
d: s
for d, s in zip(dims, selection)
}
dataset = dataset.isel(selector)
return dataset
def get_N_frames_buffer(
frame_size,
N_frames,
max_buffer_size,
min_chunk_size,
min_total_chunks,
):
chunk_size = max(min_chunk_size, frame_size) # bytes
N_frames_per_chunk = max(chunk_size // frame_size, 1)
N_frames_per_chunk = min(N_frames_per_chunk, N_frames)
# After we round, we should recompute it
chunk_size = N_frames_per_chunk * frame_size
total_chunks = max(max_buffer_size // chunk_size, min_total_chunks)
N_frames_buffer = total_chunks * N_frames_per_chunk
N_frames_buffer = min(
((N_frames + (N_frames_per_chunk - 1)) // N_frames_per_chunk) * N_frames_per_chunk,
N_frames_buffer
)
return N_frames_buffer, N_frames_per_chunk
def alloc_buffer_for_acquisition(
shape, dtype, *,
garbage_collection=True,
tqdm=None,
try_hugetlb=True,
cpu_count=None,
):
# Allocating so much memory is slow due to the fact that we have
# to cause the memory to actually be available by causing page faults.
# Therefore we want to force things to cause page faults
# immediately after this function returns
# instead of using `full`, we roll our own loop so we can
# report progress with tqdm
if try_hugetlb:
raw_data, needs_faulting = try_huge_alloc(shape, dtype=dtype)
else:
raw_data = empty_aligned(shape, dtype=dtype)
needs_faulting = True
try:
if needs_faulting:
fault_pages(raw_data, cpu_count=cpu_count, tqdm=tqdm)
except Exception:
raw_data = None
if garbage_collection:
gc.collect()
raise
return raw_data
def get_opening_parameters(dna_string, init_kwargs):
if init_kwargs is None:
init_kwargs = {}
serial_number = init_kwargs.get('serial_number', None)
original_serial_number = None
opening_parameters = get_registry_entry(dna_string, serial_number=serial_number)
if original_serial_number is not None:
opening_parameters['serial_number'] = original_serial_number
maximum_datarate_check = init_kwargs.get(
"with_maximum_datarate_check",
not RAMONA_MCAM_DISABLE_MCAM_DATARATE_CHECK
)
if not maximum_datarate_check:
opening_parameters["maximum_datarate_GiBps"] = None
# 2022/04/04 Jed/Mark
# Allow users to override opening sequence parameters using certain
# legacy parameters
opening_parameters["exif_orientation"] = init_kwargs.get(
'exif_orientation', opening_parameters["exif_orientation"])
opening_parameters["calibration_filename"] = init_kwargs.get(
"calibration_filename", opening_parameters["calibration_filename"]
)
opening_parameters["z_stage_serial_number"] = init_kwargs.get(
"with_z_stage", opening_parameters["z_stage_serial_number"]
)
opening_parameters["y_stage_serial_number"] = init_kwargs.get(
"with_y_stage", opening_parameters["y_stage_serial_number"]
)
opening_parameters["x_stage_serial_number"] = init_kwargs.get(
"with_x_stage", opening_parameters["x_stage_serial_number"]
)
opening_parameters["x_sample_stage_serial_number"] = init_kwargs.get(
"with_x_sample_stage", opening_parameters["x_sample_stage_serial_number"]
)
opening_parameters["transmission_illumination_serial_number"] = init_kwargs.get(
"with_transmission_illumination",
opening_parameters["transmission_illumination_serial_number"]
)
opening_parameters["reflection_illumination_serial_number"] = init_kwargs.get(
"with_reflection_illumination",
opening_parameters["reflection_illumination_serial_number"]
)
opening_parameters["fluorescence_illumination_serial_number"] = init_kwargs.get(
"with_fluorescence_illumination",
opening_parameters["fluorescence_illumination_serial_number"]
)
opening_parameters["temperature_monitor_serial_number"] = init_kwargs.get(
"with_temperature_monitor",
opening_parameters["temperature_monitor_serial_number"]
)
opening_parameters["plate_tapper_serial_number"] = init_kwargs.get(
"with_plate_tapper",
opening_parameters["plate_tapper_serial_number"]
)
opening_parameters["solenoid_driver_serial_number"] = init_kwargs.get(
"with_solenoid_driver",
opening_parameters["solenoid_driver_serial_number"]
)
opening_parameters["incubator_serial_number"] = init_kwargs.get(
"with_incubator",
opening_parameters["incubator_serial_number"]
)
opening_parameters["indicator_board_serial_number"] = init_kwargs.get(
"with_indicator_board",
# 2025/03/09 -- can be removed in 2025/06
# Use the getter to enable a smoother transition for people
# that don't pip install every day
opening_parameters.get("indicator_board_serial_number", False)
)
opening_parameters["output_trigger_events"] = init_kwargs.get(
"output_trigger_events",
opening_parameters.get("output_trigger_events", None),
)
# It is hard to store dictionaries in pandas.
# We therefore store a tuple, but we chose to convert it here
# To something where we can access more readily as a dictionary
virtual_selections_tuple = opening_parameters["virtual_selections"]
opening_parameters["virtual_selections"] = OrderedDict(virtual_selections_tuple)
return opening_parameters
def _maybe_slice(data, selection, dataset_selection_slice):
if np.ndim(data) == 2:
return data[dataset_selection_slice][selection]
else:
return data
def _get_lazy_instrument_properties(lazy_dict):
info = {}
with concurrent.futures.ThreadPoolExecutor() as executor:
futures_to_name = {
executor.submit(func): name
for name, func in lazy_dict.items()
}
for future in concurrent.futures.as_completed(futures_to_name):
name = futures_to_name[future]
value = future.result()
info[name] = value
return info
def _safe_temperature_monitor_read(temperature_monitor, probe_location):
try:
return temperature_monitor.temperature(probe_location)
except Exception as e:
serial_number = getattr(temperature_monitor, 'serial_number', 'unknown')
warn(
f"Failed to read {probe_location} temperature from temperature monitor "
f"{serial_number}: {e}",
stacklevel=2,
)
return np.nan
def _overlay_in_focus_plane_metadata(base_dataset, source_dataset, camera_mask):
for name, base_var in base_dataset.data_vars.items():
dims = base_var.dims
if name == 'images' or 'image_y' not in dims or 'image_x' not in dims:
continue
image_y_axis = dims.index('image_y')
if dims.index('image_x') != image_y_axis + 1:
raise NotImplementedError(
f"Variable {name!r} has non-adjacent image_y/image_x dims "
f"{dims}; per-camera metadata assembly assumes they are adjacent."
)
index = (slice(None),) * image_y_axis + (camera_mask,)
source_var = source_dataset[name]
if 'image_y' in source_var.dims:
base_var.data[index] = source_var.data[index]
else:
base_var.data[index] = source_var.data
def truthy_or_None(value):
return value or (value is None)
def _get_memory_metrics():
process_memory = {}
system_memory = {}
try:
process = psutil.Process(os.getpid())
process_memory = process.memory_info()
system_memory = psutil.virtual_memory()
except Exception:
# Ignore the errors for now, we'll report as np.nan
pass
return {
'process_memory_rss': getattr(process_memory, 'rss', -1),
'process_memory_vms': getattr(process_memory, 'vms', -1),
'system_memory_total': getattr(system_memory, 'total', -1),
'system_memory_used': getattr(system_memory, 'used', -1),
'system_memory_buffers': getattr(system_memory, 'buffers', -1),
'system_memory_cached': getattr(system_memory, 'cached', -1),
'system_memory_available': getattr(system_memory, 'available', -1),
'system_memory_free': getattr(system_memory, 'free', -1),
# This is a floating point
'system_memory_percent': getattr(system_memory, 'percent', float('nan')),
}
_default_exposure = 100E-3
_default_digital_gain = 1
_default_analog_gain = 1
[docs]
class MCAM:
"""The MCAM object.
This object allows you to interact with the MCAM as a whole, capturing
images under various conditions and saving data.
Parameters
----------
serial_number: str, or None
- If provided, the MCAM will open a particular serial number.
with_z_stage : bool, or None
- If ``True``, opening the MCAM will fail if no z-stage is found.
- If ``None``, then a warning will be issued if no z-stage is found.
- If ``False``, opening the MCAM will not attempt to open the z-stage.
with_y_stage : bool, or None
- If ``True``, opening the MCAM will fail if no y-stage is found.
- If ``None``, then a warning will be issued if no y-stage is found.
- If ``False``, opening the MCAM will not attempt to open the y-stage.
with_x_stage : bool, or None
- If ``True``, opening the MCAM will fail if no x-stage is found.
- If ``None``, then a warning will be issued if no x-stage is found.
- If ``False``, opening the MCAM will not attempt to open the x-stage.
with_x_sample_stage : bool, or None
- If ``True``, opening the MCAM will fail if no x sample stage is found.
- If ``None``, then a warning will be issued if no x sample stage is found.
- If ``False``, opening the MCAM will not attempt to open the x sample stage.
with_reflection_illumination : bool, or None
- If ``True``, opening the MCAM will fail if no reflection-illumination is found.
- If ``None``, then a warning will be issued if no reflection-illumination is found.
- If ``False``, opening the MCAM will not attempt to open the reflection-illumination.
with_transmission_illumination : bool, or None
- If ``True``, opening the MCAM will fail if no transmission-illumination is found.
- If ``None``, then a warning will be issued if no transmission-illumination is found.
- If ``False``, opening the MCAM will not attempt to open the transmission-illumination.
with_fluorescence_illumination : bool, or None
- If ``True``, opening the MCAM will fail if no fluorescence-illumination is found.
- If ``None``, then a warning will be issued if no fluorescence-illumination is found.
- If ``False``, opening the MCAM will not attempt to open the fluorescence-illumination.
with_temperature_monitor : bool, or None
- If ``True``, opening the MCAM will fail if no temperature_monitor is found.
- If ``None``, then a warning will be issued if no temperature_monitor is found.
- If ``False``, opening the MCAM will not attempt to open the temperature_monitor.
with_plate_tapper : bool, or None
- If ``True``, opening the MCAM will fail if no plate_tapper is found.
- If ``None``, then a warning will be issued if no plate_tapper is found.
- If ``False``, opening the MCAM will not attempt to open the plate_tapper.
with_solenoid_driver : bool, or None
- If ``True``, opening the MCAM will fail if no solenoid driver is found.
- If ``None``, then a warning will be issued if no solenoid driver is found.
- If ``False``, opening the MCAM will not attempt to open the solenoid driver.
with_incubator : bool, or None
- If ``True``, opening the MCAM will fail if no incubator is found.
- If ``None``, then a warning will be issued if no incubator is found.
- If ``False``, opening the MCAM will not attempt to open the incubator.
with_indicator_board: bool, or None
- If ``True``, opening the MCAM will fail if no indicator board is found.
- If ``None``, then a warning will be issued if no indicator board is found.
- If ``False``, opening the MCAM will not attempt to open the indicator board.
with_calibration : True, False, or None
- If ``False``, then no calibration file is loaded.
- If ``None``, then the default calibration file is loaded.
with_maximum_datarate_check: True, False
- If ``False``, then the MCAM will not check the data rate.
- If ``True``, the MCAM's datarate is going to be checked against the recorded
factory datarate. If it is too low, then an error will be raised prior to
opening the instrument.
enusre_homed: True, False
- If ``True``, then all stages will be homed.
- If ``False``, then the stages will not be homed. Useful for troubleshooting only.
Notes
-----
.. versionchanged :: 0.18.89
The keyword argument ``exif_orientation`` is formally deprecated and will
be removed in a future version.
"""
__slots__ = [
'_exif_orientation',
'_falcon',
'_fake_device',
'_raw_video_buffers',
'_latest_acquisition_index',
'_acquisition_count',
'_physical_acquisition_count',
'_acquisition_index',
'_physical_acquisition_index',
'_expected_fps',
'_fps_calculation_info',
'_system_calibration_data',
'_calibration_data',
'_white_light_transmission',
'_white_light_reflection',
'_stages',
'_last_set_mechanical_state',
'_available_mechanical_states',
'_available_illumination_devices',
'_last_image_acquisition_parameters',
'_own_illumination',
'_illumination_mode',
'_illumination_device',
'_device_illumination_mode',
'_external_illumination_brightness',
'_illumination_brightness_kwargs',
'_temperature_monitor',
'_own_temperature_monitor',
'_plate_tapper',
'_own_plate_tapper',
'_solenoid_driver',
'_own_solenoid_driver',
'_incubator',
'_own_incubator',
'_indicator_board',
'_own_indicator_board',
'_acquisition_mode',
'_acquisition_mode_data',
'_virtual_selection_key',
'_available_virtual_selections',
'_init_kwargs',
'_serial_number',
'_product_line',
'_fiducial_stage_position',
'_labware_acquisition_modes',
]
def __init__(self, **kwargs):
for slot in self.__slots__:
setattr(self, slot, None)
if 'exif_orientation' in kwargs:
warn("You supplied `exif_orientation` as part of the constructor."
"We are planning to deprecate this behavior. Please set "
"The desired exif orientation after opening the MCAM.",
stacklevel=2)
# Save the user's keyword arguments to repeatedly call open
self._init_kwargs = deepcopy(kwargs)
self.open(_stacklevel_increment=1)
# Rewrite the signature to expose the serial number as a valid in the
# kwargs
# I'm going to consider other keys as "advanced" and users that need
# then can read the documentation and type things out nicely
__init__.__signature__ = inspect.Signature(
parameters=(
inspect.Parameter(
'self',
kind=inspect.Parameter.POSITIONAL_OR_KEYWORD
),
inspect.Parameter(
'serial_number',
kind=inspect.Parameter.KEYWORD_ONLY,
default=None,
),
inspect.Parameter(
'kwargs',
kind=inspect.Parameter.VAR_KEYWORD,
),
)
)
[docs]
def open(self, *, _stacklevel_increment=0):
"""Open the MCAM for communication."""
try:
self._open(stacklevel_increment=_stacklevel_increment + 1)
except Exception:
# If something fails at opening, just close things so we can try
# again later
self.close()
raise
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
self.close()
@property
def _motor_current_run(self):
"""The total current required for the stages to all move together."""
current = 0
for stage, owner in self._stages.values():
current += stage.current_run
if self.motorized_illumination is not None:
current += self.motorized_illumination._motor_current_run
return current
@property
def _motor_current_hold(self):
"""The total current required for the stages to all move together."""
current = 0
for stage, owner in self._stages.values():
current += stage.current_hold
if self.motorized_illumination is not None:
current += self.motorized_illumination._motor_current_hold
return current
@property
def _motor_voltage(self):
all_voltages = [
stage.motor_voltage
for stage, _owner in self._stages.values()
]
if self.motorized_illumination is not None:
all_voltages.extend([
stage.motor_voltage
for stage in self.motorized_illumination._all_active_stages
])
if not all_voltages:
return 0.
return float(np.mean(all_voltages))
def _add_stage(self, axis, stage=None, owndevice=None, *, _stacklevel_increment=0):
if self._falcon is None:
raise RuntimeError("The usage of this function requires the MCAM to be open.")
if axis not in ['z', 'y', 'x', 'x_sample']:
raise ValueError(
f'Given `axis` value must be `z`, `y`, `x`, or `x_stage` not `{axis}`.'
)
if self._stages.get(axis) is not None:
raise ValueError(f"The {axis} stage is already connected")
if owndevice is None:
owndevice = stage is None
if stage is None:
registry_settings = get_registry_entry(
self._falcon.dna_string,
serial_number=self._serial_number
)
connection_managers = [stage.connection_manager
for _, (stage, _) in self._stages.items()]
stage_serial_number = registry_settings[f'{axis}_stage_serial_number']
stage = _connect_to_stage(
axis=axis,
stage_serial_number=stage_serial_number,
fake_device=self._fake_device,
connection_managers=connection_managers,
_stacklevel_increment=1 + _stacklevel_increment,
)
if stage is None:
raise RuntimeError(f"Could not find any {axis} stage.")
self._stages[axis] = (stage, owndevice)
@property
def z_stage(self):
stage, _ = self._stages.get('z', (None, None))
return stage
@property
def y_stage(self):
stage, _ = self._stages.get('y', (None, None))
return stage
@property
def x_stage(self):
stage, _ = self._stages.get('x', (None, None))
return stage
@property
def y_position(self):
if self.y_stage is not None:
return self.y_stage.position
else:
return None
def set_y_position(self, value, wait=True, *, anti_backlash=0.):
self.y_stage.set_position(value, anti_backlash=anti_backlash, wait=False)
if self.motorized_illumination is not None:
y_collimation_position = (
value +
self.motorized_illumination._y_collimation_from_imaging_offset
)
self.motorized_illumination.set_position_y(
y_collimation_position,
anti_backlash=anti_backlash,
wait=False,
)
if wait:
self.wait_until_idle_y()
def wait_until_idle_y(self):
if self.y_stage is not None:
self.y_stage.wait_until_idle()
if self.motorized_illumination is not None:
self.motorized_illumination.wait_until_idle_y()
@y_position.setter
def y_position(self, value):
self.set_y_position(value, wait=True)
@property
def x_position(self):
if self.x_stage is not None:
return self.x_stage.position
else:
return None
@x_position.setter
def x_position(self, value):
self.set_x_position(value, wait=True)
def set_x_position(self, value, wait=True, *, anti_backlash=0.):
self.x_stage.set_position(value, anti_backlash=anti_backlash, wait=False)
if self.motorized_illumination is not None:
x_collimation_position = (
value + self.motorized_illumination._x_collimation_from_imaging_offset)
self.motorized_illumination.set_position_x(
x_collimation_position,
anti_backlash=anti_backlash,
wait=False,
)
if wait:
self.wait_until_idle_x()
def wait_until_idle_x(self):
if self.x_stage is not None:
self.x_stage.wait_until_idle()
if self.motorized_illumination is not None:
self.motorized_illumination.wait_until_idle_x()
def wait_until_idle(self):
for stage, owner in self._stages.values():
stage.wait_until_idle()
if self.motorized_illumination is not None:
self.motorized_illumination.wait_until_idle()
def wait_until_idle_all(self):
warn("wait_until_idle_all is deprecated. Please use wait_until_idle instead.", stacklevel=2)
self.wait_until_idle()
@property
def available_mechanical_states(self):
"""A list of mechanical states currently available for the MCAM.
For MCAM systems equipped with one or more motion stages, the
MCAM may be defined with a set of pre-calibrated mechanical states.
This list reports the available states for the MCAM, taking into
account the presently connected hardware so that the users may
programmatically select the most appropriate state for their usecase.
"""
return [
key
for key, state in self._available_mechanical_states.items()
if all(
axis in self._stages
for axis in state
if axis != MECHANICAL_STATE_SOLENOID_AXIS
)
]
@property
def mechanical_state(self):
"""The mechanical state of the MCAM system.
For MCAM systems equipped with one or more motion stages, the
MCAM may be defined with a set of pre-calibrated mechanical states.
This variable reflects the current mechanical state of the system.
The initial state of the MCAM is undefined, as such, this property
will return `None` upon opening the MCAM.
"""
key = self._last_set_mechanical_state
if key is None:
return None
state = self._available_mechanical_states[key]
# This is really a heuristic. but for now, we will assume that our stages
# have accuracies better than 1 mm.
# If the stage is still within 1 mm of its position, lets consider it
# in the same state as before.
# Upon setting the state with set_state, the stage will first move to
# this position
# then move to the final position
# a 1 mm travel isn't significant.
# The ultimate challenge, is that due to the discrete nature of
# the stages, whatever floating point values we put in the registry
# will be mismatched with the actual values that the stage can go to
# so we have to have a finite tolerance.
# While 1 mm might be a lot for the high precision stages we have
# it might be just enough for the low precision stages we use
position_tolerance = 1E-3
# Check that each stage specified by the mechanical state is still
# roughly near where it is supposed to be
for axis, position in state.items():
if axis == MECHANICAL_STATE_SOLENOID_AXIS:
continue
stage, _ = self._stages.get(axis, None)
if stage is None:
# Should we set the _last_set_mechanical_state to None?
return None
if abs(stage.position - position) > position_tolerance:
# Should we set the _last_set_mechanical_state to None?
return None
# All stages specified are within their tolerance
# return the last set key
return key
@mechanical_state.setter
def mechanical_state(self, key):
if key is None:
self._last_set_mechanical_state = None
# I am using the internal available_mechanical_states
# and manually checking for incompatibility so that we can create
# a better error message for the user.
if key not in self._available_mechanical_states:
raise ValueError(f"{key} is not a known state of this MCAM.")
state = self._available_mechanical_states[key]
self._set_mechanical_state(state)
self._last_set_mechanical_state = key
def _apply_solenoid_from_mechanical_state_dict(self, state):
if self._solenoid_driver is None:
return
strength = state.get(MECHANICAL_STATE_SOLENOID_AXIS)
if strength is None:
self._solenoid_driver.energize(0.0)
else:
self._solenoid_driver.energize(float(strength))
def _apply_solenoid_for_mechanical_state_key(self, key):
if key is None or key not in self._available_mechanical_states:
if self._solenoid_driver is not None:
self._solenoid_driver.energize(0.0)
return
self._apply_solenoid_from_mechanical_state_dict(
self._available_mechanical_states[key]
)
def _set_mechanical_state(self, state):
# This is an internal function for us to easily prototype new states
# without them being in the registry.
positions = []
stages = []
for axis, position in state.items():
if axis == MECHANICAL_STATE_SOLENOID_AXIS:
continue
# I really want to know which axis is disconnected
# so I can write a nice error message to the user
if axis not in self._stages:
raise RuntimeError(
"The state requires a stage along the "
f"{axis} axis that is not connected to the MCAM.")
stages.append(self._stages[axis][0])
positions.append(position)
for position, stage in zip(positions, stages):
current_position = stage.position
# 1 um tolerance, we can adjust this later
if abs(current_position - position) > 1E-6:
stage.set_position(position, wait=False)
for stage in stages:
stage.wait_until_idle()
self._apply_solenoid_from_mechanical_state_dict(state)
def get_mechanical_state_positions(self, key):
if key not in self._available_mechanical_states:
raise ValueError(f"{key} is not a known state of this MCAM.")
state = self._available_mechanical_states[key]
return {
axis: position
for axis, position in state.items()
if axis != MECHANICAL_STATE_SOLENOID_AXIS
}
@property
def x_sample_stage(self):
stage, _ = self._stages.get('x_sample', (None, None))
return stage
[docs]
def add_z_stage(self, stage=None, owndevice=None):
"""Add a Z-stage to the MCAM.
The stage will be closed upon closing the MCAM or calling MCAM.remove_z_stage()
if ownership of the stage is explicitly passed to the MCAM.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
Parameters
----------
stage: owl.instruments.X_LSM_E
Opened stage to attach to the MCAM. If no stage is provided,
the MCAM will attempt to open the stage from the system settings.
.. versionchanged:: 0.18.373
Providing a value of None is allowed.
owndevice: bool
True if MCAM has permission to close the stage. If the stage is None,
``owndevice`` will be set to True by default.
"""
self._add_stage('z', stage, owndevice,
_stacklevel_increment=1)
[docs]
def add_y_stage(self, stage=None, owndevice=None):
"""Add a Y-stage to the MCAM.
The stage will be closed upon closing the MCAM or calling MCAM.remove_y_stage()
if ownership of the stage is explicitly passed to the MCAM.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
Parameters
----------
stage: owl.instruments.X_LSM_E
Opened stage to attach to the MCAM. If no stage is provided,
the MCAM will attempt to open the stage from the system settings.
.. versionchanged:: 0.18.373
Providing a value of None is allowed.
owndevice: bool
True if MCAM has permission to close the stage. If the stage is None,
``owndevice`` will be set to True by default.
"""
self._add_stage('y', stage, owndevice,
_stacklevel_increment=1)
[docs]
def add_x_stage(self, stage=None, owndevice=None):
"""Add an X-stage to the MCAM.
The stage will be closed upon closing the MCAM or calling MCAM.remove_x_stage()
if ownership of the stage is explicitly passed to the MCAM.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
Parameters
----------
stage: owl.instruments.X_LSM_E
Opened stage to attach to the MCAM. If no stage is provided,
the MCAM will attempt to open the stage from the system settings.
.. versionchanged:: 0.18.373
Providing a value of None is allowed.
owndevice: bool
True if MCAM has permission to close the stage. If the stage is None,
``owndevice`` will be set to True by default.
"""
self._add_stage('x', stage, owndevice,
_stacklevel_increment=1)
[docs]
def add_x_sample_stage(self, stage=None, owndevice=None):
"""Add a sample side X stage to the MCAM.
The stage will be closed upon closing the MCAM or calling
``MCAM.remove_x_sample_stage()``
if ownership of the stage is explicitly passed to the MCAM.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
Parameters
----------
stage: owl.instruments.X_LSM_E
Opened stage to attach to the MCAM. If no stage is provided,
the MCAM will attempt to open the stage from the system settings.
.. versionchanged:: 0.18.373
Providing a value of None is allowed.
owndevice: bool
True if MCAM has permission to close the stage. If the stage is None,
``owndevice`` will be set to True by default.
"""
self._add_stage('x_sample', stage, owndevice,
_stacklevel_increment=1)
[docs]
def remove_z_stage(self):
"""Remove the z-stage from the MCAM. If the MCAM has ownership of the stage,
the stage will be closed.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
"""
self._remove_stage('z')
def _remove_stage(self, axis):
if self._falcon is None:
raise RuntimeError("The usage of this function requires the MCAM to be open.")
if axis not in ['z', 'y', 'x', 'x_sample']:
raise ValueError(
f'Given `axis` value must be `z`, `y`, `x`, or `x_sample` not `{axis}`.'
)
stage, owndevice = self._stages.pop(axis, (None, None))
if stage is None:
raise ValueError(f"No {axis} stage is currently connected.")
if owndevice:
stage.close()
[docs]
def remove_y_stage(self):
"""Remove the y-stage from the MCAM. If the MCAM has ownership of the stage,
the stage will be closed.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
"""
self._remove_stage('y')
[docs]
def remove_x_stage(self):
"""Remove the x-stage from the MCAM. If the MCAM has ownership of the stage,
the stage will be closed.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
"""
self._remove_stage('x')
[docs]
def remove_x_sample_stage(self):
"""Remove the x-stage from the MCAM. If the MCAM has ownership of the stage,
the stage will be closed.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
"""
self._remove_stage('x_sample')
def remove_temperature_monitor(self):
if self._temperature_monitor is None:
raise RuntimeError("No temperature monitor is connected to the MCAM")
if self._own_temperature_monitor:
self._temperature_monitor.close()
self._temperature_monitor = None
def add_temperature_monitor(self, temperature_monitor, *, owndevice=False):
if self._temperature_monitor is not None:
raise RuntimeError("Cannot add two temperature monitors.")
if 'chamber' not in temperature_monitor.probe_locations:
raise RuntimeError("Temperature monitor must contain a probe in the chamber.")
self._temperature_monitor = temperature_monitor
self._own_temperature_monitor = owndevice
def remove_plate_tapper(self):
if self._plate_tapper is None:
raise RuntimeError("No plate tapper is connected to the MCAM")
if self._own_plate_tapper:
self._plate_tapper.close()
self._plate_tapper = None
def add_plate_tapper(self, plate_tapper, *, owndevice=False):
if self._plate_tapper is not None:
raise RuntimeError("Cannot add two plate tappers.")
self._plate_tapper = plate_tapper
self._own_plate_tapper = owndevice
def remove_solenoid_driver(self):
if self._solenoid_driver is None:
raise RuntimeError("No solenoid driver is connected to the MCAM")
if self._own_solenoid_driver:
self._solenoid_driver.close()
self._solenoid_driver = None
def add_solenoid_driver(self, solenoid_driver, *, owndevice=False):
if self._solenoid_driver is not None:
raise RuntimeError("Cannot add two solenoid drivers.")
self._solenoid_driver = solenoid_driver
self._own_solenoid_driver = owndevice
def remove_incubator(self):
if self._incubator is None:
raise RuntimeError("No incubator is connected to the MCAM")
if self._own_incubator:
self._incubator.close()
self._incubator = None
def add_incubator(self, incubator, *, owndevice=False):
if self._incubator is not None:
raise RuntimeError("Cannot add two incubators.")
self._incubator = incubator
self._own_incubator = owndevice
def add_indicator_board(self, indicator_board, *, owndevice=False):
if self._indicator_board is not None:
raise RuntimeError("Cannot add two indicator boards.")
self._indicator_board = indicator_board
self._own_indicator_board = owndevice
def remove_indicator_board(self):
if self._indicator_board is None:
raise RuntimeError("No indicator board is connected to the MCAM")
if self._own_indicator_board:
self._indicator_board.close()
self._indicator_board = None
[docs]
def add_illumination(self, light, illumination_device, *, owndevice=False):
"""Add illumination board to the MCAM.
Once the board is added, the user should not attempt to independently close it.
The board will close automatically upon closing the MCAM,
or it can be closed by calling MCAM.remove_reflection_illumination()
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
Parameters
----------
light: owl.instruments.Illuminate
Open illumination board to attach to the MCAM
"""
if self._falcon is None:
raise RuntimeError("The usage of this function requires the MCAM to be open.")
if illumination_device in self._available_illumination_devices:
raise ValueError(f"{illumination_device} illumination device is already connected.")
self._available_illumination_devices[illumination_device] = light
self._own_illumination[illumination_device] = owndevice
def _close_illumination(self, illumination_device):
light = self._available_illumination_devices.pop(illumination_device)
own_device = self._own_illumination.pop(illumination_device)
if own_device:
light.close()
[docs]
def remove_illumination(self, illumination_device):
"""Remove and close illumination board.
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
"""
if self._falcon is None:
raise RuntimeError("The usage of this function requires the MCAM to be open.")
if illumination_device not in self._available_illumination_devices:
raise ValueError(
f"No {illumination_device} illumination device is currently connected.")
self._close_illumination(illumination_device)
if self.illumination_mode:
if self.illumination_mode.startswith(illumination_device):
self.illumination_mode = self.available_illumination_modes[0]
@property
def reflection_illumination(self):
if 'reflection' in self._available_illumination_devices:
return self._available_illumination_devices['reflection']
else:
return None
@property
def transmission_illumination(self):
if 'transmission' in self._available_illumination_devices:
return self._available_illumination_devices['transmission']
elif ('motorized' in self._available_illumination_devices and
self._available_illumination_devices['motorized']._transmission is not None):
return self._available_illumination_devices['motorized']
else:
return None
@property
def fluorescence_illumination(self):
if 'fluorescence' in self._available_illumination_devices:
return self._available_illumination_devices['fluorescence']
elif ('motorized' in self._available_illumination_devices and
self._available_illumination_devices['motorized']._fluorescence is not None):
return self._available_illumination_devices['motorized']
else:
return None
@property
def motorized_illumination(self):
if 'motorized' in self._available_illumination_devices:
return self._available_illumination_devices['motorized']
else:
return None
# TODO: Remove once no longer needed in falcon_gui
[docs]
def add_reflection_illumination(self, light, *, owndevice=False):
"""Add reflection illumination board to the MCAM.
Once the board is added, the user should not attempt to independently close it.
The board will close automatically upon closing the MCAM,
or it can be closed by calling MCAM.remove_reflection_illumination()
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
Parameters
----------
light: owl.instruments.Illuminate
Open reflection illumination board to attach to the MCAM
"""
self.add_illumination(light, 'reflection', owndevice=owndevice)
# TODO: Remove once no longer needed in falcon_gui
[docs]
def add_transmission_illumination(self, light, *, owndevice=False):
"""Add transmission illumination board to the MCAM.
Once the board is added, the user should not attempt to independently close it.
The board will close automatically upon closing the MCAM,
or it can be closed by calling MCAM.remove_transmission_illumination()
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
Parameters
----------
light: owl.instruments.Illuminate
Open transmission illumination board to attach to the MCAM
"""
self.add_illumination(light, 'transmission', owndevice=owndevice)
[docs]
def remove_transmission_illumination(self):
"""Remove and close transmission illumination board.
Once the board is added, the user should not attempt to independently close it.
The board will close automatically upon closing the MCAM,
or it can be closed by calling MCAM.remove_transmission_illumination()
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
"""
self.remove_illumination('transmission')
[docs]
def remove_reflection_illumination(self):
"""Remove and close reflection illumination board.
Once the board is added, the user should not attempt to independently close it.
The board will close automatically upon closing the MCAM,
or it can be closed by calling MCAM.remove_reflection_illumination()
Note
----
This function is not thread safe. Video or image acquisition should be stopped
before calling this function.
"""
self.remove_illumination('reflection')
@property
def temperature_monitor(self):
"""The MCAM temperature monitor subcomponent, if it is connected."""
return self._temperature_monitor
@property
def plate_tapper(self):
"""The MCAM plate tapper subcomponent, if it is connected."""
return self._plate_tapper
@property
def solenoid_driver(self):
"""The MCAM solenoid driver subcomponent, if it is connected."""
return self._solenoid_driver
@property
def incubator(self):
"""The MCAM incubator subcomponent, if it is connected."""
return self._incubator
@property
def indicator_board(self):
"""The MCAM indicator board subcomponent, if it is connected."""
return self._indicator_board
@property
def illumination_device(self):
warn("The illumination_device property is deprecated. "
"Please contact help@ramonaoptics.com with a screenshot "
"of your screen including this error message to discuss your usecase.",
stacklevel=2)
return self._illumination_device
@property
def illumination_mode(self):
"""The illumination mode of the MCAM at the present state."""
return self._illumination_mode
@illumination_mode.setter
def illumination_mode(self, illumination_mode):
if self._falcon is None:
raise RuntimeError("The usage of this function requires the MCAM to be open.")
def _split_illumination_mode(illumination_mode):
illumination_device_key = illumination_mode.split('_')[0]
device_illumination_mode = "_".join(illumination_mode.split('_')[1:])
return illumination_device_key, device_illumination_mode
if illumination_mode is not None and illumination_mode != "external":
# graceful deprecation for the 0.18.159 illumination mode update can be removed later
if illumination_mode.startswith(('transmission-', 'reflection-', 'fluorescence-')):
new_illumination_mode = illumination_mode.replace('-', '_')
warn(
'Illumination modes have been changed to no longer include `-`. '
f'Instead of using the illumination mode {illumination_mode}, '
f'please use {new_illumination_mode}.')
illumination_mode = new_illumination_mode
if 'motorized' in self._available_illumination_devices:
illumination_device_key = 'motorized'
device_illumination_mode = illumination_mode
elif illumination_mode != 'fluorescence':
(illumination_device_key,
device_illumination_mode) = _split_illumination_mode(illumination_mode)
else:
illumination_device_key = 'fluorescence'
device_illumination_mode = None
if self._falcon.dna_string != '0x40020000013A554114D121C5':
if illumination_mode not in self.available_illumination_modes:
raise ValueError(
f'Illumination mode {illumination_mode} is not available.')
elif illumination_device_key != 'fluorescence':
raise ValueError(f'Illumination board {illumination_device_key} is not available.')
illumination_device = self._available_illumination_devices[illumination_device_key]
else:
illumination_device = None
device_illumination_mode = None
mode_change = illumination_mode != self._illumination_mode
# must turn off other illumination devices before changing
# to a new illumination mode
if self._illumination_device is not None and mode_change:
# TODO: Once motorized illumination is setup with the C012 board
# it will need to accept the full device mode to know if it should
# use the fluorescence board or the brightfield
if isinstance(self._illumination_device, MotorizedIllumination):
self._illumination_device.set_brightness(
0., self._device_illumination_mode[len('fluorescence_'):])
else:
self._illumination_device.set_brightness(
0., self._device_illumination_mode)
self._illumination_device = illumination_device
self._illumination_mode = illumination_mode
self._device_illumination_mode = device_illumination_mode
@property
def available_illumination(self):
"""The available illumination hardware devices connected to the MCAM.
This property reports a tuple of strings containing the available
illumination devices presently connected to the MCAM.
The capabilities of these devices, in conjunction with the MCAM
determine the available illumination modes.
See also
--------
available_illumination_modes, set_illumination_brightness
"""
return tuple(self._available_illumination_devices.keys())
@property
def available_illumination_modes(self):
"""The available illumination modes of the MCAM.
This property reports a tuple of strings containing the available
illumination modes of the MCAM given the presently connected
illumination devices.
See also
--------
available_illumination, set_illumination_brightness
"""
available_illumination_modes = []
for illumination_device, light in self._available_illumination_devices.items():
for illumination_mode in light.available_illumination_modes:
if illumination_device == 'motorized':
if illumination_mode in light.available_fluorescence_illumination_modes:
available_illumination_modes.append(f'fluorescence_{illumination_mode}')
elif illumination_mode in light.available_transmission_illumination_modes:
available_illumination_modes.append(f'transmission_{illumination_mode}')
else:
available_illumination_modes.append(
f'{illumination_device}_{illumination_mode}'
)
available_illumination_modes.append('external')
return tuple(available_illumination_modes)
[docs]
def set_illumination_brightness(self, brightness, illumination_mode=None,
**kwargs):
"""Set the brightness of the chosen illumination mode.
Parameters
----------
brightness: float
A number between 0 and 1 indicating the overall brightness of the
illumination mode.
illumination_mode: str
The illumination mode of the MCAM. This should be one of the
reported available illumination modes.
See also
--------
available_illumination_modes
"""
if illumination_mode is not None:
self.illumination_mode = illumination_mode
else:
illumination_mode = self.illumination_mode
if illumination_mode in [
'transmission_visible_pwm_fullarray',
'transmission_visible_pwm_perimeter']:
device_illumination_mode_kwargs = {
'color_ratio': self.white_light_transmission,
}
elif illumination_mode in [
'transmission_visible_analog_fullarray',
'transmission_visible_analog_perimeter']:
if self.serial_number in [
'Kestrel0069', 'Kestrel0069R',
'Kestrel0070', 'Kestrel0070R',
'Kestrel0071', 'Kestrel0071R',
]:
device_illumination_mode_kwargs = {
'color_ratio': (1.6, 0.1, 0.29)
}
else:
# 20221106 SWAG from Mark that makes things "look good"
device_illumination_mode_kwargs = {
'color_ratio': (1.5, 0.1, 0.45)
}
elif illumination_mode in [
'transmission_ir850_pwm_fullarray',
'transmission_ir850_pwm_perimeter',
'transmission_ir850_analog_fullarray',
'transmission_ir850_analog_perimeter']:
device_illumination_mode_kwargs = {
'color_ratio': self.white_light_transmission_ir850,
}
elif illumination_mode in [
'reflection_visible_pwm_fullarray',
'reflection_visible_pwm_perimeter']:
device_illumination_mode_kwargs = {
'color_ratio': self.white_light_reflection,
}
else:
device_illumination_mode_kwargs = {}
# Only store the user provided ones since the others are defined by
# the illumination mode
self._illumination_brightness_kwargs = kwargs
device_illumination_mode_kwargs.update(kwargs)
# TODO: Once motorized illumination is setup with the C012 board
# it will need to accept the full device mode to know if it should
# use the fluorescence board or the brightfield
if isinstance(self._illumination_device, MotorizedIllumination):
if (self._device_illumination_mode[len('fluorescence_'):] in
self._illumination_device.available_fluorescence_illumination_modes):
device_illumination_mode = self._device_illumination_mode[len('fluorescence_'):]
elif (self._device_illumination_mode[len('transmission_'):] in
self._illumination_device.available_transmission_illumination_modes):
device_illumination_mode = self._device_illumination_mode[len('transmission_'):]
else:
device_illumination_mode = self._device_illumination_mode
if self.illumination_mode == 'external':
self._external_illumination_brightness = brightness
else:
self._illumination_device.set_brightness(
brightness,
device_illumination_mode,
**device_illumination_mode_kwargs,
)
[docs]
def get_illumination_brightness(self):
"""The brightness of the present illumination mode as a fraction.
Returns
-------
brightness: float
The illumination brightness of the illumination modes as a number
between 0. and 1.
See also
--------
set_illumination_brightness, illumination_mode
"""
if self.illumination_mode == 'external':
return self._external_illumination_brightness
elif self.illumination_mode is None:
return None
else:
return self._illumination_device.get_brightness()
@property
def calibration_filename(self):
"""File containing calibration information specific to the MCAM device.
Returns
-------
calibration_filename: Path
Path object pointing to the calibration file. This file may
or may not exist.
"""
if self._serial_number is None:
# Maybe this should be the DNA of the FPGA instead???
calibration_id = self._falcon.serial_number
else:
# Remove the "R" at the end of the serial number.
# This would appear if we have a "Fake" MCAM.
# That said, we typically want these MCAMs to mimic real ones.
calibration_id = self._serial_number.rstrip('R')
return user_config_dir / f"calibration_board_{calibration_id}.json"
@property
def serial_number(self):
"""The serial number of the MCAM."""
value = self._serial_number
if value is None:
warn(
"The serial number of this MCAM is undefined. "
"Please contact us at help@ramonaoptics.com with a screenshot "
"of your screen including this error message for one to be assigned to you.",
stacklevel=2)
return value
@property
def product_line(self):
return self._product_line
[docs]
def save_calibration(self, calibration_data=None, calibration_filename=None):
"""Save the given calibration data at the desired location.
Parameters
----------
calibration_data : Dict
Data for alignment and photometric corrections as well as metadata.
calibration_filename : Path-like, optional
Location to save the calibration data. If no location is given
defaults to self.calibration_filename.
Returns
-------
calibration_filename : Path
Returns the saved location of the calibration
"""
if calibration_data is None:
calibration_data = self._system_calibration_data
if calibration_filename is None:
calibration_filename = self.calibration_filename
save_calibration(calibration_data, calibration_filename=calibration_filename)
return calibration_filename
def _load_calibration(self, calibration_filename=None, **kwargs):
"""Load the calibration data from the desired location.
Parameters
----------
calibration_filename : Path-like, optional
Location to save the calibration data. If no location is given
defaults to the ``calibration_filename`` property.
Returns
-------
A dictionary containing all calibration data.
"""
import owl
if calibration_filename is None:
calibration_filename = self.calibration_filename
calibration_filename = Path(calibration_filename)
if not calibration_filename.exists():
try:
calibration_filename = download_factory_calibration(
calibration_name=calibration_filename.stem,
output=calibration_filename,
force_update=False
)
except Exception as e:
warn(str(e), stacklevel=2)
return {
"__owl_version__": owl.__version__,
"__owl_settings_type__": "mcam_system_calibration",
"available_calibrations": {},
}
calibration_data = load_calibration(calibration_filename=calibration_filename, **kwargs)
# convert mcam_calibration to mcam_system_calibration here to have access to the
# available virtual selections
if '__owl_settings_type__' not in calibration_data:
return calibration_data # this is to pass roundtrip test
if calibration_data['__owl_settings_type__'] == 'mcam_calibration':
system_calibration_data = {
"__owl_version__": owl.__version__,
"__owl_settings_type__": "mcam_system_calibration",
"available_calibrations": {},
}
for virtual_selection in self.available_virtual_selections:
system_calibration_data['available_calibrations'][virtual_selection] = \
deepcopy(calibration_data)
elif calibration_data['__owl_settings_type__'] == 'mcam_system_calibration':
system_calibration_data = calibration_data
else:
system_calibration_data = {
"__owl_version__": owl.__version__,
"__owl_settings_type__": "mcam_system_calibration",
"available_calibrations": {},
}
return system_calibration_data
def _set_calibration_data(self, value):
self._system_calibration_data = value
if self._system_calibration_data is not None:
calibration_data = self._system_calibration_data['available_calibrations'].get(
self.virtual_selection, None)
acquisition_mode_data = self._system_calibration_data.get('acquisition_mode', None)
fiducial_y_stage_position = self._system_calibration_data.get(
'fiducial_y_stage_position', None)
fiducial_x_stage_position = self._system_calibration_data.get(
'fiducial_x_stage_position', None)
if (fiducial_y_stage_position is not None and
fiducial_x_stage_position is not None):
fiducial_stage_position = (
fiducial_y_stage_position, fiducial_x_stage_position
)
else:
fiducial_stage_position = None
else:
calibration_data = None
acquisition_mode_data = None
fiducial_stage_position = None
self._acquisition_mode_data = acquisition_mode_data
self._calibration_data = calibration_data
self._fiducial_stage_position = fiducial_stage_position
if (
len(self.available_acquisition_modes) == 1 or
self.acquisition_mode not in self.available_acquisition_modes
):
self.acquisition_mode = 'manual'
# This could be a setter, but we want to leave this outside the
# public API for now while we fix things up.
if (self._calibration_data is not None and
'transmission_visible_pwm_fullarray_photometric_response'
in self._calibration_data):
response = self._calibration_data[
'transmission_visible_pwm_fullarray_photometric_response']
self._white_light_transmission = define_white_light(response)
else:
self._white_light_transmission = None
if (self._calibration_data is not None and
'reflection_visible_pwm_fullarray_photometric_response' in self._calibration_data):
response = self._calibration_data[
'reflection_visible_pwm_fullarray_photometric_response']
self._white_light_reflection = define_white_light(response)
else:
self._white_light_reflection = None
def _update_labware_acquisition_modes(self):
self._labware_acquisition_modes = {}
if (self._system_calibration_data is not None and
self._fiducial_stage_position is not None):
starting_virtual_selection = self.virtual_selection
for virtual_selection in self.available_virtual_selections:
calibration_data = self._system_calibration_data['available_calibrations'].get(
virtual_selection, None)
if calibration_data is None:
warn(
f'Calibration data is not set for virtual selection {virtual_selection}. '
'Will not add labware acquisition modes.',
stacklevel=2,
)
continue
pixel_width = None
for illumination_mode in brightfield_illumination_modes:
pixel_width = calibration_data.get(f'{illumination_mode}_pixel_width', None)
if pixel_width is not None:
break
if pixel_width is None:
warn(
f'Pixel width is not set for virtual selection {virtual_selection}. '
'Will not add labware acquisition modes.',
stacklevel=2,
)
continue
self.virtual_selection = virtual_selection
sensor_fov = min(self.image_shape) * pixel_width
exif_orientation = self.exif_orientation
N_cameras = self.N_cameras
micro_camera_separation = (
self.micro_camera_separation_y,
self.micro_camera_separation_x
)
for labware_key in labware_well_plate_properties:
new_mode = f'{virtual_selection}_{labware_key}'
well_properties = labware_well_plate_properties[labware_key]
if len(self.available_virtual_selections) > 1:
description = f'{virtual_selection} - {well_properties['description']}'
else:
description = f'{well_properties['description']}'
# add a buffer of .2 mm - this was manually chosen as a value
# that gave enough padding to handle our tolerances without
# drastically increasing the number of images
well_buffer = .0002 / well_properties['well_diameter']
self._labware_acquisition_modes[new_mode] = AcquisitionModeProperties(
**create_full_coverage_acquisition_mode_from_well_properties(
well_properties,
sensor_fov,
description,
self._fiducial_stage_position,
exif_orientation,
N_cameras,
micro_camera_separation,
pixel_width,
virtual_selection,
self.camera_offset,
overlap=None,
well_buffer=well_buffer
)
)
self.virtual_selection = starting_virtual_selection
def _open(self, *, stacklevel_increment):
stacklevel = stacklevel_increment + 1
serial_number = self._init_kwargs.get(
'serial_number',
# Provide a means to specify a serial number via an environment
# variable so we can write examples more cleanly
os.environ.get('RAMONA_MCAM_SERIAL_NUMBER', None)
)
if serial_number is not None:
fake_device = (
# New style fake serial numbers end in 'R'
serial_number.endswith('R') or
# Old style, can specify the FPGA serial number
# These start with 0xC002 or 0x4EAD for fake ones
serial_number.startswith("0xC002") or
serial_number.startswith("0x4EAD")
)
# This device was the prototype for the Vireo
# and existed before the Vireo name was chosen
# We built up a lot of our testing around this device
# so we want to make sure those tests continue to work
serial_number_aliases = {
'Kestrel0101R': 'Vireo1101R',
}
if fake_device and serial_number in serial_number_aliases:
serial_number = serial_number_aliases[serial_number]
if serial_number in mcam_registry['serial_number'].to_numpy():
# We know these are unique, so just pick the first one.
entry = mcam_registry.loc[
mcam_registry['serial_number'] == serial_number
].iloc[0]
elif serial_number in mcam_registry.index:
# We know these are unique, so just pick the first one.
entry = mcam_registry.loc[serial_number]
serial_number = entry['serial_number']
if pd.isna(serial_number):
warn("You specified an old-style serial number. " # by Falcon.dna_string
"Contact Ramona Optics at help@ramonaoptics.com "
"so that we may assign you a valid serial number.",
stacklevel=2 + stacklevel_increment)
else:
warn("You specified an old-style serial number. " # by Falcon.dna_string
f'The correct serial number for this device is "{serial_number}".',
stacklevel=2 + stacklevel_increment)
else:
raise ValueError(
f"No MCAM with serial number {serial_number} could be found."
)
dna = entry.name
else:
fake_device = self._init_kwargs.get('fake_device', False)
if not fake_device:
dna = None
elif isinstance(fake_device, str):
# TODO: deprecate this later.
# for now, we have too much that uses this.
dna = fake_device
else:
# This DNA corresponds to a fake_falcon with color sensors
# Eventually we will want to enable the user to specify
# some kind of dna so that they can choose between different
# fake device, but for now, this is enough to avoid a warning
dna = '0x4EADBEEFCAFE1010BA5EBA11'
self._fake_device = fake_device
self._available_illumination_devices = {}
self._own_illumination = {}
self._stages = {}
self._available_mechanical_states = {}
self._external_illumination_brightness = 0
self._illumination_brightness_kwargs = {}
extra_kwargs = {}
# This is mostly for a new feature that allows us
# to specify the serial number of the daughter board for fake devices
# this mostly helps in testing when we want to test new system configurations
if dna is not None:
extra_kwargs['has_daughter_board'] = \
fpga_registry.loc[dna]['daughter_board_serial_number']
self._falcon = Falcon(
sensor_ini_filename=self._init_kwargs.get('config_file'),
interface_number=self._init_kwargs.get('interface_number'),
dna=dna,
_stacklevel_increment=1 + stacklevel_increment, # Pass the warnings back to the user.
**extra_kwargs,
)
opening_parameters = get_opening_parameters(
self._falcon.dna_string,
self._init_kwargs,
)
# Opening parameters will overwrite the serial number
# with the serial number associated with the MCAM that was
# actually opened. Therefore, we must create the datasets
# after this serial number is updated to maximum knowledge we have.
self._serial_number = opening_parameters['serial_number']
if opening_parameters['maximum_datarate_GiBps'] is not None:
expected_maximum_datarate = opening_parameters['maximum_datarate_GiBps'] * (1024 ** 3)
# Check if we are UNDER 10% of the expected datarate
if (
expected_maximum_datarate - self._falcon.maximum_datarate
) / expected_maximum_datarate > 0.1:
message = (
"The measured datarate of the MCAM is too low and indicates a "
"potential hardware problem. Please contact Ramona Optics for "
"help by email at help@ramonaoptics.com"
f"with the serial number of your unit: ({self._serial_number})."
)
if RAMONA_MCAM_ADVANCED_MODE:
measured_datarate_GiBps = self._falcon.maximum_datarate / (1024 ** 3)
message += (
" The measured datarate of your system is "
f"{measured_datarate_GiBps:.2f} GiB/s, while "
"the expected value of the datarate is "
f"{opening_parameters['maximum_datarate_GiBps']} GiB/s."
)
raise RuntimeError(message)
self._available_virtual_selections = opening_parameters.get('virtual_selections')
self._physical_acquisition_count = np.zeros(
self._falcon.mcam_physical_shape,
dtype=int
)
self._physical_acquisition_index = np.zeros(
self._falcon.mcam_physical_shape,
dtype=int
)
self._latest_acquisition_index = 0
available_virtual_selections = self.available_virtual_selections
if len(available_virtual_selections):
self.virtual_selection = available_virtual_selections[0]
else:
self._virtual_selection = slice(None), slice(None)
if opening_parameters['calibration_filename'] != False: # noqa
calibration_filename = opening_parameters['calibration_filename']
self._set_calibration_data(self._load_calibration(calibration_filename))
# Exif must be set before updating labware acquisition modes. Typical Vireo units use
# EXIF 7 with the HTG FPGA; Sparrow (integrated sensor board + FPGA) uses EXIF 1.
self.exif_orientation = opening_parameters["exif_orientation"]
if self.exif_orientation in (1, 7):
self._update_labware_acquisition_modes()
# Always start in the manual acquisition mode
self.acquisition_mode = 'manual'
def truthy_or_None(value):
return value or (value is None)
# Mark: 2020/07/16
# This logic is a little convoluted, because for now, I want to support
# partial system. In the future, upon not finding a requested device
# We would simply error, and close all the previous devices we have
# opened
axis_stage_serial_numbers = (
('z', opening_parameters.get('z_stage_serial_number')),
('y', opening_parameters.get('y_stage_serial_number')),
('x', opening_parameters.get('x_stage_serial_number')),
('x_sample', opening_parameters.get('x_sample_stage_serial_number')),
)
connection_managers = []
stages_to_home = []
for axis, stage_serial_number in axis_stage_serial_numbers:
if truthy_or_None(stage_serial_number):
stage = _connect_to_stage(
axis=axis,
stage_serial_number=stage_serial_number,
fake_device=fake_device,
connection_managers=connection_managers,
ensure_homed=False,
_stacklevel_increment=1 + stacklevel_increment,
)
if stage is None:
continue
stages_to_home.append(stage)
self._add_stage(axis, stage, owndevice=True,
_stacklevel_increment=1 + stacklevel_increment)
connection_managers.append(stage.connection_manager)
device_serial_numbers_keys = (
('reflection', 'reflection_illumination_serial_number'),
('transmission', 'transmission_illumination_serial_number'),
('fluorescence', 'fluorescence_illumination_serial_number'),
)
illumination_devices_to_home = []
motorized_illumination_devices = {}
for device, serial_number_key in device_serial_numbers_keys:
serial_number = opening_parameters.get(serial_number_key)
if truthy_or_None(serial_number):
if not isinstance(serial_number, str):
illumination_kwargs = {
'serial_number': None,
'_fake_device': fake_device,
}
else:
illumination_kwargs = {
'serial_number': serial_number,
'_fake_device': serial_number.endswith('R'),
}
try:
light = open_illumination(device,
**illumination_kwargs)
if light.serial_number in motorized_illumination_serial_numbers:
connection_managers = [stage.connection_manager
for _, (stage, _) in self._stages.items()]
motorized_illumination_devices[device] = light
else:
self.add_illumination(light, device, owndevice=True)
except RuntimeError:
with_illumination = opening_parameters.get(serial_number_key)
if with_illumination:
# Only raise an exception if the user explicitly asked
# for the board
raise
elif with_illumination is None:
warn(f"No {device} illumination was found. "
"To avoid this warning, provide the "
f"with_{device}_illumination "
"parameter when opening the MCAM.",
stacklevel=stacklevel)
if len(motorized_illumination_devices) > 0:
light = MotorizedIllumination(
**motorized_illumination_devices,
connection_managers=connection_managers,
ensure_homed=False,
)
illumination_devices_to_home.append(light)
self.add_illumination(light, 'motorized', owndevice=True)
if self._init_kwargs.get('ensure_homed', True):
for stage in stages_to_home + illumination_devices_to_home:
stage.ensure_homed(wait_until_idle=False)
for stage in stages_to_home + illumination_devices_to_home:
stage.wait_until_idle()
# Set up the device in the first illumination mode
self.illumination_mode = self.available_illumination_modes[0]
temperature_monitor_serial_number = opening_parameters.get(
'temperature_monitor_serial_number')
if truthy_or_None(temperature_monitor_serial_number):
if ((fake_device and not isinstance(temperature_monitor_serial_number, str))
or (
isinstance(temperature_monitor_serial_number, str) and
temperature_monitor_serial_number.endswith('R')
)):
TemperatureMonitor = FakeTemperatureMonitor
else:
TemperatureMonitor = RealTemperatureMonitor
kwargs = {}
if isinstance(temperature_monitor_serial_number, str):
kwargs['serial_number'] = temperature_monitor_serial_number
try:
temperature_monitor = TemperatureMonitor(**kwargs)
self.add_temperature_monitor(temperature_monitor, owndevice=True)
except RuntimeError:
with_temperature_monitor = opening_parameters.get(
'temperature_monitor_serial_number')
if with_temperature_monitor:
# Only raise an exception if the user explicitly asked
# for the board
raise
elif with_temperature_monitor is None:
warn("No temperature monitor was found. "
"To avoid this warning, provide the "
"with_temperature_monitor"
"parameter when opening the MCAM.",
stacklevel=stacklevel)
plate_tapper_serial_number = opening_parameters.get(
'plate_tapper_serial_number')
if truthy_or_None(plate_tapper_serial_number):
if ((fake_device and not isinstance(plate_tapper_serial_number, str))
or (
isinstance(plate_tapper_serial_number, str) and
plate_tapper_serial_number.endswith('R')
)):
PlateTapper = FakePlateTapper
else:
PlateTapper = RealPlateTapper
kwargs = {}
if isinstance(plate_tapper_serial_number, str):
kwargs['serial_number'] = plate_tapper_serial_number
try:
plate_tapper = PlateTapper(**kwargs)
self.add_plate_tapper(plate_tapper, owndevice=True)
except RuntimeError:
with_plate_tapper = opening_parameters.get(
'plate_tapper_serial_number')
if with_plate_tapper:
# Only raise an exception if the user explicitly asked
# for the board
raise
elif with_plate_tapper is None:
warn("No plate tapper was found. "
"To avoid this warning, provide the "
"with_plate_tapper"
"parameter when opening the MCAM.",
stacklevel=stacklevel)
solenoid_driver_serial_number = opening_parameters.get(
'solenoid_driver_serial_number')
if truthy_or_None(solenoid_driver_serial_number):
if ((fake_device and not isinstance(solenoid_driver_serial_number, str))
or (
isinstance(solenoid_driver_serial_number, str) and
solenoid_driver_serial_number.endswith('R')
)):
SolenoidDriver = FakeSolenoidDriver
else:
SolenoidDriver = RealSolenoidDriver
kwargs = {}
if isinstance(solenoid_driver_serial_number, str):
kwargs['serial_number'] = solenoid_driver_serial_number
try:
solenoid_driver = SolenoidDriver(**kwargs)
self.add_solenoid_driver(solenoid_driver, owndevice=True)
except RuntimeError:
with_solenoid_driver = opening_parameters.get(
'solenoid_driver_serial_number')
if with_solenoid_driver:
raise
elif with_solenoid_driver is None:
warn("No solenoid driver was found. "
"To avoid this warning, provide the "
"with_solenoid_driver"
"parameter when opening the MCAM.",
stacklevel=stacklevel)
incubator_serial_number = opening_parameters.get(
'incubator_serial_number')
if truthy_or_None(incubator_serial_number):
from .okolabboldline3 import FakeOkolabBoldLine3
from .okolabboldline3 import OkolabBoldLine3 as RealOkolabBoldLine3
from .okolabboldline3 import known_devices as known_oko_bold_line3_devices
if ((fake_device and not isinstance(incubator_serial_number, str))
or (
isinstance(incubator_serial_number, str) and
incubator_serial_number not in known_oko_bold_line3_devices.index and
incubator_serial_number.endswith('R')
)):
OkolabBoldLine3 = FakeOkolabBoldLine3
else:
OkolabBoldLine3 = RealOkolabBoldLine3
kwargs = {}
if isinstance(incubator_serial_number, str):
kwargs['serial_number'] = incubator_serial_number
try:
incubator = OkolabBoldLine3(**kwargs)
self.add_incubator(incubator, owndevice=True)
except RuntimeError:
with_incubator = opening_parameters.get(
'incubator_serial_number')
if with_incubator:
# Only raise an exception if the user explicitly asked
# for the board
raise
elif with_incubator is None:
warn("No incubator was found. "
"To avoid this warning, provide the "
"with_incubator"
"parameter when opening the MCAM.",
stacklevel=stacklevel)
indicator_board_serial_number = opening_parameters.get(
'indicator_board_serial_number')
if truthy_or_None(indicator_board_serial_number):
if ((fake_device and not isinstance(indicator_board_serial_number, str))
or (
isinstance(indicator_board_serial_number, str) and
indicator_board_serial_number.endswith('R')
)):
IndicatorBoard = FakeIndicatorBoard
else:
IndicatorBoard = RealIndicatorBoard
kwargs = {}
if isinstance(indicator_board_serial_number, str):
kwargs['serial_number'] = indicator_board_serial_number
try:
indicator_board = IndicatorBoard(**kwargs)
self.add_indicator_board(indicator_board, owndevice=True)
except RuntimeError:
with_indicator_board = opening_parameters.get(
'indicator_board_serial_number')
if with_indicator_board:
# Only raise an exception if the user explicitly asked
# for the board
raise
elif with_indicator_board is None:
warn("No indicator board was found. "
"To avoid this warning, provide the "
"with_indicator_board"
"parameter when opening the MCAM.",
stacklevel=stacklevel)
# Jed 2022/04/22 HACK HACK HACK
# Hacks to have the system not connected to illumination sources
# default to be able to attach calibration metadata
if self._falcon.dna_string == '0x40020000013A554114D121C5':
self.illumination_mode = 'fluorescence'
start_pixel = opening_parameters.get('start_pixel')
end_pixel = opening_parameters.get('end_pixel')
if start_pixel is not None and end_pixel is not None:
self.select_pixels(start_pixel, end_pixel)
# Convert from tuples to easier to use python structures
for key, state in opening_parameters.get('mechanical_states', ()):
self._available_mechanical_states[key] = {
axis: position
for axis, position in state
}
self._apply_solenoid_for_mechanical_state_key(self._last_set_mechanical_state)
self._fps_calculation_info = (0.0, None)
# Default imaging settings
self.exposure = _default_exposure
# By default, analog and digital gains are set to 1
# We need knowledge of both to compute the total gain
# and cache the result
self.digital_gain = _default_digital_gain
self.analog_gain = _default_analog_gain
if (output_trigger_events := opening_parameters["output_trigger_events"]) is not None:
self.output_trigger_events = output_trigger_events
from owl.analysis._software_registry import (
read_mcam_software_registry,
validate_software_enabled,
)
_valid_product_lines = ("vireo", "kestrel", "research", "eagle", "falcon", "sparrow")
try:
entry = read_mcam_software_registry(self.serial_number)
product_line = entry["product_line"]
if product_line not in _valid_product_lines:
raise ValueError(f"Invalid product line: {product_line}")
self._product_line = product_line
except Exception as e:
print(f"Error reading MCAM software registry: {e}")
self._product_line = "unknown"
if validate_software_enabled(self.serial_number, 'sensor_subsampling_xy_binning'):
self._falcon.sensor_subsampling_mode = 'bin'
else:
self._falcon.sensor_subsampling_mode = 'legacy'
self._warn_if_not_in_registry(stacklevel_increment=stacklevel_increment + 1)
def _warn_if_not_in_registry(self, stacklevel_increment):
# 2022/04/04
# TODO, we should really check if all the peripherals are here too. But for now
# lets just do a cursory check.
if self._falcon.dna_string not in mcam_registry.index:
suggested_registry_entry = pformat(self._registry_entry,
sort_dicts=False, indent=4)
warn(f"""
You have an MCAM that isn't registered. Please help us in getting it registered by
sending us the following email:
To: help@ramonaoptics.com
Subject: MCAM registry
Body:
Dear Ramona Optics,
I'm letting you know of the following information:
```
{{
{suggested_registry_entry[1:-1]},
}},
```
Best,
""", stacklevel=2 + stacklevel_increment)
@property
def _registry_entry(self):
if self.x_stage is not None:
x_stage_serial_number = self.x_stage.stage_serial_number
else:
x_stage_serial_number = False
if self.y_stage is not None:
y_stage_serial_number = self.y_stage.stage_serial_number
else:
y_stage_serial_number = False
if self.z_stage is not None:
z_stage_serial_number = self.z_stage.stage_serial_number
else:
z_stage_serial_number = False
if self.x_sample_stage is not None:
x_sample_stage_serial_number = self.x_sample_stage.stage_serial_number
else:
x_sample_stage_serial_number = False
if self.transmission_illumination is not None:
transmission_illumination_serial_number = self.transmission_illumination.serial_number
else:
transmission_illumination_serial_number = False
if self.reflection_illumination is not None:
reflection_illumination_serial_number = self.reflection_illumination.serial_number
else:
reflection_illumination_serial_number = False
if self.fluorescence_illumination is not None:
fluorescence_illumination_serial_number = self.fluorescence_illumination.serial_number
else:
fluorescence_illumination_serial_number = False
if self._temperature_monitor is not None:
temperature_monitor_serial_number = self._temperature_monitor.serial_number
else:
temperature_monitor_serial_number = False
if self._plate_tapper is not None:
plate_tapper_serial_number = self._plate_tapper.serial_number
else:
plate_tapper_serial_number = False
if self._solenoid_driver is not None:
solenoid_driver_serial_number = self._solenoid_driver.serial_number
else:
solenoid_driver_serial_number = False
if self._incubator is not None:
incubator_serial_number = self._incubator.serial_number
else:
incubator_serial_number = False
if self._indicator_board is not None:
indicator_board_serial_number = self._indicator_board.serial_number
else:
indicator_board_serial_number = False
date = datetime.now(get_localzone()).isoformat()
return {
'dna_string': self._falcon.dna_string,
'calibration_filename': None,
'x_stage_serial_number': x_stage_serial_number,
'y_stage_serial_number': y_stage_serial_number,
'z_stage_serial_number': z_stage_serial_number,
'x_sample_stage_serial_number': x_sample_stage_serial_number,
'transmission_illumination_serial_number': transmission_illumination_serial_number,
'reflection_illumination_serial_number': reflection_illumination_serial_number,
'fluorescence_illumination_serial_number': fluorescence_illumination_serial_number,
'temperature_monitor_serial_number': temperature_monitor_serial_number,
'plate_tapper_serial_number': plate_tapper_serial_number,
'solenoid_driver_serial_number': solenoid_driver_serial_number,
'incubator_serial_number': incubator_serial_number,
'indicator_board_serial_number': indicator_board_serial_number,
'exif_orientation': self.exif_orientation,
'start_pixel': self.start_pixel,
'end_pixel': self.end_pixel,
'date': date,
}
def _create_dataset(self, data, *, previous_acquisition_parameters=None):
from .. import mcam_data
bin_mode = self.bin_mode
if data is None:
if bin_mode == 4:
data = self._falcon.data_4x4
elif bin_mode == 2:
data = self._falcon.data_2x2
else: # if bin_mode == 1:
data = self._falcon.data_1x1
start_pixel = self._falcon.start_pixel
end_pixel = self._falcon.end_pixel
image_y = range(self.N_cameras_Y)
image_x = range(self.N_cameras_X)
y = range(start_pixel[0], end_pixel[0], bin_mode)
x = range(start_pixel[1], end_pixel[1], bin_mode)
N_cameras = (len(image_y), len(image_x))
image_shape = (len(y), len(x))
shape = N_cameras + image_shape
if shape != data.shape:
raise RuntimeError(
"Provided data doesn't have the expected shape. "
f"Expecting {shape}, got {data.shape}."
)
dataset = mcam_data.new_dataset(
array=data,
N_cameras=N_cameras,
image_shape=image_shape,
coords=dict(
image_y=image_y,
image_x=image_x,
y=y,
x=x,
)
)
dataset = self._add_system_coordinates(
dataset,
previous_acquisition_parameters=previous_acquisition_parameters,
)
return dataset
@property
def virtual_selection(self):
return self._virtual_selection_key
@virtual_selection.setter
def virtual_selection(self, virtual_selection_key):
if virtual_selection_key not in self.available_virtual_selections:
raise ValueError(f'The given virtual_selection `{virtual_selection_key}` '
'is not supported. Must select from the following options '
f'{self.available_virtual_selections}')
self._virtual_selection_key = virtual_selection_key
virtual_selection_slices = self._available_virtual_selections[virtual_selection_key]
self._virtual_selection = virtual_selection_slices
# reset calibration data to apply the calibration data for the new current virtual selection
self._set_calibration_data(self._system_calibration_data)
if self._illumination_device is None:
return
if (hasattr(self._illumination_device, 'has_circuit_control') and
self._illumination_device.has_circuit_control and
# 20260507 The alignment between the led circuits and the camera virtual selections
# is not standardized. We must handle the specific alignment for individual Vireos
# before enabling them.
self.serial_number in ('Vireo1117', 'Vireo1117R', 'Vireo1111', 'Vireo1111R')):
start = (virtual_selection_slices[1].start
if virtual_selection_slices[1].start is not None
else 0)
step = (virtual_selection_slices[1].step
if virtual_selection_slices[1].step is not None
else 1)
current_brightness = self._illumination_device.get_brightness()
starts_at_circuit_0 = (start % 4) == 0
steps_to_circuit_0 = ((start + step) % 4) == 0
enable_circuit_0 = starts_at_circuit_0 or steps_to_circuit_0
starts_at_circuit_1 = (start % 4) == 2
steps_to_circuit_1 = ((start + step) % 4) == 2
enable_circuit_1 = starts_at_circuit_1 or steps_to_circuit_1
self._illumination_device.led_circuit_0_enabled = enable_circuit_0
self._illumination_device.led_circuit_1_enabled = enable_circuit_1
self._illumination_device.set_brightness(current_brightness)
@property
def camera_offset(self):
return (
self._falcon.sensor_locations_y[0, 0],
self._falcon.sensor_locations_x[0, 0]
)
@property
def available_virtual_selections(self):
if self._available_virtual_selections is None:
return tuple()
else:
return tuple(self._available_virtual_selections.keys())
@property
def _virtual_selection(self):
"""The subset of micro-cameras that are in use for the given MCAM.
Example
-------
To choose the first ``(3, 2)`` micro-cameras for all subsequent
acquisitions, the following can be used.
>>> mcam._virtual_selection = (slice(3), slice(2))
>>> print(mcam.N_cameras)
(3, 2)
Notes
-----
This API is still very experimental and is not expected to function
correctly with calibrated images.
Please contact info@ramonaoptics.com to discuss your usecase.
"""
return self._falcon.virtual_selection
@_virtual_selection.setter
def _virtual_selection(self, value):
old_virtual_selection = self._falcon.virtual_selection
self._falcon.virtual_selection = value
if len(self._falcon.mcam_shape) != 2:
self._falcon.virtual_selection = old_virtual_selection
raise ValueError(f"The provided virtual selection ({value}) "
"resulted in a shape that is not 2 dimensional. "
"This is not supported yet. Please contact "
"info@ramonaoptics.com to discuss your usecase.")
elif self._falcon.mcam_shape[0] * self._falcon.mcam_shape[1] == 0:
self._falcon.virtual_selection = old_virtual_selection
raise ValueError(f"The provided virtual selection ({value}) "
"resulted in a shape that contains no micro-cameras."
"Please contact "
"info@ramonaoptics.com to discuss your usecase.")
self._acquisition_count = self._physical_acquisition_count[value]
self._acquisition_index = self._physical_acquisition_index[value]
self.free_video_buffer(garbage_collection=False)
# Mark - 2024/08/26
# We had this pretty old feature where when a user
# could take images of one sensor at a time
# building up a whole dataset with different exposures or digital gains
# in 2024, this is really not recommended, however we continue to support
# consistent metadata since the GUI expects this feature and continues
# to use it.
# However, when the virtual selection changes, we swap out this whole
# buffer
self._last_image_acquisition_parameters = None
@property
def available_acquisition_modes(self):
if (
(self._acquisition_mode_data is not None) and
(self.available_virtual_selections is not None)
):
modes = tuple(
key
for key, value in self._acquisition_mode_data.items()
if (
(value.virtual_selection is None) or
(value.virtual_selection in self.available_virtual_selections)
)
)
else:
modes = ()
if self._labware_acquisition_modes is not None:
labware_keys = tuple(
lw for lw in self._labware_acquisition_modes
)
else:
labware_keys = ()
return ('manual',) + modes + labware_keys
@property
def acquisition_mode(self):
return self._acquisition_mode
@acquisition_mode.setter
def acquisition_mode(self, acquisition_mode):
if acquisition_mode not in self.available_acquisition_modes:
raise ValueError(f'Acquisition mode {acquisition_mode} is not available.')
self._acquisition_mode = acquisition_mode
properties = self.get_acquisition_properties(acquisition_mode)
if properties is not None and properties.virtual_selection is not None:
self.virtual_selection = properties.virtual_selection
@property
def acquisition_properties(self):
return self.get_acquisition_properties(self.acquisition_mode)
def get_acquisition_properties(self, acquisition_mode=None):
if acquisition_mode is None:
acquisition_mode = self.acquisition_mode
if (self._labware_acquisition_modes is not None and
acquisition_mode in self._labware_acquisition_modes):
return self._labware_acquisition_modes[acquisition_mode]
if self._acquisition_mode_data is None:
return None
if acquisition_mode not in self._acquisition_mode_data:
return None
return self._acquisition_mode_data[acquisition_mode]
def _update_metadata(
self,
dataset,
acquisition_parameters,
*,
update_software_timestamp=True,
dataset_selection_slice=None,
selection=None,
):
"""Update metadata for a boolean selection."""
self._latest_acquisition_index += 1
self._acquisition_count[acquisition_parameters['selection']] += 1
self._acquisition_index[acquisition_parameters['selection']] = \
self._latest_acquisition_index
if dataset_selection_slice is None:
dataset_selection_slice = Ellipsis
# xarray doesn't know how to index in the same way as numpy arrays
# Therefore, go and get the underlying data array, before using the
# index
if selection is None:
selection = acquisition_parameters['selection']
# overwrite the exif orientation even though it seldom changes
dataset.exif_orientation.data[...] = self.exif_orientation
dataset.acquisition_count.data[selection] += 1
dataset.acquisition_index.data[selection] += 1
dataset.exposure.data[selection] = _maybe_slice(
acquisition_parameters['exposure'], selection, dataset_selection_slice)
dataset.exposure2.data[selection] = _maybe_slice(
acquisition_parameters['exposure2'], selection, dataset_selection_slice)
dataset.interlaced_hdr.data[selection] = _maybe_slice(
acquisition_parameters['interlaced_hdr'], selection, dataset_selection_slice)
dataset.analog_gain.data[selection] = _maybe_slice(
acquisition_parameters['analog_gain'], selection, dataset_selection_slice)
dataset.digital_gain.data[selection] = _maybe_slice(
acquisition_parameters['digital_gain'], selection, dataset_selection_slice)
dataset.digital_red_gain.data[selection] = _maybe_slice(
acquisition_parameters['digital_red_gain'], selection, dataset_selection_slice)
dataset.digital_green1_gain.data[selection] = _maybe_slice(
acquisition_parameters['digital_green1_gain'], selection, dataset_selection_slice)
dataset.digital_blue_gain.data[selection] = _maybe_slice(
acquisition_parameters['digital_blue_gain'], selection, dataset_selection_slice)
dataset.digital_green2_gain.data[selection] = _maybe_slice(
acquisition_parameters['digital_green2_gain'], selection, dataset_selection_slice)
dataset.start_pixel_y.data[selection] = _maybe_slice(
acquisition_parameters['start_pixel_y'], selection, dataset_selection_slice)
dataset.start_pixel_x.data[selection] = _maybe_slice(
acquisition_parameters['start_pixel_x'], selection, dataset_selection_slice)
dataset.end_pixel_y.data[selection] = _maybe_slice(
acquisition_parameters['end_pixel_y'], selection, dataset_selection_slice)
dataset.end_pixel_x.data[selection] = _maybe_slice(
acquisition_parameters['end_pixel_x'], selection, dataset_selection_slice)
dataset.frame_rate_setpoint.data[selection] = _maybe_slice(
acquisition_parameters['frame_rate_setpoint'], selection, dataset_selection_slice)
dataset.latest_acquisition_index.data[...] = self._latest_acquisition_index
dataset.acquisition_index.data[selection] = self._latest_acquisition_index
dataset['external_illumination_brightness'].data[...] = \
self._external_illumination_brightness
dataset['bayer_pattern'] = acquisition_parameters['bayer_pattern']
if 'temperature_monitor_chamber_start_temperature' in acquisition_parameters:
dataset['temperature_monitor_chamber_start_temperature'] = \
acquisition_parameters['temperature_monitor_chamber_start_temperature']
if 'incubator_start_temperature' in acquisition_parameters:
dataset['incubator_start_temperature'] = \
acquisition_parameters['incubator_start_temperature']
if 'incubator_start_room_temperature' in acquisition_parameters:
dataset['incubator_start_room_temperature'] = \
acquisition_parameters['incubator_start_room_temperature']
if 'incubator_start_lid_temperature' in acquisition_parameters:
dataset['incubator_start_lid_temperature'] = \
acquisition_parameters['incubator_start_lid_temperature']
if 'incubator_start_humidity' in acquisition_parameters:
dataset['incubator_start_humidity'] = \
acquisition_parameters['incubator_start_humidity']
if 'incubator_start_oxygen_concentration' in acquisition_parameters:
dataset['incubator_start_oxygen_concentration'] = \
acquisition_parameters['incubator_start_oxygen_concentration']
if 'incubator_start_carbon_dioxide_concentration' in acquisition_parameters:
dataset['incubator_start_carbon_dioxide_concentration'] = \
acquisition_parameters['incubator_start_carbon_dioxide_concentration']
if 'workstation_start_process_memory_rss' in acquisition_parameters:
dataset['workstation_start_process_memory_rss'] = \
acquisition_parameters['workstation_start_process_memory_rss']
if 'workstation_start_process_memory_vms' in acquisition_parameters:
dataset['workstation_start_process_memory_vms'] = \
acquisition_parameters['workstation_start_process_memory_vms']
if 'workstation_start_system_memory_total' in acquisition_parameters:
dataset['workstation_start_system_memory_total'] = \
acquisition_parameters['workstation_start_system_memory_total']
if 'workstation_start_system_memory_used' in acquisition_parameters:
dataset['workstation_start_system_memory_used'] = \
acquisition_parameters['workstation_start_system_memory_used']
if 'workstation_start_system_memory_buffers' in acquisition_parameters:
dataset['workstation_start_system_memory_buffers'] = \
acquisition_parameters['workstation_start_system_memory_buffers']
if 'workstation_start_system_memory_cached' in acquisition_parameters:
dataset['workstation_start_system_memory_cached'] = \
acquisition_parameters['workstation_start_system_memory_cached']
if 'workstation_start_system_memory_available' in acquisition_parameters:
dataset['workstation_start_system_memory_available'] = \
acquisition_parameters['workstation_start_system_memory_available']
if 'workstation_start_system_memory_free' in acquisition_parameters:
dataset['workstation_start_system_memory_free'] = \
acquisition_parameters['workstation_start_system_memory_free']
if 'workstation_start_system_memory_percent' in acquisition_parameters:
dataset['workstation_start_system_memory_percent'] = \
acquisition_parameters['workstation_start_system_memory_percent']
if update_software_timestamp:
# Timezones are stored in utc time.
# We can't assign timezone aware datetime
# because numpy has deprecated it
# https://github.com/numpy/numpy/blob/main/doc/source/release/1.11.0-notes.rst#datetime64-changes
dataset.software_timestamp.data[selection] = datetime.now(UTC).replace(tzinfo=None)
# At opening the mcam takes a full field of view acquisition setting all data to valid.
# Valid data is never return to False so any data saved is expected to be valid. We may
# change this in the future to reset valid_data to False at acquisition and then only set
# the acquires selection to True.
dataset.valid_data.data[selection] = True
for illumination_device in self._available_illumination_devices:
prefix = f'{illumination_device}_illumination'
light = getattr(self, prefix)
dataset[f'{prefix}_illumination_mode'].data[...] = \
light.get_illumination_mode()
dataset[f'{prefix}_brightness'].data[...] = \
light.get_brightness()
dataset['illumination_brightness'].data[...] = self.get_illumination_brightness()
# These are hardware properties and expected to read somewhat slowly
# We first create a lazy dictionary and execute them in parallel
# However, depending on the communication link, there may be locks in
# the communication that are enforced.
# In particular it seems like the stages take an increasing amount of time
# to communicate based on the number of hops they have in the serial communication
# daisy chain for Zaber's communication protocol
# https://gitlab.com/ramonaoptics/mcam/python-owl/-/merge_requests/6057#note_2083116199
lazy_dict = {}
for axis, (stage, _) in self._stages.items():
# each of these calls takes about 11 ms as measured on the
# Vireo1107 for the z, y, and x stage
lazy_dict[f'{axis}_stage'] = partial(getattr, stage, 'position')
if self._temperature_monitor is not None:
# This call takes about 1 ms
lazy_dict['temperature_monitor_chamber_end_temperature'] = \
partial(_safe_temperature_monitor_read, self._temperature_monitor, 'chamber')
# if the incubator is in standby at this check we don't attempt to read the values to
# avoid the cost of serial communication to try and get each value, Although the
# incubator could exit standby at any time and the values would be read at that time
if self._incubator is not None and not self._incubator.standby:
lazy_dict['incubator_end_temperature'] = \
partial(self._incubator.safe_read, 'temperature')
lazy_dict['incubator_end_room_temperature'] = \
partial(self._incubator.safe_read, 'room_temperature')
lazy_dict['incubator_end_lid_temperature'] = \
partial(self._incubator.safe_read, 'lid_temperature')
lazy_dict['incubator_end_humidity'] = \
partial(self._incubator.safe_read, 'humidity')
lazy_dict['incubator_end_oxygen_concentration'] = \
partial(self._incubator.safe_read, 'oxygen_concentration')
lazy_dict['incubator_end_carbon_dioxide_concentration'] = \
partial(self._incubator.safe_read, 'carbon_dioxide_concentration')
elif self._incubator is not None:
# Fast path to avoid multiple exceptions if the incubator is on standby
dataset['incubator_end_temperature'].data[...] = np.nan
dataset['incubator_end_room_temperature'].data[...] = np.nan
dataset['incubator_end_lid_temperature'].data[...] = np.nan
dataset['incubator_end_humidity'].data[...] = np.nan
dataset['incubator_end_oxygen_concentration'].data[...] = np.nan
dataset['incubator_end_carbon_dioxide_concentration'].data[...] = np.nan
# 2025/08 -- Jed
# Slow rollout of the ability to monitor temperature of the Vireo Devices
# TODO: expand to other illumination devices. however the transmission
# boards don't necessary have the temperatures property
if (isinstance(self._illumination_device, MotorizedIllumination) and
self._illumination_device._fluorescence is not None):
fluorescence_temperatures = self._illumination_device.temperatures
for i, temp in enumerate(fluorescence_temperatures):
dataset[f'fluorescence_temperature_{i}'].data[...] = temp
properties_dict = _get_lazy_instrument_properties(lazy_dict)
for name, value in properties_dict.items():
dataset[name].data[...] = value
memory_metrics = {
'workstation_end_' + k: v
for k, v in _get_memory_metrics().items()
}
for name, value in memory_metrics.items():
dataset[name].data[...] = value
return
[docs]
def allocate_video_buffer(
self, N_frames, *, selection=None,
garbage_collection=True, tqdm=None,
cpu_count=None,
_try_hugetlb=True):
"""Allocates a buffer for video frames.
Before allocating any new data, the reference to the old data will
be released. Upon failure, there may be no video buffer in memory.
Parameters
----------
N_frames: int
The number of frames to allocate memory for.
selection: [N_cameras_Y, N_cameras,X]
A boolean array of the image sensors to acquire from.
tqdm: tqdm-like
A tqdm-like progress bar manager.
garbage_collection: bool
Passed to ``free_video_buffer`` if one needs to reallocate memory.
cpu_count: int
The number of CPUs to use to allocate the memory. If None, all CPUs
will be used.
See Also
--------
free_video_buffer, acquire_video
"""
if self.can_reuse_video_buffer(N_frames, selection=selection):
return
self.free_video_buffer(garbage_collection=garbage_collection)
if selection is None:
selection = self.selection_all_cameras
N_micro_cameras = np.sum(selection)
micro_camera_frames = N_micro_cameras * N_frames
shape = (micro_camera_frames,) + self.image_shape
raw_data = alloc_buffer_for_acquisition(
shape, dtype=self._falcon.data.dtype,
garbage_collection=garbage_collection,
try_hugetlb=_try_hugetlb,
cpu_count=cpu_count,
tqdm=tqdm,
)
# Reshape as a 1D array, C-contiguous
# This ensures that our later implementation does not depend
# on the shape of the imaging array
raw_data = np.reshape(raw_data, -1, copy=False)
self._raw_video_buffers = raw_data
[docs]
def can_reuse_video_buffer(self, N_frames, selection=None):
"""Checks if the video buffer can be reused for the next acquisition.
This function may be used for interactive applications
to guide users in their navigation of the interface.
Returns
-------
video_buffer_reusable: bool
If True, ``MCAM.allocate_video_buffer`` will return instantly since
the previous video buffer will simply be reused.
"""
if selection is None:
selection = self.selection_all_cameras
N_micro_cameras = np.sum(selection)
micro_camera_frames = N_micro_cameras * N_frames
size = micro_camera_frames * self.image_size
if self._raw_video_buffers is None:
return False
return self._raw_video_buffers.size >= size
def __del__(self):
# You cannot run garbage collection duing interpreter shutdown
self.free_video_buffer(garbage_collection=False)
self.close(_garbage_collection=False)
@property
def video_buffer_nbytes(self):
"""The number of bytes allocated in the video acquisition buffer."""
if self._raw_video_buffers is None:
return 0
else:
return self._raw_video_buffers.nbytes
[docs]
def free_video_buffer(self, *, garbage_collection=True):
"""Free the allocated video buffer to reclaim the allocated memory.
Parameters
----------
garbage_collection: bool
If ``True``, the system will call python's ``gc.collect()`` to
attempt gargabe collection on the unused memory.
"""
self._raw_video_buffers = None
if garbage_collection:
gc.collect()
def _prepare_video_raw_buffers(self, N_frames, selection_slice):
# Xarray deals badly with None during slicing
if selection_slice is None:
selection_slice = Ellipsis
if self._raw_video_buffers is None:
raise RuntimeError(
"Must allocate video frames before acquiring video."
"Call `allocate_video_buffer` before `acquire_video`."
)
selection = selection_slice_to_selection(self.N_cameras, selection_slice)
N_micro_cameras = np.sum(selection)
size_available = self._raw_video_buffers.size
micro_camera_frames_available = size_available // self.image_size
frames_available = micro_camera_frames_available // N_micro_cameras
if N_frames > frames_available:
raise ValueError(f"{N_frames} frames have been requested, but "
f"only {frames_available} have been allocated.")
size_requested = N_micro_cameras * N_frames * self.image_size
video_shape = selection[selection_slice].shape
raw_data = self._raw_video_buffers[:size_requested]
# Reshape into the user's desired size
raw_data = np.reshape(
raw_data, (N_frames,) + video_shape + self.image_shape, copy=False)
return raw_data, selection
def _prepare_video_metadata(self, N_frames, selection_slice):
import owl
# Xarray doesn't deal well with Non as the selection
if selection_slice is None:
selection_slice = Ellipsis
# This is somewhat a roundabout way to create the sliced coordinates
# But we want
# - To delegate slicing to something that handles very generic slices
# Slices like: np.s_[2:3], np.s_[2:3, :], np.s_[:, 2:3]
# are allowed and valid according to numpy.
# we just want to let numpy and xarray handle this
# - We do not want to use the deprecated `MCAM.dataset` attribute
#
# Eventually, we will centralize the creation of the metadata axis
# in a way that handles selection slices more naturally, or think
# of an other API.
dummy_images = xr.DataArray(
np.zeros(self.N_cameras),
dims=('image_y', 'image_x'),
coords={
'image_y': np.arange(self.N_cameras_Y),
'image_x': np.arange(self.N_cameras_X),
}
)
sliced_dummy_images = dummy_images[selection_slice]
image_y = sliced_dummy_images.image_y
image_x = sliced_dummy_images.image_x
start_pixel = self._falcon.start_pixel
end_pixel = self._falcon.end_pixel
bin_mode = self._falcon.bin_mode
metadata = xr.Dataset(
data_vars={
'__sys_version__': ((), sys.version),
'__owl_sys_info__': ((), str(owl.sys_info())),
},
coords={
'frame_number': np.arange(N_frames, dtype=int),
'image_y': image_y.data,
'image_x': image_x.data,
# Depending on the pixel selection and bin mode,
# The x and y pixels are not necessary starting at 0 and
# incrementing by 1
# We are using "range" here instead of np.arange
# To keep it consistent with the loosely defined
# code in `_create_datasets`
'y': range(start_pixel[0], end_pixel[0], bin_mode),
'x': range(start_pixel[1], end_pixel[1], bin_mode),
'__owl_version__': owl.__version__,
}
)
metadata = self._add_system_coordinates(
metadata, software_timestamp_dims=('frame_number',))
metadata = add_ml_index_array(metadata)
return metadata
def _prepare_video_dataset(self, raw_data, selection_slice):
N_frames = raw_data.shape[0]
video_dataset = self._prepare_video_metadata(N_frames, selection_slice)
video_dataset['images'] = xr.DataArray(
data=raw_data,
dims=('frame_number', 'image_y', 'image_x', 'y', 'x'),
name='images',
)
# Save in 128 MB chunks?
# This helps a little bit, but really the unlimited dim code path
# just seems so slow...
video_dataset.images.encoding["chunksizes"] = (
((128 * 1024 ** 2 + raw_data[0].nbytes - 1) // raw_data[0].nbytes,)
+ raw_data.shape[1:]
)
video_buffers = np.empty(self.N_cameras, dtype=object)
for j, cam_y in enumerate(video_dataset.image_y):
for i, cam_x in enumerate(video_dataset.image_x):
video_buffers[cam_y, cam_x] = raw_data[:, j, i]
return video_dataset, video_buffers
[docs]
def acquire_high_speed_video(
self,
N_frames,
*,
selection_slice=None,
tqdm=None,
stop_event=None,
data_buffers=None,
**kwargs
):
"""Acquire a given number of video frames of all or a subset of cameras
at the maximum framerate.
Before calling this function, the user is expected to have allocated
a video buffer using the ``MCAM.allocate_video_buffer``.
>>> from owl.instruments import MCAM
>>> mcam = MCAM()
>>> mcam.bin_mode = 4
>>> mcam.frame_rate_setpoint = 30
>>> mcam.exposure = 30E-3
>>> mcam.allocate_video_buffer(N_frames=100, tqdm=tqdm)
>>> video_dataset = mcam.acquire_high_speed_video(100)
Parameters
----------
N_frames: int
Number of frames to acquire.
selection_slice: slice
Python slice in the MCAM array to acquire a video from
tqdm: tqdm object
TQDM object used to monitor the amount of time to allocate memory.
start_event: threading.Event
If provided, the method ``start_event.set()`` will be called after
video setup has been completed. The call to the ``set()`` method
should not be blocking otherwise it will incur additional delay on
the video acquisition.
.. versionadded:: 0.18.59
The ``start_event`` parameter was added.
stop_event: Event
A threading.Event-like object used to stop the acquisition.
Once the acquisition is stopped, None may be returned.
.. versionadded:: 0.18.371
The ``stop_event`` parameter was added.
data_buffers: np.ndarray
Optionally, the user may provide a pre-allocated array to store
the acquired data. If not provided, the buffer will be allocated
upon calling this function.
Returns
-------
mcam_dataset: xarray.Dataset
Dataset containing the video data acquired.
See Also
--------
allocate_video_buffer, free_video_buffer, frame_rate_setpoint, exposure
Notes
-----
.. versionchanged:: 0.18.29
In version 0.18.29 the metadata generate by this function no longer
keeps track of the ``frame_number`` dimension for many of the
properties as it is assumed that the properties are unchanging.
As of version 0.18.29, the returned ``software_timestamp`` is
independent of the image indcies ``image_y`` and ``image_x``.
"""
try:
if self._falcon._video_thread is not None:
warn("Video should be turned off before trying to do other acquisitions. "
"This will become an error in future versions of owl.",
stacklevel=2)
except Exception:
# Since video thread is a hidden variable, I don't want to be held
# against breaking that API for this warning
pass
if stop_event is None:
stop_event = Event()
from ..mcam_data._core import _update_timestamp
if data_buffers is None:
data_buffers, selection = self._prepare_video_raw_buffers(
N_frames=N_frames,
selection_slice=selection_slice,
)
else:
selection = np.zeros(shape=self.N_cameras, dtype=bool)
selection[selection_slice] = True
video_dataset, video_buffers = self._prepare_video_dataset(
data_buffers, selection_slice)
acquisition_parameters = {}
if self._temperature_monitor is not None:
acquisition_parameters |= {
'temperature_monitor_chamber_start_temperature':
_safe_temperature_monitor_read(self._temperature_monitor, 'chamber')
}
memory_metrics = _get_memory_metrics()
acquisition_parameters |= {
'workstation_start_' + k: v
for k, v in memory_metrics.items()
}
falcon_acquisition_parameters = self._falcon.acquire_video(
selection=selection,
video_buffers=video_buffers,
N_frames=N_frames,
stop_event=stop_event,
tqdm=tqdm,
**kwargs
)
if stop_event.is_set():
return
acquisition_parameters |= falcon_acquisition_parameters
self._update_metadata(
video_dataset,
acquisition_parameters=acquisition_parameters,
update_software_timestamp=False,
dataset_selection_slice=selection_slice,
selection=Ellipsis,
)
# because the metadata is stored at the end of the capture
# it actually stores the end time, not the start time
# Cannot use timezone aware objects, therefore we force
# We have since made the switch to UTC time
time_end = datetime.now(UTC)
frame_rate = acquisition_parameters['frame_rate']
_update_timestamp(video_dataset,
frame_rate=frame_rate, time_end=time_end)
return video_dataset
def _get_required_RAM_for_acquisition_to_null(
self,
N_frames,
*,
frame_rate=None,
selection_slice=None,
keypoints=None,
additional_buffer_space=None,
):
data_consumption_rate = _data_consumption_rate['to_null']
N_frames_buffer, frame_size, _ = self._get_required_buffer_frames(
N_frames=N_frames,
frame_rate=frame_rate,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
additional_buffer_space=additional_buffer_space,
)
return N_frames_buffer * frame_size
def _acquire_video_to_null(
self,
filename,
N_frames, *,
selection_slice=None,
keypoints=None,
include_timestamp=True,
temporary_buffer_size=None,
tqdm=None,
start_event=None,
stop_event=None,
metadata=None,
tqdm_save=None,
_stacklevel_increment=0,
):
def save_function(
metadata,
images_variable,
filename: Path,
frame_timeout=None,
timeout=None,
data_read_ready_semaphore=None,
data_write_ready_semaphore=None,
metadata_ready=None,
include_timestamp: bool=True,
ready_to_write_event=None,
tqdm=None,
stop_event=None,
# TODO: cleanup this API.
# We allow to absorb unknown keyword arguments to unify the api with the
# streamed video methods
**kwargs,
):
# We mostly mock a video writing method to allow us to "capture" for a
# long time using the circular buffer
from owl.util import make_timestamp
if stop_event is None:
stop_event = Event()
filename = Path(filename)
if include_timestamp:
filename = filename.parent / (filename.name + '_' + make_timestamp())
N_frames = len(metadata['frame_number'])
if frame_timeout is None:
frame_timeout = timeout
if ready_to_write_event is not None:
ready_to_write_event.set()
if tqdm is None:
tqdm = iter
# Mostly copied from netcdf writer
for i in tqdm(range(N_frames)):
# Wait for the data to be read y
if data_read_ready_semaphore is not None:
if stop_event.is_set():
return
if not data_read_ready_semaphore.acquire(
blocking=True,
timeout=frame_timeout,
):
if stop_event.is_set():
return
raise RuntimeError(
f"Timeout on obtained frames to write for frame {i}.")
# Do nothing
if data_write_ready_semaphore is not None:
data_write_ready_semaphore.release()
if stop_event.is_set():
return
if metadata_ready is not None:
if not metadata_ready.wait(
timeout=timeout
):
raise RuntimeError("Timeout obtained waiting for metadata")
return filename
data_consumption_rate = _data_consumption_rate['to_null']
return self._acquire_video_to_file_internal(
filename=filename,
N_frames=N_frames,
save_function=save_function,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
include_timestamp=include_timestamp,
temporary_buffer_size=temporary_buffer_size,
tqdm=tqdm,
start_event=start_event,
stop_event=stop_event,
metadata=metadata,
tqdm_save=tqdm_save,
_stacklevel_increment=_stacklevel_increment + 1
)
[docs]
def acquire_video_to_multi_mp4(
self,
filename,
N_frames,
*,
selection_slice=None,
keypoints=None,
include_timestamp=True,
temporary_buffer_size=None,
stop_event=None,
tqdm=None,
start_event=None,
metadata=None,
tqdm_save=None,
_stacklevel_increment=0,
**kwargs,
):
"""Record a video from saving it directly to compressed MP4 file.
This function records a certain number of frames directly to
a particular file. It uses the computer RAM as temporary storage
in case the disk file is too busy to store the incoming frames.
A larger buffer enables larger durations of temporary slowdowns on the
video encoding pipeline.
.. versionadded:: 0.18.242
First version of the method ``acquire_video_to_multi_mp4``.
Parameters
----------
filename: Path
The filename where the data should be saved. If the filename has no
extension, then an ``'.nc'`` extension is added.
N_frames: int
Number of frames to acquire.
selection_slice: slice
Python slice in the MCAM array to acquire a video from
tqdm: tqdm object
TQDM object used to monitor the amount of time to allocate memory.
start_event: threading.Event
If provided, the method ``start_event.set()`` will be called after
video setup has been completed. The call to the ``set()`` method
should not be blocking otherwise it will incur additional delay on
the video acquisition.
metadata:
Metadata to join to the MCAM dataset upon saving.
keypoints:
Well plate keypoints structure. If provided the data will be
cropped according to the provided keypoints during the saving
pipeline.
stop_event: threading.Event
If provided, the method ``stop_event.is_set()`` will be polled to
check if the video acquisition should be aborted.
.. versionadded:: 0.18.371
First version where the parameter ``stop_event`` became an
option.
tqdm_save:
A TQDM object to specifically track the progress on the data saving
portion of the acquisition. If video encoding is slightly slower than
the acquisition frame rate, encoding the final few frames may happen
after the acquisition is complete.
Returns
-------
filename: Path-like
The filename where the data has been stored. It includes the
optional timestamp, and the new suffix.
See Also
--------
acquire_high_speed_video
"""
from owl.mcam_data._mp4 import _save_video_streamed_multi_mp4
sensor_chroma = self.sensor_chroma
if sensor_chroma.startswith("bayer"):
sensor_chroma = "bayer"
save_function = partial(
_save_video_streamed_multi_mp4,
image_chroma=sensor_chroma,
)
data_consumption_rate = _data_consumption_rate['multi_mp4']
return self._acquire_video_to_file_internal(
filename=filename,
N_frames=N_frames,
save_function=save_function,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
include_timestamp=include_timestamp,
temporary_buffer_size=temporary_buffer_size,
stop_event=stop_event,
tqdm=tqdm,
start_event=start_event,
metadata=metadata,
tqdm_save=tqdm_save,
_stacklevel_increment=_stacklevel_increment + 1,
**kwargs,
)
[docs]
def acquire_video_to_tiledmp4(
self,
filename,
N_frames,
*,
selection_slice=None,
keypoints=None,
include_timestamp=True,
temporary_buffer_size=None,
stop_event=None,
tqdm=None,
start_event=None,
metadata=None,
tqdm_save=None,
_stacklevel_increment=0,
**kwargs
):
"""Record a video from saving it directly to compressed MP4 file.
This function records a certain number of frames directly to
a particular file. It uses the computer RAM as temporary storage
in case the disk file is too busy to store the incoming frames.
A larger buffer enables larger durations of temporary slowdowns on the
video encoding pipeline.
.. versionadded:: 0.18.193
First version of the method ``acquire_video_to_tiledmp4``.
Parameters
----------
filename: Path
The filename where the data should be saved. If the filename has no
extension, then an ``'.nc'`` extension is added.
N_frames: int
Number of frames to acquire.
selection_slice: slice
Python slice in the MCAM array to acquire a video from
tqdm: tqdm object
TQDM object used to monitor the amount of time to allocate memory.
start_event: threading.Event
If provided, the method ``start_event.set()`` will be called after
video setup has been completed. The call to the ``set()`` method
should not be blocking otherwise it will incur additional delay on
the video acquisition.
metadata:
Metadata to join to the MCAM dataset upon saving.
keypoints:
Well plate keypoints structure. If provided the data will be
cropped according to the provided keypoints during the saving
pipeline.
stop_event: threading.Event
If provided, the method ``stop_event.is_set()`` will be polled to
check if the video acquisition should be aborted.
.. versionadded:: 0.18.371
First version where the parameter ``stop_event`` became an
option.
tqdm_save:
A TQDM object to specifically track the progress on the data saving
portion of the acquisition. If video encoding is slightly slower than
the acquisition frame rate, encoding the final few frames may happen
after the acquisition is complete.
.. versionadded:: 0.18.224
Returns
-------
filename: Path-like
The filename where the data has been stored. It includes the
optional timestamp, and the new suffix.
See Also
--------
acquire_high_speed_video
"""
from owl.mcam_data._mp4 import _save_video_streamed
sensor_chroma = self.sensor_chroma
if sensor_chroma.startswith("bayer"):
sensor_chroma = "bayer"
save_function = partial(
_save_video_streamed,
image_chroma=sensor_chroma,
)
data_consumption_rate = _data_consumption_rate['tiledmp4']
# This isn't really the maximum width for hevc (h265)
# but I do not wanto allow greater frame rate acquisition yet...
# The choice of 6144 allows us to acquire hevc at up to 30 fps
# approximately with an RTX3060
max_total_pixels = (6144, 6144)
# We really only tested with
# 6144 x 4096 which is effectively "bin x2" for a 96 well plate.
# However, under some experiments, I've found that with our presets
# We can get to 6144 x (4688 = 4096 + 512 + 64 + 16 + 8 + 4 + 2)
return self._acquire_video_to_file_internal(
filename=filename,
N_frames=N_frames,
save_function=save_function,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
include_timestamp=include_timestamp,
temporary_buffer_size=temporary_buffer_size,
max_total_pixels=max_total_pixels,
tqdm=tqdm,
start_event=start_event,
metadata=metadata,
stop_event=stop_event,
tqdm_save=tqdm_save,
_stacklevel_increment=_stacklevel_increment + 1,
**kwargs,
)
def get_required_RAM_for_acquisition_to_file(
self,
N_frames,
*,
frame_rate=None,
selection_slice=None,
keypoints=None,
additional_buffer_space=None,
):
data_consumption_rate = _data_consumption_rate['to_file']
min_chunk_size = 128 * 1024 ** 2
min_total_chunks = 5
N_frames_buffer, frame_size, _ = self._get_required_buffer_frames(
N_frames=N_frames,
frame_rate=frame_rate,
data_consumption_rate=data_consumption_rate,
min_chunk_size=min_chunk_size,
min_total_chunks=min_total_chunks,
selection_slice=selection_slice,
keypoints=keypoints,
additional_buffer_space=additional_buffer_space,
)
return N_frames_buffer * frame_size
def get_required_RAM_for_acquisition_to_multi_mp4(
self,
N_frames,
*,
frame_rate=None,
selection_slice=None,
keypoints=None,
additional_buffer_space=None,
):
data_consumption_rate = _data_consumption_rate['multi_mp4']
N_frames_buffer, frame_size, _ = self._get_required_buffer_frames(
N_frames=N_frames,
frame_rate=frame_rate,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
additional_buffer_space=additional_buffer_space,
)
return N_frames_buffer * frame_size
def get_required_RAM_for_acquisition_to_tiledmp4(
self,
N_frames,
*,
frame_rate=None,
selection_slice=None,
keypoints=None,
additional_buffer_space=None,
):
data_consumption_rate = _data_consumption_rate['tiledmp4']
N_frames_buffer, frame_size, _ = self._get_required_buffer_frames(
N_frames=N_frames,
frame_rate=frame_rate,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
additional_buffer_space=additional_buffer_space,
)
return N_frames_buffer * frame_size
def _get_required_buffer_frames(
self,
N_frames,
*,
frame_rate=None,
min_chunk_size=None,
min_total_chunks=None,
data_consumption_rate,
selection_slice=None,
keypoints=None,
additional_buffer_space=None,
):
if additional_buffer_space is None:
additional_buffer_space = 5 * 1024 ** 3
if selection_slice is None:
selection_slice = np.s_[:, :]
selection = np.zeros(self.N_cameras, dtype=bool)
selection[selection_slice] = True
if keypoints is not None:
if not isinstance(keypoints, WellPlateKeypointState):
warn((
"Keypoints should be specified as WellPlateKeypointState "
f"not as {keypoints.__class__}."
),
stacklevel=2
)
keypoints = deepcopy(keypoints)
keypoints = WellPlateKeypointState.model_validate(keypoints)
if frame_rate is None:
frame_rate = self.frame_rate
generated_frames_per_second = frame_rate
# TODO: replace this with something else.
# THis is only used as a convenience function to compute the required size
N_cropped_image_indices = (
range(self.N_cameras[0])[selection_slice[0]],
range(self.N_cameras[1])[selection_slice[1]],
)
if keypoints is not None:
N_cropped_images = 0
keypoint_array = keypoints.keypoint_array
# 1 byte per pixel at bin x1 resolution
diameter = keypoint_array[0, 0][4] // self.bin_mode
# We crop a square region
image_nbytes = diameter ** 2
for i in np.ndindex(keypoint_array.shape[:2]):
camera_index = keypoint_array[i][:2]
if camera_index[0] not in N_cropped_image_indices[0]:
continue
if camera_index[1] not in N_cropped_image_indices[1]:
continue
# We crop out the well no matter what if the camera is active
# even if it is outside the ROI of the camera
N_cropped_images += 1
images_nbytes = N_cropped_images * image_nbytes
else:
N_cropped_images = len(N_cropped_image_indices[0]) * len(N_cropped_image_indices[1])
images_nbytes = N_cropped_images * self.image_nbytes
consumed_frames_per_second = data_consumption_rate / images_nbytes
accumulated_frames_per_second = max(
generated_frames_per_second - consumed_frames_per_second,
0,
)
accumulated_frames_total = int(
accumulated_frames_per_second * N_frames / generated_frames_per_second
+ 1
)
frame_size = self._data[selection_slice].nbytes
accumulated_bytes = accumulated_frames_total * frame_size
temporary_buffer_size = additional_buffer_space + accumulated_bytes
if min_chunk_size is not None and min_total_chunks is not None:
N_frames_buffer, N_frames_per_chunk = get_N_frames_buffer(
frame_size=frame_size,
N_frames=N_frames,
max_buffer_size=temporary_buffer_size,
min_chunk_size=128 * 1024 ** 2,
min_total_chunks=5,
)
else:
N_frames_buffer = min(temporary_buffer_size // frame_size, N_frames)
N_frames_buffer = max(N_frames_buffer, 10)
N_frames_per_chunk = 1
return N_frames_buffer, frame_size, N_frames_per_chunk
def _acquire_video_to_file_internal(
self,
filename,
N_frames, *,
save_function,
data_consumption_rate,
selection_slice=None,
keypoints=None,
include_timestamp=True,
temporary_buffer_size=None,
max_total_pixels=None,
min_chunk_size=None,
min_total_chunks=None,
tqdm=None,
start_event=None,
metadata=None,
tqdm_save=None,
stop_event=None,
_stacklevel_increment=0,
**kwargs,
):
analysis_function = kwargs.get('analysis_function', None)
from ..mcam_data._core import _update_timestamp
if stop_event is None:
stop_event = Event()
if start_event is None:
start_event = Event()
try:
if self._falcon._video_thread is not None:
warn("Video should be turned off before trying to do other acquisitions. "
"This will become an error in future versions of owl.",
stacklevel=2 + _stacklevel_increment)
except Exception:
# Since video thread is a hidden variable, I don't want to be held
# against breaking that API for this warning
pass
if selection_slice is None:
selection_slice = np.s_[:, :]
if keypoints is not None:
if not isinstance(keypoints, WellPlateKeypointState):
warn((
"Keypoints should be specified as WellPlateKeypointState "
f"not as {keypoints.__class__}."
),
stacklevel=2
)
keypoints = deepcopy(keypoints)
keypoints = WellPlateKeypointState.model_validate(keypoints)
if analysis_function is None:
# Create a do nothing analysis function. This should also serve as an
# example of the required input signature for an analysis function.
def analysis_function(
metadata,
images_variable,
ready_to_analyze_event,
data_analyze_ready_semaphore,
data_save_ready_semaphore,
stop_event,
start_event,
frame_timeout,
**kwargs,
):
N_frames = len(metadata['frame_number'])
# let the other threads know that we are ready to analyze
ready_to_analyze_event.set()
# need to wait until the acquisition has started.
# 5 seconds should be long enough...
start_event.wait(timeout=5)
for i in range(N_frames):
if stop_event.is_set():
return
if not data_analyze_ready_semaphore.acquire(
blocking=True,
timeout=frame_timeout,
):
if stop_event.is_set():
return
raise RuntimeError(
f"Timeout on obtained frames to analyze for frame {i}.")
# Do nothing
data_save_ready_semaphore.release()
frame_rate = self.frame_rate
selection = np.zeros(self.N_cameras, dtype=bool)
selection[selection_slice] = True
N_frames_buffer, _frame_size, N_frames_per_chunk = self._get_required_buffer_frames(
N_frames=N_frames,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
additional_buffer_space=temporary_buffer_size,
)
# This allocation should be quick because the buffer should be small
# no need to use tqdm here
self.allocate_video_buffer(
N_frames=N_frames_buffer,
selection=selection,
tqdm=tqdm
)
raw_data, selection = self._prepare_video_raw_buffers(
N_frames=N_frames_buffer,
selection_slice=selection_slice,
)
video_metadata = self._prepare_video_metadata(N_frames, selection_slice)
# exif orientation is important to get right for well extraction
# to label the wells correctly
# https://gitlab.com/ramonaoptics/mcam/python-owl/-/issues/1122
video_metadata['exif_orientation'] = self.exif_orientation
if metadata is not None:
video_metadata = video_metadata.merge(metadata)
video_buffers = np.empty(self.N_cameras, dtype=object)
for j, cam_y in enumerate(video_metadata.image_y):
for i, cam_x in enumerate(video_metadata.image_x):
video_buffers[cam_y, cam_x] = raw_data[:, j, i]
# 2023/02 -- Mark
# This is somewhat of a hack. This variable may not be backed by the full
# array behind it.
# This is why we don't add it back to the video_metadata above.
images_variable = xr.DataArray(
data=raw_data,
dims=('frame_number', 'image_y', 'image_x', 'y', 'x'),
name='images',
)
if keypoints is not None:
from owl.analysis.wellplate import (
SubindexedArray,
_generate_metadata,
create_wellplate_slices,
)
well_image_shape, single_well_image_shape, image_slices, well_slices = \
create_wellplate_slices(
mcam_dataset=video_metadata,
keypoints_array=keypoints.keypoint_array,
images_dims=images_variable.dims,
)
# Mark -- 2024/01
# For now, I want a slow rollout of this feature.
# This will technically adjust which sensors are turned on instead
# of turning on all contiguous sensors in the FOV.
# This is mostly only useful for the Kestrel0036 where the strange
# Organization of the well-like structure will reduce the bandwidth
# and of the system.
# By reducing the number of cameras that are turned on,
# this seems to affect how well the system can run and seems to reduce
# The error probability
if self._serial_number in ["Kestrel0036", "Kestrel0036R"]:
used_selection = np.zeros_like(selection)
for s in image_slices.flatten():
used_selection[s[:2]] = True
selection = np.logical_and(selection, used_selection)
subindexed_data = SubindexedArray(
data=raw_data,
shape=(raw_data.shape[0],) + well_image_shape[1:],
input_indexers=image_slices,
output_indexers=well_slices,
)
images_variable = xr.DataArray(
data=subindexed_data,
name='images',
dims=images_variable.dims
)
video_metadata, _image_dims = _generate_metadata(
video_metadata,
images_variable=images_variable,
keypoints_array=keypoints.keypoint_array,
square_wells=keypoints.square_wells,
_stacklevel_increment=1 + _stacklevel_increment,
)
else:
subindexed_data = None
if N_frames_per_chunk is not None:
# We are going to assuming that the well plate cropping
# is not enough to throw off the chunk size
# in terms of the optimal size for the file saving speed
chunksizes = (N_frames_per_chunk,) + images_variable.shape[1:]
images_variable.encoding["chunksizes"] = chunksizes
if max_total_pixels is not None:
tiled_pixels = (
video_metadata.sizes['image_y'] * video_metadata.sizes['y'],
video_metadata.sizes['image_x'] * video_metadata.sizes['x'],
)
if tiled_pixels[0] > max_total_pixels[0] or tiled_pixels[1] > max_total_pixels[1]:
raise RuntimeError(
"Can only capture streams that total "
f"{max_total_pixels[0]} x {max_total_pixels[1]} pixels. "
f"Requesting {tiled_pixels[0]} x {tiled_pixels[1]} pixels."
f"Consider using 'High Frame Rate' or other resolution mode "
f"or decrease the number of cameras actively streaming."
)
def save_streamed(
*,
metadata, images_variable,
filename,
data_read_ready_semaphore,
data_write_ready_semaphore,
metadata_ready,
save_file_name_queue,
bayer_pattern,
subindexed_data,
ready_to_write_event,
stop_event,
tqdm=None,
):
try:
filename = save_function(
metadata=metadata,
images_variable=images_variable,
filename=filename,
include_timestamp=include_timestamp,
data_read_ready_semaphore=data_read_ready_semaphore,
data_write_ready_semaphore=data_write_ready_semaphore,
metadata_ready=metadata_ready,
bayer_pattern=bayer_pattern,
subindexed_data=subindexed_data,
ready_to_write_event=ready_to_write_event,
stop_event=stop_event,
tqdm=tqdm,
frame_timeout=1 + 1 / frame_rate,
timeout=2,
)
save_file_name_queue.put((True, filename))
except Exception as e:
message = f"{e.__class__}: {e}"
print(message)
save_file_name_queue.put((False, e))
# trigger the stop event to stop the other threads
stop_event.set()
def analysis_streamed(
*,
metadata,
images_variable,
ready_to_analyze_event,
data_analyze_ready_semaphore,
data_save_ready_semaphore,
stop_event,
start_event,
analysis_output_queue,
**kwargs,
):
try:
analysis_function(
metadata=metadata,
images_variable=images_variable,
ready_to_analyze_event=ready_to_analyze_event,
data_analyze_ready_semaphore=data_analyze_ready_semaphore,
data_save_ready_semaphore=data_save_ready_semaphore,
stop_event=stop_event,
start_event=start_event,
frame_timeout=1 + 1 / frame_rate,
**kwargs
)
# I don't really know what I should do with this output yet... Maybe this holds the
# analysis results to be added to the metadata?
analysis_output_queue.put((True, None))
except Exception as e:
message = f"{e.__class__}: {e}"
print(message)
analysis_output_queue.put((False, e))
# trigger the stop event to stop the other threads
stop_event.set()
# Data has been acquired and is read to be read by the analysis function
data_read_ready_semaphore = Semaphore(value=0)
# Data has been analyzed and is ready to be saved
data_save_ready_semaphore = Semaphore(value=0)
# Data has been saved and buffer can now be overwritten by the acquisition
data_write_ready_semaphore = Semaphore(value=0)
metadata_ready = Event()
save_file_name_queue = Queue()
analysis_output_queue = Queue()
ready_to_write_event = Event()
ready_to_analyze_event = Event()
analysis_thread = Thread(
target=analysis_streamed,
kwargs={
'metadata': video_metadata,
'images_variable': images_variable,
'ready_to_analyze_event': ready_to_analyze_event,
'data_analyze_ready_semaphore': data_read_ready_semaphore,
'data_save_ready_semaphore': data_save_ready_semaphore,
'stop_event': stop_event,
'start_event': start_event,
'analysis_output_queue': analysis_output_queue,
}
)
save_thread = Thread(
target=save_streamed,
kwargs={
'metadata': video_metadata,
'images_variable': images_variable,
'filename': filename,
'data_read_ready_semaphore': data_save_ready_semaphore,
'data_write_ready_semaphore': data_write_ready_semaphore,
'metadata_ready': metadata_ready,
'save_file_name_queue': save_file_name_queue,
# bayer_pattern only necessary for the video functions
'bayer_pattern': self.bayer_pattern,
# Only necessary for the nc saving
'subindexed_data': subindexed_data,
'ready_to_write_event': ready_to_write_event,
'stop_event': stop_event,
'tqdm': tqdm_save,
}
)
acquisition_parameters = {}
if self._temperature_monitor is not None:
acquisition_parameters |= {
'temperature_monitor_chamber_start_temperature':
_safe_temperature_monitor_read(self._temperature_monitor, 'chamber')
}
memory_metrics = _get_memory_metrics()
acquisition_parameters |= {
'workstation_start_' + k: v
for k, v in memory_metrics.items()
}
save_thread.start()
analysis_thread.start()
# The encoder can take some time to startup, wait for it
# I've measured up to 600 ms for starting up an encoder for
# 2048 x 4096 videos
# Due to the fact that the thread will attempt to "discover"
# which encoders exist, this step may take a few seconds for large
# videos
# We use a loop so we can check for both the ready_to_write_event
# And the save_file_name_queue at the same time
for i in range(10):
ready_to_write = ready_to_write_event.wait(timeout=1)
if ready_to_write:
break
try:
# The expected result is that this will return EmptyException
# unless an exception has occurred in the thread
# If an exception has occurred the result will be
# (False, exception)
success_data = save_file_name_queue.get_nowait()
except EmptyException:
success_data = None
if success_data is not None:
# The thread will return early if an exception has occurred
# It will return (False, exception)
# We raise it to the end user.
success, exception = success_data
if not success:
raise exception
# copy the logic above to make sure the analysis thread is ready. It is expected
# this check should not have to run as many iterations since the analysis thread
# would have more time to prepare while we are waiting on the saving thread.
for i in range(10):
ready_to_analyze = ready_to_analyze_event.wait(timeout=1)
if ready_to_analyze:
break
try:
success_data = analysis_output_queue.get_nowait()
except EmptyException:
success_data = None
if success_data is not None:
success, exception = success_data
if not success:
raise exception
falcon_acquisition_parameters = self._falcon.acquire_video(
selection=selection,
video_buffers=video_buffers,
N_frames=N_frames,
stop_event=stop_event,
tqdm=tqdm,
data_read_ready_semaphore=data_read_ready_semaphore,
data_write_ready_semaphore=data_write_ready_semaphore,
start_event=start_event,
)
# because the metadata is stored at the end of the capture
# it actually stores the end time, not the start time
# Cannot use timezone aware objects, therefore we force
# We have since made the switch to UTC time
time_end = datetime.now(UTC)
if stop_event.is_set():
save_thread.join()
analysis_thread.join()
# flush the queue
success, filename_or_exception = save_file_name_queue.get(
block=True, timeout=10)
# this stop may have been triggered by an exception in the save thread
if not success:
raise filename_or_exception
# this stop may have been triggered by an exception in the analysis thread
success, output_or_exception = analysis_output_queue.get(
block=True, timeout=10)
if not success:
raise output_or_exception
return
acquisition_parameters |= falcon_acquisition_parameters
# # Wait for the analysis thread to finish to avoid any contention
# in writing to the metadata
analysis_thread.join()
success, output_or_exception = analysis_output_queue.get(
block=True, timeout=1)
if not success:
raise output_or_exception
self._update_metadata(
video_metadata,
acquisition_parameters=acquisition_parameters,
update_software_timestamp=False,
dataset_selection_slice=selection_slice,
# Gotto mock the selection for the video acquisitions...
selection=np.ones(
video_metadata.image_y.shape + video_metadata.image_x.shape,
dtype=bool,
)
)
# because the metadata is stored at the end of the capture
# it actually stores the end time, not the start time
# Cannot use timezone aware objects, therefore we force
# We have since made the switch to UTC time
time_end = datetime.now(UTC)
frame_rate = acquisition_parameters['frame_rate']
_update_timestamp(video_metadata,
frame_rate=frame_rate, time_end=time_end)
metadata_ready.set()
save_thread.join()
success, filename_or_exception = save_file_name_queue.get(
block=True, timeout=1)
if not success:
raise filename_or_exception
return filename_or_exception
[docs]
def acquire_video_to_file(
self,
filename,
N_frames,
*,
selection_slice=None,
include_timestamp=True,
temporary_buffer_size=None,
stop_event=None,
tqdm=None,
start_event=None,
metadata=None,
keypoints=None,
tqdm_save=None,
_stacklevel_increment=0,
**kwargs,
):
"""Record a video from the MCAM saving it directly to a file.
This function records a certain number of frames directly to
a particular file. It uses the computer RAM as temporary storage
in case the disk file is too busy to store the incoming frames.
A larger buffer enables larger durations of temporary slowdowns on the
disk.
.. versionadded:: 0.18.68
First version of the method ``acquire_video_to_file``.
Parameters
----------
filename: Path
The filename where the data should be saved. If the filename has no
extension, then an ``'.nc'`` extension is added.
N_frames: int
Number of frames to acquire.
selection_slice: slice
Python slice in the MCAM array to acquire a video from
tqdm: tqdm object
TQDM object used to monitor the amount of time to allocate memory.
start_event: threading.Event
If provided, the method ``start_event.set()`` will be called after
video setup has been completed. The call to the ``set()`` method
should not be blocking otherwise it will incur additional delay on
the video acquisition.
metadata:
Additional metadata to add to the saved file after the acquisition
is complete.
.. versionadded:: 0.18.191
keypoints:
Well plate keypoints structure. If provided the data will be
cropped according to the provided keypoints during the saving
pipeline.
.. versionadded:: 0.18.193
stop_event: threading.Event
If provided, the method ``stop_event.is_set()`` will be polled to
check if the video acquisition should be aborted.
.. versionadded:: 0.18.371
First version where the parameter ``stop_event`` became an
option.
tqdm_save:
A TQDM object to specifically track the progress on the data saving
portion of the acquisition. If video encoding is slightly slower than
the acquisition frame rate, encoding the final few frames may happen
after the acquisition is complete.
.. versionadded:: 0.18.224
Returns
-------
filename: Path-like
The filename where the data has been stored. It includes the
optional timestamp, and the new suffix.
See Also
--------
acquire_high_speed_video
"""
from owl.mcam_data._netcdf import _save_video_streamed as save_function
data_consumption_rate = _data_consumption_rate['to_file']
# in terms of creating a buffer, we need to match the constraints of
# the data writer:
# It is best to write in 128MB chunks or so.
# This limits the total number of chunks, while keeping each chunk
# large enough to attain full speed.
# It wants data to be ordered:
# [time, camera_index_y, camera_index_x, y, x]
# Data should be C-continguous along the time slice that is saved
# the falcon data generator:
# It wants data to be ordered:
# [camera_index, time, y, x]
# Data must only be C-contiguous along the [y, x]
# As such, we want to find what is the largest chunk close enough to
# 128 MB to allocate. Then we want to allocate a good buffer of them
# Maybe ~20 of these buffers to give some space for writing when writing
# gets a little slow.
min_chunk_size = 128 * 1024 ** 2
min_total_chunks = 5
return self._acquire_video_to_file_internal(
filename=filename,
N_frames=N_frames,
save_function=save_function,
data_consumption_rate=data_consumption_rate,
selection_slice=selection_slice,
keypoints=keypoints,
include_timestamp=include_timestamp,
temporary_buffer_size=temporary_buffer_size,
min_chunk_size=min_chunk_size,
min_total_chunks=min_total_chunks,
stop_event=stop_event,
tqdm=tqdm,
start_event=start_event,
metadata=metadata,
tqdm_save=tqdm_save,
_stacklevel_increment=_stacklevel_increment + 1,
**kwargs,
)
def acquire_video(self, N_frames, *, selection_slice=None, tqdm=tqdm):
raise RuntimeError("This function has been removed. Used acquire_high_speed_video")
@property
def image_shape(self) -> (int, int):
"""The image shape obtained from the acquisition of a single sensor.
The image shape is affected by the following imaging parameters:
* ``bin_mode``
* ``start_pixel`` (defined by ``select_pixels``)
* ``end_pixel`` (defined by ``select_pixels``)
See also
--------
bin_mode, start_pixel, end_pixel
"""
return self._falcon.image_shape
@property
def sensor_shape(self) -> (int, int):
"""A tuple containing the physical number of pixels of the sensors.
This is an inherent property of the physical sensor and unaffected
by the imaging parameters.
See also
--------
sensor_height, sensor_width, image_shape
"""
return self._falcon.sensor_shape
[docs]
def select_pixels(self, start_pixel, end_pixel, step=None):
"""Set the pixel sub-selection and bin mode of the MCAM sensors.
When selecting pixels you must follow these constraints after binning:
* The width must be greater than or equal to 320 and divisible by 8.
* The height must be greater than or equal to 8 and divisible by 16.
* The product or the width and the height must be divisible by 4096.
Calling this function is equivalent to invalidating all the data in the
dataset. The data (and metadata) stored in ``MCAM.dataset`` should be
considered invalid.
The exposure time is not guaranteed to stay constant after calling this
function.
Parameters
----------
start_pixel: (int, int)
The y and x start pixels, relative to how data is stored in software.
end_pixel: (int, int)
The y and x end pixels, relative to how data is stored in software.
step: int
The binning mode. Must be 1, 2, or 4. If None, the current bin mode
is maintained.
"""
self._falcon.select_pixels(start_pixel, end_pixel, step)
[docs]
def select_center_pixels(self, shape, step=None):
"""Select the desired pixels centered about the center of the sensor.
Parameters
----------
shape: int, int
Image shape to capture
step: 1, 2, 4, or None
Bin mode to select. If None, the current bin mode is maintained.
See also
--------
select_pixels
"""
self._falcon.select_center_pixels(shape=shape, step=step)
[docs]
def select_sensor_regions_of_interest(self, image_span, *, start_pixels, step=None):
"""Per sensor region of interest selection.
.. versionadded:: 0.18.21
Parameters
----------
image_span: (int, int)
The image space across pixels in the imaging sensors. This span
is common to all imaging sensors.
start_pixels: array-like [N_cameras, 2]
Starting pixels for each sensor in the MCAM array.
step: 1, 2, 4 or None
A valid bin mode for the imaging sensor
Note
----
This feature is still experimental and does not maintain consistency
across the entire metadata. This function's API is also subject to
change without notice.
See also
--------
bin_mode, select_pixels
"""
self._falcon.select_sensor_regions_of_interest(
image_span=image_span, start_pixels=start_pixels, step=step)
[docs]
def select_center_square_pixels(self, step=None, step_alignment=4):
"""Select the largest center square pixels supported by the MCAM.
Currently, this is the same as selecting the center square
``(3072, 3072)`` pixels.
Parameters
----------
step: 1, 2, 4, or None
The binning mode used when selecting the sensors. If None, the
current bin mode is used.
step_alignment: 4
Different bin modes may support slightly different shapes for the
maximum supported square images. By default, we use image extents
that are supported by the largest values for bin mode. This
guarantees that they are supported by the smaller values as well.
See also
--------
select_pixels
"""
self._falcon.select_center_square_pixels(step=step,
step_alignment=step_alignment)
[docs]
def select_default_4_by_3_pixels(self, step=None):
"""Select the default pixel selection supported by the MCAM.
Currently, this is the same as selecting the center
``(3120, 4096)`` pixels.
Parameters
----------
step: 1, 2, 4, or None
The binning mode used when selecting the sensors. If None, the
current bin mode is used.
See also
--------
select_pixels
"""
self._falcon.select_default_4_by_3_pixels(step=step)
@property
def sensor_width(self) -> int:
"""The width in pixels of an image from a single sensor.
This is an inherent property of the physical sensor and unaffected
by the imaging parameters.
See also
--------
sensor_shape, sensor_height, image_shape
"""
return self.image_shape[1]
@property
def sensor_height(self) -> int:
"""The height in pixels of an image from a single sensor.
This is an inherent property of the physical sensor and unaffected
by the imaging parameters.
See also
--------
sensor_shape, sensor_width, image_shape
"""
return self.image_shape[0]
@property
def image_size(self) -> int:
"""Size of image in pixels from a single sensor."""
shape = self.image_shape
return shape[0] * shape[1]
@property
def start_pixel(self):
"""Starting pixel position in the coordinate space of the physical sensor.
The value is returned as a tuple ``(start_pixel_y, start_pixel_x)``.
These values can be changed through a call to ``select_pixels``.
See also
--------
select_pixels, start_pixel, end_pixel
"""
return self._falcon.start_pixel
@property
def end_pixel(self):
"""Ending pixel position in the coordinate space of the physical sensor.
The value is returned as a tuple ``(end_pixel_y, end_pixel_x)``.
These values can be changed through a call to ``select_pixels``.
See also
--------
select_pixels, start_pixel, end_pixel
"""
return self._falcon.end_pixel
@property
def image_nbytes(self) -> int:
"""Size of image in bytes from a single sensor."""
return self._falcon.image_bytes
@property
def N_cameras(self): # noqa
"""Tuple containing the number of cameras in each direction."""
return self._falcon.N_cameras
@property
def N_cameras_Y(self): # noqa
"""The number of cameras in the Y direction."""
return self._falcon.N_cameras[0]
@property
def N_cameras_X(self): # noqa
"""The number of cameras in the X direction."""
return self._falcon.N_cameras[1]
@property
def N_cameras_total(self):
"""The total number of micro cameras in the array."""
return self._falcon.N_cameras[0] * self._falcon.N_cameras[1]
@property
def micro_camera_separation(self):
"""Separation between micro cameras in the array in meters."""
warn("This property is deprecated and will be removed in a future version of owl. Please "
"use mcam.micro_camera_separation_y or mcam.micro_camera_separation_x instead",)
return self._falcon.sensor_pitch
@property
def micro_camera_separation_y(self):
"""Separation between micro cameras in the Y direction in meters."""
skipped_cameras = getattr(self._virtual_selection[0], 'step', None)
if skipped_cameras is None:
skipped_cameras = 1
return self._falcon.sensor_pitch * skipped_cameras
@property
def micro_camera_separation_x(self):
"""Separation between micro cameras in the X direction in meters."""
skipped_cameras = getattr(self._virtual_selection[1], 'step', None)
if skipped_cameras is None:
skipped_cameras = 1
elif self._falcon.sensor_board == 'kestrel48':
skipped_cameras //= 2
return self._falcon.sensor_pitch * skipped_cameras
@property
def sensor_pixel_pitch(self):
"""Distance, in meters, between pixel centers in a sensor."""
return 1.1E-6
sensor_pixel_width = sensor_pixel_pitch
@property
def pixel_width(self):
"""The width of a pixel's FOV in meters."""
if self._calibration_data is None:
return None
return self._calibration_data.get(f'{self.illumination_mode}_pixel_width', None)
@property
def exif_orientation(self):
"""Image Orientation flag to facilitate image display.
Should be set to one of
1, 2, 3, 4, 5, 6, 7, or 8
depending on orientation of the MCAM relative to the sample [1]_.
Changing this parameter only has an effect on the next acquisition's
metadata.
References
----------
.. [1] https://sirv.com/help/articles/rotate-photos-to-be-upright/
"""
return self._exif_orientation
@exif_orientation.setter
def exif_orientation(self, value):
# if value not in (1, 2, 3, 4, 5, 6, 7, 8):
if value not in range(1, 9):
raise ValueError("exif_orientation must be either "
f"1, 2, 3, 4, 5, 6, 7, or 8, but got {value}.")
self._exif_orientation = value
@property
def frame_rate(self):
"""The sensor limited frame rate."""
return self._falcon.frame_rate
@property
def frame_rate_setpoint(self):
"""The current frame rate setpoint for high speed video acquisition.
Changing this value has a side effect of changing the exposure of the
image sensors.
The true frame rate will be limited by both the frame rate setpoint,
and the exposure of the image sensor.
See also
--------
maximum_exposure, minimum_exposure, exposure, frame_rate
"""
return self._falcon.frame_rate_setpoint
@frame_rate_setpoint.setter
def frame_rate_setpoint(self, value):
self._falcon.frame_rate_setpoint = value
@property
def exposure(self):
"""Image sensor exposure in seconds."""
return self._falcon.exposure
@exposure.setter
def exposure(self, value):
"""Image exposure, in seconds."""
self._falcon.exposure = value
@property
def exposure2(self):
"""Image sensor exposure in seconds used in iHDR acquisitions."""
return self._falcon.exposure2
@exposure2.setter
def exposure2(self, value):
"""Image exposure, in seconds."""
self._falcon.exposure2 = value
@property
def interlaced_hdr(self):
"""Interlaced HDR mode for the MCAM."""
return self._falcon.interlaced_hdr
@interlaced_hdr.setter
def interlaced_hdr(self, value):
self._falcon.interlaced_hdr = value
@property
def maximum_datarate(self):
"""The maximum acquisition datarate capable by the MCAM hardware."""
return self._falcon.maximum_datarate
@property
def maximum_frame_rate_setpoint(self):
"""The maximum frame rate setpoint as limited by the sensor hardware."""
return self._falcon.maximum_frame_rate_setpoint
@property
def minimum_frame_rate_setpoint(self):
"""The minimum frame rate setpoint as limited by the sensor hardware."""
return self._falcon.minimum_frame_rate_setpoint
[docs]
def optimize_sensor_timing(self, mode='stability'):
"""Optimize the senor timing for different acquisition modes.
Parameters
----------
mode: 'stability' or 'frame_rate'
The desired sensor timing mode.
For snapshot acquisition, 'stability' is recommended.
Those requiring high frame rate acquisition are encouraged to contact Ramona Optics at
info@ramonaoptics.com to discuss their application.
Notes
-----
The primary use of this function is to enable high speed video acquisition.
The goal is to deprecate the call to this function.
Users should contact Ramona Optics prior to using this function.
"""
warn(
(
"You should prefer to use a context manager for this instead. "
"Contact Ramona Optics for help if you see this warning."
),
stacklevel=2,
)
return self._falcon.optimize_sensor_timing(
mode=mode, _stacklevel_increment=1
)
@contextmanager
def _optimize_sensor_timing_context(
self,
sensor_timing_mode='stability',
exposure=None,
exposure2=None,
frame_rate_setpoint=None,
):
original_sensor_timing_mode = self._falcon.sensor_timing_mode
original_exposure = self.exposure
original_exposure2 = self.exposure2
original_frame_rate_setpoint = self.frame_rate_setpoint
try:
self._falcon.sensor_timing_mode = sensor_timing_mode
if frame_rate_setpoint is not None:
self._falcon.frame_rate_setpoint = frame_rate_setpoint
if exposure is not None:
self._falcon.exposure = exposure
if exposure2 is not None:
self._falcon.exposure2 = exposure2
yield
finally:
self._falcon.sensor_timing_mode = original_sensor_timing_mode
if frame_rate_setpoint is not None:
self._falcon.frame_rate_setpoint = original_frame_rate_setpoint
if exposure is not None:
self._falcon.exposure = original_exposure
if exposure2 is not None:
self._falcon.exposure2 = original_exposure2
@property
def maximum_exposure(self):
"""The maximum exposure in seconds given the current imaging configuration."""
return self._falcon.maximum_exposure
@property
def minimum_exposure(self):
"""The minimum exposure in seconds given the current imaging configuration."""
return self._falcon.minimum_exposure
@property
def maximum_digital_gain(self):
"""The maximum value that the digital gain can be set to."""
return self._falcon.maximum_digital_gain
@property
def minimum_digital_gain(self):
"""The minimum value that the digital gain can be set to."""
return self._falcon.minimum_digital_gain
@property
def maximum_analog_gain(self):
"""The maximum value that the analog gain can be set to."""
return self._falcon.maximum_analog_gain
@property
def minimum_analog_gain(self):
"""The minimum value that the analog gain can be set to."""
return self._falcon.minimum_analog_gain
@property
def use_hardware_trigger(self):
"""Use trigger pulses to control image acquisition.
Using the hardware trigger will update ``minimum_exposure``.
See also
--------
minimum_exposure
"""
return self._falcon.hardware_trigger
@use_hardware_trigger.setter
def use_hardware_trigger(self, value):
if value:
self._falcon.configure_trigger_mode()
else:
self._falcon.configure_streaming_mode()
@property
def output_trigger_events(self):
"""Events that cause an output trigger pulse to be generated.
This is only available for modules that have the hardware trigger
option.
"""
return self._falcon.output_trigger_events
@output_trigger_events.setter
def output_trigger_events(self, value):
self._falcon.output_trigger_events = value
@property
def valid_output_trigger_events(self):
"""The valid output trigger events that can be set."""
return self._falcon.valid_output_trigger_events
@property
def trigger_out(self):
return self._falcon.trigger_out
@trigger_out.setter
def trigger_out(self, value):
self._falcon.trigger_out = value
@property
def has_trigger_out(self):
return self._falcon.has_trigger_out
@property
def bin_mode(self):
"""The number of pixels binned in the X-axis and skipped in the Y-axis.
Changing the bin mode does not affect the image exposure.
The ``data`` property is updated when ``bin_mode`` is modified to
accommodate the changing image resolution.
The data acquired in previous binning modes can still be retrieved
by changing the binning mode back to its previous value.
Valid bin_mode values are either ``1``, ``2``, or ``4``.
"""
return self._falcon.bin_mode
@bin_mode.setter
def bin_mode(self, value):
self._falcon.bin_mode = value
@property
def gain(self):
"""Total gain of the sensor.
Equal to the product of ``analog_gain`` and ``digital_gain``.
Note
----
This property cannot be set. To change the gain, you must change
either the ``digital_gain`` or the ``analog_gain``.
"""
return self._falcon.analog_gain * self._falcon.digital_gain
@property
def digital_gain(self):
"""Digital gain applied globally to all pixels of all sensors."""
return self._falcon.digital_gain
@digital_gain.setter
def digital_gain(self, value):
self._falcon.digital_gain = value
@property
def digital_gain_color(self):
"""Digital gain applied to each individual color in a bayer pixel.
You can assign to this property a tuple of four values
``(red_gain, green0_gain, blue_gain, green1_gain)`` or three values
``(red_gain, green_gain, blue gain)``.
Returns
-------
digital_gain_color: (red gain, green0_gain, blue_gain, green1 gain)
Digital gain values for each color in the form of the tuple.
"""
return self._falcon.digital_gain_color
@digital_gain_color.setter
def digital_gain_color(self, value):
value_len = len(value)
if value_len not in [3, 4]:
raise ValueError(
"The provided color specific gain must be of length 3 or 4.")
value = tuple(value)
if value_len == 3:
# rgb g
value = value + (value[1],)
# value now in rgbg
self._falcon.digital_gain_color = value
@property
def analog_gain(self):
"""Analog gain applied globally to all pixels of all sensors."""
return self._falcon.analog_gain
@analog_gain.setter
def analog_gain(self, value):
self._falcon.analog_gain = value
@property
def bayer_pattern(self):
return self._falcon.bayer_pattern
@property
def sensor_chroma(self):
"""The sensor chromaticity.
This property can take on one of 5 string values:
* ``'monochrome'``
* ``'bayer_grbg'``, ``'bayer_rggb'``, ``'bayer_bggr'`` or, ``'bayer_gbrg'``
"""
return self._falcon.sensor_chroma
@property
def data(self):
# TODO: deprecate this property...
# it isn't even documented anywhere
if self._falcon is None:
raise RuntimeError(
"As of owl version 0.18.0 ccessing the data property requires "
"the MCAM to be open. Please ensure that you create a "
"deep copy of your data before closing the MCAM.")
warn("The data property has been deprecated. "
"We suggest you use other properties, and if you need access to the "
"data use the returned dataset from acquire_full_field_of_view. "
"If you have other questions or suggestions please contact "
"Ramona Optics at help@ramonaoptics.com.", stacklevel=2)
return self._data
@property
def _data(self):
bin_mode = self.bin_mode
if bin_mode == 1:
return self._falcon.data_1x1
elif bin_mode == 2:
return self._falcon.data_2x2
else: # bin_mode == 4
return self._falcon.data_4x4
@property
def dataset(self):
raise RuntimeError(
"The metadata within the dataset variable may change as the mcam's setting "
"are changed. To avoid this issue it is suggested to use the dataset object "
"returned by the acquisition functions "
"(`acquire_selection`, `acquire_full_field_of_view`, `acquire_image_from_sensor`)"
"\nTransition Guideline: mcam.dataset -> mcam.acquire_full_field_of_view()",
)
@property
def white_light_transmission(self):
"""LED color that provides white light to the MCAM in reflection.
These led values come from the photometric calibration data which models
the response of our system when used in transmission.
The user should take care to normalize these LED values to a value
that suits their application.
"""
if self._white_light_transmission is not None:
return self._white_light_transmission
else:
default_values = (0.8132269219712621, 0.16804689162505787, 0.5571554680664121)
return default_values
@property
def white_light_reflection(self):
"""LED color that provides white light to the MCAM in transmission.
These led values come from the photometric calibration data which models
the response of our system when used in transmission.
The user should take care to normalize these LED values to a value
that suits their application.
"""
if self._white_light_reflection is not None:
return self._white_light_reflection
else:
default_values = (0.8132269219712621, 0.16804689162505787, 0.5571554680664121)
return default_values
@property
def white_light_transmission_ir850(self):
return 0, 0, 1
@property
def white_light_reflection_ir850(self):
return 0, 0, 1
def _broadcast_falcon_acquisition_parameters(self, falcon_acquisition_parameters, selection):
result = {}
for key, value in falcon_acquisition_parameters.items():
if key in ['data',]:
continue
elif key in ['bayer_pattern']:
# Keep these as scalars
result[key] = value
else:
# Broadcast these as arrays since for now we still do this
# for the metadata
value = np.broadcast_to(value, selection.shape)
value = np.array(value, copy=True)
result[key] = value
return result
[docs]
def acquire_selection(self, selection, *,
skip_bad_sensors=True, data=None):
"""Acquires an image from the selected sensors.
After calling this method, the metadata included in the ``dataset``
attribute will be updated.
Parameters
----------
selection : np.array[bool]
A boolean array of shape (`N_cameras_X`, `N_cameras_Y`) with the
sensors selected for acquisition set to true.
data : array
Buffer with shape
``(N_cameras_X, N_cameras_Y, sensor_width, sensor_height)``.
to hold the acquired image. If set to ``None``, then the data
is stored in the built-in buffer ``MCAM.data``.
skip_bad_sensors : bool
Don't acquire from non-functioning sensors.
Returns
-------
dataset:
The dataset containing the MCAM metadata in addition to
the image data. For more information about the metadata
included, please see
https://docs.ramonaoptics.com/python_metadata.html This
dataset's metadata will not change when the imaging
parameters of the MCAM are modified. However the, data
included within the dataset may change unless the
``data`` parameter is provided to this function.
.. versionchanged :: 0.18.85
In version 0.18.85 the return value was changed from a
dictionary containing low level information from the MCAM
to the full dataset.
See also
--------
acquire_full_field_of_view, acquire_image_from_sensor
"""
try:
if self._falcon._video_thread is not None:
warn("Video should be turned off before trying to do other acquisitions. "
"This will become an error in future versions of owl.",
stacklevel=2)
except Exception:
# Since video thread is a hidden variable, I don't want to be held
# against breaking that API for this warning
pass
if selection is None:
selection = np.ones(self.N_cameras, dtype=bool)
if not np.any(selection):
raise RuntimeError("Must acquire from at least one image sensor.")
acquisition_parameters = {}
if self._temperature_monitor is not None:
acquisition_parameters |= {
'temperature_monitor_chamber_start_temperature':
_safe_temperature_monitor_read(self._temperature_monitor, 'chamber')
}
memory_metrics = _get_memory_metrics()
acquisition_parameters |= {
'workstation_start_' + k: v
for k, v in memory_metrics.items()
}
dataset = self._create_dataset(
data,
previous_acquisition_parameters=self._last_image_acquisition_parameters,
)
falcon_acquisition_parameters = self._acquire_selection(
selection=selection, skip_bad_sensors=skip_bad_sensors,
data_buffers=data,
)
# Get the data buffers from the acquisition parameters
# If the user provided a data buffer, this will be that
data = falcon_acquisition_parameters.pop('data')
acquisition_parameters |= falcon_acquisition_parameters
self._update_metadata(
dataset,
acquisition_parameters=acquisition_parameters
)
falcon_acquisition_parameters = self._broadcast_falcon_acquisition_parameters(
falcon_acquisition_parameters=falcon_acquisition_parameters,
selection=selection,
)
if self._last_image_acquisition_parameters is None:
self._last_image_acquisition_parameters = {}
for key, value in falcon_acquisition_parameters.items():
self._last_image_acquisition_parameters[key] = value
else:
for key, value in falcon_acquisition_parameters.items():
if key in ['bayer_pattern']:
# scalar variables
self._last_image_acquisition_parameters[key] = value
else:
self._last_image_acquisition_parameters[key][selection] = value[selection]
return dataset
def _acquire_selection(
self,
selection, *,
skip_bad_sensors,
data_buffers,
):
"""Acquires a selected number of sensors"""
acquisition_parameters = self._falcon.acquire_image(
selection,
skip_bad_sensors=skip_bad_sensors,
data_buffers=data_buffers)
return acquisition_parameters
[docs]
def acquire_full_field_of_view(self, *, skip_bad_sensors=True, data=None):
"""Acquires an image from all sensors and updates the metadata.
Parameters
----------
data : array
Buffer with shape
``(N_cameras_X, N_cameras_Y, sensor_width, sensor_height)``.
to hold the acquired image. If set to ``None``, then the data
is stored in the built-in buffer ``MCAM.data``.
.. versionadded :: 0.18.85
In version 0.18.85 the data keyword argument was added.
skip_bad_sensors : bool
Don't acquire from non-functioning sensors.
Returns
-------
dataset:
The dataset containing the MCAM metadata in addition to
the image data. For more information about the metadata
included, please see
https://docs.ramonaoptics.com/python_metadata.html This
dataset's metadata will not change when the imaging
parameters of the MCAM are modified. However the, data
included within the dataset may change unless the
``data`` parameter is provided to this function.
.. versionchanged :: 0.18.85
In version 0.18.85 the return value was changed from a
dictionary containing low level information from the MCAM
to the full dataset.
See also
--------
acquire_selection, acquire_image_from_sensor
"""
selection = self.selection_all_cameras
return self.acquire_selection(
selection,
skip_bad_sensors=skip_bad_sensors,
data=data,
)
[docs]
def acquire_light_pulsed_video(
self,
*,
number_of_pulses,
pulse_period,
illumination_warmup_time,
illumination_mode,
brightness,
selection_slice=None,
data_buffers=None,
tqdm=None,
stop_event=None,
):
"""Acquire a sequence of single-shot frames with periodic light pulses.
At each call, the lights are pulsed ``number_of_pulses`` times with
``pulse_period`` spacing. A single frame is acquired ``illumination_warmup_time``
after each lights-on command.
Designed to be passed as ``func`` into ``run_xy_scan`` — the
``data_buffers`` and ``tqdm`` kwargs are injected by the scan loop.
Parameters
----------
number_of_pulses : int
Number of light pulses (and single-shot frames) to acquire.
pulse_period : float
Period between consecutive pulses in seconds. Must satisfy
``illumination_warmup_time + exposure + 0.5 <= pulse_period``.
illumination_warmup_time : float
Time to wait between switching the lights on and triggering the
camera, in seconds.
illumination_mode : str
Illumination mode applied during each pulse (e.g.
``'fluorescence_brightfield_530nm_high_contrast'``).
brightness : float
Illumination brightness during each pulse, as a fraction between
0.0 and 1.0. Matches the scale used by
:meth:`set_illumination_brightness`.
selection_slice : tuple[slice, slice] or None
Passthrough selection. If provided, the returned dataset's
``image_y`` / ``image_x`` coordinates are sliced accordingly;
acquisition still occurs on the full array.
data_buffers : np.ndarray, optional
Pre-allocated buffer of shape
``(number_of_pulses, N_cameras_Y, N_cameras_X) + image_shape``.
Allocated internally when None. Reused by ``run_xy_scan``.
tqdm : callable, optional
Progress bar updater injected by ``run_xy_scan``.
stop_event : threading.Event, optional
Event used to abort the acquisition. Checked at each pulse
boundary and honoured by ``sleep_from``.
Returns
-------
xarray.Dataset
Dataset with dims ``(frame_number, image_y, image_x, y, x)``.
The illumination switch is guaranteed to be off on exit.
See also
--------
acquire_full_field_of_view, acquire_high_speed_video
"""
if stop_event is None:
stop_event = Event()
shape = (number_of_pulses,) + self.N_cameras + self.image_shape
if data_buffers is None:
data_buffers = alloc_buffer_for_acquisition(
shape, dtype=self._falcon.data.dtype,
)
elif data_buffers.shape != shape:
raise ValueError(
f"data_buffers has shape {data_buffers.shape} but expected {shape}"
)
video_dataset, _ = self._prepare_video_dataset(
data_buffers, selection_slice,
)
from ..mcam_data._core import add_software_timestamp
acquisition_start = datetime.now(UTC)
video_dataset = video_dataset.drop_vars(
['software_timestamp', 'software_timestamp_timezone', 'local_timezone'],
errors='ignore',
)
video_dataset = add_software_timestamp(
video_dataset,
dims=('frame_number', 'image_y', 'image_x'),
since=acquisition_start,
)
selection = np.ones(self.N_cameras, dtype=bool)
last_acquisition_parameters = {}
completed_pulses = 0
if tqdm is None:
tqdm = iter
try:
for pulse_index in tqdm(range(number_of_pulses)):
if stop_event.is_set():
break
t_on = perf_counter()
# Capture the pulse-start wall clock alongside the perf_counter
# anchor. Pulses are phase-locked to ``t_on`` via ``sleep_from``,
# so recording the timestamp here (rather than after acquisition
# completes) yields spacing that matches ``pulse_period`` and
# does not drift with exposure / readout variability.
pulse_start_timestamp = datetime.now(UTC).replace(tzinfo=None)
self.set_illumination_brightness(
brightness, illumination_mode,
)
sleep_from(t_on, illumination_warmup_time, stop_event=stop_event)
if stop_event.is_set():
break
last_acquisition_parameters = self._acquire_selection(
selection=selection,
skip_bad_sensors=True,
data_buffers=data_buffers[pulse_index],
)
video_dataset.software_timestamp.data[pulse_index, ...] = (
pulse_start_timestamp
)
self.set_illumination_brightness(
0.0, illumination_mode,
)
completed_pulses += 1
if pulse_index < number_of_pulses - 1:
sleep_from(t_on, pulse_period, stop_event=stop_event)
finally:
self.set_illumination_brightness(0.0, illumination_mode)
if completed_pulses == 0:
return video_dataset
acquisition_parameters = {}
if isinstance(last_acquisition_parameters, dict):
acquisition_parameters.update(last_acquisition_parameters)
acquisition_parameters.pop('data', None)
memory_metrics = _get_memory_metrics()
acquisition_parameters |= {
'workstation_start_' + k: v
for k, v in memory_metrics.items()
}
self._update_metadata(
video_dataset,
acquisition_parameters=acquisition_parameters,
update_software_timestamp=False,
dataset_selection_slice=selection_slice,
selection=Ellipsis,
)
# _update_metadata polls the live illumination state, which is
# always "off" between pulses (and after the last one). For a pulsed
# acquisition we want the recorded brightness to reflect what the
# lights were driven at during a pulse, not the off-state zero.
video_dataset['illumination_brightness'].data[...] = brightness
if 'motorized' in self._available_illumination_devices:
device_key = 'motorized'
else:
device_key = illumination_mode.split('_')[0]
if device_key in self._available_illumination_devices:
device_brightness_var = f'{device_key}_illumination_brightness'
if device_brightness_var in video_dataset:
video_dataset[device_brightness_var].data[...] = brightness
return video_dataset
[docs]
def acquire_in_focus_video(
self,
*,
z_positions,
number_of_frames,
valid_data=None,
selection_slice=None,
data_buffers=None,
scratch_buffers=None,
tqdm=None,
stop_event=None,
):
"""Acquire one video with every camera parked at its in-focus Z plane.
Each distinct plane among the valid entries of ``z_positions`` is visited
once: at every plane a full-array video is captured, and per camera the
frames recorded at that camera's assigned plane are kept. This collapses
the Z stack a naive capture would produce into a single in-focus video,
so no out-of-focus planes are ever saved. A single shared plane takes the
ordinary one-acquisition path with no extra Z motion.
Designed to be passed as ``func`` into ``run_xy_scan`` — the
``data_buffers`` and ``tqdm`` kwargs are injected by the scan loop.
Parameters
----------
z_positions : np.ndarray
Per-camera physical stage Z positions in meters, shaped to match
``N_cameras`` and in this MCAM's EXIF orientation. The Best Focus
tool assigns every camera a finite plane, so a camera is excluded
from acquisition through ``valid_data``, not through a sentinel here.
number_of_frames : int
Number of frames to acquire at each visited plane.
valid_data : np.ndarray, optional
Per-camera boolean mask shaped like ``z_positions``. Invalid cameras
contribute no plane to visit, so an empty well never costs an extra Z
move. When None every camera is considered valid.
selection_slice : tuple[slice, slice] or None
Passthrough selection forwarded to :meth:`acquire_high_speed_video`.
data_buffers : np.ndarray, optional
Pre-allocated buffer of shape
``(number_of_frames, N_cameras_Y, N_cameras_X) + image_shape``.
Allocated internally when None. Reused by ``run_xy_scan``.
scratch_buffers : np.ndarray, optional
A second buffer with the same shape as ``data_buffers``, used only on
the multi-plane path to hold each plane's full capture before its
in-focus frames are selected out. Allocated internally when None;
pass one in to reuse it across positions.
tqdm : callable, optional
Progress-bar factory. A single bar is created for this position's
acquisition and advanced per frame across every visited plane, using
the nested child-bar mechanism of ``tqdm_base`` (each plane's
frame-acquisition loop becomes a child that contributes one plane's
worth of progress to the shared bar). The bar therefore fills
smoothly during capture rather than jumping once per plane, and its
description reports the plane being acquired.
stop_event : threading.Event, optional
Event used to abort between planes, honoured by
:meth:`acquire_high_speed_video`.
Returns
-------
xarray.Dataset or None
Dataset with dims ``(frame_number, image_y, image_x, y, x)`` plus a
per-camera ``z_stage`` variable recording the plane each camera was
kept from. None if ``stop_event`` aborted the acquisition.
On the multi-plane path, acquisition-wide metadata that would
otherwise be shared across cameras but genuinely differs between the
separate plane captures -- such as ``software_timestamp`` -- gains
``image_y``/``image_x`` dims so each camera records the value from
the acquisition its frames were kept from.
"""
from ..mcam_data._core import add_software_timestamp
if stop_event is None:
stop_event = Event()
z_positions = np.asarray(z_positions, dtype=float)
if z_positions.shape != self.N_cameras:
raise ValueError(
f"z_positions has shape {z_positions.shape} but expected "
f"{self.N_cameras} (one Z per camera)."
)
if valid_data is not None:
valid_data = np.asarray(valid_data, dtype=bool)
if valid_data.shape != z_positions.shape:
raise ValueError(
f"valid_data has shape {valid_data.shape} but z_positions "
f"has shape {z_positions.shape}; they must match."
)
else:
valid_data = np.ones(z_positions.shape, dtype=bool)
shape = (number_of_frames,) + self.N_cameras + self.image_shape
if data_buffers is None:
data_buffers = alloc_buffer_for_acquisition(
shape,
dtype=self._falcon.data.dtype,
)
elif data_buffers.shape != shape:
raise ValueError(f"data_buffers has shape {data_buffers.shape} but expected {shape}")
planes = np.unique(z_positions[valid_data])
if planes.size == 0:
raise ValueError(
"acquire_in_focus_video found no valid Z plane to visit; every "
"camera was marked invalid via valid_data."
)
if callable(tqdm):
progress = tqdm(total=planes.size, desc="Acquiring in-focus video")
else:
progress = None
z_meta = np.zeros(z_positions.shape, dtype=float)
if planes.size == 1:
target_z = float(planes[0])
self.z_stage.set_position(target_z, wait=False)
self.wait_until_idle()
dataset = self.acquire_high_speed_video(
N_frames=number_of_frames,
selection_slice=selection_slice,
data_buffers=data_buffers,
stop_event=stop_event,
tqdm=progress,
)
if dataset is None:
return None
z_meta[...] = target_z
else:
if scratch_buffers is None:
plane_scratch = alloc_buffer_for_acquisition(
data_buffers.shape,
dtype=self._falcon.data.dtype,
)
elif scratch_buffers.shape != data_buffers.shape:
raise ValueError(
f"scratch_buffers has shape {scratch_buffers.shape} but must "
f"match data_buffers shape {data_buffers.shape}"
)
else:
plane_scratch = scratch_buffers
dataset = None
images_encoding = None
assigned = np.zeros(z_positions.shape, dtype=bool)
self.z_stage.set_position(float(planes[0]), wait=False)
for plane_index, target_z in enumerate(planes):
if stop_event.is_set():
return None
self.wait_until_idle()
if progress is not None:
progress.desc = (
f"Acquiring in-focus video "
f"(plane {plane_index + 1}/{planes.size})"
)
plane_dataset = self.acquire_high_speed_video(
N_frames=number_of_frames,
selection_slice=selection_slice,
data_buffers=plane_scratch,
stop_event=stop_event,
tqdm=progress,
)
if plane_dataset is None:
return None
if plane_index + 1 < planes.size:
self.z_stage.set_position(float(planes[plane_index + 1]), wait=False)
plane_mask = z_positions == target_z
data_buffers[:, plane_mask] = plane_dataset.images.data[:, plane_mask]
z_meta[plane_mask] = target_z
assigned |= plane_mask
if dataset is None:
dataset = plane_dataset
images_encoding = dict(plane_dataset.images.encoding)
timestamp_dims = ('image_y', 'image_x') + dataset.software_timestamp.dims
dataset = add_software_timestamp(
dataset.drop_vars('software_timestamp'), dims=timestamp_dims
)
_overlay_in_focus_plane_metadata(dataset, plane_dataset, plane_mask)
if not assigned.all():
data_buffers[:, ~assigned] = plane_scratch[:, ~assigned]
z_meta[~assigned] = float(planes[-1])
_overlay_in_focus_plane_metadata(dataset, plane_dataset, ~assigned)
dataset['images'] = (dataset.images.dims, data_buffers)
dataset.images.encoding.update(images_encoding)
if progress is not None:
progress.update_to(planes.size)
progress.close()
dataset['z_stage'] = (('image_y', 'image_x'), z_meta)
return dataset
[docs]
def acquire_image_from_sensor(self, index, *, skip_bad_sensors=True, data=None):
"""Acquires an image from a single sensor and updates the metadata.
Parameters
----------
index : tuple
Tuple containing the index of the desired sensor, (row, column).
data : array
Buffer with shape
``(N_cameras_X, N_cameras_Y, sensor_width, sensor_height)``.
to hold the acquired image. If set to ``None``, then the data
is stored in the built-in buffer ``MCAM.data``.
.. versionadded :: 0.18.85
In version 0.18.85 the data keyword argument was added.
skip_bad_sensors : bool
Don't acquire from non-functioning sensors.
Returns
-------
dataset:
The dataset containing the MCAM metadata in addition to
the image data. For more information about the metadata
included, please see
https://docs.ramonaoptics.com/python_metadata.html This
dataset's metadata will not change when the imaging
parameters of the MCAM are modified. However the, data
included within the dataset may change unless the
``data`` parameter is provided to this function.
.. versionchanged :: 0.18.85
In version 0.18.85 the return value was changed from a
dictionary containing low level information from the MCAM
to the full dataset.
See also
--------
acquire_full_field_of_view, acquire_selection
"""
selection = np.zeros(self.N_cameras, dtype=bool)
selection[index] = True
return self.acquire_selection(
selection,
skip_bad_sensors=skip_bad_sensors,
data=data,
)
acquire_new_image = acquire_image_from_sensor
@property
def selection_all_cameras(self):
"""The selection of sensors that corresponds to all cameras being used.
Returns
-------
selection: np.array[bool]
Array of bool corresponding to the selection of all sensors. In
this array, all values are ``True``. This array has shape
``MCMA.N_cameras``.
"""
return np.ones(self.N_cameras, dtype=bool)
@property
def selection_center_12(self):
"""The selection of sensors that corresponds to the center 12 cameras.
The center 4 x 3 cameras are selected.
This property return an array of bools corresponding to the selection of
the center 12 sensors.
.. deprecated:: 0.18.4
Notes
-----
This function is only supported by by micro cameras of shape (9, 6).
"""
warn("The use of the ``selection_center_12`` property is deprecated "
"and will be removed in a future version of owl.", stacklevel=2)
if self.N_cameras != (9, 6):
raise RuntimeError(
"This function is only supported by micro camera arrays of "
f"shape (9, 6). This MCAM has a shape of {self.N_cameras}.")
selection = selection_slice_to_selection(
self.N_cameras,
np.s_[3:6, 1:5])
return selection
[docs]
def close(self, *, _garbage_collection=True):
"""Close the connection to the MCAM."""
try:
if self._falcon is not None and self._falcon._video_thread is not None:
warn("Video should be turned off before trying to close the MCAM. "
"This will become an error in future versions of owl.",
stacklevel=2)
except Exception:
# Since video thread is a hidden variable, I don't want to be held
# against breaking that API for this warning
pass
# Internal state of the MCAM Instrument
self._exif_orientation = None
# The order here should be done in reverse order compared to how things
# are done in the opening sequence
# Step 1: Remove input helpers
self._fps_calculation_info = None
self._latest_acquisition_index = None
self._acquisition_count = None
self._physical_acquisition_count = None
self._last_image_acquisition_parameters = None
self._acquisition_index = None
self._physical_acquisition_index = None
# Step 2: Remove peripheral devices
if self._stages is not None:
for axis in list(self._stages.keys()):
self._remove_stage(axis)
self._stages = None
self._available_mechanical_states = None
self._last_set_mechanical_state = None
# Closing the illumination devices will change the size of this
# dictionary. So we convert the keys to a list
if self._available_illumination_devices is not None:
# This helps avoid going through the setter in the case that
# the falcon was never opened
# The setter requires that the falcon have been opened
# successfully
if self.illumination_mode is not None:
self.illumination_mode = None
for illumination_devices in list(self._available_illumination_devices.keys()):
self.remove_illumination(illumination_devices)
self._illumination_brightness_kwargs = None
if self._temperature_monitor is not None:
self.remove_temperature_monitor()
if self._plate_tapper is not None:
self.remove_plate_tapper()
if self._solenoid_driver is not None:
self.remove_solenoid_driver()
if self._incubator is not None:
self.remove_incubator()
if self._indicator_board is not None:
self.remove_indicator_board()
self._external_illumination_brightness = None
self._available_illumination_devices = None
self._own_illumination = None
self._available_virtual_selections = None
self._virtual_selection_key = None
# Step 4: Remove calibration
if self._system_calibration_data is not None:
self._set_calibration_data(None)
# Acquisition mode needs to be removed again after setting the calibration
# to None since the calibration data will go back to 'manual'
# if the current acquisition mode is not within the acquisition mode set
self._acquisition_mode = None
# Step 5: Remove falcon
if self._falcon is not None:
self._falcon.close()
self._falcon = None
self._fake_device = None
self.free_video_buffer(garbage_collection=False)
self._serial_number = None
self._product_line = None
if _garbage_collection:
gc.collect()
def _add_system_coordinates(
self,
dataset,
*,
software_timestamp_dims=None,
previous_acquisition_parameters=None,
):
from ..mcam_data._core import add_software_timestamp
dims = ('image_y', 'image_x')
shape = tuple(dataset.sizes[dim] for dim in dims)
if software_timestamp_dims is None:
software_timestamp_dims = dims
# Acquisition count has always somewhat been broken
# but try to keep it as consistent as possible
if shape == self._acquisition_count.shape:
acquisition_count = np.array(
self._acquisition_count,
copy=True,
dtype=np.int64,
)
acquisition_index = np.array(
self._acquisition_index,
copy=True,
dtype=np.int64,
)
else:
acquisition_count = np.full(
shape,
fill_value=np.max(self._acquisition_count),
dtype=np.int64
)
acquisition_index = np.full(
shape,
fill_value=np.max(self._acquisition_index),
dtype=np.int64
)
new_variables = {
'__falcon.build_info.version__': self._falcon.build_info['version'],
'__falcon.build_info.configuration__': self._falcon.build_info['configuration'],
'__falcon.build_info.branch__': self._falcon.build_info['branch'],
'__falcon.sensor_board.serial_number__': str(self._falcon.serial_number),
# While this is a number
# it is a 128 bit number, and I don't think many serialization
# formats support 96 bit numbers
# Therefore, we choose to store the string representation of the number
'__falcon.dna__': self._falcon.dna_string,
'sensor_chroma': self._falcon.sensor_chroma,
'sensor_board': self._falcon.sensor_board,
'micro_camera_separation_y': self.micro_camera_separation_y,
'micro_camera_separation_x': self.micro_camera_separation_x,
'camera_offset_y': self.camera_offset[0],
'camera_offset_x': self.camera_offset[1],
'external_illumination_brightness': 0.,
'N_cameras_Y': self._falcon.mcam_shape[0],
'N_cameras_X': self._falcon.mcam_shape[1],
'N_cameras_physical_Y': self._falcon.mcam_physical_shape[0],
'N_cameras_physical_X': self._falcon.mcam_physical_shape[1],
# Give the bayer pattern a reasonable value at the beginning
'latest_acquisition_index': 0,
'illumination_brightness': 0.,
'bayer_pattern': self.bayer_pattern,
'acquisition_index': (('image_y', 'image_x'), acquisition_index),
'acquisition_count': (('image_y', 'image_x'), acquisition_count),
'exif_orientation': self.exif_orientation,
}
# only add the serial number if we have it for the specific board
if self._serial_number is not None:
new_variables['serial_number'] = self.serial_number
# Add many coordinates for the various metadata we keep track of
new_variables.update({
name: (dims, np.zeros(shape, dtype=dtype))
for name, dtype in [
('exposure', np.float64),
# For iHDR acquisition, exposure2 is the second exposure
# that is acquired
('exposure2', np.float64),
('interlaced_hdr', bool),
('digital_red_gain', np.float64),
('digital_green1_gain', np.float64),
('digital_blue_gain', np.float64),
('digital_green2_gain', np.float64),
('analog_gain', np.float64),
('digital_gain', np.float64),
('frame_rate_setpoint', np.float64),
('start_pixel_x', np.int64),
('start_pixel_y', np.int64),
('end_pixel_x', np.int64),
('end_pixel_y', np.int64),
('valid_data', bool),
]
})
if previous_acquisition_parameters is not None:
for key in previous_acquisition_parameters:
if key in ['bayer_pattern',]:
new_variables[key] = previous_acquisition_parameters[key]
elif key in ['bin_mode', '_duration', 'selection', '_transfer_duration']:
continue
elif key in new_variables:
new_variables[key][1][...] = previous_acquisition_parameters[key]
else:
# Internal warning only, users shouldn't see this
warn(f"Key {key} not found in new_variables", stacklevel=2)
new_variables.update({
'camera_y': (dims,
np.ones(shape, dtype=np.int64) * dataset.image_y.data[..., np.newaxis]),
'camera_x': (dims,
np.ones(shape, dtype=np.int64) * dataset.image_x.data[np.newaxis]),
})
if self.virtual_selection is not None:
new_variables['virtual_selection'] = self.virtual_selection
if self.acquisition_mode is not None:
new_variables['acquisition_mode'] = self.acquisition_mode
for axis, (stage, _) in self._stages.items():
new_variables |= self._make_stage_metadata(
stage, axis, dims=dims, shape=shape)
if self._temperature_monitor is not None:
new_variables |= self._make_temperature_monitor_metadata(
self._temperature_monitor,
'temperature_monitor',
)
new_variables |= self._make_memory_metadata()
if self._plate_tapper is not None:
new_variables |= self._make_plate_tapper_metadata(
self._plate_tapper,
dataset,
)
if self._solenoid_driver is not None:
new_variables |= self._make_solenoid_driver_metadata(
self._solenoid_driver,
dataset,
)
if self._incubator is not None:
new_variables |= self._make_incubator_metadata(
self._incubator,
dims=dims,
shape=shape,
)
for name, light in self._available_illumination_devices.items():
new_variables |= self._make_illumination_metadata(
light,
f'{name}_illumination',
)
if (isinstance(light, MotorizedIllumination) and
light._fluorescence is not None):
fluorescence_temperatures = light._fluorescence.temperatures
for i, temp in enumerate(fluorescence_temperatures):
new_variables[f'fluorescence_temperature_{i}'] = ((), temp)
dataset = xr.merge([dataset, new_variables])
dataset = add_software_timestamp(
dataset,
dims=software_timestamp_dims,
)
dataset['acquisition_mode'] = self.acquisition_mode
dataset = self._adjust_per_color_digital_gain(
replace_calibration_data(
dataset, self._calibration_data, self.illumination_mode
)
)
dataset['product_line'] = self.product_line
return dataset
def _adjust_per_color_digital_gain(self, dataset):
if 'photometric_response' not in dataset:
return dataset
if self.sensor_chroma == 'monochrome':
return dataset
# Hmm... this is less than idea because of how stateful the dataset
# is for this kind of workflow.
# Ideally, we would just take the original data, and write it once to the
# dataset, and not "read it in and write it back" like the operation we
# are doing below
# However, due to the history of version 0.18 I don't think it is worth
# this cleanup
# Therefore, this function should be called whenever the digital gain
# or the illumination mode are changed
# We will replace the calibration data at that time
gain_red, gain_green1, gain_blue, gain_green2 = self.digital_gain_color
gain_green = (gain_green1 + gain_green2) / 2
norm = min(gain_red, gain_green, gain_blue)
gain_red = gain_red / norm
gain_green = gain_green / norm
gain_blue = gain_blue / norm
photometric_response = dataset['photometric_response']
# 1 for a constant
gain_matrix = np.diag(
np.asarray([gain_red, gain_green, gain_blue, 1],
dtype=photometric_response.dtype)
)
dataset['photometric_response'] = (
photometric_response.dims,
gain_matrix @ np.asarray(photometric_response)
)
return dataset
def _make_temperature_monitor_metadata(self, temperature_monitor, prefix):
metadata = {
# This data is expected to be available on every illumination device
# that connects to the MCAM
f'__{prefix}.serial_number__': ((), temperature_monitor.serial_number),
f'{prefix}_chamber_start_temperature': ((), np.array(np.nan, dtype='float32')),
f'{prefix}_chamber_end_temperature': ((), np.array(np.nan, dtype='float32')),
}
return metadata
def _make_plate_tapper_metadata(self, plate_tapper, dataset):
metadata = {
'__plate_tapper.serial_number__': ((), plate_tapper.serial_number),
}
return metadata
def _make_solenoid_driver_metadata(self, solenoid_driver, dataset):
metadata = {
'__solenoid_driver.serial_number__': ((), solenoid_driver.serial_number),
}
return metadata
def _make_memory_metadata(self):
memory_metrics = _get_memory_metrics()
metadata = {
f'workstation_start_{name}': ((), value)
for name, value in memory_metrics.items()
} | {
f'workstation_end_{name}': ((), value)
for name, value in memory_metrics.items()
}
return metadata
def _make_incubator_metadata(self, incubator, dims, shape):
serial_number = incubator.serial_number
standby = incubator.safe_read('standby')
start_temperature = incubator.safe_read('temperature')
start_room_temperature = incubator.safe_read('room_temperature')
start_lid_temperature = incubator.safe_read('lid_temperature')
start_humidity = incubator.safe_read('humidity')
start_oxygen_concentration = incubator.safe_read('oxygen_concentration')
start_carbon_dioxide_concentration = incubator.safe_read('carbon_dioxide_concentration')
metadata = {
'__incubator.serial_number__': ((), serial_number),
'incubator_standby': ((), standby),
'incubator_start_temperature': (
dims,
np.full(shape, start_temperature, dtype='float32')
),
'incubator_end_temperature': (dims, np.full(shape, np.nan, dtype='float32')),
'incubator_start_room_temperature': (
dims,
np.full(shape, start_room_temperature, dtype='float32')
),
'incubator_end_room_temperature': (dims, np.full(shape, np.nan, dtype='float32')),
'incubator_start_lid_temperature': (
dims,
np.full(shape, start_lid_temperature, dtype='float32')
),
'incubator_end_lid_temperature': (dims, np.full(shape, np.nan, dtype='float32')),
'incubator_start_humidity': (
dims,
np.full(shape, start_humidity, dtype='float32')
),
'incubator_end_humidity': (dims, np.full(shape, np.nan, dtype='float32')),
'incubator_start_oxygen_concentration': (
dims,
np.full(shape, start_oxygen_concentration, dtype='float32')
),
'incubator_end_oxygen_concentration': (dims, np.full(shape, np.nan, dtype='float32')),
'incubator_start_carbon_dioxide_concentration': (
dims,
np.full(shape, start_carbon_dioxide_concentration, dtype='float32')
),
'incubator_end_carbon_dioxide_concentration': (
dims,
np.full(shape, np.nan, dtype='float32')
),
}
return metadata
def _make_illumination_metadata(self, light, prefix):
data_vars = {
# This data is expected to be available on every illumination device
# that connects to the MCAM
f'__{prefix}.serial_number__': ((), light.serial_number),
f'__{prefix}.version__': ((), light.version),
f'__{prefix}.device_name__': ((), light.device_name),
f'{prefix}_illumination_mode': (
# We need to specify that we want a large data buffer for the
# illumination mode in case the data increases in size.
# without this, when we try to modify the data in place,
# the update will not "take" correctly
(), np.asarray(light.get_illumination_mode(), dtype='U255')
),
f'{prefix}_brightness': ((), np.array(light.get_brightness(), dtype=float)),
}
return data_vars
def _make_stage_metadata(self, stage, direction, *, dims, shape):
# We really want to call xarray merge operations as little as possible
# They are quite expensive
# Therefore, create the coordinates we want to add first
data_vars = {
f"{direction}_stage": (dims, np.zeros(shape, dtype=np.float64)),
f"__{direction}_stage.stage_serial_number__": ((), stage.stage_serial_number),
f"__{direction}_stage.communication_adapter_serial_number__": (
(), stage.communication_adapter_serial_number
),
f"__{direction}_stage.flipped__": ((), stage.flip_axis),
}
if stage.travel_range is not None:
data_vars[f"__{direction}_stage.travel_range__"] = stage.travel_range
return data_vars
[docs]
@contextmanager
def hold_state(self):
"""Context manager to ensure the state isn't mutated in a code block.
Example
-------
>>> from owl.instruments import MCAM
>>> mcam = MCAM()
>>> mcam.exposure = 100E-3
>>> with mcam.hold_state():
... mcam.exposure = 10E-3
... print(f"The MCAM exposure is temporarily set to {mcam.exposure * 1000:.2f} ms")
>>> print(f"The MCAM exposure is {mcam.exposure * 1000:.2f} ms")
"""
state = self.get_state()
try:
yield
finally:
self.set_state(state)
def get_state(self, *, strict=True):
from owl.schemas.mcam import MCAMState
if self.z_stage is None:
with_z_stage = False
else:
with_z_stage = self.z_stage.stage_serial_number
if self.y_stage is None:
with_y_stage = False
else:
with_y_stage = self.y_stage.stage_serial_number
if self.x_stage is None:
with_x_stage = False
else:
with_x_stage = self.x_stage.stage_serial_number
if self.x_sample_stage is None:
with_x_sample_stage = False
else:
with_x_sample_stage = self.x_sample_stage.stage_serial_number
if self.transmission_illumination is not None:
with_transmission_illumination = self.transmission_illumination.serial_number
else:
with_transmission_illumination = False
if self.reflection_illumination is not None:
with_reflection_illumination = self.reflection_illumination.serial_number
else:
with_reflection_illumination = False
if self.fluorescence_illumination is not None:
with_fluorescence_illumination = self.fluorescence_illumination.serial_number
else:
with_fluorescence_illumination = False
if self.temperature_monitor is not None:
with_temperature_monitor = self.temperature_monitor.serial_number
else:
with_temperature_monitor = False
if self.plate_tapper is not None:
with_plate_tapper = self.plate_tapper.serial_number
else:
with_plate_tapper = False
if self.solenoid_driver is not None:
with_solenoid_driver = self.solenoid_driver.serial_number
else:
with_solenoid_driver = False
if self.incubator is not None:
with_incubator = self.incubator.serial_number
else:
with_incubator = False
if self.indicator_board is not None:
with_indicator_board = self.indicator_board.serial_number
else:
with_indicator_board = False
state = dict(
exposure=self.exposure,
exposure2=self.exposure2,
interlaced_hdr=self.interlaced_hdr,
analog_gain=self.analog_gain,
illumination_mode=self.illumination_mode,
illumination_brightness=self.get_illumination_brightness(),
# illumination color??? extra parameters???
bin_mode=self.bin_mode,
start_pixel=self.start_pixel,
end_pixel=self.end_pixel,
exif_orientation=self.exif_orientation,
frame_rate_setpoint=self.frame_rate_setpoint,
# sensor_timing_settings
use_hardware_trigger=self.use_hardware_trigger,
output_trigger_events=list(self._falcon.output_trigger_events),
virtual_selection=self.virtual_selection,
acquisition_mode=self.acquisition_mode,
z_stage_position=None if self.z_stage is None else self.z_stage.position,
y_stage_position=self.y_position,
x_stage_position=self.x_position,
x_sample_stage_position=None if self.x_sample_stage is None else self.x_sample_stage.position, # noqa
mechanical_state=self.mechanical_state,
serial_number=self._serial_number,
falcon_dna=self._falcon.dna_string,
sensor_board_serial_number=self._falcon.serial_number,
sensor_chroma=self.sensor_chroma,
sensor_board=self._falcon.sensor_board,
with_z_stage=with_z_stage,
with_y_stage=with_y_stage,
with_x_stage=with_x_stage,
with_x_sample_stage=with_x_sample_stage,
with_transmission_illumination=with_transmission_illumination,
with_reflection_illumination=with_reflection_illumination,
with_fluorescence_illumination=with_fluorescence_illumination,
with_temperature_monitor=with_temperature_monitor,
with_plate_tapper=with_plate_tapper,
with_solenoid_driver=with_solenoid_driver,
with_incubator=with_incubator,
with_indicator_board=with_indicator_board,
)
# Do not include both the digital gain and the digital_gain_color
# in the final schema. Users that might try to inspect the generated json
# may get confused as to which one they should edit
digital_gain = self.digital_gain
digital_gain_color = self.digital_gain_color
if all(g == digital_gain for g in digital_gain_color):
state['digital_gain'] = digital_gain
else:
state['digital_gain_color'] = digital_gain_color
# Do not unnecessarily pollute the illumination brightness kwargs
if len(self._illumination_brightness_kwargs) != 0:
state['illumination_brightness_kwargs'] = self._illumination_brightness_kwargs
return MCAMState(**state)
[docs]
def set_state(self, state, *, strict=True):
"""Set the MCAM to a previously saved state.
Parameters
----------
state:
The desired MCAM state.
strict: bool
If ``True``, the state will be validated for the appropriate serial
numbers, and for the validity of all other parameters.
If ``False``, the serial numbers of the MCAM and the state will not
be compared, and invalid parameters will be replaced by ``None``
and the MCAM will retain the previously applied state.
"""
# TODO, we should do a real check before applying any settings that all
# the settings are valid for this MCAM
if strict:
self._validate_mcam_state_matches_serial_numbers(state)
self._validate_applicability_of_state(state)
else:
state = self._remove_invalid_state_keys(state)
@contextmanager
def raise_if_strict():
try:
yield
except Exception:
if strict:
raise
if state.mechanical_state is not None:
with raise_if_strict():
self.mechanical_state = state.mechanical_state
if state.x_sample_stage_position is not None and self.x_sample_stage is not None:
with raise_if_strict():
self.x_sample_stage.position = state.x_sample_stage_position
if state.x_stage_position is not None and self.x_stage is not None:
with raise_if_strict():
self.x_position = state.x_stage_position
if state.y_stage_position is not None and self.y_stage is not None:
with raise_if_strict():
self.y_position = state.y_stage_position
if state.z_stage_position is not None and self.z_stage is not None:
with raise_if_strict():
self.z_stage.position = state.z_stage_position
if state.virtual_selection in self.available_virtual_selections:
with raise_if_strict():
self.virtual_selection = state.virtual_selection
# set the acquisition mode after the virtual selection
# since the available acquisition modes are dependent on the virtual
# selection
if state.acquisition_mode in self.available_acquisition_modes:
with raise_if_strict():
self.acquisition_mode = state.acquisition_mode
if state.use_hardware_trigger is not None:
with raise_if_strict():
self.use_hardware_trigger = state.use_hardware_trigger
if state.output_trigger_events is not None:
with raise_if_strict():
self.output_trigger_events = state.output_trigger_events
if (
state.start_pixel is not None or
state.end_pixel is not None or
state.bin_mode is not None
):
if state.start_pixel is not None:
start_pixel = state.start_pixel
else:
start_pixel = self.start_pixel
if state.end_pixel is not None:
end_pixel = state.end_pixel
else:
end_pixel = self.end_pixel
if state.bin_mode is not None:
bin_mode = state.bin_mode
else:
bin_mode = self.bin_mode
with raise_if_strict():
self.select_pixels(start_pixel, end_pixel, step=bin_mode)
if state.illumination_mode is not None or state.illumination_brightness is not None:
if state.illumination_brightness is not None:
brightness = state.illumination_brightness
else:
brightness = 0.
if state.illumination_brightness_kwargs is None:
illumination_brightness_kwargs = {}
else:
illumination_brightness_kwargs = state.illumination_brightness_kwargs
with raise_if_strict():
self.set_illumination_brightness(
brightness,
state.illumination_mode,
**illumination_brightness_kwargs,
)
if state.digital_gain is not None:
with raise_if_strict():
self.digital_gain = state.digital_gain
# The digital gain will overwrite all 3 (rgb) (or 4 rgbg) color of digital gain
# so set the per color digital gain after the standard digital gain is set
if state.digital_gain_color is not None:
with raise_if_strict():
self.digital_gain_color = state.digital_gain_color
if state.analog_gain is not None:
with raise_if_strict():
self.analog_gain = state.analog_gain
# frame_rate_setpoint should be set first since it determines the
# allowable bounds for the exposure
if state.frame_rate_setpoint is not None:
with raise_if_strict():
self.frame_rate_setpoint = state.frame_rate_setpoint
if state.exposure is not None:
with raise_if_strict():
self.exposure = state.exposure
if state.exposure2 is not None:
with raise_if_strict():
self.exposure2 = state.exposure2
if state.interlaced_hdr is not None:
with raise_if_strict():
self.interlaced_hdr = state.interlaced_hdr
if state.exif_orientation is not None:
with raise_if_strict():
self.exif_orientation = state.exif_orientation
def _validate_mcam_state_matches_serial_numbers(self, state):
if state.serial_number is not None and (
self._serial_number not in [state.serial_number, state.serial_number + 'R']
):
raise ValueError("Provided MCAMState serial number does not match the given MCAM.")
if state.falcon_dna is not None and (
self._falcon.dna_string not in [state.falcon_dna, state.falcon_dna.replace('0x4002', '0xC002')] # noqa
):
raise ValueError("Provided MCAMState falcon dna does not match the given MCAM.")
if state.sensor_chroma is not None and self.sensor_chroma != state.sensor_chroma:
raise ValueError("Provided MCAMState sensor chroma does not match the given MCAM.")
if state.with_z_stage is not None:
if isinstance(state.with_z_stage, str):
if (
self.z_stage is None or
self.z_stage.stage_serial_number not in [state.with_z_stage, state.with_z_stage + 'R'] # noqa
):
raise ValueError("Provided MCAMState z stage serial number does not match "
"the given MCAM.")
elif (self.z_stage is not None) != state.with_z_stage:
raise ValueError("Provided MCAMState z stage serial number does not match "
"the given MCAM.")
if state.with_y_stage is not None:
if isinstance(state.with_y_stage, str):
if (
self.y_stage is None or
self.y_stage.stage_serial_number not in [state.with_y_stage, state.with_y_stage + 'R'] # noqa
):
raise ValueError("Provided MCAMState y stage serial number does not match the "
"given MCAM.")
elif (self.y_stage is not None) != state.with_y_stage:
raise ValueError("Provided MCAMState y stage serial number does not match the "
"given MCAM.")
if state.with_x_stage is not None:
if isinstance(state.with_x_stage, str):
if (
self.x_stage is None or
self.x_stage.stage_serial_number not in [state.with_x_stage, state.with_x_stage + 'R'] # noqa
):
raise ValueError("Provided MCAMState x stage serial number does not match the "
"given MCAM.")
elif (self.x_stage is not None) != state.with_x_stage:
raise ValueError("Provided MCAMState x stage serial number does not match the "
"given MCAM.")
if state.with_x_sample_stage is not None:
if isinstance(state.with_x_sample_stage, str):
if (
self.x_sample_stage is None or
self.x_sample_stage.stage_serial_number not in [state.with_x_sample_stage, state.with_x_sample_stage + 'R'] # noqa
):
raise ValueError("Provided MCAMState x sample stage serial number does not "
"match the given MCAM.")
elif (self.x_sample_stage is not None) != state.with_x_sample_stage:
raise ValueError("Provided MCAMState x sample stage serial number does not match "
"the given MCAM.")
if state.with_transmission_illumination is not None:
if isinstance(state.with_transmission_illumination, str):
if (
self.transmission_illumination is None or
self.transmission_illumination.serial_number not in [
state.with_transmission_illumination, state.with_transmission_illumination + 'R' # noqa
]
):
raise ValueError("Provided MCAMState transmission illumination serial number "
"does not match the given MCAM.")
elif (self.transmission_illumination is not None) != state.with_transmission_illumination: # noqa
raise ValueError("Provided MCAMState transmission illumination serial number does "
"not match the given MCAM.")
if state.with_reflection_illumination is not None:
if isinstance(state.with_reflection_illumination, str):
if (
self.reflection_illumination is None or
self.reflection_illumination.serial_number not in [
state.with_reflection_illumination, state.with_reflection_illumination + 'R' # noqa
]
):
raise ValueError("Provided MCAMState reflection illumination serial number "
"does not match the given MCAM.")
elif (self.reflection_illumination is not None) != state.with_reflection_illumination: # noqa
raise ValueError("Provided MCAMState reflection illumination serial number does "
"not match the given MCAM.")
if state.with_fluorescence_illumination is not None:
if isinstance(state.with_fluorescence_illumination, str):
if (
self.fluorescence_illumination is None or
self.fluorescence_illumination.serial_number not in [
state.with_fluorescence_illumination, state.with_fluorescence_illumination + 'R' # noqa
]
):
raise ValueError("Provided MCAMState fluorescence illumination serial number "
"does not match the given MCAM.")
elif (self.fluorescence_illumination is not None) != state.with_fluorescence_illumination: # noqa
raise ValueError("Provided MCAMState fluorescence illumination serial number does "
"not match the given MCAM.")
if state.with_temperature_monitor is not None:
if isinstance(state.with_temperature_monitor, str):
if (
self.temperature_monitor is None or
self.temperature_monitor.serial_number not in [
state.with_temperature_monitor, state.with_temperature_monitor + 'R'
]
):
raise ValueError("Provided MCAMState temperature monitor serial number does "
"not match the given MCAM.")
elif (self.temperature_monitor is not None) != state.with_temperature_monitor:
raise ValueError("Provided MCAMState temperature monitor serial number does not "
"match the given MCAM.")
if state.with_plate_tapper is not None:
if isinstance(state.with_plate_tapper, str):
if (
self.plate_tapper is None or
self.plate_tapper.serial_number not in [
state.with_plate_tapper, state.with_plate_tapper + 'R'
]
):
raise ValueError("Provided MCAMState plate tapper serial number does not match"
" the given MCAM.")
elif (self.plate_tapper is not None) != state.with_plate_tapper:
raise ValueError("Provided MCAMState plate tapper serial number does not match"
" the given MCAM.")
if state.with_solenoid_driver is not None:
if isinstance(state.with_solenoid_driver, str):
if (
self.solenoid_driver is None or
self.solenoid_driver.serial_number not in [
state.with_solenoid_driver, state.with_solenoid_driver + 'R'
]
):
raise ValueError("Provided MCAMState solenoid driver serial number does not "
"match the given MCAM.")
elif (self.solenoid_driver is not None) != state.with_solenoid_driver:
raise ValueError("Provided MCAMState solenoid driver serial number does not match"
" the given MCAM.")
# The incubator is still experimental. Do not check the serial number so strictly
if False and state.with_incubator is not None:
if isinstance(state.with_incubator, str):
if (
self.incubator is None or
self.incubator.serial_number not in [
state.with_incubator, state.with_incubator + 'R'
]
):
raise ValueError("Provided MCAMState incubator serial number does not match"
" the given MCAM.")
elif (self.incubator is not None) != state.with_incubator:
raise ValueError("Provided MCAMState incubator serial number does not match"
" the given MCAM.")
# The indicator board is still experimental. Do not check the serial number so strictly
if False and state.with_indicator_board is not None:
if isinstance(state.with_indicator_board, str):
if (
self.indicator_board is None or
self.indicator_board.serial_number not in [
state.with_indicator_board, state.with_indicator_board + 'R'
]
):
raise ValueError(
"Provided MCAMState indicator board serial number does not match"
" the given MCAM.")
elif (self.indicator_board is not None) != state.with_indicator_board:
raise ValueError(
"Provided MCAMState indicator board serial number does not match"
" the given MCAM.")
def _validate_applicability_of_state(self, state):
if state.illumination_mode is not None:
if state.illumination_mode not in self.available_illumination_modes:
raise ValueError(
f"Provided illumination_mode {state.illumination_mode} "
"is not valid for current MCAM")
if state.mechanical_state is not None:
if state.mechanical_state not in self.available_mechanical_states:
raise ValueError(
f"Provided mechanical_state {state.mechanical_state} "
"is not valid for current MCAM")
if state.acquisition_mode is not None:
if state.acquisition_mode not in self.available_acquisition_modes:
raise ValueError(
f"Provided acquisition_mode {state.acquisition_mode} "
"is not valid for current MCAM")
if self.z_stage is not None and state.z_stage_position is not None:
if (state.z_stage_position > self.z_stage.maximum_position or
state.z_stage_position < self.z_stage.minimum_position):
raise ValueError(
f"z_stage_position is out of bounds {state.z_stage_position} not in "
f"[self.z_stage.minimum_position, {self.z_stage.maximum_position})"
)
if self.y_stage is not None and state.y_stage_position is not None:
if (state.y_stage_position > self.y_stage.maximum_position or
state.y_stage_position < self.y_stage.minimum_position):
raise ValueError(
f"y_stage_position is out of bounds {state.y_stage_position} not in "
f"[self.y_stage.minimum_position, {self.y_stage.maximum_position})"
)
if self.x_stage is not None and state.x_stage_position is not None:
if (state.x_stage_position > self.x_stage.maximum_position or
state.x_stage_position < self.x_stage.minimum_position):
raise ValueError(
f"x_stage_position is out of bounds {state.x_stage_position} not in "
f"[self.x_stage.minimum_position, {self.x_stage.maximum_position})"
)
if self.x_sample_stage is not None and state.x_sample_stage_position is not None:
if (state.x_sample_stage_position > self.x_sample_stage.maximum_position or
state.x_sample_stage_position < self.x_sample_stage.minimum_position):
raise ValueError(
f"x_sample_stage_position is out of bounds {state.x_sample_stage_position} "
f"not in [self.x_sample_stage.minimum_position, {self.x_sample_stage.maximum_position})" # noqa
)
def _remove_invalid_state_keys(self, state):
# TODO: consider using the update field in `model_copy` to make this "easier"
valid_state = state.model_copy(deep=True)
if valid_state.illumination_mode not in self.available_illumination_modes:
valid_state.illumination_mode = None
if valid_state.mechanical_state not in self.available_mechanical_states:
valid_state.mechanical_state = None
if valid_state.acquisition_mode not in self.available_acquisition_modes:
valid_state.acquisition_mode = None
if self.z_stage is not None and valid_state.z_stage_position is not None:
if (valid_state.z_stage_position > self.z_stage.maximum_position or
valid_state.z_stage_position < 0.):
valid_state.z_stage_position = None
if self.y_stage is not None and valid_state.y_stage_position is not None:
if (valid_state.y_stage_position > self.y_stage.maximum_position or
valid_state.y_stage_position < 0.):
valid_state.y_stage_position = None
if self.x_stage is not None and valid_state.x_stage_position is not None:
if (valid_state.x_stage_position > self.x_stage.maximum_position or
valid_state.x_stage_position < 0.):
valid_state.x_stage_position = None
if self.x_sample_stage is not None and valid_state.x_sample_stage_position is not None:
if (valid_state.x_sample_stage_position > self.x_sample_stage.maximum_position or
valid_state.x_sample_stage_position < 0.):
valid_state.x_sample_stage_position = None
return valid_state
def _connect_to_stage_autodetect(axis, fake_device, connection_managers, ensure_homed):
"""Open a stage by axis-only autodetection (no user-supplied serial)."""
for StageClass in [X_LSM_E, SmartAxis]:
try:
return StageClass.by_axis(
axis=axis,
fake_device=fake_device,
connection_managers=connection_managers,
ensure_homed=ensure_homed,
)
except Exception:
continue
return None
def _connect_to_stage_by_serial(axis, stage_serial_number, connection_managers, ensure_homed):
"""Open a stage by axis + a registry-known serial number."""
fake_device = stage_serial_number.endswith('R')
for StageClass in [X_LSM_E, SmartAxis]:
try:
return StageClass.by_axis(
axis=axis,
stage_serial_number=stage_serial_number,
fake_device=fake_device,
connection_managers=connection_managers,
ensure_homed=ensure_homed,
)
except Exception:
continue
return None
def _connect_to_stage_with_config(
axis, config, fake_device, connection_managers, ensure_homed,
_stacklevel_increment,
):
"""Open a stage from a user-supplied parameter dict, bypassing the registry.
The base classes (``X_LSM_E``, ``SmartAxis``) take explicit scalar
parameters; the iteration over connection managers belongs here, at the
MCAM dispatch layer. Try every ``(StageClass, connection_manager)`` pair
using each class's singular connection-manager kwarg name, ending with
``None`` for autoconnect. The first pair that succeeds wins.
This path exists because ``by_axis`` rejects any serial number that is
not already in the device registry — but the GUI's Advanced Settings
lets users type arbitrary serials.
"""
kwargs = dict(config)
if 'axis' in kwargs:
warn(
f"Passing 'axis' inside the with_{axis}_stage configuration "
"dict is not supported and the entry will be ignored. "
f"To autodetect a stage by axis, pass with_{axis}_stage=True "
"instead.",
stacklevel=2 + _stacklevel_increment,
)
kwargs.pop('axis')
if 'fake_device' not in kwargs:
sn = kwargs.get('stage_serial_number')
if isinstance(sn, str):
kwargs['fake_device'] = sn.endswith('R')
else:
kwargs['fake_device'] = fake_device
stage_class_specs = [
(X_LSM_E, 'connection_manager'),
(SmartAxis, 'bus'),
]
for StageClass, cm_kwarg in stage_class_specs:
for cm in [*connection_managers, None]:
try:
return StageClass(
ensure_homed=ensure_homed,
**{cm_kwarg: cm},
**kwargs,
)
except Exception:
continue
return None
def _connect_to_stage(
axis,
stage_serial_number,
fake_device,
connection_managers,
*,
ensure_homed=False,
_stacklevel_increment=0
):
# ``stage_serial_number`` is the value the user passed for ``with_{axis}_stage``.
# It may be ``True``/``None`` for autodetection, a string serial number for
# registry-driven lookup, or a dict of construction kwargs that bypasses the
# registry (the GUI's Advanced Settings path).
if isinstance(stage_serial_number, dict):
stage = _connect_to_stage_with_config(
axis=axis,
config=stage_serial_number,
fake_device=fake_device,
connection_managers=connection_managers,
ensure_homed=ensure_homed,
_stacklevel_increment=_stacklevel_increment + 1,
)
elif isinstance(stage_serial_number, str):
stage = _connect_to_stage_by_serial(
axis=axis,
stage_serial_number=stage_serial_number,
connection_managers=connection_managers,
ensure_homed=ensure_homed,
)
else:
stage = _connect_to_stage_autodetect(
axis=axis,
fake_device=fake_device,
connection_managers=connection_managers,
ensure_homed=ensure_homed,
)
if stage is not None:
return stage
if stage_serial_number:
# Only raise an exception if the user explicitly asked for the stage
raise RuntimeError(f"No {axis} stage was found. "
f"To avoid this warning, provide the "
f"with_{axis}_stage "
f"parameter when opening the MCAM.")
warn(f"No {axis} stage was found. "
"To avoid this warning, provide the "
f"with_{axis}_stage "
"parameter when opening the MCAM.",
stacklevel=2 + _stacklevel_increment)
return None
def open_illumination(illumination_device, serial_number=None, _fake_device=False):
if not _fake_device:
Illuminate = RealIlluminate
Fluorescence = RealFluorescence
else:
Illuminate = FakeIlluminate
Fluorescence = FakeFluorescence
if illumination_device == 'transmission':
exceptions = []
for device_name in [
# The last device is our "priority"
# And will report the error message
'c-008-falcon-transmission',
'c-011-vireo-brightfield',
'c-012-vireo-brightfield'
]:
try:
light = Illuminate.by_device_name(
device_name,
serial_number=serial_number)
except Exception as e:
exceptions.append(e)
continue
break
else:
msg = f'Could not find transmission device with {serial_number}'
if exceptions:
raise Exception(msg) from exceptions[0]
else:
raise Exception(msg)
if illumination_device == 'reflection':
light = Illuminate.by_device_name('c-007-falcon',
serial_number=serial_number)
if illumination_device in ('rails', 'fluorescence'):
light = Fluorescence(serial_number=serial_number)
return light
def get_device_illumination_mode(illumination_mode):
if illumination_mode not in supported_illumination_modes:
raise ValueError(
f"Illumination mode ``{illumination_mode}`` is not a supported illumination mode")
if illumination_mode.startswith('reflection_'):
return illumination_mode[len('reflection_'):]
elif illumination_mode.startswith('transmission_'):
return illumination_mode[len('transmission_'):]
elif illumination_mode.startswith('fluorescence_'):
return illumination_mode[len('fluorescence_'):]
else: # for 'external' or 'fluorescence' illumination modes
return None