barcode.py

import threading
import serial.tools.list_ports
from serial import Serial, SerialException

from loguru import logger


class SerialWorker(threading.Thread):
    def __init__(self, port, baudrate, queue):
        super().__init__()
        self.port = port
        self.baudrate = baudrate
        self.queue = queue
        self._stop_event = threading.Event()

    @staticmethod
    def get_ports_list():
        ports = serial.tools.list_ports.comports()
        return {port.description: port.device for port in ports}

    def run(self):
        try:
            with Serial(self.port, self.baudrate, timeout=1) as serial_conn:
                logger.success(f"Соединение установлено. Порт {serial_conn.port} открыт.")
                while not self._stop_event.is_set():
                    if serial_conn.in_waiting:
                        data = serial_conn.readline().decode('utf-8').strip()
                        logger.info(f"Код считан: {data}")
                        self.queue.put(f"Received: {data}")
        except SerialException as e:
            logger.error(e)
            self.queue.put(f"Error: {e}")
        finally:
            logger.info(f"Поток SerialWorker завершён")

    def stop(self):
        self._stop_event.set()