OSC streaming with Python-OSC

[BoyNeedsHelp]

Thank you,

I'll run it from command prompt and see what arguments I can run. But could you tell me how I'd go about doing that? Or tell me where to look for examples? Sorry for all these questions, I just have no idea how to get what I need done.

[BoyNeedsHelp]

Thank you,

I'll run it from command prompt and see what arguments I can run. But could you tell me how I'd go about doing that? Or tell me where to look for examples? Sorry for all these questions, I just have no idea how to get what I need done.

[James]

You can Google "how to run command line programs" for a bunch of examples depending on what operating system you're using.

What exactly is your end goal? Perhaps there is an easier way? I find it a little unusual for someone to be programming python, but not sure how to run the code. If this is your very first time programming, then I would recommend that you go through some YouTube programming tutorials before jumping into EEG data processing over UDP network streams... it's jumping in the deep end a bit!

[BoyNeedsHelp]

Yeah, this is definitely a leap for a first project in python. I watched tutorials on Udemy (Zero to mastery in Machine Learning) and got into data analysis, but these are with CSV's and the project requires me to have live data fed and analysed in real time. So this is why I am trying to figure how to get OSC data into python. If you're interested in helping, I could go into more detail, but I get that it's a big ask. So thank you for being so responsive and helpful! Really appreciate it mate

[antonello]

Hi there, I managed to make things work. I'm using another python OSC library (https://github.com/kivy/oscpy) and it works well for my purposes. This is a little snippet I use to acquire EEG signals from the MindMonitor App, it could be a good starting point, but as pointed out by James, if this will be your first Python program it will be quite difficult to understand at the beginning.
Good Luck

Code: Select all

from oscpy.server import OSCThreadServer
from time import sleep

def eeg_callback(*values):
	print("got values: {}".format(values))

osc = OSCThreadServer()  # See sources for all the arguments
sock = osc.listen(address='your_ip_address', port=7000, default=True)
osc.bind(b'/muse/eeg', eeg_callback)

sleep(1000)

osc.stop()  # Stop the default socket
 
osc.stop_all()  # Stop all sockets
 
# Here the server is still alive, one might call osc.listen() again
 
osc.terminate_server()  # Request the handler thread to stop looping
 
osc.join_server()  # Wait for the handler thread to finish pending tasks and exit

[BoyNeedsHelp]

Hi Antonello,

Thanks so much, I'll be sure to try it out and get as much help as I can from people at my college.

[BoyNeedsHelp]

Is there any way you could share a screenshot of you running this? Just don't know what I'm supposed to be looking at and whether or not I've put everything in the right place.

[Peter Gamma]

Is there a way to stream sensor data of the Muse 1 to Linux? For the Muse 2 there is Alexandre Brachands Muse LSL which also works under Linux. But for Muse 1? If there would be an easy understandable way to receive OSC in Python, I would change to the Linux Pinephone.

[hobbyhack]

In case someone else lands here, here is some starter python server code using the python-osc library.

There are examples of handling specific messages and also some commented out lines if you want to catch everything. This is just a start and won't really do much except tell you when you blink or clench your teeth. However, that should be plenty to know the server is working. Obviously, install the python-osc library change the ip address to your server. No other python3 dependencies were needed.

Code: Select all

from pythonosc.dispatcher import Dispatcher
from pythonosc.osc_server import BlockingOSCUDPServer

IP = "192.168.1.100"
PORT = 5000

def muse_handler(address, *args):
    """handle all muse data"""
    print(f"{address}: {args}")

def eeg_handler(address, *args):
    """handle raw eeg data"""
    print(f"{address}: {args}")

def battery_handler(address, *args):
    """handle battery data"""
    level, batt_voltage, adc_voltage, temp = args
    print(f"Battery level: {level}")

def blink_handler(address, *args):
    """handle blink data"""

    # Note: when the jaw stays clenched blink message is spammed

    # it seems to only send a 1 when a blink is detected
    # so no reason to do much to process the data
    print("Blink detected")

def jaw_handler(address, *args):
    """handle jaw clench data"""

    # it seems to only send a 1 when a clench is detected
    # so no reason to do much to process the data
    print("Jaw Clench detected")

def marker_handler(address, *args):
    """handle marker data"""

    # this didn't do anything for me.  Might be because I have
    # the S model
    print(f"{address}: {args}")

def default_handler(address, *args):
    print(f"DEFAULT {address}: {args}")

def get_dispatcher():
    dispatcher = Dispatcher()
    # dispatcher.map("/muse/batt", battery_handler)
    dispatcher.map("/muse/elements/blink", blink_handler)
    dispatcher.map("/muse/elements/jaw_clench", jaw_handler)
    # dispatcher.map("/Marker/*", marker_handler)
    
    # this will handle any unidentified messages if you would like
    # dispatcher.set_default_handler(default_handler)

    # use this dispatched to handle anythin muse
    # dispatcher.map("/muse/*", muse_handler)

    return dispatcher

    # TODO
    # dispatcher.map("/muse/*", muse_handler)
    # dispatcher.map("/muse/eeg", eeg_handler)
    # dispatcher.map("/muse/elements/delta_absolute", delta_absolute_handler)
    # dispatcher.map("/muse/elements/theta_absolute", theta_absolute_handler)
    # dispatcher.map("/muse/elements/alpha_absolute", alpha_absolute_handler)
    # dispatcher.map("/muse/elements/beta_absolute", beta_absolute_handler)
    # dispatcher.map("/muse/elements/gamma_absolute", gamma_absolute_handler)
    # dispatcher.map("/muse/elements/horseshoe", horseshoe_handler)
    # dispatcher.map("/muse/elements/touching_forehead", touching_forehead_handler)
    # dispatcher.map("/muse/gyro", gyro_handler)
    # dispatcher.map("/muse/acc", accel_handler)
    # dispatcher.set_default_handler(default_handler)


def start_blocking_server(ip, port):
    server = BlockingOSCUDPServer((ip, port), dispatcher)
    server.serve_forever()  # Blocks forever

if __name__ == '__main__':
    dispatcher = get_dispatcher()
    start_blocking_server(IP, PORT)

I hope this is useful to someone.

[James]

If you set it to listen on "127.0.0.1" it should listen on all available network interfaces.