Session

SessionConfig dataclass

SessionConfig(user_name: str = None, user_age: int = None, user_gender: str = None, classes: list = lambda: ['default'](), trials: int = 1, baseline_duration: float = 15, rest_duration: float = 2, ready_duration: float = 1, cue_duration: float = 1.5, motor_duration: float = 4, extra_duration: float = 0, save_dir: str = './sessions/', capture_blinks: bool = False)

Configuration settings for a session.

Attributes:

Name	Type	Description
user_name	str	Name of the user.
user_age	int	Age of the user.
user_gender	str	Gender of the user.
classes	list	List of class names for the session.
trials	int	Number of trials per class.
baseline_duration	float	Duration of the baseline phase in seconds.
rest_duration	float	Duration of the rest phase in seconds.
ready_duration	float	Duration of the ready phase in seconds.
cue_duration	float	Duration of the cue phase in seconds.
motor_duration	float	Duration of the motor phase in seconds.
extra_duration	float	Additional random duration added to motor phase.
save_dir	str	Directory to save session data.
capture_blinks	bool	Whether to capture blink events.

Session

Session(headset: MindWaveMobile2, config: SessionConfig, lazy_start: bool = True)

Manages EEG headset data collection sessions.

This class handles the configuration, execution, and data storage of EEG data collection sessions.

It also emits signals at different stages of the session, this can be used to build a UI on top of it.

Session flow:

• Session Start signal: Indicates the start of the session.
• Baseline Start signal: Indicates the start of the baseline phase.
• Baseline End signal: Indicates the end of the baseline phase. for trial in range(trials*len(classes)):
  • Trial Start signal: Indicates the start of a new trial associated with a class.
  • Rest signal: Indicates the rest phase before starting the motor imagery task.
  • Ready signal: Indicates the ready phase before the cue.
  • Cue signal: Indicates the cue phase for the motor imagery task.
  • Motor signal: Indicates the start of the motor imagery task.
  • Trial End signal: Indicates the end of the trial associated with a class.
• Session End signal: Indicates the end of the session.

Parameters:

Name	Type	Description	Default
headset	MindWaveMobile2	The MindWave Mobile 2 headset instance.	required
config	SessionConfig	The configuration settings for the session.	required
lazy_start	bool	Whether to start the session immediately after initialization.	True

Source code in mindwave/session.py

def __init__(
    self,
    headset: MindWaveMobile2,
    config: SessionConfig,
    lazy_start: bool = True,
) -> None:
    """Initializes a new Session instance.

    Args:
        headset (MindWaveMobile2): The MindWave Mobile 2 headset instance.
        config (SessionConfig): The configuration settings for the session.
        lazy_start (bool, optional): Whether to start the session immediately after initialization.
    """
    self._logger = Logger.get_logger(self.__class__.__name__)
    self.headset = headset
    self.config = config

    self.start_time = None
    self.end_time = None
    self.is_active = False
    self.is_finished = False

    self._event_manager = EventManager()
    self._save_dir = None
    self._data_subscription = None
    self._blinks_subscription = None
    self._data = []
    self._events = []
    self._stop_flag = threading.Event()

    if not lazy_start:
        self.start()

    random.seed(time.perf_counter())

start

start() -> None

Starts the data collection session.

NOTE: The headset must be running before starting the session.

Source code in mindwave/session.py

def start(self) -> None:
    """Starts the data collection session.

    NOTE: The headset must be running before starting the session.
    """
    if not self.headset.is_running:
        self._logger.info("Headset is not running!, start headset before starting the session")
        return
    if self.is_active:
        self._logger.info("Session is already active!")
        return
    if self.is_finished:
        self._logger.info("Session is finished!")
        return

    self._data_subscription = self.headset.on_data(self._data_collator)
    if self.config.capture_blinks:
        self._blinks_subscription = self.headset.on_blink(self._data_collator)

    self._create_user_dir()
    self._save_info()

    self.is_active = True
    self.start_time = datetime.now()
    self._stop_flag.clear()

    thread = threading.Thread(target=self._events_processor, daemon=True)
    thread.start()

    self._logger.info(f"New Session started at {self.start_time.strftime('%H:%M:%S')}")

stop

stop() -> None

Stops the data collection session.

Source code in mindwave/session.py

def stop(self) -> None:
    """Stops the data collection session."""
    if not self.is_active:
        self._logger.info("Session is not active!")
        return

    self._stop_flag.set()
    self.end_time = datetime.now()

    self.is_active = False
    self.is_finished = True

    self._data_subscription.detach()
    if self.config.capture_blinks:
        self._blinks_subscription.detach()

    self._logger.info(f"Session ended at {self.end_time.strftime('%H:%M:%S')}")

save

save()

Saves the collected session data and events to disk.

Source code in mindwave/session.py

def save(self):
    """Saves the collected session data and events to disk."""
    if self.is_active:
        self._logger.info("Session is still active!, stop the session before saving")
        return
    if len(self) == 0:
        self._logger.info("Session data is empty!, nothing to save")
        return

    if not self._save_dir:
        self._create_user_dir()

    filename = f"{self._save_dir}/data.csv"
    with open(filename, mode="w", newline="", encoding="utf-8") as file:
        writer = csv.DictWriter(file, fieldnames=self._data[0].keys())
        writer.writeheader()
        writer.writerows(self._data)

    self._save_events()

    self._logger.info(f"Session collected data is saved to {self._save_dir}")

on_signal

on_signal(listener: Callable[[SessionEvent], Any]) -> Subscription

Registers a listener for session signals.

Parameters:

Name	Type	Description	Default
listener	Callable[[SessionEvent], Any]	The listener function to be called when a session event is emitted.	required

Returns:

Name	Type	Description
Subscription	Subscription	A Subscription object that can be used to unsubscribe the listener.

Source code in mindwave/session.py

def on_signal(self, listener: Callable[[SessionEvent], Any]) -> Subscription:
    """Registers a listener for session signals.

    Args:
        listener: The listener function to be called when a session event is emitted.

    Returns:
        Subscription: A Subscription object that can be used to unsubscribe the listener.
    """
    return self._event_manager.add_listener(EventType.SessionEvent, listener)