So i am looking into multi threading for a purpose, The Raspberry pi Pico, so am not starting on micropython as its rather limited and little info is about for the Pico right now so yeah, pi-top[4] it is and also this helps others out
Idea
So the idea is to have 1 thread to get data from a sensor and another thread to print the data to the console for step 1 and did not want it to be bloated.
Solution
The solution to this is, obviously use multi threading but also add a queue system
Thread 1: Get sensor data and pass it to the queue
Thread 2: wait for data to be added to the queue and print the data to the console
I have never used threading or queue before so a lot of this was trial and error, starting with a counter and printing the count though threading and queue. and I worked it all out.
Result
Thread 1: gets GPS data (GPGGA, GPGSA, GPRMC, GPZDA) and add them to the queue
Thread 2: waits till the queue is not empty and prints each queue item to the console (GPS data) if the queue is empty then print “waiting for work”, and had to add a sleep timer in there to slow down the thread a little
What i have done here is add some logging to see what is going on and here is the code that was run
import queue, threading, io, serial, pynmea2
from time import sleep
import logging
# Serial setup
gpsSerial = serial.Serial('/dev/ttyS0', 9600, timeout=1.)
gpsIO = io.TextIOWrapper(io.BufferedRWPair(gpsSerial, gpsSerial))
# Enable only NMEA sentences required
gpsSerial.write(b'$PMTK314,0,1,0,1,1,0,0,0,0,0,0,0,0,0,0,0,0,1*34\r\n')
nmea_list = ['GGA', 'GSA', 'RMC', 'ZDA']
q = queue.Queue()
def SensorThread(q,name):
while True:
logging.debug(f'{name} Thread: Started')
try:
logging.info(f'{name} Thread: Reading GPS line Data')
line = gpsIO.readline()
logging.info(f'{name} Thread: Parsing GPS line Data')
msg = pynmea2.parse(line)
logging.info(f'{name} Thread: Checking if NMEA Sentence is wanted')
if msg.sentence_type in nmea_list:
logging.info(f'{name} Thread: Adding NMEA Sentence to work queue')
q.put(msg)
except serial.SerialException as e:
print('Device error: {}'.format(e))
break
except (pynmea2.ParseError, AttributeError) as e:
continue
def PrintThread(q,name):
logging.info(f'{name} Thread: Started')
while True:
logging.info(f'{name} Thread: checking to see if the work queue is empty')
if not q.empty():
logging.info(f'{name} Thread: printing work')
print(f'Print Thread: {q.get()}')
sleep(0.1)
if __name__ == "__main__":
format = "%(asctime)s: %(message)s"
logging.basicConfig(format=format, level=logging.INFO,
datefmt="%H:%M:%S")
logging.info(f'Creating Sensor Thread')
qt = threading.Thread(target=SensorThread, args=(q,'Sensor'))
logging.info(f'Creating Print Thread')
pt = threading.Thread(target=PrintThread, args = (q,'Print'))
logging.info(f'Starting Sensor Thread')
qt.start()
logging.info(f'Starting Sensor Thread')
pt.start()
More development for this will continue on this till get it to display GPS information on the display, may even add functionality to use the mini screen buttons to scroll though different data
