Source code for brainaccess_board.stream

import time

from pylsl import StreamInfo, StreamOutlet




class Stimulation:

    def __init__(
        self, name: str = "BrainAccessMarkers", source_id: str = "BrainAccessMarkers"
    ) -> None:
        """Initialize LSL marker device.

        Parameters
        ----------
        name : str, optional
            Name of the LSL stream, by default "BrainAccessMarkers"
        source_id : str, optional
            Source ID of the LSL stream, by default "BrainAccessMarkers"

        """
        self._info = StreamInfo(
            name=name,
            type="Markers",
            channel_count=1,
            nominal_srate=0,
            channel_format=3,  # "string",
            source_id=source_id,
        )
        self._outlet = StreamOutlet(self._info)
        time.sleep(1)



    def annotate(self, msg: str) -> None:
        """Send a marker to the LSL stream

        Parameters
        ----------
        msg : str
            Marker message to send to the LSL stream.

        """
        self._outlet.push_sample([msg])




    def have_consumers(self) -> bool:
        """Check if there are any consumers connected to the LSL stream

        Returns
        -------
        bool
            True if there are consumers connected to the LSL stream, False otherwise.
        """
        return self._outlet.have_consumers()