Online Multimodal Fusion

bcipy/task/paradigm/matrix/__init__.py

@@ -2,3 +2,4 @@
 from .calibration import MatrixCalibrationTask  # noqa
 from .copy_phrase import MatrixCopyPhraseTask  # noqa
 from .timing_verification import MatrixTimingVerificationCalibration  # noqa
+from .copy_phrase import RSVPCopyPhraseTask  # noqa

bcipy/task/paradigm/matrix/copy_phrase.py

@@ -1,14 +1,36 @@
 """Defines the Copy Phrase Task which uses a Matrix display"""
+import logging
+from typing import Any, List, Optional, Tuple
+
 from psychopy import visual
+from psychopy.visual import Window
 
-from bcipy.display import InformationProperties, StimuliProperties
+from bcipy.acquisition import ClientManager
+from bcipy.config import (SESSION_LOG_FILENAME, WAIT_SCREEN_MESSAGE)
+from bcipy.display import (InformationProperties, StimuliProperties,
+                           init_display_window)
 from bcipy.display.components.task_bar import CopyPhraseTaskBar
 from bcipy.display.main import PreviewParams
 from bcipy.display.paradigm.matrix.display import MatrixDisplay
+from bcipy.exceptions import TaskConfigurationException
+from bcipy.feedback.visual.visual_feedback import VisualFeedback
+from bcipy.helpers.acquisition import (LslDataServer, active_content_types,
+                                       init_acquisition)
+from bcipy.helpers.clock import Clock
+from bcipy.helpers.copy_phrase_wrapper import CopyPhraseWrapper
+from bcipy.helpers.language_model import init_language_model
+from bcipy.io.load import choose_signal_models
+from bcipy.core.parameters import Parameters
+from bcipy.core.stimuli import InquirySchedule, StimuliOrder
+from bcipy.helpers.task import trial_complete_message, get_user_input
+from bcipy.core.triggers import (Trigger, TriggerType,
+                                 convert_timing_triggers, offset_label)
+from bcipy.language.main import LanguageModel
+from bcipy.signal.model import SignalModel
 from bcipy.task import TaskMode
 from bcipy.task.paradigm.rsvp.copy_phrase import RSVPCopyPhraseTask
-from bcipy.core.parameters import Parameters
-from bcipy.helpers.clock import Clock
+
+logger = logging.getLogger(SESSION_LOG_FILENAME)
 
 
 class MatrixCopyPhraseTask(RSVPCopyPhraseTask):
@@ -99,13 +121,203 @@
         "trigger_type",
     ]
 
+    def __init__(self, parameters: Parameters,
+                 file_save: str, fake: bool = False):
+        super(MatrixCopyPhraseTask, self).__init__(parameters, file_save, fake)
+        self.matrix = self.init_display()
+
     def init_display(self) -> MatrixDisplay:
         """Initialize the Matrix display"""
-        return init_display(self.parameters, self.window,
-                            self.experiment_clock, self.spelled_text)
+        return _init_matrix_display(self.parameters,
+                                    self.window,
+                                    self.experiment_clock,
+                                    self.spelled_text,
+                                    )
+
+    def setup(
+            self,
+            parameters: Parameters,
+            data_save_location: str,
+            fake: bool = False) -> Tuple[ClientManager, List[LslDataServer], Window]:
+        # Initialize Acquisition
+        daq, servers = init_acquisition(
+            parameters, data_save_location, server=fake)
+
+        # Initialize Display
+        display = init_display_window(parameters)
+        self.initalized = True
+
+        return daq, servers, display
+
+    def get_language_model(self) -> LanguageModel:
+        return init_language_model(self.parameters)
+
+    def get_signal_models(self) -> Optional[List[SignalModel]]:
+        if not self.fake:
+            try:
+                signal_models = choose_signal_models(
+                    active_content_types(self.parameters['acq_mode']))
+                assert signal_models, "No signal models selected"
+            except Exception as error:
+                logger.exception(
+                    f'Cannot load signal models. Exiting. {error}')
+                raise error
+            return signal_models
+        return []
+
+    def cleanup(self):
+        return super().cleanup()
+
+    def save_session_data(self):
+        return super().save_session_data()
+
+    def init_copy_phrase_task(self) -> None:
+        """Initialize the CopyPhraseWrapper.
+
+        Returns:
+        --------
+        initialized CopyPhraseWrapper
+        """
+
+        self.copy_phrase_task = CopyPhraseWrapper(
+            self.parameters["min_inq_len"],
+            self.parameters["max_inq_per_series"],
+            lmodel=self.language_model,
+            alp=self.alp,
+            evidence_names=self.evidence_types,
+            task_list=[(str(self.copy_phrase), self.spelled_text)],
+            is_txt_stim=self.parameters["is_txt_stim"],
+            stim_timing=[
+                self.parameters["time_fixation"],
+                self.parameters["time_flash"],
+            ],
+            decision_threshold=self.parameters["decision_threshold"],
+            backspace_prob=self.parameters["lm_backspace_prob"],
+            backspace_always_shown=self.parameters["backspace_always_shown"],
+            stim_length=self.parameters["stim_length"],
+            stim_jitter=self.parameters["stim_jitter"],
+            stim_order=StimuliOrder(self.parameters["stim_order"]),
+        )
+
+    def user_wants_to_continue(self) -> bool:
+        """Check if user wants to continue or terminate.
+
+        Returns
+        -------
+        - `True` to continue
+        - `False` to finish the task.
+        """
+        should_continue = get_user_input(
+            self.matrix,
+            WAIT_SCREEN_MESSAGE,
+            self.parameters["stim_color"],
+            first_run=self.first_run,
+        )
+        if not should_continue:
+            logger.info("User wants to exit.")
+        return should_continue
+
+    def present_inquiry(
+        self, inquiry_schedule: InquirySchedule
+    ) -> Tuple[List[Tuple[str, float]], bool]:
+        """Present the given inquiry and return the trigger timing info.
+
+        Parameters
+        ----------
+        - inquiry_schedule : schedule for next sequences of stimuli to present.
+        Currently, only the first list of stims in the schedule is used.
+
+        Returns
+        -------
+        Tuple of `(stim_times, proceed)`
+
+        - stim_times : list of tuples representing the stimulus and time that
+        it was presented relative to the experiment clock. Non-stim triggers
+        may be also be included in the list ('calibration', etc).
+        - proceed : indicates whether to proceed with evaluating eeg evidence.
+        a value of False indicates that the inquiry was previewed but not
+        presented in sequence.
+        """
+        # Update task state and reset the static
+        self.matrix.update_task_bar(self.spelled_text)
+        self.matrix.draw_static()
+        self.window.flip()
+
+        self.wait()
+
+        # Setup the new stimuli
+        self.matrix.schedule_to(
+            stimuli=inquiry_schedule.stimuli[0],
+            timing=inquiry_schedule.durations[0],
+            colors=(
+                inquiry_schedule.colors[0] if self.parameters["is_txt_stim"] else None
+            ),
+        )
+
+        stim_times = self.matrix.do_inquiry()
+        proceed = True  # `proceed` is always True for matrix task, since it does't have inquiry preview
+
+        return stim_times, proceed
+
+    def exit_display(self) -> None:
+        """Close the UI and cleanup."""
+        # Update task state and reset the static
+        self.matrix.update_task_bar(text=self.spelled_text)
+
+        # Say Goodbye!
+        self.matrix.info_text = trial_complete_message(
+            self.window, self.parameters)
+        self.matrix.draw_static()
+        self.window.flip()
+
+        # Give the system time to process
+        self.wait()
+
+    def write_offset_trigger(self) -> None:
+        """Append the offset to the end of the triggers file."""
+        # To help support future refactoring or use of lsl timestamps only
+        # we write only the sample offset here.
+        triggers = []
+        for content_type, client in self.daq.clients_by_type.items():
+            label = offset_label(content_type.name, prefix="daq_sample_offset")
+            time = client.offset(self.matrix.first_stim_time)
+            triggers.append(Trigger(label, TriggerType.SYSTEM, time))
+
+        self.trigger_handler.add_triggers(triggers)
+        self.trigger_handler.close()
+
+    def write_trigger_data(
+        self, stim_times: List[Tuple[str, float]], target_stimuli: str
+    ) -> None:
+        """Save trigger data to disk.
+
+        Parameters
+        ----------
+        - stim_times : list of (stim, clock_time) tuples
+        - target_stimuli : current target stimuli
+        """
+
+        if self.first_run:
+            # offset will factor in true offset and time relative from
+            # beginning
+            offset_triggers = []
+            for content_type, client in self.daq.clients_by_type.items():
+                label = offset_label(content_type.name)
+                time = (
+                    client.offset(self.matrix.first_stim_time) -
+                    self.matrix.first_stim_time
+                )
+                offset_triggers.append(
+                    Trigger(label, TriggerType.OFFSET, time))
+            self.trigger_handler.add_triggers(offset_triggers)
+
+        triggers = convert_timing_triggers(
+            stim_times, target_stimuli, self.trigger_type
+        )
+        self.trigger_handler.add_triggers(triggers)
 
 
-def init_display(
+def _init_matrix_display(
         parameters: Parameters,
         win: visual.Window,
         experiment_clock: Clock,