"""Tests for ibllib.pipes.mesoscope_tasks."""
import sys
import unittest
from unittest import mock
import tempfile
import json
from itertools import chain
from pathlib import Path
import subprocess
from copy import deepcopy
from one.api import ONE
import numpy as np
from ibllib.pipes.mesoscope_tasks import MesoscopePreprocess, MesoscopeFOV, \
find_triangle, surface_normal, _nearest_neighbour_1d
from ibllib.io.extractors import mesoscope
from ibllib.tests import TEST_DB
# Mock suit2p which is imported in MesoscopePreprocess
attrs = {'default_ops.return_value': {}}
sys.modules['suite2p'] = mock.MagicMock(**attrs)
[docs]
class TestMesoscopePreprocess(unittest.TestCase):
"""Test for MesoscopePreprocess task."""
[docs]
def setUp(self) -> None:
self.td = tempfile.TemporaryDirectory()
self.session_path = Path(self.td.name).joinpath('subject', '2020-01-01', '001')
self.img_path = self.session_path.joinpath('raw_imaging_data_00')
self.img_path.mkdir(parents=True)
self.task = MesoscopePreprocess(self.session_path, one=ONE(**TEST_DB))
self.img_path.joinpath('_ibl_rawImagingData.meta.json').touch()
self.tifs = [self.img_path.joinpath(f'2024-01-01_1_subject_00001_0000{i}.tif') for i in range(5)]
for file in self.tifs:
file.touch()
[docs]
def test_get_default_tau(self):
"""Test for MesoscopePreprocess.get_default_tau method."""
subject_detail = {'genotype': [{'allele': 'Cdh23', 'zygosity': 1},
{'allele': 'Ai95-G6f', 'zygosity': 1},
{'allele': 'Camk2a-tTa', 'zygosity': 1}]}
with mock.patch.object(self.task.one.alyx, 'rest', return_value=subject_detail):
self.assertEqual(self.task.get_default_tau(), .7)
subject_detail['genotype'].pop(1)
self.assertEqual(self.task.get_default_tau(), 1.5) # return the default value
[docs]
def test_consolidate_exptQC(self):
"""Test for MesoscopePreprocess._consolidate_exptQC method."""
exptQC = [
{'frameQC_names': np.array(['ok', 'PMT off', 'galvos fault', 'high signal'], dtype=object),
'frameQC_frames': np.array([0, 0, 0, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 4])},
{'frameQC_names': np.array(['ok', 'PMT off', 'foo', 'galvos fault', np.array([])], dtype=object),
'frameQC_frames': np.array([0, 0, 1, 1, 2, 2, 2, 2, 3, 4])},
{'frameQC_names': 'ok', # check with single str instead of array
'frameQC_frames': np.array([0, 0])}
]
# Check concatinates frame QC arrays
frame_qc, frame_qc_names, bad_frames = self.task._consolidate_exptQC(exptQC)
# Check frame_qc array
expected_frames = [
0, 0, 0, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 5, 0, 0, 1, 1, 4, 4, 4, 4, 2, 5, 0, 0]
np.testing.assert_array_equal(expected_frames, frame_qc)
# Check bad_frames array
expected = [3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 18, 19, 20, 21, 22, 23, 24, 25]
np.testing.assert_array_equal(expected, bad_frames)
# Check frame_qc_names data frame
self.assertCountEqual(['qc_values', 'qc_labels'], frame_qc_names.columns)
self.assertEqual(list(range(6)), frame_qc_names['qc_values'].tolist())
expected = ['ok', 'PMT off', 'galvos fault', 'high signal', 'foo', 'unknown']
self.assertCountEqual(expected, frame_qc_names['qc_labels'].tolist())
[docs]
def test_setup_uncompressed(self):
"""Test set up behaviour when raw tifs present."""
# Test signature when clobber = True
self.task.overwrite = True
raw = self.task.signature['input_files'][1]
self.assertEqual(2, len(raw.identifiers))
self.assertEqual('*.tif', raw.identifiers[0][-1])
# When clobber is False, a data.bin datasets are included as input
self.task.overwrite = False
raw = self.task.signature['input_files'][1]
self.assertEqual(3, len(raw.identifiers))
self.assertEqual('data.bin', raw.identifiers[0][-1])
# After setup and teardown the tif files should not have been removed
self.task.setUp()
self.task.tearDown()
self.assertTrue(all(map(Path.exists, self.tifs)), 'tifs unexpectedly removed')
[docs]
def test_setup_compressed(self):
"""Test set up behaviour when only compressed tifs present."""
# Make compressed file
outfile = self.img_path.joinpath('imaging.frames.tar.bz2')
cmd = 'tar -cjvf "{output}" "{input}"'.format(
output=outfile.relative_to(self.img_path),
input='" "'.join(str(x.relative_to(self.img_path)) for x in self.tifs))
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=self.img_path)
info, error = process.communicate() # b'2023-02-17_2_test_2P_00001_00001.tif\n'
assert process.returncode == 0, f'compression failed: {error.decode()}'
for file in self.tifs:
file.unlink()
self.task.setUp()
self.assertTrue(all(map(Path.exists, self.tifs)))
self.assertTrue(self.img_path.joinpath('imaging.frames.tar.bz2').exists())
self.task.tearDown()
self.assertFalse(any(map(Path.exists, self.tifs)))
[docs]
def test_roi_detection(self):
"""Test roi_detection method.
This simply tests that the input ops are modified and that suite2p is called
and it's return value is returned.
"""
run_plane_mock = sys.modules['suite2p'].run_plane
run_plane_mock.reset_mock()
run_plane_mock.return_value = {'foo': 'bar'}
ret = self.task.roi_detection({'do_registration': True, 'bar': 'baz'})
self.assertEqual(ret, {'foo': 'bar'}, 'failed to return suite2p function return value')
run_plane_mock.assert_called_once_with({'do_registration': False, 'bar': 'baz', 'roidetect': True})
[docs]
def test_image_motion_registration(self):
"""Test image_motion_registration method."""
motion_reg_mock = sys.modules['suite2p'].run_plane
motion_reg_mock.reset_mock()
ops = {'foo': 'bar'}
ret = {'regDX': np.array([2, 3, 4]), 'regPC': np.array([4, 5, 6]), 'tPC': 5}
motion_reg_mock.return_value = ret
metrics = self.task.image_motion_registration(ops)
expected = ('regDX', 'regPC', 'tPC', 'reg_metrics_avg', 'reg_metrics_max')
self.assertCountEqual(expected, metrics.keys())
self.assertEqual(3, metrics['reg_metrics_avg'])
self.assertEqual(4, metrics['reg_metrics_max'])
motion_reg_mock.assert_called_once_with(
{'foo': 'bar', 'do_registration': True, 'do_regmetrics': True, 'roidetect': False})
[docs]
def test_get_plane_paths(self):
"""Test _get_plane_paths method."""
path = self.session_path.joinpath('suite2p')
self.assertEqual([], self.task._get_plane_paths(path))
path.mkdir()
for i in range(13):
path.joinpath(f'plane{i}').mkdir()
plane_paths = self.task._get_plane_paths(path)
self.assertEqual(13, len(plane_paths))
self.assertTrue(all(isinstance(x, Path) for x in plane_paths))
expected = ['plane9', 'plane10', 'plane11', 'plane12']
actual = [str(p.relative_to(path)) for p in plane_paths[-4:]]
self.assertEqual(expected, actual, 'failed to nat sort')
[docs]
def tearDown(self) -> None:
self.td.cleanup()
[docs]
class TestMesoscopeFOV(unittest.TestCase):
"""Test for MesoscopeFOV task and associated functions."""
[docs]
def test_get_provenance(self):
"""Test for MesoscopeFOV.get_provenance method."""
filename = 'mpciMeanImage.mlapdv_estimate.npy'
provenance = MesoscopeFOV.get_provenance(filename)
self.assertEqual('ESTIMATE', provenance.name)
filename = 'mpciROIs.brainLocation_ccf_2017.npy'
provenance = MesoscopeFOV.get_provenance(filename)
self.assertEqual('HISTOLOGY', provenance.name)
[docs]
def test_find_triangle(self):
"""Test for find_triangle function."""
points = np.array([[2.435, -3.37], [2.435, -1.82], [2.635, -2.], [2.535, -1.7]])
connectivity_list = np.array([[0, 1, 2], [1, 2, 3], [2, 3, 4], [3, 4, 5]], dtype=np.intp)
point = np.array([2.6, -1.9])
self.assertEqual(1, find_triangle(point, points, connectivity_list))
point = np.array([3., 1.]) # outside of defined vertices
self.assertEqual(-1, find_triangle(point, points, connectivity_list))
[docs]
def test_surface_normal(self):
"""Test for surface_normal function."""
vertices = np.array([[0, 1, 0], [0, 0, 0], [1, 0, 0]])
expected = np.array([0, 0, 1])
np.testing.assert_almost_equal(surface_normal(vertices), expected)
# Test against multiple triangles
vertices = np.r_[vertices[np.newaxis, :, :], [[[0, 0, 0], [0, 2, 0], [2, 0, 0]]]]
expected = np.array([[0, 0, 1], [0, 0, -1]])
np.testing.assert_almost_equal(surface_normal(vertices), expected)
# Some real data
vertices = np.array([[2.435, -1.82, -0.53], [2.635, -2., -0.58], [2.535, -1.7, -0.58]])
expected = np.array([0.33424239, 0.11141413, 0.93587869])
np.testing.assert_almost_equal(surface_normal(vertices), expected)
# Test input validation
self.assertRaises(ValueError, surface_normal, np.array([[1, 2, 3, 4]]))
[docs]
def test_nearest_neighbour_1d(self):
"""Test for _nearest_neighbour_1d function."""
x = np.array([2., 1., 4., 5., 3.])
x_new = np.array([-3, 0, 1.2, 3, 3, 2.5, 4.7, 6])
val, ind = _nearest_neighbour_1d(x, x_new)
np.testing.assert_array_equal(val, [1., 1., 1., 3., 3., 2., 5., 5.])
np.testing.assert_array_equal(ind, [1, 1, 1, 4, 4, 0, 3, 3])
[docs]
def test_update_surgery_json(self):
"""Test for MesoscopeFOV.update_surgery_json method.
Here we mock the Alyx object and simply check the method's calls.
"""
one = ONE(**TEST_DB)
task = MesoscopeFOV('/foo/bar/subject/2020-01-01/001', one=one)
record = {'json': {'craniotomy_00': {'center': [1., -3.]}, 'craniotomy_01': {'center': [2.7, -1.3]}}}
normal_vector = np.array([0.5, 1., 0.])
meta = {'centerMM': {'ML': 2.7, 'AP': -1.30000000001}}
with mock.patch.object(one.alyx, 'rest', return_value=[record, {}]), \
mock.patch.object(one.alyx, 'json_field_update') as mock_rest:
task.update_surgery_json(meta, normal_vector)
expected = {'craniotomy_01': {'center': [2.7, -1.3],
'surface_normal_unit_vector': (0.5, 1., 0.)}}
mock_rest.assert_called_once_with('subjects', 'subject', data=expected)
# Check errors and warnings
# No matching craniotomy center
with self.assertLogs('ibllib.pipes.mesoscope_tasks', 'ERROR'), \
mock.patch.object(one.alyx, 'rest', return_value=[record, {}]):
task.update_surgery_json({'centerMM': {'ML': 0., 'AP': 0.}}, normal_vector)
# No matching surgery records
with self.assertLogs('ibllib.pipes.mesoscope_tasks', 'ERROR'), \
mock.patch.object(one.alyx, 'rest', return_value=[]):
task.update_surgery_json(meta, normal_vector)
# ONE offline
one.mode = 'local'
try:
with self.assertLogs('ibllib.pipes.mesoscope_tasks', 'WARNING'):
task.update_surgery_json(meta, normal_vector)
finally:
# ONE function is cached so we must reset the mode for other tests
one.mode = 'auto'
[docs]
class TestRegisterFOV(unittest.TestCase):
"""Test for MesoscopeFOV.register_fov method."""
[docs]
def setUp(self) -> None:
self.one = ONE(**TEST_DB)
tmpdir = tempfile.TemporaryDirectory()
self.addCleanup(tmpdir.cleanup)
self.session_path = Path(tmpdir.name, 'subject', '2020-01-01', '001')
self.session_path.joinpath('alf', 'FOV_00').mkdir(parents=True)
filename = self.session_path.joinpath('alf', 'FOV_00', 'mpciMeanImage.brainLocationIds_ccf_2017_estimate.npy')
np.save(filename, np.array([0, 1, 2, 2, 4, 7], dtype=int))
[docs]
def test_register_fov(self):
"""Test MesoscopeFOV.register_fov method.
Note this doesn't actually hit Alyx. Also this doesn't test stack creation.
"""
task = MesoscopeFOV(self.session_path, device_collection='raw_imaging_data', one=self.one)
mlapdv = {'topLeft': [2317.2, -1599.8, -535.5], 'topRight': [2862.7, -1625.2, -748.7],
'bottomLeft': [2317.3, -2181.4, -466.3], 'bottomRight': [2862.7, -2206.9, -679.4],
'center': [2596.1, -1900.5, -588.6]}
meta = {'FOV': [{'MLAPDV': mlapdv, 'nXnYnZ': [512, 512, 1], 'roiUUID': 0}]}
with unittest.mock.patch.object(task.one.alyx, 'rest') as mock_rest:
task.register_fov(meta, 'estimate')
calls = mock_rest.call_args_list
self.assertEqual(3, len(calls))
args, kwargs = calls[1]
self.assertEqual(('fields-of-view', 'create'), args)
expected = {'data': {'session': None, 'imaging_type': 'mesoscope', 'name': 'FOV_00', 'stack': None}}
self.assertEqual(expected, kwargs)
args, kwargs = calls[2]
self.assertEqual(('fov-location', 'create'), args)
expected = ['field_of_view', 'default_provenance', 'coordinate_system', 'n_xyz', 'provenance', 'x', 'y', 'z',
'brain_region']
self.assertCountEqual(expected, kwargs.get('data', {}).keys())
self.assertEqual(5, len(kwargs['data']['brain_region']))
self.assertEqual([512, 512, 1], kwargs['data']['n_xyz'])
self.assertIs(kwargs['data']['field_of_view'], mock_rest().get('id'))
self.assertEqual('E', kwargs['data']['provenance'])
self.assertEqual([2317.2, 2862.7, 2317.3, 2862.7], kwargs['data']['x'])
# Check dry mode with suffix input = None
for file in self.session_path.joinpath('alf', 'FOV_00').glob('mpciMeanImage.*'):
file.replace(file.with_name(file.name.replace('_estimate', '')))
task.one.mode = 'local'
with unittest.mock.patch.object(task.one.alyx, 'rest') as mock_rest:
out = task.register_fov(meta, None)
mock_rest.assert_not_called()
self.assertEqual(1, len(out))
self.assertEqual('FOV_00', out[0].get('name'))
locations = out[0]['location']
self.assertEqual(1, len(locations))
self.assertEqual('L', locations[0].get('provenance', 'L'))
[docs]
def tearDown(self) -> None:
"""
The ONE function is cached and therefore the One object persists beyond this test.
Here we return the mode back to the default after testing behaviour in offline mode.
"""
self.one.mode = 'auto'