You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
175 lines
7.9 KiB
Python
175 lines
7.9 KiB
Python
1 year ago
|
# read the values from ESP32 and call functions to generate ps-file:
|
||
|
|
import serial # import serial library to read the values from ESP32 via serial (RX) connection
|
||
|
|
import os # library to use bash commands
|
||
|
|
|
||
|
|
#LED connect to GND and pin 23
|
||
|
|
import RPi.GPIO as GPIO # GPIO pins
|
||
|
|
import time
|
||
|
|
import logging
|
||
|
|
#logging gives information for debugging purposes
|
||
|
|
logging.getLogger().setLevel(5)
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
from print_button import print_button
|
||
|
|
from preview_button import preview_button
|
||
|
|
from friction_label import friction_label
|
||
|
|
from future_relics import future_relics
|
||
|
|
from weaver import weaver
|
||
|
|
from whoosh import whoosh
|
||
|
|
from dynamic_glyph import dynamic_glyph
|
||
|
|
from print_or_preview import preview_output, print_output
|
||
|
|
|
||
|
|
function_index = {
|
||
|
|
"FL": friction_label,
|
||
|
|
"FR": future_relics,
|
||
|
|
"WV": weaver,
|
||
|
|
"WH": whoosh,
|
||
|
|
"DG": dynamic_glyph
|
||
|
|
}
|
||
|
|
|
||
|
|
module_index = {
|
||
|
|
"FL": "friction_label",
|
||
|
|
"FR": "future_relics",
|
||
|
|
"WV": "weaver",
|
||
|
|
"WH": "whoosh",
|
||
|
|
"DG": "dynamic_glyph"
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
#LED
|
||
|
|
GPIO.setmode(GPIO.BOARD) #Use the physical board number (not GPIO numbering)
|
||
|
|
GPIO.setup(23, GPIO.OUT, initial=GPIO.LOW) # Set pin 23 to be an output pin and set initial value to low (off)
|
||
|
|
GPIO.setwarnings(False) #Ignore warnings
|
||
|
|
|
||
|
|
|
||
|
|
preview_button_count = 0
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
ser = serial.Serial('/dev/ttyS0', 9600, timeout=0.1) # establish a serial connection using Serial Pi Zero (S0), using the RX pin
|
||
|
|
ser.reset_input_buffer()
|
||
|
|
|
||
|
|
try:
|
||
|
|
while True:
|
||
|
|
#for some reason the whole ED setup has to be repeated here:
|
||
|
|
GPIO.setmode(GPIO.BOARD) #Use the physical board number (not GPIO numbering
|
||
|
|
GPIO.setup(23, GPIO.OUT, initial=GPIO.LOW) # Set pin 23 to be an output pin and set initial value to low (off)
|
||
|
|
GPIO.output(23, GPIO.LOW)
|
||
|
|
|
||
|
|
if ser.in_waiting > 0:
|
||
|
|
print(ser.in_waiting)
|
||
|
|
#the longer serial data is received, the bigger the data in the buffer
|
||
|
|
#set a max amount of buffer data, then reset the buffer:
|
||
|
|
if ser.in_waiting > 400:
|
||
|
|
ser.reset_input_buffer()
|
||
|
|
GPIO.output(23, GPIO.HIGH) # Turn on while serial data is incoming
|
||
|
|
|
||
|
|
line = ser.readline() # reads bytes
|
||
|
|
class theEnd(Exception): pass #defines an exception for the following try (in case, it is executed at the end).
|
||
|
|
try:
|
||
|
|
line = line.decode("utf-8", 'ignore').rstrip() # converts bytes to string
|
||
|
|
line = line.replace("\r\n", "") # replaces "\r\n" (end of line) with nothing
|
||
|
|
line = line.replace("\r", "") # replaces "\r" (between two modules) with nothing
|
||
|
|
#print(line)
|
||
|
|
if not line.startswith("&&&") and not line.endswith("&&&"): # break when line of data is not complete
|
||
|
|
#print(line)
|
||
|
|
#print("break")
|
||
|
|
break
|
||
|
|
line = line.replace("&&&", "") # delete all &&&
|
||
|
|
line = line.split("#") # splits the line into a list of lists
|
||
|
|
# after this split the list begins with an empty list, so delete it:
|
||
|
|
if line[0] == "": #if first list in lists is empty:
|
||
|
|
del line[0] #delete the first element
|
||
|
|
# print(line)
|
||
|
|
modules = []
|
||
|
|
for module in line:
|
||
|
|
values = module.split(",") # splits each module into values
|
||
|
|
modules.append(values)
|
||
|
|
print("These are the values of all connected modules ")
|
||
|
|
print(modules)
|
||
|
|
|
||
|
|
#for each module: check if all values are being received, if not jump to theEnd execption at the end of this code
|
||
|
|
for module in modules:
|
||
|
|
if module[0] == "FL" and len(module) != 9:
|
||
|
|
raise theEnd
|
||
|
|
if module[0] == "FR" and len(module) != 6:
|
||
|
|
raise theEnd
|
||
|
|
if module[0] == "WV" and len(module) != 6:
|
||
|
|
raise theEnd
|
||
|
|
if module[0] == "WH" and len(module) != 10:
|
||
|
|
raise theEnd
|
||
|
|
if module[0] == "DG" and len(module) != 2:
|
||
|
|
raise theEnd
|
||
|
|
#ser.reset_input_buffer()
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
if preview_button() == True:
|
||
|
|
|
||
|
|
#increase the preview button count to know if it is a preview start or a preview refresh
|
||
|
|
preview_button_count = preview_button_count + 1
|
||
|
|
|
||
|
|
previous_module = "" #define previous_module as non-existent
|
||
|
|
|
||
|
|
for module in modules: #for each module inside the list of modules
|
||
|
|
if module[0] not in module_index: #this if break is added to avoid the script from breaking at the start (which happens when serial is not yet receiving all values, but only fragments)
|
||
|
|
# print("break")
|
||
|
|
break # break and start the for loop again
|
||
|
|
|
||
|
|
current_module = module_index[module[0]] #save the name of the current module, use module_index to replace abbr. with full nam
|
||
|
|
function_index[module[0]](previous_module, current_module, module) #use function_index to run functions for each module #example: weaver(previous_module, current_module, module)
|
||
|
|
previous_module = current_module #previous_module is now redefined, bacause current_module becomes previous_module
|
||
|
|
|
||
|
|
if module == modules[-1]: #if module is the last module in list of modules
|
||
|
|
last_module = current_module
|
||
|
|
preview_output(preview_button_count, last_module) #call preview function
|
||
|
|
|
||
|
|
#delete text from text file that was written by dynamic glyph module
|
||
|
|
open('dynamic_glyph.txt', 'w').close()
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
if print_button() == True:
|
||
|
|
|
||
|
|
previous_module = "" #define previous_module as non-existent
|
||
|
|
|
||
|
|
for module in modules: #for each module inside the list of modules
|
||
|
|
if module[0] not in module_index: #this if break is added to avoid the script from breaking at the start (which happens when serial is not yet receiving all values, but only fragments)
|
||
|
|
# print("break")
|
||
|
|
break # break and start the for loop again
|
||
|
|
|
||
|
|
current_module = module_index[module[0]] #save the name of the current module, use module_index to replace abbr. with full name
|
||
|
|
function_index[module[0]](previous_module, current_module, module) #use function_index to run functions for each module #example: weaver(previous_module, current_module, module)
|
||
|
|
previous_module = current_module #previous_module is now redefined, bacause current_module becomes previous_module
|
||
|
|
|
||
|
|
if module == modules[-1]: #if module is the last module in list of modules
|
||
|
|
last_module = current_module #define current_module as last module to know what file to send to the preview/printer
|
||
|
|
print_output(last_module) #call print function
|
||
|
|
|
||
|
|
#delete text from text file that was written by dynamic glyph module
|
||
|
|
open('dynamic_glyph.txt', 'w').close()
|
||
|
|
|
||
|
|
#ser.flush()
|
||
|
|
|
||
|
|
|
||
|
|
except theEnd:
|
||
|
|
print("error reading") #prints error if serial does not receive all values of each module
|
||
|
|
|
||
|
|
except OSError as e:
|
||
|
|
print("Error: could not decode bytes")
|
||
|
|
print(e)
|
||
|
|
|
||
|
|
|
||
|
|
#ser.flush()
|
||
|
|
|
||
|
|
|
||
|
|
except KeyboardInterrupt: #is the script is interrupted, turn the LED off
|
||
|
|
GPIO.output(23, GPIO.LOW)
|
||
|
|
|
||
|
|
# finally:
|
||
|
|
# GPIO.cleanup()
|