import functools
import hashlib
import importlib
import os
import platform
import sys
from collections import defaultdict
from datetime import UTC
from datetime import datetime
from datetime import datetime as _datetime
from pathlib import Path
from threading import Event
from time import perf_counter, sleep
from warnings import warn
import numpy as np
from appdirs import AppDirs
from multiuserfilelock import tmpdir as _multiuserfilelock_tmpdir
from packaging.version import Version
from .._version import __version__
IS_LINUX = platform.system() == 'Linux'
IS_OSX = platform.system() == 'Darwin'
if IS_OSX:
# Group used for MultiUserFileLock files. `dialout` is the conventional Linux
# group for serial/tty access, but that group does not exist on macOS; use
# `localaccounts` (which every local user belongs to) there instead.
MULTIUSERFILELOCK_GROUP = 'localaccounts'
# Directory used as the base for MultiUserFileLock lock files. We hard-code
# /tmp instead of tempfile.gettempdir() because on macOS the latter returns a
# per-user path (/var/folders/...), which defeats cross-user file locking.
MULTIUSERFILELOCK_TMPDIR = Path('/tmp')
# Hmm temporary directory on osx. assumed to be a single user system
_MCAM_data = str(Path.home() / 'MCAM_data')
else:
MULTIUSERFILELOCK_GROUP = 'dialout'
MULTIUSERFILELOCK_TMPDIR = _multiuserfilelock_tmpdir
_MCAM_data = '/MCAM_data'
app = AppDirs('mcam', 'ramonaoptics')
# We could optionally allow for an environment variable to override
# each of these paths
user_config_dir = Path(app.user_config_dir)
calibration_filename = user_config_dir / 'calibration.json'
gui_configuration_filename = user_config_dir / 'gui_config.json'
web_configuration_filename = user_config_dir / 'web_mcam_config.json'
cache_dir = Path(os.environ.get(
'RAMONA_MCAM_CACHE_DIR',
app.user_cache_dir,
))
# Persistent storage for the QtWebEngine profile used during Auth0 sign-in
# (cookies, localStorage, HTTP cache). Kept under cache_dir so it stays
# isolated from the user's system browser.
AUTH_BROWSER_PROFILE_DIR = cache_dir / 'auth_browser_profile'
# Directory for application log files (see owl.logging.SimpleLogger). Defaults
# to a 'logs' subdirectory of cache_dir; RAMONA_MCAM_LOG_DIRECTORY overrides it
# so a system (or CI) can send logs to a writable location independent of the
# dataset cache.
log_dir = Path(os.environ.get(
'RAMONA_MCAM_LOG_DIRECTORY',
cache_dir / 'logs',
))
# ML Pipeline environment variables
RAMONA_MCAM_STORAGE_PATH = Path(os.environ.get(
'RAMONA_MCAM_STORAGE_PATH',
_MCAM_data + '/ml_data',
))
RAMONA_MCAM_USE_NPU = os.environ.get('RAMONA_MCAM_USE_NPU', 'False').lower() in ('true', '1')
# Enables or disables certain preview features
# mostly in the GUI
RAMONA_MCAM_ADVANCED_MODE = (
os.environ.get('RAMONA_MCAM_ADVANCED_MODE', 'False').lower() in ('true', '1')
)
RAMONA_MCAM_VIEWER_PREFETCHER_ENABLED = (
os.environ.get('RAMONA_MCAM_VIEWER_PREFETCHER_ENABLED', 'True').lower() in ('true', '1')
)
# Optional SAM image-encoder optimizations. Opt-in (default off) so they can
# be validated internally before being enabled in production.
# - RAMONA_MCAM_SAM_COMPILE: torch.compile the encoder; compiled artifacts are
# cached on disk, keyed by model and input resolution.
# Runtime dtype (float32/bfloat16/float16) is controlled via the ``dtype=``
# kwarg to setup_predictor / setup_video_predictor and by the GUI "Compute
# Precision" dropdown. A boolean env-var shortcut was removed in favour of
# this explicit API.
RAMONA_MCAM_SAM_COMPILE = (
os.environ.get('RAMONA_MCAM_SAM_COMPILE', 'False').lower() in ('true', '1')
)
RAMONA_MCAM_DISABLE_MCAM_DATARATE_CHECK = (
os.environ.get('RAMONA_MCAM_DISABLE_MCAM_DATARATE_CHECK', 'False').lower() in ('true', '1')
)
RAMONA_MCAM_GOOGLE_DRIVE_CREDENTIALS_PATH = os.environ.get(
'RAMONA_MCAM_GOOGLE_DRIVE_CREDENTIALS_PATH',
None
)
RAMONA_MCAM_STUDIO_RESOLUTION_LIMIT = (
os.environ.get('RAMONA_MCAM_STUDIO_RESOLUTION_LIMIT', None)
)
RAMONA_MCAM_SKIP_AUTH = (
os.environ.get('RAMONA_MCAM_SKIP_AUTH', 'False').lower() in ('true', '1', 'yes')
)
# Headless/CI mode: GUI code that would otherwise block on a modal dialog
# (e.g. a confirmation popup) should instead fall through with a safe default,
# so an automated run never hangs waiting on .exec(). Enabled in pytest.ini.
RAMONA_MCAM_HEADLESS_TESTING = (
os.environ.get('RAMONA_MCAM_HEADLESS_TESTING', 'False').lower() in ('true', '1')
)
RAMONA_MCAM_STUDIO_IDLE_ENABLED = (
os.environ.get('RAMONA_MCAM_STUDIO_IDLE_ENABLED', 'True').lower() in ('true', '1')
)
RAMONA_MCAM_STUDIO_FILE_BROWSER_REFRESH_INTERVAL: float = max(60.0, float(
os.environ.get('RAMONA_MCAM_STUDIO_FILE_BROWSER_REFRESH_INTERVAL', '900')
))
RAMONA_MCAM_TEST_DIR = Path(os.environ.get('MCAM_TEST_DIR', _MCAM_data + '/mcam_test'))
try:
if RAMONA_MCAM_STUDIO_RESOLUTION_LIMIT is not None:
RAMONA_MCAM_STUDIO_RESOLUTION_LIMIT = int(RAMONA_MCAM_STUDIO_RESOLUTION_LIMIT)
except ValueError:
RAMONA_MCAM_STUDIO_RESOLUTION_LIMIT = None
RAMONA_MCAM_STUDIO_DATA_DIR = Path(os.environ.get(
'RAMONA_MCAM_STUDIO_DATA_DIR', _MCAM_data
))
try:
# port 80 is privileged
RAMONA_MCAM_STUDIO_PORT = int(
os.environ.get('RAMONA_MCAM_STUDIO_PORT', '8830')
)
except Exception:
RAMONA_MCAM_STUDIO_PORT = 8830
RAMONA_MCAM_STUDIO_AUTO_OPEN = (
os.environ.get('RAMONA_MCAM_STUDIO_AUTO_OPEN', 'False').lower() in ('true', '1')
)
try:
RAMONA_MCAM_STUDIO_MAX_SESSIONS = int(
os.environ.get('RAMONA_MCAM_STUDIO_MAX_SESSIONS', '5')
)
except ValueError:
RAMONA_MCAM_STUDIO_MAX_SESSIONS = int('5')
# Annotate this name exactly once: re-annotating it (`name: float = ...`) on
# each reassignment makes conda-forge Cython 3.2.7 (a 3.3.0a2.dev0 snapshot)
# hang forever in its type-inference pass while compiling this module.
RAMONA_MCAM_AXIS_SETTLE_WAIT_TIMEOUT: float = float(
os.environ.get('RAMONA_MCAM_AXIS_SETTLE_WAIT_TIMEOUT', '10.0')
)
try:
RAMONA_MCAM_AXIS_SETTLE_WAIT_TIMEOUT = float(RAMONA_MCAM_AXIS_SETTLE_WAIT_TIMEOUT)
except Exception:
RAMONA_MCAM_AXIS_SETTLE_WAIT_TIMEOUT = 10.0
AUTH_TOKEN_STORAGE_DIR = Path.home() / '.ramona' / 'auth'
# The order of the dictionary dictates the order in which the operating
# modes appear. Change with caution
visual_illumination_modes = {
'reflection_visible_pwm_fullarray': 'Reflection Brightfield',
'reflection_ir850_pwm_fullarray': 'Reflection IR850',
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
'transmission_ir850_analog_fullarray': 'Transmission IR850',
# Disabled until further notice by Mark
# Transmission Darkfield hasn't been useful as an operating mode
# from 2022 -> 2025
# Will be brought back when we have the max current fixed
# 'transmission_visible_analog_perimeter': 'Transmission Darkfield',
'fluorescence_380nm': 'Fluorescence Ex380nm — DAPI',
'fluorescence_410nm': 'Fluorescence Ex410nm — CFP',
'fluorescence_440nm': 'Fluorescence Ex440nm — GFP FITC',
'fluorescence_470nm': 'Fluorescence Ex470nm — GFP FITC',
'fluorescence_500nm': 'Fluorescence Ex500nm — YFP',
'fluorescence_530nm': 'Fluorescence Ex530nm — tdTom mCherry',
'fluorescence_540nm': 'Fluorescence Ex540nm — tdTom mCherry',
'fluorescence_590nm': 'Fluorescence Ex590nm — RFP mCherry',
'fluorescence_610nm': 'Fluorescence Ex610nm — mCherry',
'fluorescence_633nm': 'Fluorescence Ex633nm — Cy5',
'fluorescence_650nm': 'Fluorescence Ex650nm — Cy5',
'fluorescence_850nm': 'Brightfield (IR 850nm)',
'fluorescence_unknown': 'Fluorescence',
'fluorescence_brightfield_530nm': 'Brightfield (Green)',
'fluorescence_brightfield_530nm_high_contrast': 'High Contrast Brightfield',
'fluorescence_brightfield_530nm_focused': 'Focused Brightfield',
'fluorescence_white_5650K': 'Brightfield (White)',
'fluorescence_brightfield_530nm_diffuser': 'Brightfield',
'fluorescence_brightfield_540nm': 'Brightfield (Green)',
'fluorescence_brightfield_540nm_high_contrast': 'High Contrast Brightfield',
'fluorescence_brightfield_540nm_focused': 'Focused Brightfield',
'fluorescence_brightfield_540nm_diffuser': 'Brightfield',
'fluorescence_brightfield_590nm': 'Brightfield (Amber)',
'fluorescence_brightfield_590nm_high_contrast': 'High Contrast Brightfield',
'fluorescence_brightfield_590nm_focused': 'Focused Brightfield',
'fluorescence_brightfield_590nm_diffuser': 'Brightfield',
'transmission_visible_analog_535nm_diffuser': 'Brightfield',
'transmission_visible_analog_535nm_focused': 'Focused Brightfield',
# 'transmission_visible_pwm_535nm_diffuser': 'Brightfield',
# 'transmission_visible_pwm_535nm_focused': 'Focused Brightfield',
}
per_device_visual_illumination_modes = defaultdict(
lambda: visual_illumination_modes.copy()
)
per_device_visual_illumination_modes['Vireo1114'] = {
'transmission_visible_analog_535nm_diffuser': 'Brightfield (analog)',
'transmission_visible_analog_535nm_focused': 'Focused Brightfield (analog)',
'transmission_visible_pwm_535nm_diffuser': 'Brightfield (pwm)',
'transmission_visible_pwm_535nm_focused': 'Focused Brightfield (pwm)',
'fluorescence_380nm': 'Fluorescence Ex380nm — DAPI',
'fluorescence_470nm': 'Fluorescence Ex470nm — GFP FITC',
'fluorescence_540nm': 'Fluorescence Ex540nm — tdTom mCherry',
'fluorescence_590nm': 'Fluorescence Ex590nm — RFP mCherry',
'fluorescence_633nm': 'Fluorescence Ex633nm — Cy5',
'fluorescence_brightfield_530nm': 'Brightfield (Green)',
'fluorescence_brightfield_530nm_high_contrast': 'High Contrast Brightfield',
'fluorescence_brightfield_530nm_focused': 'Focused Brightfield',
'fluorescence_brightfield_590nm': 'Brightfield (Amber)',
'fluorescence_brightfield_590nm_high_contrast': 'High Contrast Brightfield',
'fluorescence_brightfield_590nm_focused': 'Focused Brightfield',
'fluorescence_brightfield_590nm_diffuser': 'Brightfield',
'fluorescence_white_5650K': 'Brightfield (White)',
'fluorescence_brightfield_530nm_diffuser': 'Brightfield',
'fluorescence_brightfield_540nm': 'Brightfield (Green)',
'fluorescence_brightfield_540nm_high_contrast': 'High Contrast Brightfield',
'fluorescence_brightfield_540nm_focused': 'Focused Brightfield',
'fluorescence_brightfield_540nm_diffuser': 'Brightfield',
}
per_device_visual_illumination_modes['Kestrel0010'] = {
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
'transmission_ir850_analog_fullarray': 'Transmission IR850',
# Mark - 2023
# We have given Kestrel0010 a dual 380nm/440nm illumination unit, but they only
# use 440nm. Blocking the 380nm mode to avoid confusion.
# 'fluorescence_380nm': 'Fluorescence Ex380nm — DAPI',
'fluorescence_410nm': 'Fluorescence Ex410nm — CFP',
'fluorescence_440nm': 'Fluorescence Ex440nm — GFP FITC',
'fluorescence_500nm': 'Fluorescence Ex500nm — YFP',
'fluorescence_530nm': 'Fluorescence Ex530nm — tdTom mCherry',
'fluorescence_590nm': 'Fluorescence Ex590nm — RFP mCherry',
'fluorescence_610nm': 'Fluorescence Ex610nm — mCherry',
'fluorescence_633nm': 'Fluorescence Ex633nm — Cy5',
}
per_device_visual_illumination_modes['Kestrel0004'] = {
'transmission_red_analog_fullarray': 'Brightfield (Red)',
'transmission_green_analog_fullarray': 'Brightfield (Green)',
'transmission_blue_analog_fullarray': 'Brightfield (Blue)',
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
'transmission_ir850_analog_fullarray': 'Transmission IR850',
'fluorescence_380nm': 'Fluorescence Ex380nm — DAPI',
'fluorescence_410nm': 'Fluorescence Ex410nm — CFP',
'fluorescence_440nm': 'Fluorescence Ex440nm — GFP FITC',
'fluorescence_500nm': 'Fluorescence Ex500nm — YFP',
'fluorescence_530nm': 'Fluorescence Ex530nm — tdTom mCherry',
'fluorescence_590nm': 'Fluorescence Ex590nm — RFP mCherry',
'fluorescence_610nm': 'Fluorescence Ex610nm — mCherry',
'fluorescence_633nm': 'Fluorescence Ex633nm — Cy5',
}
per_device_visual_illumination_modes['Kestrel0041'] = {
# 'transmission_red_analog_fullarray': 'Brightfield (Red)',
# 'transmission_green_analog_fullarray': 'Brightfield (Green)',
# 'transmission_blue_analog_fullarray': 'Brightfield (Blue)',
# 'transmission_visible_analog_fullarray': 'Brightfield (White)',
# This system was delivered before we used the individual Red/Green/Blue
# illumination modes.
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
}
per_device_visual_illumination_modes['Kestrel0042'] = {
# 'transmission_red_analog_fullarray': 'Brightfield (Red)',
# 'transmission_green_analog_fullarray': 'Brightfield (Green)',
# 'transmission_blue_analog_fullarray': 'Brightfield (Blue)',
# 'transmission_visible_analog_fullarray': 'Brightfield (White)',
# This system was delivered before we used the individual Red/Green/Blue
# illumination modes.
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
}
per_device_visual_illumination_modes['Kestrel0101'] = {
'transmission_green_analog_fullarray': 'Brightfield (Green)',
'transmission_visible_analog_fullarray': 'Brightfield (White)',
}
per_device_visual_illumination_modes['Vireo1101'] = {
'transmission_green_analog_fullarray': 'Brightfield (Green)',
'transmission_visible_analog_fullarray': 'Brightfield (White)',
}
per_device_visual_illumination_modes['Vireo1105'] = {
'fluorescence_380nm': 'Fluorescence Ex380nm — DAPI',
'fluorescence_440nm': 'Fluorescence Ex440nm — GFP FITC',
'fluorescence_530nm': 'Fluorescence Ex530nm — tdTom mCherry',
'fluorescence_633nm': 'Fluorescence Ex633nm — Cy5',
# This mode just doesn't work very well. and any small misalignment causes
# it to have large glares in well plates
# 'fluorescence_brightfield_530nm_high_contrast': 'High Contrast Brightfield',
'fluorescence_brightfield_530nm_focused_diffuser': 'Focused Brightfield Diffuser',
# 'fluorescence_brightfield_530nm_focused_diffuser': 'Brightfield',
'fluorescence_brightfield_530nm_diffuser': 'High Diffusion Brightfield',
'fluorescence_brightfield_530nm_focused': 'Focused Brightfield',
# For debugging during initial setup
'transmission_green_analog_fullarray': 'Brightfield (Green)',
'transmission_visible_analog_fullarray': 'Brightfield (White)',
}
per_device_visual_illumination_modes['Vireo1109'] = {
# July 2025 -- Mark
# While this hardware no longer has the "fluorescence" module installed
# Some old data that we use in visuals for demos was acquired with this system
# And these visual names still appear there.
'fluorescence_brightfield_530nm_focused': 'Focused Brightfield',
'fluorescence_brightfield_530nm_diffuser': 'Brightfield',
# We removed the white illumination from this MCAM
# so just name this cleaner
# 'transmission_green_analog_fullarray': 'Brightfield (Green)',
# 'transmission_visible_analog_fullarray': 'Brightfield (White)',
'transmission_green_analog_fullarray': 'Brightfield',
}
per_device_visual_illumination_modes['Vireo1113'] = {
# We removed the white illumination from this MCAM
# so just name this cleaner
# 'transmission_green_analog_fullarray': 'Brightfield (Green)',
# 'transmission_visible_analog_fullarray': 'Brightfield (White)',
'transmission_green_analog_fullarray': 'Brightfield',
}
per_device_visual_illumination_modes['Kestrel0017'] = {
'fluorescence_530nm': 'Green (530 nm)',
'fluorescence_405nm': 'Blue (405 nm)',
}
per_device_visual_illumination_modes['Kestrel0066'] = {
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
'transmission_ir850_analog_fullarray': 'Transmission IR850',
'fluorescence_440nm': 'Fluorescence Ex440nm — GFP FITC',
'fluorescence_590nm': 'Fluorescence Ex590nm — RFP mCherry',
}
per_device_visual_illumination_modes['Eagle0030'] = {
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
'transmission_red_analog_fullarray': 'Transmission Brightfield (Red)',
'transmission_green_analog_fullarray': 'Transmission Brightfield (Green)',
'transmission_blue_analog_fullarray': 'Transmission Brightfield (Blue)',
}
per_device_visual_illumination_modes['Eagle0070'] = {
'transmission_visible_analog_fullarray': 'Transmission Brightfield',
'transmission_red_analog_fullarray': 'Transmission Brightfield (Red)',
'transmission_green_analog_fullarray': 'Transmission Brightfield (Green)',
'transmission_blue_analog_fullarray': 'Transmission Brightfield (Blue)',
'fluorescence_unknown': 'Fluorescence',
}
per_device_visual_illumination_modes['Vireo1120'] = {
'transmission_visible_analog_535nm_focused': 'Focused Brightfield',
'transmission_visible_analog_535nm_diffuser': 'Brightfield',
}
device_keys = list(per_device_visual_illumination_modes.keys())
for device in device_keys:
# Also populate the fake devices
per_device_visual_illumination_modes[device + 'R'] = {
**per_device_visual_illumination_modes[device]
}
__all__ = [
'sys_info',
'mkdir_and_open',
'arrange_pyramid_as_single',
'cache_dir',
'user_config_dir',
'gui_configuration_filename',
'make_timestamp',
'calibration_filename',
'is_stable_version',
'sleep_from',
]
def mkdir_and_open(
file, mode='r', buffering=-1, encoding=None, errors=None, newline=None,
closefd=True, opener=None,
):
"""Like the built-in :func:`open`, but create the parent directory first.
Takes the same arguments as :func:`open` and returns the same file
object, after ensuring ``file``'s parent directory exists (created with
``parents=True, exist_ok=True``). This lets callers write to config or
calibration paths on a fresh install without a separate ``mkdir`` step,
while keeping the familiar ``open`` calling convention.
"""
Path(file).parent.mkdir(parents=True, exist_ok=True)
return open(
file, mode, buffering, encoding, errors, newline, closefd, opener,
)
def is_stable_version(version=None):
if version is None:
version = __version__
return '+' not in version
@functools.lru_cache
def _read_lsb_release_description():
description = None
try:
with open('/etc/lsb-release', encoding='utf-8') as file:
for line in file:
if line.startswith('DISTRIB_DESCRIPTION='):
description = line.split('=', 1)[1].strip().strip('"')
break
except FileNotFoundError:
# Can happen on non-Linux systemsc
pass
except Exception:
pass
return description
[docs]
def sys_info(*, force_all_modules=False):
r"""Collect relevant system and debugging information.
Returns
-------
system_information: dictionary
Version information of many packages, optional and required, to run
owl.
Example
-------
>>> owl.sys_info()
{'python': '3.8.5 | packaged by conda-forge | (default, Sep 27 2020, 15:56:17) \n[GCC 7.5.0]',
'owl': '0.15.18',
'sys.executable': '/home/ramona/falcongui/bin/python',
'sys.exec_prefix': '/home/mark/falcongui',
'platform.node': 'falcon',
'mcam_sdk': 'is not installed',
'pyilluminate': '0.6.5',
'teensytoany': '0.0.24',
'opencv': '4.5.2',
'numpy': '1.21.4',
'scipy': '1.7.0',
'imageio': '2.9.0',
'scikit-image': '0.18.2',
'xarray': '0.18.2',
'dask': '2021.07.0',
'vispy': '0.7.0',
'tqdm': '4.61.2',
'pillow': '8.3.1',
'zaber-motion': '2.4.1',
'ro_json': '3.15.5'}
"""
versions = {}
# I want these two to show up first
versions['python'] = sys.version
versions['owl'] = __version__
versions['sys.executable'] = sys.executable
versions['sys.exec_prefix'] = sys.exec_prefix
versions['platform.node'] = platform.node()
versions['platform.release'] = platform.release()
if IS_LINUX:
versions['lsb_release'] = _read_lsb_release_description()
for req in [
'numpy',
'pexpect',
'desktop-app',
'pygments',
'scipy',
'imageio',
'imageio-ffmpeg',
'av',
'scikit-image',
'tzlocal',
'xarray',
'pandas',
'netCDF4',
'h5py',
'h5netcdf',
'opencv',
'psutil',
'ro-json',
'dask',
'cytoolz',
'pyilluminate',
'tqdm',
'pyyaml',
'appdirs',
'teensytoany',
'zaber-motion',
'tifffile',
'pooch',
'gdown',
'boto3',
'xxhash',
'multiuserfilelock',
'tabulate',
'pydantic',
# We don't directly depend on pillow but it is an implicit dependency
# from many of our packages
'pillow',
]:
# Annoying modules that have different common names than import names
if req == 'opencv':
import_name = 'cv2'
elif req == 'pillow':
import_name = 'PIL'
elif req == 'scikit-image':
import_name = 'skimage'
elif req == 'pyyaml':
import_name = 'yaml'
elif req == 'wgpu-py':
import_name = 'wgpu'
else:
import_name = req.replace('-', '_')
# some modules do not use the standard python __version__ attribute
version_attribute = '__version__'
if req == 'xxhash':
version_attribute = 'VERSION'
try:
module = importlib.import_module(import_name)
version = getattr(
module,
version_attribute,
'is installed but the version could not be determined'
)
except ModuleNotFoundError:
version = 'is not installed'
except Exception:
version = 'could not be determined'
versions[req] = version
# click will issue a warning if you access __version__ and don't plan
# to undeprecate this
# https://github.com/pallets/click/issues/2900
versions['click'] = importlib.metadata.version('click')
# Packages like qt can have big side effects on the open application.
# Even the use of importlib was found to slow this down a lot. (150 us -> 1.5 ms)
# Thus for performance reasons, we choose to only get the versions if these
# modules have already been loaded
for req in [
'matplotlib',
'pyside6',
'pygfx',
'wgpu-py',
'pyqt',
'cupy',
'pytorch',
'fastapi',
'uvicorn'
]:
if req == 'pyside6':
import_name = 'PySide6'
elif req == 'pytorch':
import_name = 'torch'
elif req == 'wgpu-py':
import_name = 'wgpu'
else:
import_name = req
# some modules do not use the standard python __version__ attribute
version_attribute = '__version__'
if force_all_modules or import_name in sys.modules:
try:
module = importlib.import_module(import_name)
version = getattr(
module,
version_attribute,
'is installed but the version could not be determined'
)
except ModuleNotFoundError:
version = 'is not installed'
except Exception:
version = 'could not be determined'
else:
version = 'is not loaded'
versions[req] = version
# Reporting the library versions is useful when troubleshooting
# ffmpeg encoding error
try:
# Lazyload av so we don't need it installed (heavy dependency)
# when downloading the cache
import av
versions |= {
name: '.'.join(str(v) for v in version)
for name, version in av.library_versions.items()
}
versions['ffmpeg'] = av.ffmpeg_version_info
except Exception:
versions['ffmpeg'] = 'is not installed'
return versions
def make_timestamp(datetime=None):
"""Make a timestamp with millisecond accuracy.
This function returns the value of
>>> from datetime import datetime
>>> datetime.now(None).strftime('%Y%m%d_%H%M%S_%f')[:-3]
'20201013_100337_415'
Parameters
----------
datetime:
datetime object from which to create the timestamp from. If ``None``
then the current time will be used.
Returns
-------
timestamp: str
Timestamp as a string with millisecond precision.
"""
if datetime is None:
# We specify None as the time zone so our linter doesn't pick this up.
# Ultimately, this was designed to be a human readable timezone
# with the localtime
# If we want to change this behavior, we would have to deprecate this
# function
datetime = _datetime.now(None) # noqa: DTZ005
return datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]
def arrange_pyramid_as_single(pyramid):
"""Create a single mosaic from an image pyramid to quickly visualize it.
Takes a pyramid of images, and creates a stack placing the [0] image on
the left, then all the other images on the right, stacked one over
the other.
Parameters
----------
pyramid: array like
Pyramid of images (2D or 2D+Channels)
Returns
-------
A single composite image as described above.
"""
rows, cols = pyramid[0].shape[:2]
pyramid_shape = np.asarray([i.shape for i in pyramid])
total_rows = np.maximum(np.sum(pyramid_shape[1:, 0]), rows)
composite_shape = (total_rows, pyramid[0].shape[1] + pyramid[1].shape[1])
if len(pyramid[0].shape) == 3:
composite_shape = (*composite_shape, pyramid[0].shape[2])
composite_image = np.zeros(composite_shape, dtype=pyramid[0].dtype)
composite_image[:rows, :cols, ...] = pyramid[0]
i_row = 0
for p in pyramid[1:]:
n_rows, n_cols = p.shape[:2]
composite_image[i_row:i_row + n_rows, cols:cols + n_cols, ...] = p
i_row += n_rows
return composite_image
def sleep_from(previous_time, seconds, *, stop_event=None):
"""Sleep for a given amount of time relative to a previous time point.
Unlike the call to sleep, this will account for time spent doing other
tasks that have unpredictable timing and sleep for the remaining requested
time.
Sleep can be aborted through the use of the ``stop_event`` parameter. If
this parameter is used, it is the caller's duty to check whether or not the
``stop_event`` has been set before continuing.
The return value will not change in the case that the ``stop_event`` was set.
If the required time to sleep is negative, the function will return
immediately.
Example
-------
The following example shows how to control the periodicity of a for loop
such that an event occurs every 200 ms.
>>> from time import perf_counter
>>> from owl.util import sleep_from
>>> current_time = perf_counter()
>>> for i in range(10):
... pass # Some time consuming task
... current_time = sleep_from(current_time, 200E-3)
The following example shows an example of how to provide a stop event to
cancel sleeping.
>>> from threading import Thread, Event
>>> from owl.util import sleep_from
>>> from time import sleep, perf_counter
>>> stop_event = Event()
>>> current_time = perf_counter()
>>> def set_stop_event(stop_event):
... sleep(5)
... stop_event.set()
>>> stop_thread = Thread(target=set_stop_event, args=(stop_event,))
>>> stop_thread.start()
>>> for i in range(10):
... pass # Some time consuming task
... current_time = sleep_from(current_time, 2, stop_event=stop_event)
Parameters
----------
previous_time: float
The returned value from a previous call of ``time.perf_counter()``.
seconds: float
The number of seconds relative to the ``previous_time`` before
returning.
stop_event: threading.Event
An optional parameter specifying an event that can cause this method to
return early. Note that even in the case that this method returns early,
the returned value will not be affected.
Returns
-------
current_time:
Result of ``previous_time + seconds``. Always.
"""
get_to_time = previous_time + seconds
required_sleep_time = get_to_time - perf_counter()
if required_sleep_time > 0:
if stop_event is None:
# Do not use multi-threading primitives if the user didn't want to
sleep(required_sleep_time)
else:
# Event.wait will always return True except if a timeout is given
# and the operation times out.
stop_event.wait(required_sleep_time)
return get_to_time
def tqdm_sleep(seconds, desc='Sleeping', tqdm=None, stop_event=None):
"""Sleep with a progress bar based on tqdm.
Parameters
----------
seconds: int
Amount of seconds to sleep for.
desc: string, optional
Description passed to tqdm.
tqdm: tqdm-like
A tqdm constructor object.
"""
if tqdm is None:
def tqdm(*args, **kwargs):
return args[0]
if stop_event is None:
stop_event = Event()
tenths_of_seconds = int(seconds * 10)
remaining_time = seconds - tenths_of_seconds / 10
current_time = perf_counter()
if tenths_of_seconds > 0:
# The if statemetn above avoids a call to tqdm's initializer
# which stops this progress bar from appearing
# for very short sleep durations
for i in tqdm(range(tenths_of_seconds), desc=desc):
current_time = sleep_from(current_time, 0.1, stop_event=stop_event)
current_time = sleep_from(current_time, remaining_time, stop_event=stop_event)
def to_datetime(date):
timestamp = ((date - np.datetime64('1970-01-01T00:00:00', "ns"))
/ np.timedelta64(1, 's'))
return datetime.fromtimestamp(timestamp, UTC)
def pooch_downloader(fname, output_file, pooch, tqdm=None):
if (
fname.startswith('https://drive.google.com/uc?id=') or
fname.startswith('https://drive.google.com/file/d/')
):
if os.environ.get('RCLONE_FALCON_LEGACY_DATA_READ_ONLY_FILENAME', None):
try:
return _rclone_downloader(fname, output_file, pooch, tqdm=tqdm)
except Exception:
pass
return _gdown_downloader(fname, output_file, pooch, tqdm=tqdm)
if fname.startswith('http://') or fname.startswith('https://'):
return _http_downloader(fname, output_file, pooch, tqdm=tqdm)
# Fall back to pooch for any exotic schemes (ftp, sftp, doi, ...). Its own
# progressbar is already byte-scaled.
from pooch.downloaders import choose_downloader
downloader = choose_downloader(fname, progressbar=tqdm is not None)
return downloader(fname, output_file, pooch)
def _http_downloader(url, output_file, pooch, tqdm=None):
# pooch's HTTPDownloader reuses a passed-in tqdm instance as-is and never
# gives it byte units, so the bar reads raw counts with a nonsense rate.
# Stream it ourselves and drive tqdm the same way the gdown/rclone paths do.
import requests
response = requests.get(url, stream=True, timeout=(10, 60))
response.raise_for_status()
total = int(response.headers.get('content-length', 0)) or None
pbar = tqdm(total=total, unit='B', unit_scale=True) if tqdm is not None else None
try:
with open(output_file, 'wb') as f:
for chunk in response.iter_content(chunk_size=1024 * 1024):
f.write(chunk)
if pbar is not None:
pbar.update(len(chunk))
finally:
if pbar is not None:
pbar.close()
def _gdown_downloader(fname, output_file, pooch, tqdm=None):
import gdown
if fname.startswith('https://drive.google.com/file/d/'):
# Mark - 2025/11/03
# Not super sure why by pooch really wants the uc?id= format
fname.replace(
'https://drive.google.com/file/d/',
'https://drive.google.com/uc?id='
)
kwargs = {}
pbar = None
if tqdm is not None:
if Version(gdown.__version__) >= Version('6.0.0'):
pbar = tqdm(unit='B', unit_scale=True)
def progress(bytes_so_far, bytes_total):
if pbar.total is None and bytes_total is not None:
pbar.total = bytes_total
pbar.update(bytes_so_far - pbar.n)
kwargs['progress'] = progress
else:
warn(f"""Progress bar requires gdown>=6.0. Please update gdown with
conda install --prefix {sys.prefix} "gdown>=6.0.0"
""")
try:
gdown.download(fname, output_file, quiet=True, **kwargs)
finally:
if pbar is not None:
pbar.close()
def _rclone_downloader(fname, output_path, pooch, tqdm=None):
import subprocess
file_id = fname.replace(
'https://drive.google.com/uc?id=', ''
).replace(
'https://drive.google.com/file/d/', ''
)
credentials = os.environ['RCLONE_FALCON_LEGACY_DATA_READ_ONLY_FILENAME']
command = [
"rclone",
# Use the copyid backend in order to use the gdrive_fileid directly
"backend", "copyid",
f"--config={credentials}",
]
target = [
"Falcon_Legacy_Data_Read_Only:",
f"{file_id}",
f"{output_path}",
]
if tqdm is None:
subprocess.check_call(command + target)
return
# rclone has no Python progress hook, but --use-json-log makes it emit a JSON
# stats line every interval. We parse stats.bytes / stats.totalBytes out of
# those and drive a tqdm bar so the progress matches every other downloader.
import json
command += ["--use-json-log", "--stats=200ms", "--stats-log-level=NOTICE"]
process = subprocess.Popen(command + target, stderr=subprocess.PIPE, text=True)
pbar = tqdm(unit='B', unit_scale=True)
try:
for line in process.stderr:
try:
stats = json.loads(line).get('stats')
except json.JSONDecodeError:
continue
if not stats:
continue
total = stats.get('totalBytes') or None
if total and pbar.total != total:
pbar.total = total
pbar.update(stats['bytes'] - pbar.n)
finally:
pbar.close()
returncode = process.wait()
if returncode != 0:
raise subprocess.CalledProcessError(returncode, command + target)
class AnyEvent:
def __init__(self, events):
"""Takes a list of events and makes them behave as a single event.
The methods ``set`` and ``clear`` get passed through to only the first
event in the iterable of events provided.
"""
events = tuple(events)
if len(events) == 0:
raise ValueError("There must be at least 1 event")
self._events = events
def is_set(self):
return any(x.is_set() for x in self._events)
def set(self):
self._events[0].set()
def clear(self):
self._events[0].clear()
def compute_sha256(file_path):
"""
Computes the SHA-256 hash of a given file.
Parameters:
file_path (str): The path to the file to be hashed.
Returns:
str: The SHA-256 hash of the file in hexadecimal format.
"""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
sha256_hash.update(chunk)
return sha256_hash.hexdigest()
def compute_sha256_in_memory(data):
sha256_hash = hashlib.sha256()
sha256_hash.update(data)
return sha256_hash.hexdigest()
def format_version_string(serial_number=None):
version_split = __version__.split('.')
version = '.'.join(version_split[:3])
if len(version_split) >= 4:
version += ';' if RAMONA_MCAM_ADVANCED_MODE else '.'
elif RAMONA_MCAM_ADVANCED_MODE:
version += ','
if serial_number is not None:
# Removing any trailing 'R' characters. These indicate
# "fake devices" that we use for testing.
serial_number = serial_number.rstrip('R')
version += f" ({serial_number})"
return version