Блог

Я использую pySerial для связи с микроконтроллером через USB, большая часть связи инициируется настольным скриптом python, который отправляет пакет команд и ожидает ответа.

Но есть также пакет предупреждений, который может быть отправлен микроконтроллером без команды из скрипта python. В этом случае мне нужно отслеживать поток чтения для любых предупреждений.

Для обработки предупреждений я выделяю отдельный процесс для вызова readline() и обхода его, например:

     def serialMonitor(self):

        while not self.stopMonitor:
            self.lock.acquire()
            message = self.stream.readline()
            self.lock.release()
            self.callback(message)
  
 

внутри класса. Затем функция запускается в отдельном процессе с помощью

 self.monitor = multiprocessing.Process(target = SerialManager.serialMonitor, args = [self])
 

Всякий раз, когда отправляется командный пакет, командная функция должна вернуть контроль над потоком, для чего она должна прервать readline() вызов, находящийся в блокировке. Как мне прервать readline() вызов? Есть ли какой-либо способ безопасно завершить процесс?

Вы можете завершить многопроцессорный процесс с .terminate() помощью . Это безопасно? Вероятно, это нормально для случая с читаемой строкой.

Однако я бы не стал так поступать здесь. Когда я читаю ваш сценарий, есть две возможности:

• MCU инициирует пакет предупреждений
• Компьютер отправляет данные в микроконтроллер (и микроконтроллер, возможно, отвечает)

Я предполагаю, что MCU не отправит пакет предупреждений, пока происходит обмен, инициированный компьютером.

Поэтому я бы просто инициировал последовательный объект с небольшим таймаутом и оставил его в цикле, когда я его не использую. Мой общий поток будет выглядеть так:

 ser = Serial(PORT, timeout=1)
response = None
command_to_send = None
running = True
while running: # event loop
    while running and not command_to_send and not line:
        try:
            line = ser.readline()
        except SerialTimeoutException:
            pass

    if not command_to_send:
        process_mcu_alert(line)

    else:
        send_command(command_to_send)
        command_to_send = None
        response = ser.readline()
 

Это всего лишь набросок, так как его нужно будет запускать в потоке или подпроцессе, поскольку readline() он действительно блокируется, поэтому вам нужен какой-то потокобезопасный способ настройки command_to_send и running (используется для изящного выхода) и получения response , и вы, вероятно, захотите объединить все это состояние в класс. Точная реализация зависит от того, что вы делаете, но принцип остается тем же самым—есть один цикл, который обрабатывает чтение и запись в последовательный порт, это тайм-аут, чтобы реагировать достаточно быстро (вы можете установить меньший тайм-аут, Если вам нужно), и он не подвергайте какой интерфейс вы можете справиться.

К сожалению, насколько мне известно, в python нет совместимой с asyncio последовательной библиотеки, иначе такой подход казался бы более аккуратным.