polycortex / pymuse Goto Github PK

View Code? Open in Web Editor NEW

58.0 18.0 16.0 12.18 MB

Python tools associated with Muse headband

Home Page: http://polycortex.polymtl.ca/

License: MIT License

Python 100.00%

muse-headset python eeg-displayer muse-data

pymuse's Introduction

pyMuse

This repository contains tools for gathering and processing Muse signals using Python. With pyMuse you can configure and run an EEG pipeline faster than light.

Features

🚧 Pymuse is still a work in progress. But we aim to have a flexible and extensible architecture so you can use it to your own sauce.

For now, pyMuse already contains a lot of features:

EEG acquisition methods like MuseOSCInputStream to get data from MuseDirect.
Pipeline creation is extensible and modulable to your own needs. If you need a need a custom stage for your pipeline, just inherit from PipelineStage.
Pipeline can be forked to process multiple outputs from a same source signal.
Each PipelineStage has its own thread and are automatically linked together by thread-safe queues. They provides hooks for their initialization and their shutdown.

Of course, many more will come in the following months. Stay tuned.

Example

from pymuse.pipeline import Pipeline
from pymuse.configureshutdown import configure_shutdown
from pymuse.inputstream.muse_osc_input_stream import MuseOSCInputStream
from pymuse.pipelinestages.outputstream.muse_csv_output_stream import MuseCSVOutputStream

muse_osc_input_stream = MuseOSCInputStream(['eeg', 'beta_relative']) # Signal acquisition module
pipeline = Pipeline( # Pipeline modules are automagically linked together
muse_osc_input_stream.get_signal('eeg'),
MuseCSVOutputStream("recorded_eeg.csv")
)
# Ensure resources are freed when application is shutted down
configure_shutdown(muse_osc_input_stream, pipeline)
pipeline.start()
muse_osc_input_stream.start()

Installation & dependencies

You will need a few tools to get started with the Muse headset and pyMuse:

MuseDirect (Windows)
Python 3.7

Do not hesitate to visit the Muse Developer website for additional information and to access the docs.

Installation of this package

Ensure you have downloaded the correct Python distribution and the pyMuse package (e.g., using git clone).

Open a terminal, go into the package and type:

pip install -r requirements.txt.

The installer should install all requirements, including:

numpy
scipy
matplotlib
python-osc > 1.7.0

pymuse's People

Contributors

Stargazers

Watchers

Forkers

gabriel-amyot twuilliam cypherkhris silvernader pumpikano ma5onic endeffects openmindrobotika osasai sky7360-zz appliedneuro alexandremarcotte lblogan14 lisabpink xczhanjun neurux

pymuse's Issues

Add a visualization interface

The visualization interface should be able to generate visual stimuli and coordinate with the analysis pipeline.

Typical use for the P300 project, the visualization window would contain several boxes with a letter or a drawing, flashing at different (or unique) frequencies. The flashing sequence should be transmitted to the analysis pipeline in order to train a event detection model, based on the P300 ERP phenomenon.

Related to issue #9.

Add event detection in analysis pipeline

Event detection can be made by learning features based on a machine learning system.
We should add two steps:

feature extraction module
learning event detection model

Add IO for OpenBCI

Take inspiration from:

import sys;

sys.path.append('..')  # help python find open_bci_v3.py relative to scripts folder
import open_bci_v3 as bci
import os
import logging
import time
import timeit
from threading import Thread

# counter for sampling rate
nb_samples_out = -1


# try to ease work for main loop
class Monitor(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.nb_samples_out = -1

        # Init time to compute sampling rate
        self.tick = timeit.default_timer()
        self.start_tick = self.tick
        self.polling_interval = 10

    def run(self):
        while True:
            # check FPS + listen for new connections
            new_tick = timeit.default_timer()
            elapsed_time = new_tick - self.tick
            current_samples_out = nb_samples_out
            print "--- at t: ", (new_tick - self.start_tick), " ---"
            print "elapsed_time: ", elapsed_time
            print "nb_samples_out: ", current_samples_out - self.nb_samples_out
            sampling_rate = (current_samples_out - self.nb_samples_out) / elapsed_time
            print "sampling rate: ", sampling_rate
            self.tick = new_tick
            self.nb_samples_out = nb_samples_out
            time.sleep(self.polling_interval)


tick = timeit.default_timer()
start_tick = tick

def printData(sample):
    global tick, start_tick
    # os.system('clear')
    print "----------------"
    print("%f" % (sample.id))
    print sample.channel_data
    print sample.aux_data
    print "----------------"

    new_tick = timeit.default_timer()
    elapsed_time = new_tick - tick
    current_samples_out = nb_samples_out
    print "--- at t: ", (new_tick - start_tick), " ---"
    print "elapsed_time: ", elapsed_time
    print "nb_samples_out: ", current_samples_out - nb_samples_out
    sampling_rate = (current_samples_out - nb_samples_out) / elapsed_time
    print "sampling rate: ", sampling_rate
    tick = new_tick


if __name__ == '__main__':
    port = '/dev/tty.usbserial-DB00MF30'
    # port = '/dev/tty.OpenBCI-DN0096XA'
    baud = 115200
    logging.basicConfig(filename="test.log", format='%(asctime)s - %(levelname)s : %(message)s', level=logging.DEBUG)
    logging.info('---------LOG START-------------')

    monit = Monitor()
    # if len(self.args) > 0:
    #   monit.polling_interval = float(self.args[0])
    # daemonize thread to terminate it altogether with the main when time will come
    monit.daemon = True
    monit.start()

    board = bci.OpenBCIBoard(port=port, scaled_output=False, log=True)
    print("Board Instantiated")
    board.ser.write('v')
    print 'Sample Rate: ', board.getSampleRate()
    time.sleep(10)

    tick = timeit.default_timer()
    start_tick = tick

    board.start_streaming(printData)
    # board.print_bytes_in()