Add initial audio device test

tests/audio_device.py

@@ -0,0 +1,98 @@
+import time
+
+import numpy as np
+
+from open_ephys.control import OpenEphysHTTPServer
+from open_ephys.analysis import Session
+
+"""
+Test Name: Audio Device Test
+Test Description: Configures the buffer size and sample rate of the audio device and validates recordings
+"""
+
+def test(gui, params):
+
+    results = {}
+
+    if params['fetch']:
+
+        def record():
+            gui.record()
+            time.sleep(params['rec_time'])
+
+        gui.clear_signal_chain()
+        gui.add_processor("File Reader")
+        gui.add_processor("Record Node")
+
+        test_name = 'Prevent buffer size change during acquisition'
+
+        gui.acquire()
+
+        original_buffer_size = gui.get_audio_settings('buffer_size')
+        gui.set_buffer_size(512)
+        time.sleep(1)
+        new_buffer_size = gui.get_audio_settings('buffer_size')
+
+        if new_buffer_size != original_buffer_size:
+            results[test_name] = "FAILED\n\tExpected: %d\n\tActual: %d" % (original_buffer_size, new_buffer_size)
+        else:
+            results[test_name] = "PASSED"
+
+        test_name = 'Prevent sample rate change during acquisition'
+
+        original_sample_rate = gui.get_audio_settings('sample_rate')
+        gui.set_sample_rate(48000)
+        time.sleep(1)
+        new_sample_rate = gui.get_audio_settings('sample_rate')
+
+        if new_sample_rate != original_sample_rate:
+            results[test_name] = "FAILED\n\tExpected: %d\n\tActual: %d" % (original_sample_rate, new_sample_rate)
+        else:
+            results[test_name] = "PASSED"
+
+        gui.idle()
+
+    #TODO: Validate recorded data with various buffer sizes and samples ratess
+
+    return results
+
+'''
+================================================================================================================================================
+'''
+import os
+import sys
+import argparse
+import platform
+
+from pathlib import Path
+
+if platform.system() == 'Windows':
+    RECORD_PATH = 'C:\\open-ephys\\data'
+elif platform.system() == 'Linux':
+    RECORD_PATH = '<path/to/linux/runner>' #TODO
+else:
+    RECORD_PATH = '<path/to/mac/runner>' #TODO
+
+if __name__ == '__main__':
+
+    parser = argparse.ArgumentParser(description='Process some integers.')
+    parser.add_argument('--mode', required=True, choices={'local', 'githubactions'})
+    parser.add_argument('--fetch', required=False, type=int, default=1)
+    parser.add_argument('--address', required=False, type=str, default='http://127.0.0.1')
+    parser.add_argument('--cfg_path', required=False, type=str, default=os.path.join(Path(__file__).resolve().parent, '../configs/file_reader_config.xml'))
+    parser.add_argument('--acq_time', required=False, type=int, default=2)
+    parser.add_argument('--rec_time', required=False, type=int, default=5)
+    parser.add_argument('--num_rec', required=False, type=int, default=1)
+    parser.add_argument('--num_exp', required=False, type=int, default=1)
+    parser.add_argument('--prepend_text', required=False, type=str, default='')
+    parser.add_argument('--base_text', required=False, type=str, default='')
+    parser.add_argument('--append_text', required=False, type=str, default='')
+    parser.add_argument('--parent_directory', required=False, type=str, default=RECORD_PATH)
+    parser.add_argument('--engine', required=False, type=str, default='engine=0')
+
+    params = vars(parser.parse_args(sys.argv[1:]))
+
+    results = test(OpenEphysHTTPServer(), params)
+
+    for test, result in results.items():
+        print(test, '-'*(80-len(test)), result)

tests/basic_record.py

@@ -1,10 +1,10 @@
 import time
 
+import numpy as np
+
 from open_ephys.control import OpenEphysHTTPServer
 from open_ephys.analysis import Session
 
-import numpy as np
-
 """
 Test Name: Add/Delete Processor
 Test Description: Add and delete a processor from the signal chain