Skip to content

Commit

Permalink
Merge pull request #363 from CAMBI-tech/saved-device-status
Browse files Browse the repository at this point in the history 
Saved Device Status
  • Loading branch information
@lawhead
lawhead authored Nov 13, 2024
2 parents ad64938 + 725f6f5 commit d1be1f9
Show file tree
Hide file tree
Showing 2 changed files with 186 additions and 56 deletions.
24 changes: 15 additions & 9 deletions bcipy/helpers/acquisition.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,9 @@
discover_device_spec)
from bcipy.acquisition.devices import (DeviceSpec, DeviceStatus,
preconfigured_device, with_content_type)
from bcipy.config import BCIPY_ROOT, RAW_DATA_FILENAME, SESSION_LOG_FILENAME
from bcipy.config import BCIPY_ROOT
from bcipy.config import DEFAULT_DEVICE_SPEC_FILENAME as spec_name
from bcipy.config import RAW_DATA_FILENAME, SESSION_LOG_FILENAME
from bcipy.helpers.save import save_device_specs

logger = logging.getLogger(SESSION_LOG_FILENAME)
Expand Down Expand Up @@ -62,9 +63,7 @@ def init_acquisition(
# Start the server before init_device so it is discoverable.
await_start(dataserver)

device_spec = init_device(content_type, device_name)
if status:
device_spec.status = status
device_spec = init_device(content_type, device_name, status)
raw_data_name = raw_data_filename(device_spec)

client = init_lsl_client(parameters, device_spec, save_folder, raw_data_name)
Expand Down Expand Up @@ -93,7 +92,8 @@ def raw_data_filename(device_spec: DeviceSpec) -> str:


def init_device(content_type: str,
device_name: Optional[str] = None) -> DeviceSpec:
device_name: Optional[str] = None,
status_override: Optional[DeviceStatus] = None) -> DeviceSpec:
"""Initialize a DeviceSpec for the given content type.
If a device_name is provided, the DeviceSpec will be looked up from the list
Expand All @@ -108,12 +108,18 @@ def init_device(content_type: str,
content_type - LSL content type (EEG, Gaze, etc).
device_name - optional; name of the device. If provided, the DeviceSpec
must be a preconfigured device.
status - optional; if provided this value will be used to override the
preconfigured status
"""
if device_name:
return preconfigured_device(device_name, strict=True)
discovered_spec = discover_device_spec(content_type)
configured_spec = preconfigured_device(discovered_spec.name, strict=False)
return configured_spec or discovered_spec
spec = preconfigured_device(device_name, strict=True)
else:
discovered_spec = discover_device_spec(content_type)
configured_spec = preconfigured_device(discovered_spec.name, strict=False)
spec = configured_spec or discovered_spec
if status_override is not None:
spec.status = status_override
return spec


def server_spec(content_type: str,
Expand Down
218 changes: 171 additions & 47 deletions bcipy/helpers/tests/test_acquisition.py
View file
Original file line number Diff line number Diff line change
@@ -1,62 +1,147 @@
"""Tests for acquisition helper."""
import logging
import shutil
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from unittest.mock import Mock, call, patch

from bcipy.acquisition.devices import DeviceSpec, DeviceStatus
from bcipy.config import DEFAULT_PARAMETERS_PATH
from bcipy.config import DEFAULT_DEVICE_SPEC_FILENAME as spec_name
from bcipy.helpers.acquisition import (RAW_DATA_FILENAME, StreamType,
active_content_types, init_acquisition,
init_device, is_stream_type_active,
max_inquiry_duration, parse_stream_type,
raw_data_filename, server_spec,
stream_types)
from bcipy.helpers.load import load_json_parameters
from bcipy.helpers.save import init_save_data_structure
from bcipy.helpers.parameters import Parameters


class TestAcquisition(unittest.TestCase):
"""Unit tests for acquisition helper"""

def setUp(self):
"""set up the needed path for load functions."""
self.parameters_used = DEFAULT_PARAMETERS_PATH
self.parameters = load_json_parameters(self.parameters_used,
value_cast=True)
self.data_save_path = 'data/'
self.user_information = 'test_user_001'
self.task = 'RSVP Calibration'

self.save = init_save_data_structure(self.data_save_path,
self.user_information,
self.parameters_used, self.task)

def tearDown(self):
"""Override; teardown test"""
shutil.rmtree(self.save)

def test_init_acquisition(self):
"""Unit tests for acquisition helper main method"""

@patch('bcipy.helpers.acquisition.save_device_specs')
@patch('bcipy.helpers.acquisition.start_viewer')
@patch('bcipy.helpers.acquisition.init_lsl_client')
@patch('bcipy.helpers.acquisition.init_device')
@patch('bcipy.helpers.acquisition.await_start')
@patch('bcipy.helpers.acquisition.LslDataServer')
@patch('bcipy.helpers.acquisition.server_spec')
@patch('bcipy.helpers.acquisition.ClientManager')
def test_init_acquisition(self, make_client_manager, server_spec_mock,
make_lsl_data_server, await_start_mock,
init_device_mock, init_lsl_client_mock,
start_viewer_mock, save_device_specs_mock):
"""Test init_acquisition with LSL client."""

params = self.parameters
logger = Mock(spec=logging.Logger)
logger.info = lambda x: x
params['acq_mode'] = 'EEG:passive/DSI-24'

client, servers = init_acquisition(params, self.save, server=True)

client.stop_acquisition()
client.cleanup()
for server in servers:
server.stop()

# Function parameters
params = Parameters.from_cast_values(acq_mode='EEG:passive/DSI-24',
acq_show_viewer=False)
save_folder = "temp"

# Mock objects
manager = Mock()
server_device = Mock()
eeg_device = DeviceSpec(content_type='EEG',
name='DSI-24',
status=DeviceStatus.PASSIVE,
channels=['ch1', 'ch2', 'ch3'],
sample_rate=300.0)
server = Mock()
lsl_client = Mock()

# Mock the functions/constructors
make_client_manager.return_value = manager
server_spec_mock.return_value = server_device
make_lsl_data_server.return_value = server
init_device_mock.return_value = eeg_device
init_lsl_client_mock.return_value = lsl_client

client_manager, servers = init_acquisition(params,
save_folder=save_folder,
server=True)

# Assertions
self.assertEqual(client_manager, manager)
self.assertEqual(1, len(servers))
self.assertEqual(client.device_spec.name, 'DSI-24')
self.assertFalse(client.device_spec.is_active)
self.assertEqual(server, servers[0])
init_device_mock.assert_called_with('EEG', 'DSI-24',
DeviceStatus.PASSIVE)
await_start_mock.assert_called_once()
start_viewer_mock.assert_not_called()
manager.add_client.assert_called_with(lsl_client)
save_device_specs_mock.assert_called_with(manager.device_specs,
save_folder, spec_name)

@patch('bcipy.helpers.acquisition.save_device_specs')
@patch('bcipy.helpers.acquisition.start_viewer')
@patch('bcipy.helpers.acquisition.init_lsl_client')
@patch('bcipy.helpers.acquisition.init_device')
@patch('bcipy.helpers.acquisition.await_start')
@patch('bcipy.helpers.acquisition.LslDataServer')
@patch('bcipy.helpers.acquisition.server_spec')
@patch('bcipy.helpers.acquisition.ClientManager')
def test_init_acquisition_multiple_devices(
self, make_client_manager, server_spec_mock, make_lsl_data_server,
await_start_mock, init_device_mock, init_lsl_client_mock,
start_viewer_mock, save_device_specs_mock):
"""Test init acquisition with multiple devices."""
# Function parameters
params = Parameters.from_cast_values(
acq_mode='EEG/DSI-24+Eyetracker:passive',
acq_show_viewer=True,
stim_screen=0,
parameter_location=".")
save_folder = "temp"

# Mock objects
manager = Mock()
# preconfigured devices
eeg_device = DeviceSpec(content_type='EEG',
name='DSI-24',
status=DeviceStatus.ACTIVE,
channels=['ch1', 'ch2', 'ch3'],
sample_rate=300.0)
gaze_device = DeviceSpec(content_type='Eyetracker',
name='Tobii Nano',
status=DeviceStatus.PASSIVE,
channels=['left_pos', 'right_pos'],
sample_rate=60)
eeg_server = Mock()
gaze_server = Mock()
eeg_client = Mock()
gaze_client = Mock()

# Mock the functions/constructors
make_client_manager.return_value = manager
server_spec_mock.side_effect = [eeg_device, gaze_device]
make_lsl_data_server.side_effect = [eeg_server, gaze_server]

# get preconfigured device from devices.json
init_device_mock.size_effect = [eeg_device, gaze_device]
init_lsl_client_mock.side_effect = [eeg_client, gaze_client]

client_manager, servers = init_acquisition(params,
save_folder=save_folder,
server=True)

# Assertions
self.assertEqual(client_manager, manager)
self.assertEqual(2, len(servers))
self.assertEqual(eeg_server, servers[0])
self.assertEqual(gaze_server, servers[1])

self.assertEqual(init_device_mock.call_args_list, [
call('EEG', 'DSI-24', None),
call('Eyetracker', None, DeviceStatus.PASSIVE)
])
self.assertEqual(await_start_mock.call_count, 2)

start_viewer_mock.assert_called_once()
self.assertEqual(
manager.add_client.call_args_list,
[call(eeg_client), call(gaze_client)])
save_device_specs_mock.assert_called_with(manager.device_specs,
save_folder, spec_name)

self.assertTrue(Path(self.save, 'devices.json').is_file())

class TestAcquisitionHelpers(unittest.TestCase):
"""Unit tests for acquisition helper functions"""

def test_max_inquiry_duration(self):
"""Test the max inquiry duration function"""
Expand Down Expand Up @@ -149,6 +234,49 @@ def test_init_device_with_named_device(self, preconfigured_device_mock,
preconfigured_device_mock.assert_called_with('DSI-24', strict=True)
self.assertEqual(device_spec, device_mock)

@patch('bcipy.helpers.acquisition.discover_device_spec')
@patch('bcipy.helpers.acquisition.preconfigured_device')
def test_init_device_with_status_override(self, preconfigured_device_mock,
discover_spec_mock):
"""Test device initialization where the provided status is different than
the preconfigured status."""

preconf_device = DeviceSpec(content_type='EEG',
name='DSI-24',
status=DeviceStatus.ACTIVE,
channels=['ch1', 'ch2', 'ch3'],
sample_rate=300.0)

preconfigured_device_mock.return_value = preconf_device

device_spec = init_device('EEG', 'DSI-24', DeviceStatus.PASSIVE)

discover_spec_mock.assert_not_called()
preconfigured_device_mock.assert_called_with('DSI-24', strict=True)

self.assertEqual(device_spec.name, preconf_device.name)
self.assertEqual(device_spec.content_type, preconf_device.content_type)
self.assertEqual(device_spec.status, DeviceStatus.PASSIVE)

@patch('bcipy.helpers.acquisition.discover_device_spec')
@patch('bcipy.helpers.acquisition.preconfigured_device')
def test_init_device_with_active_status_override(self,
preconfigured_device_mock,
discover_spec_mock):
"""Test device initialization where the provided active status is
different than the preconfigured status."""

preconf_device = DeviceSpec(content_type='EEG',
name='DSI-24',
status=DeviceStatus.PASSIVE,
channels=['ch1', 'ch2', 'ch3'],
sample_rate=300.0)
preconfigured_device_mock.return_value = preconf_device

device_spec = init_device('EEG', 'DSI-24', DeviceStatus.ACTIVE)
self.assertEqual(device_spec.status, DeviceStatus.ACTIVE)
discover_spec_mock.assert_not_called()

def test_parse_stream_type(self):
"""Test function to split the stream type into content_type, name,
and status"""
Expand Down Expand Up @@ -198,10 +326,6 @@ def test_raw_data_filename_eye_tracker(self):
self.assertEqual(raw_data_filename(device),
'eyetracker_data_tobii-p0.csv')


class TestAcquisitionHelpers(unittest.TestCase):
"""Unit tests for acquisition helper functions"""

def test_stream_type_active_given_status(self):
"""Test function to test if a StreamType is active given the provided
status."""
Expand Down

0 comments on commit d1be1f9

Please sign in to comment.