Skip to content
Snippets Groups Projects
Commit b90afb9e authored by Kai Junge's avatar Kai Junge
Browse files

Removed testing code

parent a6201e65
Branches
Tags
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Testing code/comms_wrapper_tester.py deleted 100644 → 0
+ 0
− 34
View file @ a6201e65
from matplotlib.pyplot import disconnect
from comms_wrapper_v1 import *
def main():
key = Key()
arduino = Arduino(descriptiveDeviceName="Uno1", portName="/dev/ttyACM0", baudrate=115200)
arduino.connect_and_handshake()
arduino.add_new_message_name("item1")
arduino.add_new_message_name("item2")
arduino.add_new_message_name("item3")
while 1:
if arduino.recieve_message():
print(arduino.recievedMessages)
time.sleep(0.2)
# simple key commands for testing
if key.keyPressValue == keyboard.Key.esc:
print("break from loop")
arduino.disconnect_arduino()
break
if key.keyPressValue == "y":
arduino.send_message("On")
if key.keyPressValue == "t":
arduino.send_message("Off")
if __name__ == '__main__':
main()
\ No newline at end of file
Testing code/comms_wrapper_v1.py deleted 100644 → 0
+ 0
− 214
View file @ a6201e65
import serial
import serial.tools.list_ports
from threading import Thread
import time
from pynput import keyboard
########
### Arduino communication
########
class Arduino:
def __init__(self, descriptiveDeviceName, portName, baudrate):
self.descriptiveDeviceName = descriptiveDeviceName
self.portName = portName
self.baudrate = baudrate
self._rawRecievedMessage = None
self._prevMessage = None
self.recievedMessages = {}
self.messageNames = list(self.recievedMessages.keys())
self.arduino = None
self.handshakeStatus = False
def _serial_readline(self):
while 1:
try:
self._rawRecievedMessage = self.arduino.readline().decode('utf-8')[:-2]
except:
pass
time.sleep(0.001)
def _startReadingThread(self):
thread = Thread(target=self._serial_readline)
thread.start()
def _serial_write(self, msg):
if self.arduino is not None:
self.arduino.write(bytes(msg, 'utf-8'))
def add_new_message_name(self, msgName):
if msgName in self.recievedMessages:
print("!! Message name ", msgName, " already exists so it cannot be added. !!")
else:
self.recievedMessages[msgName] = None
self.messageNames = list(self.recievedMessages.keys())
def delete_message_name(self, msgName):
if msgName in self.recievedMessages:
del self.recievedMessages[msgName]
self.messageNames = list(self.recievedMessages.keys())
else:
print("!! Message name ", msgName, " does not exist so it cannot be deleted. !!")
def _just_connect(self):
# Connect to the arduino device
try:
self.arduino = serial.Serial(port=self.portName, baudrate=self.baudrate)
# toggle dtr to reset the arduino
self.arduino.dtr = True
self.arduino.dtr = False
print("Successfully connected to " + self.descriptiveDeviceName)
except:
print("!! Cannot connect to " + self.descriptiveDeviceName + " !!")
return False
self.arduino.dtr = True
self.arduino.dtr = False
def disconnect_arduino(self):
self.arduino.close()
def connect_and_handshake(self):
# Connect to the arduino device
try:
self.arduino = serial.Serial(port=self.portName, baudrate=self.baudrate)
# toggle dtr to reset the arduino
self.arduino.dtr = True
self.arduino.dtr = False
print("Successfully connected to " + self.descriptiveDeviceName)
except:
print("!! Cannot connect to " + self.descriptiveDeviceName + " !!")
return False
# Start the reading thread
self._startReadingThread()
# Wait for a bit for the arduino to initialise nicely
time.sleep(0.5)
# Conduct the handshake process
timeoutTimer = time.time()
handshakeTimeoutSec = 5
self.arduino.reset_input_buffer()
self.arduino.reset_output_buffer()
while time.time() - timeoutTimer < handshakeTimeoutSec:
self._serial_write("handshake1\n")
time.sleep(0.1)
if self._rawRecievedMessage == "handshake2":
self.handshakeStatus = True
break
if self.handshakeStatus:
print("Successfull handshake with " + self.descriptiveDeviceName)
else:
print("!! Handshake failed with " + self.descriptiveDeviceName + " !!")
return self.handshakeStatus
def send_message(self, msg):
self._serial_write( msg + "\n")
def recieve_message(self):
if not self.handshakeStatus:
print("!! Handshake not completed !!")
return False
else:
isMessageValid = True
msg = self._rawRecievedMessage
# sanity check 1: check if ends of the message are < and >
try:
if msg[0] == "<" and msg[-1] == ">":
pass
else:
isMessageValid = False
except:
isMessageValid = False
if isMessageValid:
isMessageValid = self.disect_and_save_message(msg)
else:
pass
return True
def disect_and_save_message(self, msg):
recievedMessageTemp = self.recievedMessages
if msg[-2] != ":":
return False
msg = msg[1:-2]
splitMsg = msg.split(":")
for singleMsg in splitMsg:
if len(singleMsg.split(";")) == 2:
msgName = singleMsg.split(";")[0]
msgPayload = singleMsg.split(";")[1]
if msgName not in self.messageNames:
return False
recievedMessageTemp[msgName] = msgPayload
else:
return False
self.recievedMessages = recievedMessageTemp
return True
class Key():
def __init__(self):
self.keyPressValue = None
self.keyReleaseValue = None
self.start_keyboard_listener()
def on_press(self, key):
try:
self.keyPressValue = key.char
except AttributeError:
self.keyPressValue = key
def on_release(self, key):
try:
self.keyReleaseValue = key.char
except AttributeError:
self.keyReleaseValue = key
def start_keyboard_listener(self):
listener = keyboard.Listener(on_press=self.on_press, on_release=self.on_release)
listener.start()
print("keyboard listener started")
########
### Other useful functions
########
def list_connected_serial_devices():
devices = list(serial.tools.list_ports.comports())
for device in devices:
print(device)
\ No newline at end of file
Testing code/comms_wrapper_v1/comms_wrapper_v1.ino deleted 100644 → 0
+ 0
− 105
View file @ a6201e65
#define LEDPIN 13
void setup() {
pinMode(LEDPIN, OUTPUT);
digitalWrite(LEDPIN, LOW);
Serial.begin(115200);
init_python_communiction();
}
bool LED = false;
long timer = millis();
long globalTimer = millis();
String msgName[] = {"pressure", "time", "command"};
String Data[] = {"Null", "Null", "Null"};
int numOfMsg = 3;
String rawMsg = "NULL";
void loop() {
Data[0] = "20";
//Data[1] = String(millis() - globalTimer);
Data[2] = "go";
timer = millis();
rawMsg = recieve_msg(rawMsg);
Data[1] = String(millis() - timer);
if(rawMsg != "None") {
if(rawMsg == "On") {
digitalWrite(LEDPIN, HIGH);
}
else {
digitalWrite(LEDPIN, LOW);
}
}
Data[2] = rawMsg;
send_msg(msgName, Data, numOfMsg);
}
void send_msg(String* msgName, String* msg, int numOfMsg) {
// If we have the same number of data compared to the message
String payload = "<";
for(int i = 0; i < numOfMsg; i++) {
payload.concat(msgName[i]);
payload.concat(";");
payload.concat(msg[i]);
payload.concat(":");
}
payload.concat(">");
Serial.println(payload);
}
String recieve_msg(String prevMsg) {
String msg = "";
while(Serial.available()) {
if(Serial.available()) {
msg = Serial.readStringUntil('\n');
}
}
if(msg == "") {
msg = prevMsg;
}
return msg;
}
void init_python_communiction() {
while(true){
// if the python side sent a message
if(Serial.available() > 0) {
String rawMsg = Serial.readStringUntil('\n');
if(rawMsg == "handshake1") {
break;
}
}
}
long timer = millis();
while(millis() - timer < 1000) {
Serial.println("handshake2");
}
while(Serial.available()) {
Serial.read();
}
}
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment