International Brain Laboratory

https://docs.internationalbrainlab.org
Admin
International Brain Laboratory is a global neuroscience collaboration providing an open software...
Tokens:1,368,233
Snippets:7,093
Trust Score:7.5
Update:4 weeks ago
Show doc for...
Tokens:
Raw
### Module docstring setup and example usage

Source: https://docs.internationalbrainlab.org/_modules/brainbox/metrics/single_units.html

Setup instructions and example usage for the module. It imports brainbox, one.alf.io, numpy, matplotlib, and ibllib.ephys.spikes, converts ks2 output to alf, loads spikes and clusters objects, and computes a units bunch.

```python
>>> import brainbox as bb
>>> import one.alf.io as aio
>>> import numpy as np
>>> import matplotlib.pyplot as plt
>>> import ibllib.ephys.spikes as e_spks
# (*Note, if there is no 'alf' directory, make 'alf' directory from 'ks2' output directory):
>>> e_spks.ks2_to_alf(path_to_ks_out, path_to_alf_out)
# Load the alf spikes bunch and clusters bunch, and get a units bunch.
>>> spks_b = aio.load_object(path_to_alf_out, 'spikes')
>>> clstrs_b = aio.load_object(path_to_alf_out, 'clusters')
>>> units_b = bb.processing.get_units_bunch(spks_b)  # may take a few mins to compute
```

--------------------------------

### setUp(**kwargs)

Source: https://docs.internationalbrainlab.org/_autosummary/ibllib.pipes.tasks.html

Gets the data handler and ensures all data is available locally to run the task.

```APIDOC
## setUp(**kwargs)

### Description
Get the data handler and ensure all data is available locally to run task.
```

--------------------------------

### Example entry point for creating a communicator

Source: https://docs.internationalbrainlab.org/_modules/iblutil/io/net/app.html

This async main function serves as an example entry point for creating a server or client communicator. It checks the role argument: for 'server', it starts an EchoProtocol server and serves for 1 hour if using UDP, otherwise serves forever; for 'client', it starts a client and sends a message. It raises a ValueError for unknown roles.

```python
async def main(role, server_uri, name=None, **kwargs):
    """An example of an entry point for creating an individual communicator."""
    if role == 'server':
        print('Starting server')
        com = await EchoProtocol.server(server_uri, name=name, **kwargs)
        try:
            if com.server_uri.startswith('udp'):
                await asyncio.sleep(60 * 60)  # Serve for 1 hour.
            else:
                await com.Server.serve_forever()
        finally:
            com.close()
    elif role == 'client':
        print('Starting client')
        com = await EchoProtocol.client(server_uri, name=name)
        try:
            # Here you would send a message
            await com.start('2022-01-01_1_subject')
        finally:
            com.close()

    else:
        raise ValueError(f'Unknown role "{role}"')
```

--------------------------------

### setUp: Prepare sync test with mocked video metadata

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/tasks/test_widefield_tasks.html

Sets up the sync test by locating the session path, creating an empty video file, mocking get_video_meta to return a video_meta with length 2032, and starting the mock. Initializes ONE with the test database.

```python
    def setUp(self):
        self.session_path = self.data_path.joinpath(
            'widefield', 'widefieldChoiceWorld', 'JC076', '2022-02-04', '002')
        if not self.session_path.exists():
            return
        self.alf_folder = self.session_path.joinpath('alf', 'widefield')
        self.video_file = self.session_path.joinpath('raw_widefield_data', 'imaging.frames.mov')
        self.video_file.touch()
        self.video_meta.length = 2032
        self.patch = unittest.mock.patch('ibllib.io.extractors.widefield.get_video_meta',
                                         return_value=self.video_meta)
        self.patch.start()
        self.one = ONE(**base.TEST_DB, mode='local')
```

--------------------------------

### Setup and data loading for brainbox.plot docstring examples

Source: https://docs.internationalbrainlab.org/_modules/brainbox/plot.html

Commands to set up the workspace for running docstring examples. Imports required modules, converts ks2 output to alf format, and loads spikes, clusters, and units bunches.

```python
>>> from brainbox import processing
>>> import one.alf.io as alfio
>>> import numpy as np
>>> import matplotlib.pyplot as plt
>>> import ibllib.ephys.spikes as e_spks
# (*Note, if there is no 'alf' directory, make 'alf' directory from 'ks2' output directory):
>>> e_spks.ks2_to_alf(path_to_ks_out, path_to_alf_out)
# Load the alf spikes bunch and clusters bunch, and get a units bunch.
>>> spks_b = alfio.load_object(path_to_alf_out, 'spikes')
>>> clstrs_b = alfio.load_object(path_to_alf_out, 'clusters')
>>> units_b = processing.get_units_bunch(spks_b)  # may take a few mins to compute
```

--------------------------------

### Test SDSCDataHandler setUp and cleanUp methods

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/test_oneibl.html

Creates a ChoiceWorldTrialsBpod task, gets signatures, instantiates SDSCDataHandler, adds files, calls setUp, verifies symlinks and session path, then calls cleanUp and checks links are removed.

```python
task = ChoiceWorldTrialsBpod(self.session_path, one=self.one, collection='raw_behavior_data')
task.get_signatures()
handler = handlers.SDSCDataHandler(self.session_path, task.signature, self.one)
handler.patch_path = self.patch_path
handler.root_path = self.root_path
# Add some files on disk to check they are symlinked by setUp method
for uid, rel_path in handler.getData().rel_path.items():
    filepath = self.session_path.joinpath(rel_path)
    filepath.parent.mkdir(exist_ok=True, parents=True)
    filepath.with_stem(f'{filepath.stem}.{uid}').touch()
# Check setUp does the symlinks and updates the task session path
handler.setUp(task=task)
expected = self.patch_path.joinpath('ChoiceWorldTrialsBpod', *self.session_path.parts[-3:])
self.assertEqual(expected, task.session_path, 'failed to update task session path')
linked = list(expected.glob('raw_behavior_data/*.*.*'))
self.assertTrue(all(f.is_symlink for f in linked), 'failed to sym link input files')
self.assertEqual(len(task.input_files), len(linked), 'unexpected number of linked patch files')
# Check all links to root session
session_path = self.session_path.resolve()  # NB: GitHub CI uses linked temp dir so we resolve here
self.assertTrue(all(f.resolve().is_relative_to(session_path) for f in linked))
# Check sym link doesn't contain UUID
self.assertTrue(len(linked[0].resolve().stem.split('.')[-1]) == 36, 'source path missing UUID')
self.assertFalse(len(linked[0].stem.split('.')[-1]) == 36, 'symlink contains UUID')
# Check cleanUp removes links
handler.cleanUp(task=task)
self.assertFalse(any(Path.exists, linked))
# Check no root files were deleted
self.assertEqual(len(task.input_files), len(list(self.session_path.glob('raw_behavior_data/*.*.*'))))
```

--------------------------------

### start(exp_ref, data=None, addr=None)

Source: https://docs.internationalbrainlab.org/_modules/iblutil/io/net/app.html

Starts an experiment by sending a start message to the remote host. Requires an experiment reference string and optionally extra data.

```APIDOC
## start(exp_ref, data=None, addr=None)

### Description
Start an experiment. Send a stop message to the remote host.

### Method
start

### Parameters
#### Required Parameters
- **exp_ref** (str) - Required - An experiment reference string in the form yyyy-mm-dd_n_subject.

#### Optional Parameters
- **data** (any) - Optional - Optional extra data to send to the remote host.
- **addr** ((str, int)) - Optional - The remote host address and port. Only required in server role.

### Raises
- **TimeoutError** - Remote host failed to echo the message within the timeout period.

```

--------------------------------

### Imports and test class setup for audio tasks

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/tasks/test_audio_tasks.html

Imports required modules and defines the TestAudioCompress class with setUp and tearDown methods. The setUp copies raw behavior data to a temporary session path and initializes a local ONE instance.

```python
import logging
import shutil
import tempfile
from pathlib import Path

from one.api import ONE
from ibllib.pipes.audio_tasks import AudioCompress, AudioSync

from ibllib.tests import base

_logger = logging.getLogger('ibllib')




[docs]
class TestAudioCompress(base.IntegrationTest):


[docs]
    def setUp(self) -> None:
        self.data_path = self.default_data_root().joinpath(
            'ephys', 'ephys_video_init', 'ZM_1735', '2019-08-01', '001', 'raw_behavior_data')
        self.temp_dir = Path(tempfile.TemporaryDirectory().name)
        self.session_path = self.temp_dir.joinpath('ZM_1735', '2019-08-01', '001')
        shutil.copytree(self.data_path, self.session_path.joinpath('raw_behavior_data'))
        self.one = ONE(**base.TEST_DB, mode='local')





[docs]
    def test_compress(self):
        task = AudioCompress(self.session_path, one=self.one, device_collection='raw_behavior_data')
        status = task.run()
        assert status == 0

        self.assertIsNone(next(self.session_path.rglob('*.wav'), None))
        self.assertTrue(next(self.session_path.rglob('*.flac')) == task.outputs[0])





[docs]
    def tearDown(self) -> None:
        shutil.rmtree(self.temp_dir)
```

--------------------------------

### TestVideoSyncQCBpod setUp method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/tasks/test_video_tasks.html

Sets up a temporary session directory by copying data from a test folder, initializes ONE with a local mode, and runs VideoCompress with the left camera.

```python
def setUp(self) -> None:
    self.folder_path = self.data_path.joinpath('Subjects_init', 'ZM_1085', '2019-02-12', '002')
    tmp = tempfile.TemporaryDirectory()
    self.addCleanup(tmp.cleanup)
    self.temp_dir = Path(tmp.name)
    self.session_path = self.temp_dir.joinpath('ZM_1085', '2019-02-12', '002')
    shutil.copytree(self.folder_path, self.session_path)
    self.one = ONE(**base.TEST_DB, mode='local')
    task = VideoCompress(self.session_path, one=self.one, device_collection='raw_video_data', cameras=['left'])
    task.run()
```

--------------------------------

### TestVideoSyncQcCamlog tearDown method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/tasks/test_video_tasks.html

Stops the patch started in setUp.

```python
def tearDown(self):
    self.patch.stop()
```

--------------------------------

### setUp: Initialize ONE and create session path

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/test_pipes.html

Sets up the test environment by initializing the ONE client with TEST_DB, creating a temporary directory, and creating a session path for the test subject.

```python
def setUp(self) -> None:
    self.one = ONE(**TEST_DB)
    self.tempdir = tempfile.TemporaryDirectory()
    self.addCleanup(self.tempdir.cleanup)
    self.session_path = Path(self.tempdir.name).joinpath('subject', '2023-01-01', '001')
    self.session_path.mkdir(parents=True)
```

--------------------------------

### TestVideoSyncQcCamlog setUp method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/tasks/test_video_tasks.html

Sets up a temporary session directory by copying raw video, sync, and behavior data, touches a placeholder MP4 file, and patches get_video_length to return a fixed video length.

```python
def setUp(self) -> None:
    data_path = self.data_path.joinpath('widefield', 'widefieldChoiceWorld', 'FD_01', '2022-08-04', '001')
    tmp = tempfile.TemporaryDirectory()
    self.addCleanup(tmp.cleanup)
    self.temp_dir = Path(tmp.name)
    self.session_path = self.temp_dir.joinpath('FD_01', '2022-08-04', '002')
    shutil.copytree(data_path.joinpath('raw_video_data'), self.session_path.joinpath('raw_video_data'))
    shutil.copytree(data_path.joinpath('raw_sync_data'), self.session_path.joinpath('raw_sync_data'))
    shutil.copytree(data_path.joinpath('raw_behavior_data'), self.session_path.joinpath('raw_behavior_data'))
    self.session_path.joinpath('raw_video_data', '_iblrig_leftCamera.raw.mp4').touch()
    self.video_length = 244162

    self.patch = unittest.mock.patch('ibllib.io.extractors.camera.get_video_length',
                                     return_value=self.video_length)
    self.patch.start()
    self.one = ONE(**base.TEST_DB, mode='local')
```

--------------------------------

### TaskQCTestData setUp method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/qc/test_task_metrics.html

Sets up test data by loading fake bpod data and wheel data, and initializing the wheel gain.

```python
[docs]
class TaskQCTestData(unittest.TestCase):


[docs]
    def setUp(self):
        self.data = self.load_fake_bpod_data()
        self.wheel_gain = 4
        wheel_data = self.load_fake_wheel_data(self.data, wheel_gain=self.wheel_gain)
        self.data.update(wheel_data)
```

--------------------------------

### TestVideoSyncQCNidq setUp method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/tasks/test_video_tasks.html

Sets up a temporary session directory by symlinking files from a source folder, skipping ALF files that are not DLC or ROIMotionEnergy.

```python
def setUp(self) -> None:

    self.folder_path = self.data_path.joinpath('ephys', 'choice_world_init', 'KS022', '2019-12-10', '001')
    tmp = tempfile.TemporaryDirectory()
    self.addCleanup(tmp.cleanup)
    self.temp_dir = Path(tmp.name)
    self.session_path = self.temp_dir.joinpath('KS022', '2019-12-10', '001')
    self.one = ONE(**base.TEST_DB, mode='local')

    for ff in self.folder_path.rglob('*.*'):
        link = self.session_path.joinpath(ff.relative_to(self.folder_path))
        if 'alf' in link.parts and not ('dlc' in link.name or 'ROIMotionEnergy' in link.name):
            continue
        link.parent.mkdir(exist_ok=True, parents=True)
        link.symlink_to(ff)
```

--------------------------------

### Get next training status

Source: https://docs.internationalbrainlab.org/_autosummary/brainbox.behavior.training.html

Shows how to get the next training status after a given status by sorting the enum members and selecting the first one greater than the current status. The example uses a generator expression with the sorted() function and the > operator.

```python
>>> next(member for member in sorted(TrainingStatus) if member > TrainingStatus[status.upper()])
<TrainingStatus.READY4RECORDING: 128>

```

--------------------------------

### Set-up workspace

Source: https://docs.internationalbrainlab.org/_autosummary/brainbox.plot.html

Run this code to set up the workspace for the docstring examples. It imports required modules and functions. If there is no 'alf' directory, make one from the 'ks2' output directory using ks2_to_alf. Then load spikes, clusters, and units bunches.

```python
>>> from brainbox import processing
>>> import one.alf.io as alfio
>>> import numpy as np
>>> import matplotlib.pyplot as plt
>>> import ibllib.ephys.spikes as e_spks
# (*Note, if there is no ‘alf’ directory, make ‘alf’ directory from ‘ks2’ output directory):
>>> e_spks.ks2_to_alf(path_to_ks_out, path_to_alf_out)
# Load the alf spikes bunch and clusters bunch, and get a units bunch.
>>> spks_b = alfio.load_object(path_to_alf_out, ‘spikes’)
>>> clstrs_b = alfio.load_object(path_to_alf_out, ‘clusters’)
>>> units_b = processing.get_units_bunch(spks_b) # may take a few mins to compute
```

--------------------------------

### FTPPatcher.setup(par=None, silent=False)

Source: https://docs.internationalbrainlab.org/_autosummary/ibllib.oneibl.patcher.html

Sets up (and saves) FTP login parameters.

```APIDOC
## FTPPatcher.setup(par=None, silent=False)

### Description
Set up (and save) FTP login parameters.

### Parameters
- **par** - A parameters object to modify, if None the default Webclient parameters are loaded.
- **silent** (bool) - If true, the defaults are used with no user input prompt.

### Returns
The modified parameters object.
```

--------------------------------

### Compare TrainingStatus enum values

Source: https://docs.internationalbrainlab.org/_modules/brainbox/behavior/training.html

Examples from the TrainingStatus enum docstring showing how to compare training statuses using the IntFlag operators. The first example demonstrates assertions for checking if a status is ready for delay, failed, trained, in training, and using bitwise NOT and XOR. The second example shows how to get the next training status after a given status, with the output shown as <TrainingStatus.READY4RECORDING: 128>.

```python
>>> status = 'ready4delay'
... assert TrainingStatus[status.upper()] is TrainingStatus.READY4DELAY
... assert TrainingStatus[status.upper()] not in TrainingStatus.FAILED, 'Subject failed training'
... assert TrainingStatus[status.upper()] >= TrainingStatus.TRAINED, 'Subject untrained'
... assert TrainingStatus[status.upper()] > TrainingStatus.IN_TRAINING, 'Subject untrained'
... assert TrainingStatus[status.upper()] in ~TrainingStatus.FAILED, 'Subject untrained'
... assert TrainingStatus[status.upper()] in TrainingStatus.TRAINED ^ TrainingStatus.READY
```

```python
>>> next(member for member in sorted(TrainingStatus) if member > TrainingStatus[status.upper()])
<TrainingStatus.READY4RECORDING: 128>
```

--------------------------------

### Get unique labs and search sessions with one.search

Source: https://docs.internationalbrainlab.org/notebooks_external/data_download.html

Lists all sessions details, extracts unique labs, picks an example lab name, and searches for sessions with datasets='spikes' and that lab.

```python
# List details of all sessions (returns a list of dictionaries)
_, det = one.search(details=True)
labs = set(d['lab'] for d in det)  # Get the set of unique labs

# Example lab name
lab_name = list(labs)[0]

# Searching for RS sessions with specific lab name
sessions_lab = one.search(datasets='spikes', lab=lab_name)

```

--------------------------------

### Get remote stub path for a device

Source: https://docs.internationalbrainlab.org/_autosummary/ibllib.io.session_params.html

Returns the full file path to the remote experiment description stub as a pathlib.Path. The example shows the output for a given session path and host name.

```python
>>> get_remote_stub_name(Path.home().joinpath('subject', '2020-01-01', '001'), 'host-123')
Path.home() / 'subject/2020-01-01/001/_devices/2020-01-01_1_subject@host-123.yaml'
```

--------------------------------

### Get notes with pattern

Source: https://docs.internationalbrainlab.org/_autosummary/ibllib.qc.critical_reasons.html

Example of retrieving notes from Alyx using the REST API with a Django filter for text containing a static string and object ID, then parsing the first note's text as JSON.

```python
>>> notes = alyx.rest('notes', 'list',
...                   django=f'text__icontains,{STR_NOTES_STATIC},'
...                          f'object_id,{uuid}')
>>> test_json_read = json.loads(notes[0]['text'])
```

--------------------------------

### TestRegistration setUp method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/test_oneibl.html

Sets up test environment by creating a temporary directory, session path, and ALF directory with random spike data. Also creates a revision subdirectory with additional spike data.

```python
def setUp(self) -> None:
    self.settings = get_mock_session_settings(self.subject)
    self.session_dict = {
        'subject': self.subject,
        'start_time': '2018-04-01T12:48:26.795526',
        'number': 2,
        'users': [self.settings['PYBPOD_CREATOR'][0]],
    }

    # makes sure tests start without session created
    self.td = tempfile.TemporaryDirectory()
    self.session_path = Path(self.td.name).joinpath(self.subject, '2018-04-01', '002')
    self.alf_path = self.session_path.joinpath('alf')
    self.alf_path.mkdir(parents=True)
    np.save(self.alf_path.joinpath('spikes.times.npy'), np.random.random(500))
    np.save(self.alf_path.joinpath('spikes.amps.npy'), np.random.random(500))
    self.revision = ''.join(random.choices(string.ascii_letters, k=3))
    self.rev_path = self.alf_path.joinpath(f'#{self.revision}#')
    self.rev_path.mkdir(parents=True)
    np.save(self.rev_path.joinpath('spikes.times.npy'), np.random.random(300))
    np.save(self.rev_path.joinpath('spikes.amps.npy'), np.random.random(300))
```

--------------------------------

### setup_logger(name='ibl', level=0, file=None, no_color=False)

Source: https://docs.internationalbrainlab.org/_autosummary/iblutil.util.html

Set up a log for IBL packages. Uses date time, calling function and distinct colours for levels.

```APIDOC
## setup_logger(name='ibl', level=0, file=None, no_color=False)

### Description
Set up a log for IBL packages. Uses date time, calling function and distinct colours for levels. Sets the name if not set already and add a stream handler. If the stream handler already exists, does not duplicate. The naming/level allows not to interfere with third-party libraries when setting level.

### Parameters
- **name** (str) - Optional - Logger name. Default 'ibl'.
- **level** (int) - Optional - Logging level. Default 0.
- **file** - Optional - File to log to.
- **no_color** (bool) - Optional - Disable colored output. Default False.
```

--------------------------------

### Compute waveform similarity for a unit using brainbox metrics

Source: https://docs.internationalbrainlab.org/_modules/brainbox/metrics/single_units.html

Shows how to compute similarity between two sets of waveforms for a unit using `bb.metrics.wf_similarity`. The example selects channels around the max amplitude channel and uses `bb.io.extract_waveforms` to get waveforms from two timestamp ranges.

```python
>>> max_ch = clstrs_b['channels'][1]
>>> if max_ch < 10:  # take only channels greater than `max_ch`.
>>>     ch = np.arange(max_ch, max_ch + 20)
>>> elif (max_ch + 10) > 385:  # take only channels less than `max_ch`.
>>>    ch = np.arange(max_ch - 20, max_ch)
>>> else:  # take `n_c_ch` around `max_ch`.
>>>     ch = np.arange(max_ch - 10, max_ch + 10)
>>> ts1 = units_b['times']['1'][:100]
>>> ts2 = units_b['times']['1'][-100:]
>>> wf1 = bb.io.extract_waveforms(path_to_ephys_file, ts1, ch)
>>> wf2 = bb.io.extract_waveforms(path_to_ephys_file, ts2, ch)
>>> s = bb.metrics.wf_similarity(wf1, wf2)
```

--------------------------------

### start

Source: https://docs.internationalbrainlab.org/_modules/iblutil/io/net/app.html

Starts an experiment by sending a start signal to remote services and awaiting responses.

```APIDOC
## start

### Description
Starts an experiment by sending a start signal to remote services and awaiting responses.

### Method
N/A (Python async method)

### Endpoint
N/A

### Parameters
- **exp_ref** (str) - Required - A reference to the experiment to start.
- **data** (any) - Optional - Optional extra data to send to the remote host.
- **concurrent** (bool) - Optional - If false, wait for response from each service before communicating with the next.

### Request Example
N/A

### Response
#### Success Response (N/A)
- **dict of str** - A dictionary of service names and the response data received.

#### Response Example
N/A

### Errors
- **TimeoutError** - Remote host failed to echo the message within the timeout period. Remote host failed to respond within response period.
```

--------------------------------

### TestLocalServer setUp method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/test_pipes.html

Creates a temporary directory and populates it with fake raw behavior data folders for two sessions, including raw_session.flag files and task settings patches.

```python
def setUp(self):
    tmp = tempfile.TemporaryDirectory()
    self.tmpdir = Path(tmp.name)
    self.addCleanup(tmp.cleanup)
    raw_behaviour_data = fu.create_fake_raw_behavior_data_folder(self.tmpdir / 'subject/2020-01-01/001', task='ephys')
    raw_behaviour_data.parent.joinpath('raw_session.flag').touch()
    fu.populate_task_settings(raw_behaviour_data, patch={'PYBPOD_PROTOCOL': '_iblrig_ephysChoiceWorld5.2.1'})
    raw_behaviour_data = fu.create_fake_raw_behavior_data_folder(self.tmpdir / 'subject/2020-01-01/002')
    raw_behaviour_data.parent.joinpath('raw_session.flag').touch()
    fu.populate_task_settings(raw_behaviour_data, patch={'PYBPOD_PROTOCOL': 'ephys_optoChoiceWorld6.0.1'})
```

--------------------------------

### Get lateralised Allen atlas IDs

Source: https://docs.internationalbrainlab.org/notebooks_external/atlas_mapping.html

Uses mapping='Allen-lr' to get lateralised atlas IDs. Left hemisphere clusters get negative IDs, right hemisphere clusters get positive IDs.

```python
atlas_id_Allen = ba.get_labels(mlapdv, mapping='Allen-lr')
atlas_id_Allen_rh = ba.get_labels(mlapdv_rh, mapping='Allen-lr')
print(f'Lateralised Allen mapping ids using mlapdv: {atlas_id_Allen[0:10]}')
print(f'Lateralised Allen mapping ids using mlapdv_rh: {atlas_id_Allen_rh[0:10]}')
```

```output
Lateralised Allen mapping ids using mlapdv: [-596 -596 -596 -596 -596 -596 -596 -596 -596 -596]
Lateralised Allen mapping ids using mlapdv_rh: [ 596  596  596  596  596 -596 -596 -596 -596 -596]
```

--------------------------------

### Test class TestDLC with setUp

Source: https://docs.internationalbrainlab.org/_modules/brainbox/tests/test_behavior.html

Defines the TestDLC unittest class and its setUp method. The setUp method currently passes without action.

```python
class TestDLC(unittest.TestCase):

    def setUp(self):
        pass
```

--------------------------------

### Set up test data in setUpClass method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/qc/test_alignment_qc.html

Loads fixture data for manual and existing alignment QC, creates a probe insertion and trajectory on Alyx, and initializes an AlignmentQC object. It also resolves manual alignments and prints the file paths.

```python
[docs]
    @classmethod
    def setUpClass(cls) -> None:
        data = np.load(
            Path(Path(__file__).parent.parent.joinpath('fixtures', 'qc', 'data_alignmentqc_manual.npz')), allow_pickle=True
        )
        cls.xyz_picks = (data['xyz_picks'] * 1e6).tolist()
        cls.alignments = data['alignments'].tolist()
        cls.cluster_chns = data['cluster_chns']

        data = np.load(
            Path(Path(__file__).parent.parent.joinpath('fixtures', 'qc', 'data_alignmentqc_existing.npz')), allow_pickle=True
        )
        insertion = data['insertion'].tolist()
        insertion['json'] = {'xyz_picks': cls.xyz_picks}
        # makes sure there is no existing probe insertion before creating a new one
        for ins in one.alyx.rest('insertions', 'list', session=insertion['session'], name=insertion['name'], no_cache=True):
            one.alyx.rest('insertions', 'delete', id=ins['id'])
        probe_insertion = one.alyx.rest('insertions', 'create', data=insertion)
        cls.probe_id = probe_insertion['id']
        cls.probe_name = probe_insertion['name']
        cls.trajectory = data['trajectory'].tolist()
        cls.trajectory.update({'probe_insertion': cls.probe_id})
        cls.trajectory.update({'chronic_insertion': None})
        cls.trajectory.update({'json': cls.alignments})
        cls.traj = one.alyx.rest('trajectories', 'create', data=cls.trajectory)

        cls.align_qc = AlignmentQC(cls.probe_id, one=one, brain_atlas=brain_atlas, channels=False)
        cls.align_qc.load_data(
            prev_alignments=cls.traj['json'],
            xyz_picks=np.array(cls.xyz_picks) / 1e6,
            cluster_chns=cls.cluster_chns,
            depths=SITES_COORDINATES[:, 1],
            chn_coords=SITES_COORDINATES,
        )
        cls.file_paths = cls.align_qc.resolve_manual(
            '2020-09-28T15:57:25_mayo', update=True, upload_alyx=True, upload_flatiron=False
        )
        print(cls.file_paths)
```

--------------------------------

### start(exp_ref, data=None)

Source: https://docs.internationalbrainlab.org/_autosummary/iblutil.io.net.base.html

Abstract method to start an experiment. Intended to specify the expected message signature; subclassed methods should serialize the returned values and pass them to the transport layer.

```APIDOC
## start(exp_ref, data=None)

### Description
Start an experiment. This is intended to specify the expected message signature. The subclassed method should serialize the returned values and pass them to the transport layer.

### Parameters
- **exp_ref** (str) - Required - An experiment reference string in the form yyyy-mm-dd_n_subject.
- **data** (any) - Optional - Optional extra data to send to the remote server.

### Returns
- **ExpMessage.EXPSTART** - The EXPSTART event.
- **str** - The experiment reference string.
- **any, None** - Optional extra data.
```

--------------------------------

### get_protocol_period(session_path, protocol_number, bpod_sync, exclude_empty_periods=True, include_spacer=True)

Source: https://docs.internationalbrainlab.org/_autosummary/ibllib.io.extractors.ephys_fpga.html

Return the start and end time of the protocol number. Note that the start time is the start of the spacer pulses and the end time is either None if the protocol is the final one, or the start of the next spacer.

```APIDOC
## get_protocol_period(session_path, protocol_number, bpod_sync, exclude_empty_periods=True, include_spacer=True)

### Description
Return the start and end time of the protocol number. Note that the start time is the start of the spacer pulses and the end time is either None if the protocol is the final one, or the start of the next spacer.

### Parameters
- **session_path** (str, pathlib.Path) – Required - The absolute session path, i.e. '/path/to/subject/yyyy-mm-dd/nnn'.
- **protocol_number** (int) – Required - The order that the protocol was run in, counted from 0.
- **bpod_sync** (dict) – Required - The sync times and polarities for Bpod BNC1.
- **exclude_empty_periods** (bool) – Optional - When true, spacers are ignored if no bpod pulses are detected between periods.
- **include_spacer** (bool) – Optional - When true, the start time returned is the start of the spacer pulse. If false, the start time is the end of the spacer pulse and the end time is the start of the next spacer pulse.

### Returns
- float – The time of the detected spacer for the protocol number.
- float, None – The time of the next detected spacer or None if this is the last protocol run.
```

--------------------------------

### async start(exp_ref, data=None, concurrent=True)

Source: https://docs.internationalbrainlab.org/_modules/iblutil/io/net/app.html

Starts an experiment by sending an EXPSTART signal to remote services and awaiting responses. Accepts an experiment reference string and optional data, with a concurrency flag to control whether services are contacted in parallel.

```APIDOC
## async start(exp_ref, data=None, concurrent=True)

### Description
Starts an experiment by sending an EXPSTART signal to remote services and awaiting responses.

### Method
Async method

### Endpoint
`start`

### Parameters
- **exp_ref** (str) - Required - An experiment reference string in the form yyyy-mm-dd_n_subject.
- **data** (any) - Optional - Optional extra data to send to the remote host.
- **concurrent** (bool) - Optional - If false, wait for response from each service before communicating with the next.

### Returns
- **dict of str** - A dictionary of service names and the response data received.

### Raises
- **TimeoutError** - Remote host failed to echo the message within the timeout period. Remote host failed to respond within response period.
```

--------------------------------

### TestPipesMisc setUp method

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/test_pipes.html

Sets up a temporary directory with fake session folders, populates raw SpikeGLX data for 3A and 3B probes, and moves a folder to simulate IBL protocol.

```python
def setUp(self):
    # Data emulating local rig data
    self.root_test_folder = tempfile.TemporaryDirectory()

    # Create two rig sessions, one with 3A probe data and one with 3B probe data
    self.session_path_3A = fu.create_fake_session_folder(self.root_test_folder.name)
    # fu.create_fake_raw_behavior_data_folder(self.session_path_3A)
    self.session_path_3B = fu.create_fake_session_folder(self.root_test_folder.name)
    fu.create_fake_raw_behavior_data_folder(self.session_path_3B)

    # Make some files
    fu.populate_raw_spikeglx(self.session_path_3B / 'raw_ephys_data', '3B', n_probes=3)

    ephys_folder = self.session_path_3A / 'raw_ephys_data'
    fu.populate_raw_spikeglx(ephys_folder, '3A', legacy=True, n_probes=1)
    # IBL protocol is for users to copy data to the right probe folder
    shutil.move(ephys_folder.joinpath('raw_ephys_folder'), ephys_folder.joinpath('my_run_probe00'))
```

--------------------------------

### Compute inter-trial interval with load_iti

Source: https://docs.internationalbrainlab.org/notebooks_external/loading_trials_data.html

Computes the inter-trial interval (ITI) using brainbox.io.one.load_iti. The ITI is the period of open-loop grey screen from stimulus off until the quiescent period at the start of the following trial. The example loads trials for a specific eid, adds the 'iti' column, and prints the first five rows. The output includes an ALFWarning about multiple revisions and a table showing the computed ITI values.

```python
from brainbox.io.one import load_iti
eid = 'ae8787b1-4229-4d56-b0c2-566b61a25b77'
trials = one.load_object(eid, 'trials')
trials['iti'] = load_iti(trials)
print(trials.to_df().iloc[:5, -5:])
```

```text
/opt/hostedtoolcache/Python/3.12.14/x64/lib/python3.12/site-packages/one/util.py:464: ALFWarning: Multiple revisions: "", "2025-03-03"
  warnings.warn(f'Multiple revisions: {rev_list}', alferr.ALFWarning)
```

```text
   rewardVolume  firstMovement_times  intervals_0  intervals_1       iti
0           1.5           117.642774   116.695529   119.305634  1.191781
1           1.5           127.491774   119.997332   129.171944  1.223111
2           1.5           131.802774   129.895043   133.568441  1.202380
3           0.0           136.639774   134.270738   139.385252  1.255541
4           1.5           141.399774   140.140751   143.068359  1.167783
```

--------------------------------

### TestAudioSync class with setUp and tearDown

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/tasks/test_audio_tasks.html

Defines the TestAudioSync class with setUp and tearDown methods. The setUp copies raw behavior data to a temporary session path and initializes a local ONE instance.

```python
class TestAudioSync(base.IntegrationTest):



[docs]
    def setUp(self) -> None:
        self.data_path = self.default_data_root().joinpath(
            'ephys', 'ephys_video_init', 'ZM_1735', '2019-08-01', '001', 'raw_behavior_data')
        self.temp_dir = Path(tempfile.TemporaryDirectory().name)
        self.session_path = self.temp_dir.joinpath('ZM_1735', '2019-08-01', '001')
        shutil.copytree(self.data_path, self.session_path.joinpath('raw_behavior_data'))
        self.one = ONE(**base.TEST_DB, mode='local')





[docs]
    def test_audiosync(self):
        task = AudioSync(
            self.session_path, one=self.one, device_collection='raw_behavior_data', collection='raw_behavior_data', sync='bpod')
        status = task.run()
        assert status == 0

        self.assertIsNone(next(self.session_path.rglob('*.wav'), None))





[docs]
    def test_audiosync_fpga(self):
        task = AudioSync(
            self.session_path, one=self.one, device_collection='raw_behavior_data', collection='raw_behavior_data', sync='fpga')
        status = task.run()
        assert status == 0
        assert task.outputs is None
        print(task.log)
        assert 'Audio Syncing not yet implemented for FPGA' in task.log





[docs]
    def check_files(self, task):
        for exp_files in task.signature['output_files']:
            file = next(self.session_path.joinpath(exp_files[1]).glob(exp_files[0]), None)
            assert file.exists()
            assert file in task.outputs





[docs]
    def tearDown(self) -> None:
        shutil.rmtree(self.temp_dir)
```

--------------------------------

### TestTaskQCObject class setup

Source: https://docs.internationalbrainlab.org/_modules/ibllib/tests/integration/test_task_qc_extractors.html

Sets up the test environment by initializing the ONE connection, defining the session fields, and creating a ChoiceWorldTrialsBpod task to run QC. Includes imports and required files for integration testing.

```python
"""Tests TaskQC object and Bpod extractors.

NB: FPGA TaskQC extractor is tested in test_ephys_extraction_choiceWorld.

This module uses ZM_1150/2019-05-07/001 ('b1c968ad-4874-468d-b2e4-5ffa9b9964e9').
"""
import unittest.mock

from ibllib.pipes.behavior_tasks import ChoiceWorldTrialsBpod
from one.api import ONE
from one.alf import spec
from ibllib.tests import base

one = ONE(**base.TEST_DB)




[docs]
class TestTaskQCObject(base.IntegrationTest):

    required_files = ['training/ZM_1150/2019-05-07/001']



[docs]
    @classmethod
    def setUpClass(cls):
        cls.one = one
        cls.eid = 'b1c968ad-4874-468d-b2e4-5ffa9b9964e9'
        # Make sure the data exists locally
        data_path = base.IntegrationTest.default_data_root()
        cls.session_path = data_path.joinpath('training', 'ZM_1150', '2019-05-07', '001')
        task = ChoiceWorldTrialsBpod(cls.session_path, collection='raw_behavior_data', one=one)
        cls.qc = task.run_qc(update=False)
        assert cls.qc.one is one
```
Tokens:
Raw