Skip to content

Porting to python 3.11 #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
A-Akhil wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Porting to python 3.11 #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3367f8e
Changes the dict
A-Akhil Mar 15, 2024
539ee4a
Changes in encode decode
A-Akhil Mar 15, 2024
27ba34c
Changes in test.py
A-Akhil Mar 15, 2024
1c888d5
Update in README.md
A-Akhil Mar 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 73 additions & 33 deletions NeuroPy/NeuroPy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
STANDBY_SCAN = '\xd4'
RAW_VALUE = '\x80'


class NeuroPy(object):
"""NeuroPy libraby, to get data from neurosky mindwave.
Initialising: object1=NeuroPy("COM6",57600) #windows
Expand Down Expand Up @@ -80,13 +81,14 @@ class NeuroPy(object):

callBacksDictionary = {} # keep a track of all callbacks

def __init__(self, port = None, baudRate=57600, devid=None):
def __init__(self, port=None, baudRate=57600, devid=None):
if port == None:
platform = sys.platform
if platform == "win32":
port = "COM6"
elif platform.startswith("linux") or platform == 'darwin':
port = "/dev/rfcomm0"
print("in init only" + port)

self.__devid = devid
self.__serialPort = port
Expand All @@ -105,26 +107,41 @@ def __del__(self):
def disconnect(self):
self.__srl.write(DISCONNECT)

# def connect(self):
# if not self.__devid:
# self.__connected = True
# return # Only connect RF devices

# self.__srl.write(''.join([CONNECT, self.__devid.decode('hex')]))

def connect(self):
if not self.__devid:
self.__connected = True
return # Only connect RF devices
return # Only connect RF devices

encoded_devid = self.__devid.encode('utf-8')

self.__srl.write(''.join([CONNECT, self.__devid.decode('hex')]))
self.__srl.write(CONNECT.encode('utf-8') + encoded_devid)

def start(self):
# Try to connect to serial port and start a separate thread
# for data collection
if self.__threadRun == True:
print "Mindwave has already started!"
print("Mindwave has already started!")
return

if self.__srl == None:
# try:
# self.__srl = serial.Serial(
# self.__serialPort, self.__serialBaudRate)
try:
self.__srl = serial.Serial(
self.__serialPort, self.__serialBaudRate)
except serial.serialutil.SerialException, e:
print str(e)
# except serial.serialutil.SerialException, e:
# print str(e)
# return
except serial.serialutil.SerialException as e:
print(str(e))
return
else:
self.__srl.open()
Expand All @@ -142,25 +159,35 @@ def start(self):
def __packetParser(self):
"packetParser runs continously in a separate thread to parse packets from mindwave and update the corresponding variables"
while self.__threadRun:
p1 = self.__srl.read(1).encode("hex") # read first 2 packets
p2 = self.__srl.read(1).encode("hex")
# p1 = self.__srl.read(1).encode("hex") # read first 2 packets
# p2 = self.__srl.read(1).encode("hex")

p1 = self.__srl.read(1).hex() # read first 2 packets
p2 = self.__srl.read(1).hex()

while (p1 != 'aa' or p2 != 'aa') and self.__threadRun:
p1 = p2
p2 = self.__srl.read(1).encode("hex")
# p2 = self.__srl.read(1).encode("hex")
p2 = self.__srl.read(1).hex()

else:
if self.__threadRun == False:
break
# a valid packet is available
self.__packetsReceived += 1
payload = []
checksum = 0
payloadLength = int(self.__srl.read(1).encode("hex"), 16)
# payloadLength = int(self.__srl.read(1).encode("hex"), 16)
payloadLength = int(self.__srl.read(1).hex(), 16)
for i in range(payloadLength):
tempPacket = self.__srl.read(1).encode("hex")
tempPacket = self.__srl.read(1).hex()
# tempPacket = self.__srl.read(1).encode("hex")
payload.append(tempPacket)
checksum += int(tempPacket, 16)
checksum = ~checksum & 0x000000ff
if checksum == int(self.__srl.read(1).encode("hex"), 16):
# if checksum == int(self.__srl.read(1).encode("hex"), 16):
if checksum == int(self.__srl.read(1).hex(),
16): # read checksum byte and compare with calculated checksum
i = 0

while i < payloadLength:
Expand All @@ -171,28 +198,28 @@ def __packetParser(self):
elif (code == 'd1'):
print("Headset not found, reconnecting")
self.connect()
elif(code == 'd2'):
elif (code == 'd2'):
print("Disconnected!")
self.connect()
elif(code == 'd3'):
elif (code == 'd3'):
print("Headset denied operation!")
elif(code == 'd4'):
elif (code == 'd4'):
if payload[2] == 0 and not self.__connected:
print("Idle, trying to reconnect");
self.connect();
elif(code == '02'): # poorSignal
elif (code == '02'): # poorSignal
i = i + 1
self.poorSignal = int(payload[i], 16)
elif(code == '04'): # attention
elif (code == '04'): # attention
i = i + 1
self.attention = int(payload[i], 16)
elif(code == '05'): # meditation
elif (code == '05'): # meditation
i = i + 1
self.meditation = int(payload[i], 16)
elif(code == '16'): # blink strength
elif (code == '16'): # blink strength
i = i + 1
self.blinkStrength = int(payload[i], 16)
elif(code == '80'): # raw value
elif (code == '80'): # raw value
i = i + 1 # for length/it is not used since length =1 byte long and always=2
i = i + 1
val0 = int(payload[i], 16)
Expand Down Expand Up @@ -306,7 +333,8 @@ def attention(self):
def attention(self, value):
self.__attention = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("attention"):
# if self.callBacksDictionary.has_key("attention"):
if "attention" in self.callBacksDictionary:
self.callBacksDictionary["attention"](self.__attention)

# meditation
Expand All @@ -319,7 +347,8 @@ def meditation(self):
def meditation(self, value):
self.__meditation = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("meditation"):
# if self.callBacksDictionary.has_key("meditation"):
if "meditation" in self.callBacksDictionary:
self.callBacksDictionary["meditation"](self.__meditation)

# rawValue
Expand All @@ -332,7 +361,8 @@ def rawValue(self):
def rawValue(self, value):
self.__rawValue = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("rawValue"):
# if self.callBacksDictionary.has_key("rawValue"):
if "rawValue" in self.callBacksDictionary:
self.callBacksDictionary["rawValue"](self.__rawValue)

# delta
Expand All @@ -345,7 +375,8 @@ def delta(self):
def delta(self, value):
self.__delta = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("delta"):
# if self.callBacksDictionary.has_key("delta"):
if "theta" in self.callBacksDictionary:
self.callBacksDictionary["delta"](self.__delta)

# theta
Expand All @@ -358,7 +389,8 @@ def theta(self):
def theta(self, value):
self.__theta = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("theta"):
# if self.callBacksDictionary.has_key("theta"):
if "theta" in self.callBacksDictionary:
self.callBacksDictionary["theta"](self.__theta)

# lowAlpha
Expand All @@ -371,7 +403,8 @@ def lowAlpha(self):
def lowAlpha(self, value):
self.__lowAlpha = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("lowAlpha"):
# if self.callBacksDictionary.has_key("lowAlpha"):
if "lowAlpha" in self.callBacksDictionary:
self.callBacksDictionary["lowAlpha"](self.__lowAlpha)

# highAlpha
Expand All @@ -384,7 +417,8 @@ def highAlpha(self):
def highAlpha(self, value):
self.__highAlpha = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("highAlpha"):
# if self.callBacksDictionary.has_key("highAlpha"):
if "highAlpha" in self.callBacksDictionary:
self.callBacksDictionary["highAlpha"](self.__highAlpha)

# lowBeta
Expand All @@ -397,7 +431,8 @@ def lowBeta(self):
def lowBeta(self, value):
self.__lowBeta = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("lowBeta"):
# if self.callBacksDictionary.has_key("lowBeta"):
if "lowBeta" in self.callBacksDictionary:
self.callBacksDictionary["lowBeta"](self.__lowBeta)

# highBeta
Expand All @@ -410,7 +445,8 @@ def highBeta(self):
def highBeta(self, value):
self.__highBeta = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("highBeta"):
# if self.callBacksDictionary.has_key("highBeta"):
if "highBeta" in self.callBacksDictionary:
self.callBacksDictionary["highBeta"](self.__highBeta)

# lowGamma
Expand All @@ -423,7 +459,8 @@ def lowGamma(self):
def lowGamma(self, value):
self.__lowGamma = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("lowGamma"):
# if self.callBacksDictionary.has_key("lowGamma"):
if "lowGamma" in self.callBacksDictionary:
self.callBacksDictionary["lowGamma"](self.__lowGamma)

# midGamma
Expand All @@ -436,7 +473,8 @@ def midGamma(self):
def midGamma(self, value):
self.__midGamma = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("midGamma"):
# if self.callBacksDictionary.has_key("midGamma"):
if "midGamma" in self.callBacksDictionary:
self.callBacksDictionary["midGamma"](self.__midGamma)

# poorSignal
Expand All @@ -449,7 +487,8 @@ def poorSignal(self):
def poorSignal(self, value):
self.__poorSignal = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("poorSignal"):
if "poorSignal" in self.callBacksDictionary:
# if self.callBacksDictionary.has_key("poorSignal"):
self.callBacksDictionary["poorSignal"](self.__poorSignal)

# blinkStrength
Expand All @@ -462,5 +501,6 @@ def blinkStrength(self):
def blinkStrength(self, value):
self.__blinkStrength = value
# if callback has been set, execute the function
if self.callBacksDictionary.has_key("blinkStrength"):
# if self.callBacksDictionary.has_key("blinkStrength"):
if "blinkStrength" in self.callBacksDictionary:
self.callBacksDictionary["blinkStrength"](self.__blinkStrength)
38 changes: 20 additions & 18 deletions NeuroPy/test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,27 +4,26 @@
import time
from tqdm import tqdm


att = tqdm(total=100, desc="attention")
med = tqdm(total=100, desc="meditation")

object1=NeuroPy("/dev/ttyUSB0", 115200, '7d55')
object1 = NeuroPy("/dev/rfcomm0", 115200, '7d55')

volume = [50, 50, 50, 50, 50]

volume = [ 50, 50, 50, 50, 50]

def attention_callback(value):
"this function will be called everytime NeuroPy has a new value for attention"
# os.system("amixer sset 'Master' " + str(value) + "% > /dev/null")
# att.moveto(value)
# os.system("amixer sset 'Master' " + str(value) + "% > /dev/null")
# att.moveto(value)

# print
#do other stuff (fire a rocket), based on the obtained value of attention_value
#do some more stuff
# print
# do other stuff (fire a rocket), based on the obtained value of attention_value
# do some more stuff
return None


def meditation_callback(value):

volume.append(value)
volume.pop(0)
med.update(value)
Expand All @@ -35,19 +34,22 @@ def meditation_callback(value):
value = value + v
value = value / len(volume)

os.system("amixer sset 'Master' " + str(value) + "% > /dev/null")
#do other stuff (fire a rocket), based on the obtained value of attention_value
#do some more stuff
# os.system("amixer sset 'Master' " + str(value) + "% > /dev/null")
os.system("amixer sset 'PCM' " + str(value) + "% > /dev/null")

# do other stuff (fire a rocket), based on the obtained value of attention_value
# do some more stuff
return None

def blink_cb(value):
print "Blink ", value
print("Blink ", value)


#set call back:
object1.setCallBack("attention",attention_callback)
object1.setCallBack("meditation",meditation_callback)
object1.setCallBack("blinkStrength",blink_cb)
#call start method
# set call back:
object1.setCallBack("attention", attention_callback)
object1.setCallBack("meditation", meditation_callback)
object1.setCallBack("blinkStrength", blink_cb)
# call start method
object1.start()

try:
Expand Down
2 changes: 2 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ NeuroPy library written in python to connect, interact and get data from **Neuro

This library is based on the mindwave mindset communication protocol published by [Neurosky](http:://neurosky.com) and is tested with Neurosky Mindwave EEG headset.

This fork is working on python3 and works with mindwave mobile2 model

## Installation

1. Download the source distribution (zip file) from [dist directory](https://github.com/lihas/NeuroPy/tree/master/dist) or from [PyPi](https://pypi.python.org/pypi/NeuroPy/0.1)
Expand Down