Merge pull request #543 from int-brain-lab/iblrigv8dev

8.12.3

CHANGELOG.md

@@ -1,6 +1,13 @@
 Changelog
 ---------
 
+8.12.3
+------
+* bugfix: getting training status of subject not present on local server
+* skipping of bpod initialization now optional (used in GUI)
+* disable button for status LED if not supported by hardware
+* tests, type-hints, removal of dead code
+
 8.12.2
 ------
 * bugfix: rollback skipping of bpod initialization (possible source of integer overflow)

iblrig/base_choice_world.py

@@ -430,7 +430,9 @@ def draw_next_trial_info(self, pleft=0.5, contrast=None, position=None):
             contrast = misc.draw_contrast(self.task_params.CONTRAST_SET, self.task_params.CONTRAST_SET_PROBABILITY_TYPE)
         assert len(self.task_params.STIM_POSITIONS) == 2, "Only two positions are supported"
         position = position or int(np.random.choice(self.task_params.STIM_POSITIONS, p=[pleft, 1 - pleft]))
-        quiescent_period = self.task_params.QUIESCENT_PERIOD + misc.texp(factor=0.35, min_=0.2, max_=0.5)
+        quiescent_period = self.task_params.QUIESCENT_PERIOD + misc.truncated_exponential(scale=0.35,
+                                                                                          min_value=0.2,
+                                                                                          max_value=0.5)
         self.trials_table.at[self.trial_num, 'quiescent_period'] = quiescent_period
         self.trials_table.at[self.trial_num, 'contrast'] = contrast
         self.trials_table.at[self.trial_num, 'stim_phase'] = random.uniform(0, 2 * math.pi)
@@ -711,10 +713,10 @@ def new_block(self):
         if self.task_params.BLOCK_INIT_5050 and self.block_num == 0:
             block_len = 90
         else:
-            block_len = int(misc.texp(
-                factor=self.task_params.BLOCK_LEN_FACTOR,
-                min_=self.task_params.BLOCK_LEN_MIN,
-                max_=self.task_params.BLOCK_LEN_MAX
+            block_len = int(misc.truncated_exponential(
+                scale=self.task_params.BLOCK_LEN_FACTOR,
+                min_value=self.task_params.BLOCK_LEN_MIN,
+                max_value=self.task_params.BLOCK_LEN_MAX
             ))
         if self.block_num == 0:
             if self.task_params.BLOCK_INIT_5050:

iblrig/gui/wizard.py

@@ -232,6 +232,10 @@ def __init__(self, *args, **kwargs):
         # anydesk_worker.signals.result.connect(lambda var: print(f'Your AnyDesk ID: {var:s}'))
         # QThreadPool.globalInstance().tryStart(anydesk_worker)
 
+        # disable control of LED if Bpod does not have the respective capability
+        bpod = Bpod(self.model.hardware_settings['device_bpod']['COM_BPOD'], skip_initialization=True)
+        self.uiPushStatusLED.setEnabled(bpod.can_control_led)
+
         # check for update
         update_worker = Worker(check_for_updates)
         update_worker.signals.result.connect(self._on_check_update_result)
@@ -651,7 +655,7 @@ def flush(self):
         self.enable_UI_elements()
 
         try:
-            bpod = Bpod(self.model.hardware_settings['device_bpod']['COM_BPOD'])  # bpod is a singleton
+            bpod = Bpod(self.model.hardware_settings['device_bpod']['COM_BPOD'], skip_initialization=True)
             bpod.manual_override(bpod.ChannelTypes.OUTPUT, bpod.ChannelNames.VALVE, 1, self.uiPushFlush.isChecked())
         except (OSError, exceptions.bpod_error.BpodErrorException):
             print(traceback.format_exc())
@@ -668,7 +672,7 @@ def toggle_status_led(self, is_toggled: bool):
         self.enable_UI_elements()
 
         try:
-            bpod = Bpod(self.model.hardware_settings['device_bpod']['COM_BPOD'])
+            bpod = Bpod(self.model.hardware_settings['device_bpod']['COM_BPOD'], skip_initialization=True)
             bpod.set_status_led(is_toggled)
         except (OSError, exceptions.bpod_error.BpodErrorException, AttributeError):
             self.uiPushStatusLED.setChecked(False)

iblrig/hardware.py

@@ -28,10 +28,10 @@
 
 
 class Bpod(BpodIO):
+    can_control_led = True
     _instances = {}
     _lock = threading.Lock()
     _is_initialized = False
-    _can_control_led = True
 
     def __new__(cls, *args, **kwargs):
         serial_port = args[0] if len(args) > 0 else ''
@@ -44,10 +44,11 @@ def __new__(cls, *args, **kwargs):
             Bpod._instances[serial_port] = instance
             return instance
 
-    def __init__(self, *args, **kwargs):
-        # # skip initialization if it has already been performed before
-        # if self._is_initialized:
-        #     return
+    def __init__(self, *args, skip_initialization: bool = False, **kwargs):
+        # skip initialization if it has already been performed before
+        # IMPORTANT: only use this for non-critical tasks (e.g., flushing valve from GUI)
+        if skip_initialization and self._is_initialized:
+            return
 
         # try to instantiate once for nothing
         try:
@@ -65,7 +66,7 @@ def __init__(self, *args, **kwargs):
                     "Please unplug the Bpod USB cable from the computer and plug it back in to start the task. ") from e
         self.default_message_idx = 0
         self.actions = Bunch({})
-        self._can_control_led = self.set_status_led(True)
+        self.can_control_led = self.set_status_led(True)
         self._is_initialized = True
 
     def close(self) -> None:
@@ -182,7 +183,7 @@ def toggle_valve(self, duration=None):
 
     @static_vars(supported=True)
     def set_status_led(self, state: bool) -> bool:
-        if self._can_control_led and self._arcom is not None:
+        if self.can_control_led and self._arcom is not None:
             try:
                 log.info(f'{"en" if state else "dis"}abling Bpod Status LED')
                 command = struct.pack("cB", b":", state)
@@ -193,7 +194,7 @@ def set_status_led(self, state: bool) -> bool:
                 pass
             self._arcom.serial_object.reset_input_buffer()
             self._arcom.serial_object.reset_output_buffer()
-            log.error('Bpod device does not support control of the status LED. Please update firmware.')
+            log.warning('Bpod device does not support control of the status LED. Please update firmware.')
         return False
 
     def valve(self, valve_id: int, state: bool):