9032957364

Committed 10 May 2024 01:25PM UTC coverage: 48.538% (+1.7%) from 46.79%

Build # 9032957364

Build Type

Pull #643

github

Specific Base 74d2ec

Committed by

web-flow

Commit Message

Merge aebf2c9af into ec2d8e4fe

Pull Request Pull Request #643: 8.19.0

Run Details

377 of 1074 new or added lines in 38 files covered. (35.1%)

977 existing lines in 19 files now uncovered.

3253 of 6702 relevant lines covered (48.54%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.2

/iblrig/video.py

import argparse
import contextlib
import logging
import os
import subprocess
import sys
import zipfile
from pathlib import Path
from urllib.error import URLError

import yaml

from ibllib.io.raw_data_loaders import load_embedded_frame_data
from ibllib.io.video import get_video_meta, label_from_path
from ibllib.pipes.misc import load_params_dict
from iblrig.base_tasks import EmptySession
from iblrig.constants import HARDWARE_SETTINGS_YAML, HAS_PYSPIN, HAS_SPINNAKER, RIG_SETTINGS_YAML
from iblrig.path_helper import load_pydantic_yaml, patch_settings
from iblrig.pydantic_definitions import HardwareSettings
from iblrig.tools import ask_user, call_bonsai
from iblrig.transfer_experiments import VideoCopier
from iblutil.io import (
    hashfile,  # type: ignore
    params,
)
from iblutil.util import setup_logger
from one.converters import ConversionMixin
from one.webclient import AlyxClient, http_download_file  # type: ignore

with contextlib.suppress(ImportError):
    from iblrig import video_pyspin

SPINNAKER_ASSET = 59586
SPINNAKER_FILENAME = 'SpinnakerSDK_FULL_3.2.0.57_x64.exe'
SPINNAKER_MD5 = 'aafc07c858dc2ab2e2a7d6ef900ca9a7'

PYSPIN_ASSET = 59584111
PYSPIN_FILENAME = 'spinnaker_python-3.2.0.57-cp310-cp310-win_amd64.zip'
PYSPIN_MD5 = 'f93294208e0ecec042adb2f75cb72609'

log = logging.getLogger(__name__)


def _download_from_alyx_or_flir(asset: int, filename: str, target_md5: str) -> Path:
    """
    Download a file from Alyx or FLIR server and verify its integrity using MD5 checksum.

    Parameters
    ----------
    asset : int
        The asset identifier for the file on FLIR server.
    filename : str
        The name of the file to be downloaded.
    target_md5 : str
        The expected MD5 checksum value for the downloaded file.

    Returns
    -------
    Path
        The path to the downloaded file.

    Raises
    ------
    Exception
        If the downloaded file's MD5 checksum does not match the expected value.
    """
    print(f'Downloading {filename} ...')
    out_dir = Path.home().joinpath('Downloads')
    out_file = out_dir.joinpath(filename)
    options = {'target_dir': out_dir, 'clobber': True, 'return_md5': True}
    if out_file.exists() and hashfile.md5(out_file) == target_md5:
        return out_file
    try:
        tmp_file, md5_sum = AlyxClient().download_file(f'resources/spinnaker/{filename}', **options)
    except (OSError, AttributeError, URLError) as e1:
        try:
            url = f'https://flir.netx.net/file/asset/{asset}/original/attachment'
            tmp_file, md5_sum = http_download_file(url, **options)
        except OSError as e2:
            raise e2 from e1
    os.rename(tmp_file, out_file)
    if md5_sum != target_md5:
        raise Exception(f'`{filename}` does not match the expected MD5')
    return out_file


def install_spinnaker():
    """
    Install the Spinnaker SDK for Windows.

    Raises
    ------
    Exception
        If the function is not run on Windows.
    """

    # Check prerequisites
    if os.name != 'nt':
        raise Exception('install_spinnaker can only be run on Windows.')

    # Display some information
    print('This script will try to automatically download & install Spinnaker SDK for Windows')
    input('Press [ENTER] to continue.\n')

    # Check for existing installation
    if HAS_SPINNAKER and not ask_user('Spinnaker SDK for Windows is already installed. Do you want to continue anyways?'):
        return

    # Download & install Spinnaker SDK
    file_winsdk = _download_from_alyx_or_flir(SPINNAKER_ASSET, SPINNAKER_FILENAME, SPINNAKER_MD5)
    print('Installing Spinnaker SDK for Windows ...')
    input(
        'Please select the "Application Development" Installation Profile. Everything else can be left at '
        'default values. Press [ENTER] to continue.'
    )
    return_code = subprocess.check_call(file_winsdk)
    if return_code == 0:
        print('Installation of Spinnaker SDK was successful.')
    os.unlink(file_winsdk)


def install_pyspin():
    """
    Install PySpin to the IBLRIG Python environment.

    Raises
    ------
    Exception
        If the function is not run on Windows.
        If the function is not started in the IBLRIG virtual environment.
    """

    # Check prerequisites
    if os.name != 'nt':
        raise Exception('install_pyspin can only be run on Windows.')
    if sys.base_prefix == sys.prefix:
        raise Exception('install_pyspin needs to be started in the IBLRIG venv.')

    # Display some information
    print('This script will try to automatically download & install PySpin to the IBLRIG Python environment')
    input('Press [ENTER] to continue.\n')

    # Download & install PySpin
    if HAS_PYSPIN:
        print('PySpin is already installed.')
    else:
        file_zip = _download_from_alyx_or_flir(PYSPIN_ASSET, PYSPIN_FILENAME, PYSPIN_MD5)
        print('Installing PySpin ...')
        with zipfile.ZipFile(file_zip, 'r') as f:
            file_whl = f.extract(file_zip.stem + '.whl', file_zip.parent)
        return_code = subprocess.check_call([sys.executable, '-m', 'pip', 'install', file_whl])
        if return_code == 0:
            print('Installation of PySpin was successful.')
        os.unlink(file_whl)
        file_zip.unlink()


def patch_old_params(remove_old=False, update_paths=True):
    """
    Update old video parameters.

    Parameters
    ----------
    remove_old : bool
        If true, removes the old video pc settings file.
    update_paths : bool
        If true, replace data paths in iblrig settings with those in old video pc settings file.

    """
    if not (old_file := Path(params.getfile('videopc_params'))).exists():
        return
    old_settings = load_params_dict('videopc_params')

    # Update hardware settings
    if HARDWARE_SETTINGS_YAML.exists():
        with open(HARDWARE_SETTINGS_YAML) as fp:
            hardware_settings = patch_settings(yaml.safe_load(fp), HARDWARE_SETTINGS_YAML)
    else:
        hardware_settings = {}
    cams = hardware_settings.get('device_cameras', {})
    for v in cams.values():
        for cam in filter(lambda k: k in v, ('left', 'right', 'body')):
            v[cam]['INDEX'] = old_settings.get(cam.upper() + '_CAM_IDX')

    # Save hardware settings
    hardware_settings['device_cameras'] = cams
    log.debug('Saving %s', HARDWARE_SETTINGS_YAML)
    with open(HARDWARE_SETTINGS_YAML, 'w') as fp:
        yaml.safe_dump(hardware_settings, fp)

    # Update other settings
    if update_paths:
        if RIG_SETTINGS_YAML.exists():
            with open(RIG_SETTINGS_YAML) as fp:
                rig_settings = yaml.safe_load(fp)
        else:
            rig_settings = {}
        path_map = {'iblrig_local_data_path': 'DATA_FOLDER_PATH', 'iblrig_remote_data_path': 'REMOTE_DATA_FOLDER_PATH'}
        for new_key, old_key in path_map.items():
            rig_settings[new_key] = old_settings[old_key].rstrip('\\')
            if rig_settings[new_key].endswith(r'\Subjects'):
                rig_settings[new_key] = rig_settings[new_key][: -len(r'\Subjects')]
            else:  # Add a 'subjects' key so that '\Subjects' is not incorrectly appended
                rig_settings[new_key.replace('data', 'subjects')] = rig_settings[new_key]
        log.debug('Saving %s', RIG_SETTINGS_YAML)
        with open(RIG_SETTINGS_YAML, 'w') as fp:
            yaml.safe_dump(rig_settings, fp)

    if remove_old:
        # Deleting old file
        log.info('Removing %s', old_file)
        old_file.unlink()


def prepare_video_session_cmd():
    if not HAS_SPINNAKER:
        if ask_user("Spinnaker SDK doesn't seem to be installed. Do you want to install it now?"):
            install_spinnaker()
        return
    if not HAS_PYSPIN:
        if ask_user("PySpin doesn't seem to be installed. Do you want to install it now?"):
            install_pyspin()
        return

    parser = argparse.ArgumentParser(prog='start_video_session', description='Prepare video PC for video recording session.')
    parser.add_argument('subject_name', help='name of subject')
    parser.add_argument('profile', help='camera configuration name, found in "device_cameras" map of hardware_settings.yaml')
    parser.add_argument('--debug', action='store_true', help='enable debugging mode')
    args = parser.parse_args()
    setup_logger(name='iblrig', level='DEBUG' if args.debug else 'INFO')
    prepare_video_session(args.subject_name, args.profile, debug=args.debug)


def validate_video_cmd():
    parser = argparse.ArgumentParser(prog='validate_video', description='Validate video session.')
    parser.add_argument('video_path', help='Path to the video file', type=str)
    parser.add_argument(
        'configuration', help='name of the configuration (default: default)', nargs='?', default='default', type=str
    )
    parser.add_argument('camera_name', help='name of the camera (default: left)', nargs='?', default='left', type=str)
    args = parser.parse_args()

    hwsettings: HardwareSettings = load_pydantic_yaml(HardwareSettings)
    file_path = Path(args.video_path)
    configuration = hwsettings.device_cameras.get(args.configuration, None)
    camera = configuration.get(args.camera_name, None) if configuration is not None else None

    if not file_path.exists():
        print(f'File not found: {file_path}')
    elif not file_path.is_file() or file_path.suffix != '.avi':
        print(f'Not a video file: {file_path}')
    elif configuration is None:
        print(f'No such configuration: {configuration}')
    elif configuration is None:
        print(f'No such camera: {camera}')
    else:
        validate_video(video_path=file_path, config=camera)


def validate_video(video_path, config):
    """
    Check raw video file saved as expected.

    Parameters
    ----------
    video_path : pathlib.Path
        Path to the video file.
    config : iblrig.pydantic_definitions.HardwareSettingsCamera
        The expected video configuration.

    Returns
    -------
    bool
        True if all checks pass.
    """
    ref = ConversionMixin.path2ref(video_path, as_dict=False)
    log.info('Checking %s camera for session %s', label_from_path(video_path), ref)
    if not video_path.exists():
        log.critical('Raw video file does not exist: %s', video_path)
        return False
    elif video_path.stat().st_size == 0:
        log.critical('Raw video file empty: %s', video_path)
        return False
    try:
        meta = get_video_meta(video_path)
        duration = meta.duration.total_seconds()
        ok = meta.length > 0 and duration > 0.0
        log.log(20 if meta.length > 0 else 40, 'N frames = %i', meta.length)
        log.log(20 if duration > 0 else 40, 'Duration = %.2f', duration)
        if config.HEIGHT and meta.height != config.HEIGHT:
            ok = False
            log.warning('Frame height = %i; expected %i', config.HEIGHT, meta.height)
        if config.WIDTH and meta.width != config.WIDTH:
            log.warning('Frame width = %i; expected %i', config.WIDTH, meta.width)
            ok = False
        if config.FPS and meta.fps != config.FPS:
            log.warning('Frame rate = %i; expected %i', config.FPS, meta.fps)
            ok = False
    except AssertionError:
        log.critical('Failed to open video file: %s', video_path)
        return False

    # Check frame data
    count, gpio = load_embedded_frame_data(video_path.parents[1], label_from_path(video_path))
    dropped = count[-1] - (meta.length - 1)
    if dropped != 0:  # Log ERROR if > .1% frames dropped, otherwise log WARN
        pct_dropped = dropped / (count[-1] + 1) * 100
        level = 30 if pct_dropped < 0.1 else 40
        log.log(level, 'Missed frames (%.2f%%) - frame data N = %i; video file N = %i', pct_dropped, count[-1] + 1, meta.length)
        ok = False
    if len(count) != meta.length:
        log.critical('Frame count / video frame mismatch - frame counts = %i; video frames = %i', len(count), meta.length)
        ok = False
    if config.SYNC_LABEL:
        min_events = 10  # The minimum expected number of GPIO events
        if all(ch is None for ch in gpio):
            log.error('No GPIO events detected.')
            ok = False
        else:
            for i, ch in enumerate(gpio):
                if ch:
                    log.log(30 if len(ch['indices']) < min_events else 20, '%i event(s) on GPIO #%i', len(ch['indices']), i + 1)
    return ok


def prepare_video_session(subject_name: str, config_name: str, debug: bool = False):
    """
    Setup and record video.

    Parameters
    ----------
    subject_name : str
        A subject name.
    config_name : str
        Camera configuration name, found in "device_cameras" map of hardware_settings.yaml.
    debug : bool
        Bonsai debug mode and verbose logging.
    """
    assert HAS_SPINNAKER
    assert HAS_PYSPIN

    # Initialize a session for paths and settings
    session = EmptySession(subject=subject_name, interactive=False)
    session_path = session.paths.SESSION_FOLDER
    raw_data_folder = session_path.joinpath('raw_video_data')

    # Fetch camera configuration from hardware settings file
    try:
        config = session.hardware_settings.device_cameras[config_name]
    except AttributeError as ex:
        if hasattr(value_error := ValueError('"No camera config in hardware_settings.yaml file."'), 'add_note'):
            value_error.add_note(HARDWARE_SETTINGS_YAML)  # py 3.11
        raise value_error from ex
    except KeyError as ex:
        raise ValueError(f'Config "{config_name}" not in "device_cameras" hardware settings.') from ex
    workflows = config.pop('BONSAI_WORKFLOW')
    cameras = [k for k in config if k != 'BONSAI_WORKFLOW']
    params = {f'{k.capitalize()}CameraIndex': config[k].INDEX for k in cameras}
    raw_data_folder.mkdir(parents=True, exist_ok=True)

    # align cameras
    if workflows.setup:
        video_pyspin.enable_camera_trigger(enable=False)
        call_bonsai(workflows.setup, params, debug=debug)

    # record video
    filenamevideo = '_iblrig_{}Camera.raw.avi'
    filenameframedata = '_iblrig_{}Camera.frameData.bin'
    for k in map(str.capitalize, cameras):
        params[f'FileName{k}'] = str(raw_data_folder / filenamevideo.format(k.lower()))
        params[f'FileName{k}Data'] = str(raw_data_folder / filenameframedata.format(k.lower()))
    video_pyspin.enable_camera_trigger(enable=True)
    bonsai_process = call_bonsai(workflows.recording, params, wait=False, debug=debug)
    input('PRESS ENTER TO START CAMERAS')
    # Save the stub files locally and in the remote repo for future copy script to use
    copier = VideoCopier(session_path=session_path, remote_subjects_folder=session.paths.REMOTE_SUBJECT_FOLDER)
    copier.initialize_experiment(acquisition_description=copier.config2stub(config, raw_data_folder.name))

    video_pyspin.enable_camera_trigger(enable=False)
    log.info('To terminate video acquisition, please stop and close Bonsai workflow.')
    bonsai_process.wait()
    log.info('Video acquisition session finished.')

    # Check video files were saved and configured correctly
    for video_file in (Path(v) for v in params.values() if isinstance(v, str) and v.endswith('.avi')):
        validate_video(video_file, config[label_from_path(video_file)])

    session_path.joinpath('transfer_me.flag').touch()
    # remove empty-folders and parent-folders
    if not any(raw_data_folder.iterdir()):
        os.removedirs(raw_data_folder)

1	import argparse	2✔
2	import contextlib	2✔
3	import logging	2✔
4	import os	2✔
5	import subprocess	2✔
6	import sys	2✔
7	import zipfile	2✔
8	from pathlib import Path	2✔
9	from urllib.error import URLError	2✔
10
11	import yaml	2✔
12
13	from ibllib.io.raw_data_loaders import load_embedded_frame_data	2✔
14	from ibllib.io.video import get_video_meta, label_from_path	2✔
15	from ibllib.pipes.misc import load_params_dict	2✔
16	from iblrig.base_tasks import EmptySession	2✔
17	from iblrig.constants import HARDWARE_SETTINGS_YAML, HAS_PYSPIN, HAS_SPINNAKER, RIG_SETTINGS_YAML	2✔
18	from iblrig.path_helper import load_pydantic_yaml, patch_settings	2✔
19	from iblrig.pydantic_definitions import HardwareSettings	2✔
20	from iblrig.tools import ask_user, call_bonsai	2✔
21	from iblrig.transfer_experiments import VideoCopier	2✔
22	from iblutil.io import (	2✔
23	hashfile, # type: ignore
24	params,
25	)
26	from iblutil.util import setup_logger	2✔
27	from one.converters import ConversionMixin	2✔
28	from one.webclient import AlyxClient, http_download_file # type: ignore	2✔
29
30	with contextlib.suppress(ImportError):	2✔
31	from iblrig import video_pyspin	2✔
32
33	SPINNAKER_ASSET = 59586	2✔
34	SPINNAKER_FILENAME = 'SpinnakerSDK_FULL_3.2.0.57_x64.exe'	2✔
35	SPINNAKER_MD5 = 'aafc07c858dc2ab2e2a7d6ef900ca9a7'	2✔
36
37	PYSPIN_ASSET = 59584111	2✔
38	PYSPIN_FILENAME = 'spinnaker_python-3.2.0.57-cp310-cp310-win_amd64.zip'	2✔
39	PYSPIN_MD5 = 'f93294208e0ecec042adb2f75cb72609'	2✔
40
41	log = logging.getLogger(__name__)	2✔
42
43
44	def _download_from_alyx_or_flir(asset: int, filename: str, target_md5: str) -> Path:	2✔
45	"""
46	Download a file from Alyx or FLIR server and verify its integrity using MD5 checksum.
47
48	Parameters
49	----------
50	asset : int
51	The asset identifier for the file on FLIR server.
52	filename : str
53	The name of the file to be downloaded.
54	target_md5 : str
55	The expected MD5 checksum value for the downloaded file.
56
57	Returns
58	-------
59	Path
60	The path to the downloaded file.
61
62	Raises
63	------
64	Exception
65	If the downloaded file's MD5 checksum does not match the expected value.
66	"""
67	print(f'Downloading {filename} ...')	2✔
68	out_dir = Path.home().joinpath('Downloads')	2✔
69	out_file = out_dir.joinpath(filename)	2✔
70	options = {'target_dir': out_dir, 'clobber': True, 'return_md5': True}	2✔
71	if out_file.exists() and hashfile.md5(out_file) == target_md5:	2✔
72	return out_file	×
73	try:	2✔
74	tmp_file, md5_sum = AlyxClient().download_file(f'resources/spinnaker/{filename}', **options)	2✔
75	except (OSError, AttributeError, URLError) as e1:	×
76	try:	×
77	url = f'https://flir.netx.net/file/asset/{asset}/original/attachment'	×
78	tmp_file, md5_sum = http_download_file(url, **options)	×
79	except OSError as e2:	×
80	raise e2 from e1	×
81	os.rename(tmp_file, out_file)	2✔
82	if md5_sum != target_md5:	2✔
NEW 83	raise Exception(f'`{filename}` does not match the expected MD5')	×
84	return out_file	2✔
85
86
87	def install_spinnaker():	2✔
88	"""
89	Install the Spinnaker SDK for Windows.
90
91	Raises
92	------
93	Exception
94	If the function is not run on Windows.
95	"""
96
97	# Check prerequisites
98	if os.name != 'nt':	×
99	raise Exception('install_spinnaker can only be run on Windows.')	×
100
101	# Display some information
102	print('This script will try to automatically download & install Spinnaker SDK for Windows')	×
103	input('Press [ENTER] to continue.\n')	×
104
105	# Check for existing installation
106	if HAS_SPINNAKER and not ask_user('Spinnaker SDK for Windows is already installed. Do you want to continue anyways?'):	×
107	return	×
108
109	# Download & install Spinnaker SDK
NEW 110	file_winsdk = _download_from_alyx_or_flir(SPINNAKER_ASSET, SPINNAKER_FILENAME, SPINNAKER_MD5)	×
111	print('Installing Spinnaker SDK for Windows ...')	×
112	input(	×
113	'Please select the "Application Development" Installation Profile. Everything else can be left at '
114	'default values. Press [ENTER] to continue.'
115	)
116	return_code = subprocess.check_call(file_winsdk)	×
117	if return_code == 0:	×
118	print('Installation of Spinnaker SDK was successful.')	×
119	os.unlink(file_winsdk)	×
120
121
122	def install_pyspin():	2✔
123	"""
124	Install PySpin to the IBLRIG Python environment.
125
126	Raises
127	------
128	Exception
129	If the function is not run on Windows.
130	If the function is not started in the IBLRIG virtual environment.
131	"""
132
133	# Check prerequisites
134	if os.name != 'nt':	×
135	raise Exception('install_pyspin can only be run on Windows.')	×
136	if sys.base_prefix == sys.prefix:	×
137	raise Exception('install_pyspin needs to be started in the IBLRIG venv.')	×
138
139	# Display some information
140	print('This script will try to automatically download & install PySpin to the IBLRIG Python environment')	×
141	input('Press [ENTER] to continue.\n')	×
142
143	# Download & install PySpin
144	if HAS_PYSPIN:	×
145	print('PySpin is already installed.')	×
146	else:
NEW 147	file_zip = _download_from_alyx_or_flir(PYSPIN_ASSET, PYSPIN_FILENAME, PYSPIN_MD5)	×
148	print('Installing PySpin ...')	×
149	with zipfile.ZipFile(file_zip, 'r') as f:	×
150	file_whl = f.extract(file_zip.stem + '.whl', file_zip.parent)	×
151	return_code = subprocess.check_call([sys.executable, '-m', 'pip', 'install', file_whl])	×
152	if return_code == 0:	×
153	print('Installation of PySpin was successful.')	×
154	os.unlink(file_whl)	×
155	file_zip.unlink()	×
156
157
158	def patch_old_params(remove_old=False, update_paths=True):	2✔
159	"""
160	Update old video parameters.
161
162	Parameters
163	----------
164	remove_old : bool
165	If true, removes the old video pc settings file.
166	update_paths : bool
167	If true, replace data paths in iblrig settings with those in old video pc settings file.
168
169	"""
170	if not (old_file := Path(params.getfile('videopc_params'))).exists():	2✔
171	return	2✔
172	old_settings = load_params_dict('videopc_params')	2✔
173
174	# Update hardware settings
175	if HARDWARE_SETTINGS_YAML.exists():	2✔
176	with open(HARDWARE_SETTINGS_YAML) as fp:	2✔
177	hardware_settings = patch_settings(yaml.safe_load(fp), HARDWARE_SETTINGS_YAML)	2✔
178	else:
179	hardware_settings = {}	2✔
180	cams = hardware_settings.get('device_cameras', {})	2✔
181	for v in cams.values():	2✔
182	for cam in filter(lambda k: k in v, ('left', 'right', 'body')):	2✔
183	v[cam]['INDEX'] = old_settings.get(cam.upper() + '_CAM_IDX')	2✔
184
185	# Save hardware settings
186	hardware_settings['device_cameras'] = cams	2✔
187	log.debug('Saving %s', HARDWARE_SETTINGS_YAML)	2✔
188	with open(HARDWARE_SETTINGS_YAML, 'w') as fp:	2✔
189	yaml.safe_dump(hardware_settings, fp)	2✔
190
191	# Update other settings
192	if update_paths:	2✔
193	if RIG_SETTINGS_YAML.exists():	2✔
194	with open(RIG_SETTINGS_YAML) as fp:	2✔
195	rig_settings = yaml.safe_load(fp)	2✔
196	else:
197	rig_settings = {}	×
198	path_map = {'iblrig_local_data_path': 'DATA_FOLDER_PATH', 'iblrig_remote_data_path': 'REMOTE_DATA_FOLDER_PATH'}	2✔
199	for new_key, old_key in path_map.items():	2✔
200	rig_settings[new_key] = old_settings[old_key].rstrip('\\')	2✔
201	if rig_settings[new_key].endswith(r'\Subjects'):	2✔
202	rig_settings[new_key] = rig_settings[new_key][: -len(r'\Subjects')]	2✔
203	else: # Add a 'subjects' key so that '\Subjects' is not incorrectly appended
204	rig_settings[new_key.replace('data', 'subjects')] = rig_settings[new_key]	2✔
205	log.debug('Saving %s', RIG_SETTINGS_YAML)	2✔
206	with open(RIG_SETTINGS_YAML, 'w') as fp:	2✔
207	yaml.safe_dump(rig_settings, fp)	2✔
208
209	if remove_old:	2✔
210	# Deleting old file
211	log.info('Removing %s', old_file)	2✔
212	old_file.unlink()	2✔
213
214
215	def prepare_video_session_cmd():	2✔
216	if not HAS_SPINNAKER:	×
217	if ask_user("Spinnaker SDK doesn't seem to be installed. Do you want to install it now?"):	×
218	install_spinnaker()	×
219	return	×
220	if not HAS_PYSPIN:	×
221	if ask_user("PySpin doesn't seem to be installed. Do you want to install it now?"):	×
222	install_pyspin()	×
223	return	×
224
225	parser = argparse.ArgumentParser(prog='start_video_session', description='Prepare video PC for video recording session.')	×
226	parser.add_argument('subject_name', help='name of subject')	×
227	parser.add_argument('profile', help='camera configuration name, found in "device_cameras" map of hardware_settings.yaml')	×
228	parser.add_argument('--debug', action='store_true', help='enable debugging mode')	×
229	args = parser.parse_args()	×
230	setup_logger(name='iblrig', level='DEBUG' if args.debug else 'INFO')	×
231	prepare_video_session(args.subject_name, args.profile, debug=args.debug)	×
232
233
234	def validate_video_cmd():	2✔
NEW 235	parser = argparse.ArgumentParser(prog='validate_video', description='Validate video session.')	×
NEW 236	parser.add_argument('video_path', help='Path to the video file', type=str)	×
NEW 237	parser.add_argument(	×
238	'configuration', help='name of the configuration (default: default)', nargs='?', default='default', type=str
239	)
NEW 240	parser.add_argument('camera_name', help='name of the camera (default: left)', nargs='?', default='left', type=str)	×
NEW 241	args = parser.parse_args()	×
242
NEW 243	hwsettings: HardwareSettings = load_pydantic_yaml(HardwareSettings)	×
NEW 244	file_path = Path(args.video_path)	×
NEW 245	configuration = hwsettings.device_cameras.get(args.configuration, None)	×
NEW 246	camera = configuration.get(args.camera_name, None) if configuration is not None else None	×
247
NEW 248	if not file_path.exists():	×
NEW 249	print(f'File not found: {file_path}')	×
NEW 250	elif not file_path.is_file() or file_path.suffix != '.avi':	×
NEW 251	print(f'Not a video file: {file_path}')	×
NEW 252	elif configuration is None:	×
NEW 253	print(f'No such configuration: {configuration}')	×
NEW 254	elif configuration is None:	×
NEW 255	print(f'No such camera: {camera}')	×
256	else:
NEW 257	validate_video(video_path=file_path, config=camera)	×
258
259
260	def validate_video(video_path, config):	2✔
261	"""
262	Check raw video file saved as expected.
263
264	Parameters
265	----------
266	video_path : pathlib.Path
267	Path to the video file.
268	config : iblrig.pydantic_definitions.HardwareSettingsCamera
269	The expected video configuration.
270
271	Returns
272	-------
273	bool
274	True if all checks pass.
275	"""
276	ref = ConversionMixin.path2ref(video_path, as_dict=False)	2✔
277	log.info('Checking %s camera for session %s', label_from_path(video_path), ref)	2✔
278	if not video_path.exists():	2✔
279	log.critical('Raw video file does not exist: %s', video_path)	2✔
280	return False	2✔
281	elif video_path.stat().st_size == 0:	2✔
282	log.critical('Raw video file empty: %s', video_path)	2✔
283	return False	2✔
284	try:	2✔
285	meta = get_video_meta(video_path)	2✔
286	duration = meta.duration.total_seconds()	2✔
287	ok = meta.length > 0 and duration > 0.0	2✔
288	log.log(20 if meta.length > 0 else 40, 'N frames = %i', meta.length)	2✔
289	log.log(20 if duration > 0 else 40, 'Duration = %.2f', duration)	2✔
290	if config.HEIGHT and meta.height != config.HEIGHT:	2✔
291	ok = False	2✔
292	log.warning('Frame height = %i; expected %i', config.HEIGHT, meta.height)	2✔
293	if config.WIDTH and meta.width != config.WIDTH:	2✔
294	log.warning('Frame width = %i; expected %i', config.WIDTH, meta.width)	2✔
295	ok = False	2✔
296	if config.FPS and meta.fps != config.FPS:	2✔
297	log.warning('Frame rate = %i; expected %i', config.FPS, meta.fps)	2✔
298	ok = False	2✔
299	except AssertionError:	2✔
300	log.critical('Failed to open video file: %s', video_path)	2✔
301	return False	2✔
302
303	# Check frame data
304	count, gpio = load_embedded_frame_data(video_path.parents[1], label_from_path(video_path))	2✔
305	dropped = count[-1] - (meta.length - 1)	2✔
306	if dropped != 0: # Log ERROR if > .1% frames dropped, otherwise log WARN	2✔
307	pct_dropped = dropped / (count[-1] + 1) * 100	2✔
308	level = 30 if pct_dropped < 0.1 else 40	2✔
309	log.log(level, 'Missed frames (%.2f%%) - frame data N = %i; video file N = %i', pct_dropped, count[-1] + 1, meta.length)	2✔
310	ok = False	2✔
311	if len(count) != meta.length:	2✔
312	log.critical('Frame count / video frame mismatch - frame counts = %i; video frames = %i', len(count), meta.length)	2✔
313	ok = False	2✔
314	if config.SYNC_LABEL:	2✔
315	min_events = 10 # The minimum expected number of GPIO events	2✔
316	if all(ch is None for ch in gpio):	2✔
317	log.error('No GPIO events detected.')	2✔
318	ok = False	2✔
319	else:
320	for i, ch in enumerate(gpio):	2✔
321	if ch:	2✔
322	log.log(30 if len(ch['indices']) < min_events else 20, '%i event(s) on GPIO #%i', len(ch['indices']), i + 1)	2✔
323	return ok	2✔
324
325
326	def prepare_video_session(subject_name: str, config_name: str, debug: bool = False):	2✔
327	"""
328	Setup and record video.
329
330	Parameters
331	----------
332	subject_name : str
333	A subject name.
334	config_name : str
335	Camera configuration name, found in "device_cameras" map of hardware_settings.yaml.
336	debug : bool
337	Bonsai debug mode and verbose logging.
338	"""
339	assert HAS_SPINNAKER	2✔
340	assert HAS_PYSPIN	2✔
341
342	# Initialize a session for paths and settings
343	session = EmptySession(subject=subject_name, interactive=False)	2✔
344	session_path = session.paths.SESSION_FOLDER	2✔
345	raw_data_folder = session_path.joinpath('raw_video_data')	2✔
346
347	# Fetch camera configuration from hardware settings file
348	try:	2✔
349	config = session.hardware_settings.device_cameras[config_name]	2✔
350	except AttributeError as ex:	2✔
351	if hasattr(value_error := ValueError('"No camera config in hardware_settings.yaml file."'), 'add_note'):	2✔
352	value_error.add_note(HARDWARE_SETTINGS_YAML) # py 3.11	×
353	raise value_error from ex	2✔
354	except KeyError as ex:	2✔
355	raise ValueError(f'Config "{config_name}" not in "device_cameras" hardware settings.') from ex	2✔
356	workflows = config.pop('BONSAI_WORKFLOW')	2✔
357	cameras = [k for k in config if k != 'BONSAI_WORKFLOW']	2✔
358	params = {f'{k.capitalize()}CameraIndex': config[k].INDEX for k in cameras}	2✔
359	raw_data_folder.mkdir(parents=True, exist_ok=True)	2✔
360
361	# align cameras
362	if workflows.setup:	2✔
363	video_pyspin.enable_camera_trigger(enable=False)	2✔
364	call_bonsai(workflows.setup, params, debug=debug)	2✔
365
366	# record video
367	filenamevideo = '_iblrig_{}Camera.raw.avi'	2✔
368	filenameframedata = '_iblrig_{}Camera.frameData.bin'	2✔
369	for k in map(str.capitalize, cameras):	2✔
370	params[f'FileName{k}'] = str(raw_data_folder / filenamevideo.format(k.lower()))	2✔
371	params[f'FileName{k}Data'] = str(raw_data_folder / filenameframedata.format(k.lower()))	2✔
372	video_pyspin.enable_camera_trigger(enable=True)	2✔
373	bonsai_process = call_bonsai(workflows.recording, params, wait=False, debug=debug)	2✔
374	input('PRESS ENTER TO START CAMERAS')	2✔
375	# Save the stub files locally and in the remote repo for future copy script to use
376	copier = VideoCopier(session_path=session_path, remote_subjects_folder=session.paths.REMOTE_SUBJECT_FOLDER)	2✔
377	copier.initialize_experiment(acquisition_description=copier.config2stub(config, raw_data_folder.name))	2✔
378
379	video_pyspin.enable_camera_trigger(enable=False)	2✔
380	log.info('To terminate video acquisition, please stop and close Bonsai workflow.')	2✔
381	bonsai_process.wait()	2✔
382	log.info('Video acquisition session finished.')	2✔
383
384	# Check video files were saved and configured correctly
385	for video_file in (Path(v) for v in params.values() if isinstance(v, str) and v.endswith('.avi')):	2✔
386	validate_video(video_file, config[label_from_path(video_file)])	2✔
387
388	session_path.joinpath('transfer_me.flag').touch()	2✔
389	# remove empty-folders and parent-folders
390	if not any(raw_data_folder.iterdir()):	2✔
391	os.removedirs(raw_data_folder)	2✔

int-brain-lab / iblrig / 9032957364

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous