Возникли проблемы с настройкой последовательного соединения в Python ⇐ Python
-
Anonymous
Возникли проблемы с настройкой последовательного соединения в Python
Я создаю приложение Python, которое подключается к специальному оборудованию, работающему на Raspberry Pi. Я написал небольшую библиотеку, которая переводит все сообщения, поступающие туда и обратно на последовательное устройство. Последовательное устройство представляет собой простой адаптер USB-Serial (RS485). Проблема, с которой я столкнулся, связана с тем, что когда я пытаюсь отключить и снова подключить адаптер, код больше не работает.
[*]Сначала я извлекаю доступные устройства из /dev/serial/by-id:
импортировать ОС импортировать Serial.tools.list_ports как порты защита get_serial_number_for_device(device_name): """ Получите серийный номер конкретного устройства по его имени. Аргументы: имя_устройства (str): имя устройства, для которого нужно найти серийный номер. Возврат: str или None: серийный номер устройства, если оно найдено, или None, если оно не найдено. """ # Получить все текущие данные из последовательной библиотеки Python компорты = порты.comports() # Просматриваем текущие порты для порта в компортах: # Если имя устройства предоставленного аргумента совпадает с именем в списке портов если порт.устройство == имя_устройства: # возвращаем серийный номер порта вернуть порт.серийный_номер # Если устройство не найдено, верните None возврат Нет защита get_serial_devices(): """ Получите список действительных серийных устройств и соответствующие им серийные номера. Возврат: список: список кортежей, содержащий абсолютные пути к действительным последовательным устройствам и соответствующие им серийные номера. """ # Создайте пустой список для хранения последовательных устройств серийные_устройства = [] # Создайте статический путь для каталога '/dev/serial/by-id/' by_id_path = '/dev/serial/by-id/' пытаться: # Вывести список всех файлов в каталоге '/dev/serial/by-id/' файлы = os.listdir(by_id_path) кроме FileNotFoundError: #print('В /dev/serial/by-id не найдено последовательных устройств') вернуть серийные_устройства # Просматриваем список файлов в каталоге '/dev/serial/by-id/' для файла в файлах: # Получить абсолютный путь к символической ссылке в '/dev/serial/by-id/' symlink_path = os.path.join(by_id_path, file) # Проверьте, указывает ли символическая ссылка на действительный последовательный порт и существует ли реальный путь если os.path.islink(symlink_path) и os.path.exists(symlink_path): реальный_путь = os.path.realpath(symlink_path) usb_serial = get_serial_number_for_device (реальный_путь) # Добавьте действительный путь к последовательному устройству и соответствующий серийный номер. Serial_devices.append((os.path.join(by_id_path,file), usb_serial)) # Возвращаем список созданных последовательных устройств вернуть серийные_устройства [*]Я запускаю следующий код внутри потока: def Monitor_serial_devices(self): """ Отслеживайте новые подключенные последовательные устройства и создавайте для них потоки. Эта функция работает в бесконечном цикле и периодически проверяет наличие новых подключенных устройств. При обнаружении нового устройства оно создает соответствующий поток и добавляет его в список. из comport_threads. Примечание. Эту функцию можно запустить в отдельном потоке или как часть основного цикла приложения. """ пока правда: пытаться: # Получить список подключенных в данный момент последовательных устройств current_serial_devices = get_serial_devices() #self.logger.debug('Текущие последовательные устройства: %s' % self.comport_threads) # Перебираем список потоков и удаляем отключенные устройства отключенные_порты = set(self.connected_serial_ports) - set([последовательный_порт для последовательного_порта, _ в текущих_последовательных_устройствах]) для отключенного_порта в отключенных_портах: self.remove_thread_for_serial_port(отключенный_порт) # Сравните с ранее обнаруженными устройствами, чтобы найти новые подключенные new_devices = [(serial_port, usb_device_id) для серийного_порта, usb_device_id в current_serial_devices, если серийный_порт отсутствует в self.connected_serial_ports] # Создавать темы для вновь подключенных устройств для серийного_порта — usb_device_id в new_devices: #serialThread = CU16CCommunicationThread(self.config, серийный_порт, usb_device_id, self.device_manager, self.logger) SerialThread = CU16CCommunicationThread(self.config, серийный_порт, usb_device_id, self.device_manager, self.logger) SerialThread.daemon = Истина серийныйПоток.start() self.comport_threads.append(serialThread) self.connected_serial_ports.add(serial_port) # Поспите немного, прежде чем снова проверить time.sleep(15) # отрегулируйте интервал опроса по мере необходимости кроме исключения, как например: self.logger.error("MonitorSerialDevices: Произошла ошибка: %s" % бывший) def remove_thread_for_serial_port(self, серийный_порт): """ Удалите поток и закройте связанный последовательный порт для отключенного устройства. Аргументы: Serial_port (str): Имя последовательного порта, который нужно удалить. """ threads_to_remove = [] для потока в self.comport_threads: если поток.последовательный_порт == последовательный_порт: thread.stop() # Потокобезопасный механизм остановки threads_to_remove.append(поток) для потока в threads_to_remove: thread.join() # Дождитесь завершения потоков, прежде чем удалять их self.comport_threads.remove(нить) self.connected_serial_ports.remove(thread.serial_port) Как видите, сначала я создаю список последовательных устройств, расположенных в /dev/serial/by-id/ Каждое последовательное устройство, указанное в этой папке, представляет собой адаптер USB-последовательный порт. Второй цикл for в этой функции создает экземпляр класса threading.Thread, который обрабатывает связь для каждого из устройств USB-последовательный порт, подключенных к моему Raspberry Pi, и создает экземпляр связи с ним. Если я оставлю все как есть и подключу хотя бы одно устройство USB-последовательный порт к Raspberry Pi, оно будет работать нормально, но если я отключу его, а затем снова подключу, соединение не сможет восстановить соединение. к устройству USB-последовательный порт. Следующий код обрабатывает подключение к устройству USB-Serial:
импорт потоков импортный серийный номер очередь импорта время импорта класс CU16CCommunicationThread(threading.Thread): def __init__(self, config, серийный_порт, usb_device_id, device_manager, регистратор): супер(CU16CCommunicationThread, self).__init__() self.name = '%s' % серийный_порт self.logger = регистратор self.config = конфигурация self.logging_enabled = config.getboolean('LoggingParams', 'cu16c_logging') self.serial_port = серийный_порт self.usb_device_id = usb_device_id self.device_manager = диспетчер_устройства self.ser = None # Экземпляр последовательного порта self.is_running = Истина self.write_thread = Нет self.read_thread = Нет # Очередь, обрабатывающая полученные сообщения. self.receive_queue = очередь.Очередь() self.receive_queue_lock = threading.Lock() # Очередь, обрабатывающая сообщения, отправляемые на последовательное устройство. self.send_queue = очередь.Очередь() self.send_queue_lock = threading.Lock() защита запуска (сам): # Подключите последовательный порт self._connect_serial_port() # Запускаем поток для использования очереди отправки send_consumer_thread = threading.Thread(target=self._consume_send_queue) send_consumer_thread.daemon = Истина send_consumer_thread.start() # Запускаем поток для использования очереди приема get_consumer_thread = threading.Thread(target=self._consume_receive_queue) get_consumer_thread.daemon = Истина get_consumer_thread.start() self.read_thread = ReadThread(self, self.ser, self.receive_queue, self.is_running, self.logger, self.usb_device_id, self.logging_enabled, self.name, self.serial_port) self.read_thread.daemon = Истина self.read_thread.start() пока правда: время.сон(0,5) #time.sleep(0.1) # Установите соединение с последовательным портом, если оно еще не подключено #self._connect_serial_port() защита _connect_serial_port(self): """ Устанавливает соединение с последовательным портом. """ # Продолжайте попытки установить соединение, пока экземпляр последовательного порта не будет инициализирован. в то время как self.ser имеет значение None: пытаться: # Зарегистрируйте попытку подключения с идентификатором USB-устройства и именем последовательного порта. self.logger.info('%s теперь %s' % (self.usb_device_id, self.serial_port)) # Создайте экземпляр последовательного порта с указанными настройками self.ser = серийный.Serial(self.serial_port, скорость передачи данных=19200, размер байта=serial.EIGHTBITS, четность=serial.PARITY_NONE, стопбиты=serial.STOPBITS_ONE) # Зарегистрируйте успешное соединение self.logger.info('Последовательный порт %s успешно подключен.' % self.serial_port) # Обработка исключений, которые могут возникнуть при подключении к последовательному порту кроме серийного.SerialException: # Зарегистрируйте критическую ошибку и укажите повторную попытку через 5 секунд. self.logger.critical('Не удалось подключиться к последовательному порту %s. Повторная попытка через 5 секунд...' % self.usb_device_id) # Подождите 5 секунд, прежде чем повторить попытку подключения. время.сон(5) def _send(self, hex_message): """ Отправляет шестнадцатеричное сообщение через последовательный порт. Аргументы: hex_message (список строк): список шестнадцатеричных значений для отправки. """ пытаться: # Если экземпляр последовательного порта не инициализирован, установите соединение если self.ser имеет значение None: self._connect_serial_port() # Преобразуем список шестнадцатеричных значений в байты byte_data = байты (int(hex_val, 16) для hex_val в hex_message) # Записываем байтовые данные в последовательный порт self.ser.write(byte_data) кроме исключения, как например: self.logger.debug("CU16C: Ошибка отправки сообщения: %s" % (ex)) защита _consume_send_queue (сам): """ Получает сообщения из очереди отправки и отправляет их. """ пытаться: # Постоянно потреблять сообщения из очереди отправки во время работы потока в то время как self.is_running: # Получить следующее шестнадцатеричное сообщение из очереди отправки с таймаутом (1 секунда) пытаться: hex_message = self.send_queue.get() кроме очереди.Пусто: продолжать # Проверяем, указывает ли полученное сообщение на сигнал завершения если hex_message имеет значение Нет: перерыв # Отправляем шестнадцатеричное сообщение через последовательный порт self._send(hex_message) # Отметить задачу как выполненную в очереди отправки self.send_queue.task_done() # Обработка исключений, которые могут возникнуть во время обработки очереди кроме исключения, как например: # Зарегистрировать исключение и продолжить обработку self.logger.error("Ошибка в _consume_send_queue: %s" % ex) защита send_message(self, hex_message): """ Добавляет сообщение в очередь отправки. Аргументы: hex_message (str): шестнадцатеричное сообщение для отправки. """ если self.logging_enabled: self.logger.debug('CU16C: отправлен %s в CU16C с использованием %s' % (hex_message, self.name)) self.send_queue.put(hex_message) защита _consume_receive_queue(self): """ Получает сообщения из очереди приема и обрабатывает их. """ пытаться: # Постоянно потреблять сообщения из очереди приема во время работы потока в то время как self.is_running: # Получить следующее сообщение из очереди приема с таймаутом (1 секунда) пытаться: usb_device_id, hex_values = self.receive_queue.get() кроме очереди.Пусто: продолжать # Проверяем, указывает ли полученное сообщение на сигнал завершения если hex_values имеет значение Нет: перерыв # Добавляем новое устройство CU16C с полученными данными в диспетчер устройств, также обновляем устройство CU16C, если оно уже там self.device_manager.add_new_cu16c_device(usb_device_id, hex_values) # Отметить задачу как выполненную в очереди приема self.receive_queue.task_done() # Обработка исключений, которые могут возникнуть во время обработки очереди кроме исключения, как например: # Зарегистрировать исключение и продолжить обработку self.logger.Exception("Ошибка в _consume_receive_queue: %s" % бывший) класс ReadThread(threading.Thread): def __init__(self, original_thread, серийный номер, get_queue, isrunning, logger, usb_device_id, logging_enabled, name, серийный_порт): супер(ReadThread, self).__init__() self.original_thread = исходная_нить self.ser = серийный self.receive_queue = получение_очереди self.is_running = работает self.logger = регистратор self.usb_device_id = usb_device_id self.logging_enabled = logging_enabled self.name = имя self.serial_port = серийный_порт защита read_serial_message (сам): """ Считывает полное последовательное сообщение из последовательного порта побайтно. Возврат: байты: полное последовательное сообщение. """ #start_time = timeit.default_timer() # Указываем константу длины сообщения MESSAGE_LENGTH = 12 # Логическое значение, которое принимает значение True, если найден указанный стартовый байт (0x02). start_byte_found = Ложь # Буфер сообщений, содержащий сообщение размером 12 байт. message_buffer = байтовый массив() в то время как self.is_running: # Чтение по одному байту из последовательного порта новый_байт = self.ser.read(1) если не новый_байт: # Обработать, если новые данные не получены продолжать # Проверяем, является ли новый байт стартовым если не start_byte_found и new_byte == b'\x02': start_byte_found = Истина # Добавляем байт в буфер сообщения, если найден стартовый байт если start_byte_found: message_buffer.append(new_byte[0]) # Если буфер сообщения достигает ожидаемой длины, мы имеем полное сообщение если len(message_buffer) == MESSAGE_LENGTH: # Проверяем, действительно ли полученное сообщение адрес_байт = буфер_сообщения[1] командный_байт = буфер_сообщения[2] data_bytes = message_buffer[3:9] стоп_байт = буфер_сообщения[10] sum_byte = буфер_сообщения[11] #end_time = timeit.default_timer() #elapsed_time = конечное_время – начальное_время #self.logger.debug('%s:%s: получение сообщения заняло: %s секунд' % (address_byte, self.usb_device_id, elapsed_time)) вернуть байты (message_buffer) # Сбрасываем буфер сообщений и начинаем искать следующее сообщение message_buffer = байтовый массив() start_byte_found = Ложь защита _read_serial_data(self): """ Считывает и обрабатывает входящие последовательные данные. """ пытаться: # Прочитайте полное последовательное сообщение из последовательного порта полученные_данные = self.read_serial_message() # Если данные получены, обработайте и обработайте их если полученные_данные: # Преобразуем полученные байты в список целочисленных значений hex_values = [int(byte_val) для byte_val в полученных_данных] self.logger.debug('ReceiveQueue: USB ID: %s, шестнадцатеричные значения: %s' % (self.usb_device_id, hex_values)) # Поместите полученные данные и связанный с ними идентификатор USB-устройства в очередь приема self.receive_queue.put((self.usb_device_id, hex_values)) # Если ведение журнала включено, записывайте полученные данные в шестнадцатеричном формате. #if self.logging_enabled: # hex_values = [hex(byte_val) для byte_val в hex_values] # self.logger.debug("CU16C: Получен %s от %s" % (hex_values, self.name)) # Обработка исключений, которые могут возникнуть при чтении из последовательного порта кроме (serial.SerialException, OSError): # Зарегистрировать критическую ошибку, указывающую на потерю соединения self.logger.critical('Потеряно соединение с последовательным портом %s. Повторное подключение...' % self.serial_port) # Закрываем текущий экземпляр последовательного порта self.ser.close() # Установите для экземпляра последовательного порта значение «Нет», чтобы инициировать переподключение. self.ser = Нет # Повторно подключаемся к последовательному порту self._connect_serial_port() защита запуска (сам): в то время как self.is_running: self._read_serial_data() Как я могу сделать так, чтобы при отключении адаптера USB-последовательный порт и последующем его повторном подключении мое соединение с последовательным устройством восстанавливалось? Кажется, я не могу найти решение этой проблемы и не могу найти решение уже несколько недель.
Я создаю приложение Python, которое подключается к специальному оборудованию, работающему на Raspberry Pi. Я написал небольшую библиотеку, которая переводит все сообщения, поступающие туда и обратно на последовательное устройство. Последовательное устройство представляет собой простой адаптер USB-Serial (RS485). Проблема, с которой я столкнулся, связана с тем, что когда я пытаюсь отключить и снова подключить адаптер, код больше не работает.
[*]Сначала я извлекаю доступные устройства из /dev/serial/by-id:
импортировать ОС импортировать Serial.tools.list_ports как порты защита get_serial_number_for_device(device_name): """ Получите серийный номер конкретного устройства по его имени. Аргументы: имя_устройства (str): имя устройства, для которого нужно найти серийный номер. Возврат: str или None: серийный номер устройства, если оно найдено, или None, если оно не найдено. """ # Получить все текущие данные из последовательной библиотеки Python компорты = порты.comports() # Просматриваем текущие порты для порта в компортах: # Если имя устройства предоставленного аргумента совпадает с именем в списке портов если порт.устройство == имя_устройства: # возвращаем серийный номер порта вернуть порт.серийный_номер # Если устройство не найдено, верните None возврат Нет защита get_serial_devices(): """ Получите список действительных серийных устройств и соответствующие им серийные номера. Возврат: список: список кортежей, содержащий абсолютные пути к действительным последовательным устройствам и соответствующие им серийные номера. """ # Создайте пустой список для хранения последовательных устройств серийные_устройства = [] # Создайте статический путь для каталога '/dev/serial/by-id/' by_id_path = '/dev/serial/by-id/' пытаться: # Вывести список всех файлов в каталоге '/dev/serial/by-id/' файлы = os.listdir(by_id_path) кроме FileNotFoundError: #print('В /dev/serial/by-id не найдено последовательных устройств') вернуть серийные_устройства # Просматриваем список файлов в каталоге '/dev/serial/by-id/' для файла в файлах: # Получить абсолютный путь к символической ссылке в '/dev/serial/by-id/' symlink_path = os.path.join(by_id_path, file) # Проверьте, указывает ли символическая ссылка на действительный последовательный порт и существует ли реальный путь если os.path.islink(symlink_path) и os.path.exists(symlink_path): реальный_путь = os.path.realpath(symlink_path) usb_serial = get_serial_number_for_device (реальный_путь) # Добавьте действительный путь к последовательному устройству и соответствующий серийный номер. Serial_devices.append((os.path.join(by_id_path,file), usb_serial)) # Возвращаем список созданных последовательных устройств вернуть серийные_устройства [*]Я запускаю следующий код внутри потока: def Monitor_serial_devices(self): """ Отслеживайте новые подключенные последовательные устройства и создавайте для них потоки. Эта функция работает в бесконечном цикле и периодически проверяет наличие новых подключенных устройств. При обнаружении нового устройства оно создает соответствующий поток и добавляет его в список. из comport_threads. Примечание. Эту функцию можно запустить в отдельном потоке или как часть основного цикла приложения. """ пока правда: пытаться: # Получить список подключенных в данный момент последовательных устройств current_serial_devices = get_serial_devices() #self.logger.debug('Текущие последовательные устройства: %s' % self.comport_threads) # Перебираем список потоков и удаляем отключенные устройства отключенные_порты = set(self.connected_serial_ports) - set([последовательный_порт для последовательного_порта, _ в текущих_последовательных_устройствах]) для отключенного_порта в отключенных_портах: self.remove_thread_for_serial_port(отключенный_порт) # Сравните с ранее обнаруженными устройствами, чтобы найти новые подключенные new_devices = [(serial_port, usb_device_id) для серийного_порта, usb_device_id в current_serial_devices, если серийный_порт отсутствует в self.connected_serial_ports] # Создавать темы для вновь подключенных устройств для серийного_порта — usb_device_id в new_devices: #serialThread = CU16CCommunicationThread(self.config, серийный_порт, usb_device_id, self.device_manager, self.logger) SerialThread = CU16CCommunicationThread(self.config, серийный_порт, usb_device_id, self.device_manager, self.logger) SerialThread.daemon = Истина серийныйПоток.start() self.comport_threads.append(serialThread) self.connected_serial_ports.add(serial_port) # Поспите немного, прежде чем снова проверить time.sleep(15) # отрегулируйте интервал опроса по мере необходимости кроме исключения, как например: self.logger.error("MonitorSerialDevices: Произошла ошибка: %s" % бывший) def remove_thread_for_serial_port(self, серийный_порт): """ Удалите поток и закройте связанный последовательный порт для отключенного устройства. Аргументы: Serial_port (str): Имя последовательного порта, который нужно удалить. """ threads_to_remove = [] для потока в self.comport_threads: если поток.последовательный_порт == последовательный_порт: thread.stop() # Потокобезопасный механизм остановки threads_to_remove.append(поток) для потока в threads_to_remove: thread.join() # Дождитесь завершения потоков, прежде чем удалять их self.comport_threads.remove(нить) self.connected_serial_ports.remove(thread.serial_port) Как видите, сначала я создаю список последовательных устройств, расположенных в /dev/serial/by-id/ Каждое последовательное устройство, указанное в этой папке, представляет собой адаптер USB-последовательный порт. Второй цикл for в этой функции создает экземпляр класса threading.Thread, который обрабатывает связь для каждого из устройств USB-последовательный порт, подключенных к моему Raspberry Pi, и создает экземпляр связи с ним. Если я оставлю все как есть и подключу хотя бы одно устройство USB-последовательный порт к Raspberry Pi, оно будет работать нормально, но если я отключу его, а затем снова подключу, соединение не сможет восстановить соединение. к устройству USB-последовательный порт. Следующий код обрабатывает подключение к устройству USB-Serial:
импорт потоков импортный серийный номер очередь импорта время импорта класс CU16CCommunicationThread(threading.Thread): def __init__(self, config, серийный_порт, usb_device_id, device_manager, регистратор): супер(CU16CCommunicationThread, self).__init__() self.name = '%s' % серийный_порт self.logger = регистратор self.config = конфигурация self.logging_enabled = config.getboolean('LoggingParams', 'cu16c_logging') self.serial_port = серийный_порт self.usb_device_id = usb_device_id self.device_manager = диспетчер_устройства self.ser = None # Экземпляр последовательного порта self.is_running = Истина self.write_thread = Нет self.read_thread = Нет # Очередь, обрабатывающая полученные сообщения. self.receive_queue = очередь.Очередь() self.receive_queue_lock = threading.Lock() # Очередь, обрабатывающая сообщения, отправляемые на последовательное устройство. self.send_queue = очередь.Очередь() self.send_queue_lock = threading.Lock() защита запуска (сам): # Подключите последовательный порт self._connect_serial_port() # Запускаем поток для использования очереди отправки send_consumer_thread = threading.Thread(target=self._consume_send_queue) send_consumer_thread.daemon = Истина send_consumer_thread.start() # Запускаем поток для использования очереди приема get_consumer_thread = threading.Thread(target=self._consume_receive_queue) get_consumer_thread.daemon = Истина get_consumer_thread.start() self.read_thread = ReadThread(self, self.ser, self.receive_queue, self.is_running, self.logger, self.usb_device_id, self.logging_enabled, self.name, self.serial_port) self.read_thread.daemon = Истина self.read_thread.start() пока правда: время.сон(0,5) #time.sleep(0.1) # Установите соединение с последовательным портом, если оно еще не подключено #self._connect_serial_port() защита _connect_serial_port(self): """ Устанавливает соединение с последовательным портом. """ # Продолжайте попытки установить соединение, пока экземпляр последовательного порта не будет инициализирован. в то время как self.ser имеет значение None: пытаться: # Зарегистрируйте попытку подключения с идентификатором USB-устройства и именем последовательного порта. self.logger.info('%s теперь %s' % (self.usb_device_id, self.serial_port)) # Создайте экземпляр последовательного порта с указанными настройками self.ser = серийный.Serial(self.serial_port, скорость передачи данных=19200, размер байта=serial.EIGHTBITS, четность=serial.PARITY_NONE, стопбиты=serial.STOPBITS_ONE) # Зарегистрируйте успешное соединение self.logger.info('Последовательный порт %s успешно подключен.' % self.serial_port) # Обработка исключений, которые могут возникнуть при подключении к последовательному порту кроме серийного.SerialException: # Зарегистрируйте критическую ошибку и укажите повторную попытку через 5 секунд. self.logger.critical('Не удалось подключиться к последовательному порту %s. Повторная попытка через 5 секунд...' % self.usb_device_id) # Подождите 5 секунд, прежде чем повторить попытку подключения. время.сон(5) def _send(self, hex_message): """ Отправляет шестнадцатеричное сообщение через последовательный порт. Аргументы: hex_message (список строк): список шестнадцатеричных значений для отправки. """ пытаться: # Если экземпляр последовательного порта не инициализирован, установите соединение если self.ser имеет значение None: self._connect_serial_port() # Преобразуем список шестнадцатеричных значений в байты byte_data = байты (int(hex_val, 16) для hex_val в hex_message) # Записываем байтовые данные в последовательный порт self.ser.write(byte_data) кроме исключения, как например: self.logger.debug("CU16C: Ошибка отправки сообщения: %s" % (ex)) защита _consume_send_queue (сам): """ Получает сообщения из очереди отправки и отправляет их. """ пытаться: # Постоянно потреблять сообщения из очереди отправки во время работы потока в то время как self.is_running: # Получить следующее шестнадцатеричное сообщение из очереди отправки с таймаутом (1 секунда) пытаться: hex_message = self.send_queue.get() кроме очереди.Пусто: продолжать # Проверяем, указывает ли полученное сообщение на сигнал завершения если hex_message имеет значение Нет: перерыв # Отправляем шестнадцатеричное сообщение через последовательный порт self._send(hex_message) # Отметить задачу как выполненную в очереди отправки self.send_queue.task_done() # Обработка исключений, которые могут возникнуть во время обработки очереди кроме исключения, как например: # Зарегистрировать исключение и продолжить обработку self.logger.error("Ошибка в _consume_send_queue: %s" % ex) защита send_message(self, hex_message): """ Добавляет сообщение в очередь отправки. Аргументы: hex_message (str): шестнадцатеричное сообщение для отправки. """ если self.logging_enabled: self.logger.debug('CU16C: отправлен %s в CU16C с использованием %s' % (hex_message, self.name)) self.send_queue.put(hex_message) защита _consume_receive_queue(self): """ Получает сообщения из очереди приема и обрабатывает их. """ пытаться: # Постоянно потреблять сообщения из очереди приема во время работы потока в то время как self.is_running: # Получить следующее сообщение из очереди приема с таймаутом (1 секунда) пытаться: usb_device_id, hex_values = self.receive_queue.get() кроме очереди.Пусто: продолжать # Проверяем, указывает ли полученное сообщение на сигнал завершения если hex_values имеет значение Нет: перерыв # Добавляем новое устройство CU16C с полученными данными в диспетчер устройств, также обновляем устройство CU16C, если оно уже там self.device_manager.add_new_cu16c_device(usb_device_id, hex_values) # Отметить задачу как выполненную в очереди приема self.receive_queue.task_done() # Обработка исключений, которые могут возникнуть во время обработки очереди кроме исключения, как например: # Зарегистрировать исключение и продолжить обработку self.logger.Exception("Ошибка в _consume_receive_queue: %s" % бывший) класс ReadThread(threading.Thread): def __init__(self, original_thread, серийный номер, get_queue, isrunning, logger, usb_device_id, logging_enabled, name, серийный_порт): супер(ReadThread, self).__init__() self.original_thread = исходная_нить self.ser = серийный self.receive_queue = получение_очереди self.is_running = работает self.logger = регистратор self.usb_device_id = usb_device_id self.logging_enabled = logging_enabled self.name = имя self.serial_port = серийный_порт защита read_serial_message (сам): """ Считывает полное последовательное сообщение из последовательного порта побайтно. Возврат: байты: полное последовательное сообщение. """ #start_time = timeit.default_timer() # Указываем константу длины сообщения MESSAGE_LENGTH = 12 # Логическое значение, которое принимает значение True, если найден указанный стартовый байт (0x02). start_byte_found = Ложь # Буфер сообщений, содержащий сообщение размером 12 байт. message_buffer = байтовый массив() в то время как self.is_running: # Чтение по одному байту из последовательного порта новый_байт = self.ser.read(1) если не новый_байт: # Обработать, если новые данные не получены продолжать # Проверяем, является ли новый байт стартовым если не start_byte_found и new_byte == b'\x02': start_byte_found = Истина # Добавляем байт в буфер сообщения, если найден стартовый байт если start_byte_found: message_buffer.append(new_byte[0]) # Если буфер сообщения достигает ожидаемой длины, мы имеем полное сообщение если len(message_buffer) == MESSAGE_LENGTH: # Проверяем, действительно ли полученное сообщение адрес_байт = буфер_сообщения[1] командный_байт = буфер_сообщения[2] data_bytes = message_buffer[3:9] стоп_байт = буфер_сообщения[10] sum_byte = буфер_сообщения[11] #end_time = timeit.default_timer() #elapsed_time = конечное_время – начальное_время #self.logger.debug('%s:%s: получение сообщения заняло: %s секунд' % (address_byte, self.usb_device_id, elapsed_time)) вернуть байты (message_buffer) # Сбрасываем буфер сообщений и начинаем искать следующее сообщение message_buffer = байтовый массив() start_byte_found = Ложь защита _read_serial_data(self): """ Считывает и обрабатывает входящие последовательные данные. """ пытаться: # Прочитайте полное последовательное сообщение из последовательного порта полученные_данные = self.read_serial_message() # Если данные получены, обработайте и обработайте их если полученные_данные: # Преобразуем полученные байты в список целочисленных значений hex_values = [int(byte_val) для byte_val в полученных_данных] self.logger.debug('ReceiveQueue: USB ID: %s, шестнадцатеричные значения: %s' % (self.usb_device_id, hex_values)) # Поместите полученные данные и связанный с ними идентификатор USB-устройства в очередь приема self.receive_queue.put((self.usb_device_id, hex_values)) # Если ведение журнала включено, записывайте полученные данные в шестнадцатеричном формате. #if self.logging_enabled: # hex_values = [hex(byte_val) для byte_val в hex_values] # self.logger.debug("CU16C: Получен %s от %s" % (hex_values, self.name)) # Обработка исключений, которые могут возникнуть при чтении из последовательного порта кроме (serial.SerialException, OSError): # Зарегистрировать критическую ошибку, указывающую на потерю соединения self.logger.critical('Потеряно соединение с последовательным портом %s. Повторное подключение...' % self.serial_port) # Закрываем текущий экземпляр последовательного порта self.ser.close() # Установите для экземпляра последовательного порта значение «Нет», чтобы инициировать переподключение. self.ser = Нет # Повторно подключаемся к последовательному порту self._connect_serial_port() защита запуска (сам): в то время как self.is_running: self._read_serial_data() Как я могу сделать так, чтобы при отключении адаптера USB-последовательный порт и последующем его повторном подключении мое соединение с последовательным устройством восстанавливалось? Кажется, я не могу найти решение этой проблемы и не могу найти решение уже несколько недель.
