Как открыть последовательный порт из потока в Python

Question

В течение нескольких недель я застрял, как открыть последовательный COM-порт из потока, , чтобы иметь возможность писать/читать его из другого потока. В моем примере, когда я пишу в порт из другого потока, есть отказ в доступе. Когда я открываю порт из основного графического интерфейса, он отлично работает, я могу написать его из потока, но мне нужно дать пользователю возможность выбрать номер COM. Это мой код, если кто-нибудь может посмотреть и помочь, это было бы здорово ...

class Connect(QThread): 
    connectResult = QtCore.pyqtSignal(str) 
    position1 = QtCore.pyqtSignal(str) 
    actuPosResult = QtCore.pyqtSignal(str) 
    def __init__(self, myvar, parent=None): 
     QThread.__init__(self, parent) 
     self.myvar = str(myvar) # value from spinbox 
    def run(self): 
     self.pserial = serial.Serial() 
     try: 
      COMnumber= self.myvar 
      self.pserial = serial.Serial('COM'+COMnumber, 115200,timeout=None) 
      r='COM '+COMnumber+' connected.' 
      self.pserial.write("c".encode('ascii')+"\n".encode('ascii')) 
      incomingByte = self.pserial.read() 
      decodedByte = incomingByte.decode("utf-8") 
      if decodedByte == ('c'): 
       r='Atmega On-Line' 
       self.connectResult.emit(r) 
       pos1='---' 
       self.position1.emit(pos1) 
      else : 
       r=' No answer from Atmega.' 
       self.connectResult.emit(r) 
    def stop(self): 
     self.terminate() 
class ReadingEncoder(QThread): 
    actuPosResult = QtCore.pyqtSignal(str) 
    def __init__(self, mojazmienna, parent=None): 
     QThread.__init__(self, parent) 
     self.mojazmienna = str(mojazmienna) 
    def run(self): 
    Try: 
      self.pserial = serial.Serial() 
      self.pserial = serial.Serial('COM3', 115200,timeout=1) 
      self.pserial.write("p".encode('ascii')+"\n".encode('ascii')) 
      incomingByte = self.pserial.read() 
      decodedByte = incomingByte.decode("utf-8") 
      actualPos = '' 
     if decodedByte == ('a'):   
       while decodedByte != ('\n'): 
        incomingByte = self.pserial.read() 
        decodedByte = incomingByte.decode("utf-8") 
        actualPos = actualPos + decodedByte 
       pos= actualPos.rstrip('\n')# pozycja w formacie string 
       print(pos) 
      self.actuPosResult.emit(pos) 
     except (EOFError, OSError, IOError, ValueError, RuntimeError, BrokenPipeError, InterruptedError, TimeoutError): 
      print('Thread readingEncoder error') 
     self.pserial.close() 

Answer 1

Вы пытаетесь открыть порт несколько раз.

Вместо этого у вас есть несколько альтернатив;

• Создайте объект Serial один раз и передайте его в поток, который должен с ним работать.
• Укажите имя последовательного порта, который будет использоваться для потока, который должен работать с ним, и пусть он откроет порт.

В обоих случаях убедитесь, что ваша нить правильно закрывает порт, когда он выходит!

Answer 2

Вы просто открываете последовательный порт и начинаете тему.

import atexit 

class SerialPort(QThread): 
    connectResult = QtCore.pyqtSignal(str) 
    position1 = QtCore.pyqtSignal(str) 
    actuPosResult = QtCore.pyqtSignal(str) 

    def __init__(self, port=None, baud=115200, timeout=1): 
     super().__init__() 

     self.ser = serial.Serial() 
     self.ser.port = port 
     self.ser.baudrate = baud 
     self.ser.timeout = timeout 
     self.running = False 
     atexit.register(self.ser.close) # Make sure the serial port closes when you quit the program. 

    def set_port(port_num): 
     self.ser.port = "COM"+str(port_num) 

    def start(self, *args, **kwargs): 
     self.running = True 
     self.ser.open() 
     super().start() 
     self.ser.write("c\n".encode("ascii")) 

    def run(self): 
     while self.running: 
      try: 
       incomingByte = self.ser.read() 
       decodedByte = incomingByte.decode("utf-8") 
       if decodedByte == ('c'): 
        r='Atmega On-Line' 
        self.connectResult.emit(r) 
        pos1='---' 
        self.position1.emit(pos1) 
       else: 
        r=' No answer from Atmega.' 
        self.connectResult.emit(r) 
      except: 
       pass 
      # time.sleep(0.01) # You may want to sleep or use readline 

Вы используете этот класс, нажав кнопку «Пуск».

serial = SerialPort("COM3") 
btn = QtGui.QPushButton("Connect") 
btn.clicked.connect(serial.start) 

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

Последовательный порт существует в основном потоке и существует все время. Вам не нужно постоянно открывать и закрывать последовательный порт. Все показания происходят в потоке. Чтобы написать только позвонить serial.write(b"c\n"). Вам не нужно, чтобы запись происходила в потоке.

Чтобы пользователь выбрал COM-порт, просто используйте QLineEdit.

myserial = QtGui.QLineEdit("3") 
myserial.textChanged.connect(serial.set_port)