High-level abstraction for serial communication

小能豆

High-level abstraction for serial communication

On a hardware device with serial communication, I need to continously query “foo”, forever, every second (for example to log temperature of the device). At a random timing, another thread may query “bar”.

One option is to interrupt the “foo” queries, ask for “bar”, and once done, re-start the “foo” queries again - but I’m looking for a more general solution.

Is there a way to make a high-level abstraction, allowing us to just do:

def thread1():
    while True:
        serial_device_abstraction.get("foo")
        time.sleep(1)

def thread2()
    time.sleep(random.random())
    serial_device_abstraction.get("bar")

threading.Thread(target=thread1).start()
threading.Thread(target=thread2).start()

and letting the high-level abstraction handle the low-level concurrency problems?

NB: obviously, with

def get(query):                      
   serial_port.write(query)          
   data = self.serial_port.read(8)

it won’t work, because if two threads send bits through the serial wire at the same time, it will send/receive corrupted data.

阅读 92

2023-11-27

共1个答案

小能豆

To handle concurrency issues when dealing with serial communication, you can use locks to ensure that only one thread at a time accesses the serial port. Python provides the threading.Lock class for this purpose. Here’s an example of how you can modify your code:

import threading
import time
import random

class SerialDeviceAbstraction:
    def __init__(self):
        self.serial_lock = threading.Lock()

    def get(self, query):
        with self.serial_lock:
            # Perform serial communication safely
            serial_port.write(query)
            data = serial_port.read(8)
            return data

serial_device_abstraction = SerialDeviceAbstraction()

def thread1():
    while True:
        result = serial_device_abstraction.get("foo")
        # Process result as needed
        time.sleep(1)

def thread2():
    time.sleep(random.random())
    result = serial_device_abstraction.get("bar")
    # Process result as needed

# Start threads
threading.Thread(target=thread1).start()
threading.Thread(target=thread2).start()

In this example, the SerialDeviceAbstraction class includes a lock (self.serial_lock). The get method uses a with statement to acquire and release the lock automatically, ensuring that only one thread can access the serial port at any given time.

This approach helps prevent data corruption and ensures safe concurrent access to the serial device. Keep in mind that the specifics may depend on your serial port implementation and whether it supports concurrent access.

2023-11-27