Python serial 模块，STOPBITS_TWO 实例源码

def test_init_sets_the_correct_attrs(self):
        """__init__ sets the fields that get_settings reads"""
        for setting, value in (
                ('baudrate', 57600),
                ('timeout', 7),
                ('write_timeout', 12),
                ('inter_byte_timeout', 15),
                ('stopbits', serial.STOPBITS_TWO),
                ('bytesize', serial.SEVENBITS),
                ('parity', serial.PARITY_ODD),
                ('xonxoff', True),
                ('rtscts', True),
                ('dsrdtr', True)):
            kwargs = {'do_not_open': True, setting: value}
            ser = serial.serial_for_url(PORT, **kwargs)
            d = ser.get_settings()
            self.assertEqual(getattr(ser, setting), value)
            self.assertEqual(d[setting], value)

def log(options):
    # configure the serial connections (the parameters differs on the device you are connecting to)
    ser = serial.Serial(
        port=options.device,
        baudrate=9600,
        parity=serial.PARITY_ODD,
        stopbits=serial.STOPBITS_TWO,
        bytesize=serial.SEVENBITS
    )

    prog = Progress()

    with open(options.outfile, 'w') as the_file:
        the_file.write('timestamp;value\n')
        while True:
            # \r\n is for device terminators set to CR LF
            ser.write((':FETCh?\r\n'))

            # wait one second before reading output.
            time.sleep(options.interval)
            out = ''
            while ser.inWaiting() > 0:
                out += ser.read(1)
            if out != '':
                out = out.rstrip()
                res = "%s;%s\n" % (time.time(), float(out))
                the_file.write(res)
                the_file.flush()
                prog.show()

def open_port(self, port, listening_serial_thread):
                if not self.fake:
                        self.ser = serial.Serial(port,
                                                 baudrate=115200,
                                                 bytesize=serial.EIGHTBITS,
                                                 parity=serial.PARITY_NONE,
                                                 stopbits=serial.STOPBITS_TWO,
                                                 #rtscts=True,
                                                 timeout=1000)
                if listening_serial_thread:
                    SerialThread.SerialThread(self.ser).start()

def __init__(self, device='/dev/ttyUSB0', baudrate=9600, rst='-rts', debug=False):
        self._sl = serial.Serial(
                port = device,
                parity = serial.PARITY_EVEN,
                bytesize = serial.EIGHTBITS,
                stopbits = serial.STOPBITS_TWO,
                timeout = 1,
                xonxoff = 0,
                rtscts = 0,
                baudrate = baudrate,
            )
        self._rst_pin = rst
        self._debug = debug

def __init__(self,
                 port: str,
                 baudrate: int,
                 bytesize: int,
                 stopbits: float,
                 parity: str,
                 timeout: float,
                 xonxoff: bool,
                 rtscts: bool):
        """Converts data from JSON format to serial.Serial format."""
        self._port = port
        self._baudrate = baudrate
        self._bytesize = {
            5: serial.FIVEBITS,
            6: serial.SIXBITS,
            7: serial.SEVENBITS,
            8: serial.EIGHTBITS
        }[bytesize]
        self._stopbits = {
            1: serial.STOPBITS_ONE,
            1.5: serial.STOPBITS_ONE_POINT_FIVE,
            2: serial.STOPBITS_TWO
        }[stopbits]
        self._parity = {
            'none': serial.PARITY_NONE,
            'even': serial.PARITY_EVEN,
            'odd': serial.PARITY_ODD,
            'mark': serial.PARITY_MARK,
            'space': serial.PARITY_SPACE
        }[parity]
        self._timeout = timeout
        self._xonxoff = xonxoff
        self._rtscts = rtscts

def openSession(self, portname):
        self.serialport = serial.Serial(port=portname,
                                 parity=serial.PARITY_EVEN,
                                 bytesize=serial.EIGHTBITS,
                                 stopbits=serial.STOPBITS_TWO,
                                 timeout=1,
                                 xonxoff=0,
                                 rtscts=0,
                                 baudrate=9600)
        if (not self.serialport):
            return 1

        # reset it!
        #self.serialport.setRTS(0)
        self.serialport.setRTS(1)
        self.serialport.setDTR(1)
        time.sleep(0.01)  # 10ms?
        self.serialport.flushInput()
        #self.serialport.setRTS(1)
        self.serialport.setRTS(0)
        self.serialport.setDTR(0)

        ts = self.serialport.read()
        if ts == None:
            return 2             # no card?
        if ord(ts) != 0x3B:
            return 3             # bad ATR byte 
        # ok got 0x3B
        print "TS: 0x%x Direct convention" % ord(ts)

        t0 = chr(0x3B)
        while ord(t0) == 0x3b:
            t0 = self.serialport.read()

        if t0 == None:
            return 2
        print "T0: 0x%x" % ord(t0)

        # read interface bytes
        if (ord(t0) & 0x10):
            print "TAi = %x" % ord(self.serialport.read())
        if (ord(t0) & 0x20):
            print "TBi = %x" % ord(self.serialport.read())
        if (ord(t0) & 0x40):
            print "TCi = %x" % ord(self.serialport.read())
        if (ord(t0) & 0x80):
            tdi = self.serialport.read()
            print "TDi = %x" % ord(tdi)

        for i in range(0, ord(t0) & 0xF):
            x = self.serialport.read()
            print "Historical: %x" % ord(x)

        while 1:
            x = self.serialport.read()
            if (x == ""):
                break
            print "read: %x" % ord(x)

        return 0

def openSerial(self):
        #Set up the relationship between what the user enters and what the API calls for
        bytedic = {'5':serial.FIVEBITS,
                   '6':serial.SIXBITS,
                   '7':serial.SEVENBITS,
                   '8':serial.EIGHTBITS}
        bytesize = bytedic[str(self.root.variables['databits'])]

        paritydict = {'None':serial.PARITY_NONE,
                      'Even':serial.PARITY_EVEN,
                      'Odd' :serial.PARITY_ODD,
                      'Mark':serial.PARITY_MARK,
                      'Space':serial.PARITY_SPACE}
        parity=paritydict[self.root.variables['parity']]

        stopbitsdict = {'1':serial.STOPBITS_ONE,
                        '2':serial.STOPBITS_TWO}
        stopbits = stopbitsdict[str(self.root.variables['stopbits'])]

        #Open the serial port given the settings, store under the root
        if os.name == 'nt':
            port = self.root.variables['COMport'][0:5].strip()
            self.root.ser = serial.Serial(\
                port=port,\
                baudrate=str(self.root.variables['baud']),\
                bytesize=bytesize, parity=parity, stopbits=stopbits, timeout=0.5) 
        else:
            first_space = self.root.variables['COMport'].index(' ')
            port = self.root.variables['COMport'][0:first_space].strip()
            # Parameters necessary due to https://github.com/pyserial/pyserial/issues/59
            self.root.ser = serial.Serial(\
                port=port,\
                baudrate=str(self.root.variables['baud']),\
                bytesize=bytesize, parity=parity, stopbits=stopbits, timeout=0.5, rtscts=True, dsrdtr=True) 
        io.DEFAULT_BUFFER_SIZE = 5000

        #Purge the buffer of any previous data
        if float(serial.VERSION[0:3]) < 3:
            #If we're executing with pySerial 2.x
            serial.Serial.flushInput(self.root.ser)
            serial.Serial.flushOutput(self.root.ser)
        else:
            #Otherwise we're using pySerial 3.x
            serial.Serial.reset_input_buffer(self.root.ser)
            serial.Serial.reset_output_buffer(self.root.ser)

def predict_result():
    ser = serial.Serial(  # ????????????
        port='COM6',
        baudrate=1200,
        parity=serial.PARITY_ODD,
        stopbits=serial.STOPBITS_TWO,
        bytesize=serial.SEVENBITS
    )
    serial_data = []
    plt.xlim(0, 100)
    plt.ylim(300, 700)
    plt.title('GSR')
    plt.ion()
    i = 0
    j = 0
    id = 0
    while True:
        line = ser.readline()
        line = int(line)
        serial_data.append(line)
        if i > 100:
            plt.xlim(i - 100, i)
        plt.plot(serial_data)
        i += 1
        j += 1
        if j >= 50:
            clf = joblib.load('model\\happy_model.m')
            select = joblib.load('model\\vector_select.m')
            vector = getattr.get_vector(serial_data)
            new_vector = select.transform(vector)
            print(new_vector)
            result = clf.predict(new_vector)
            if result[0] == '2':
                clf = joblib.load('model\\sad_model.m')
                result = clf.predict(new_vector)
            j = 0
            plt.plot([i, i], [300, 700], 'r--')
            if result[0] == '1':
                plt.annotate('happy', xy=(i, 600), xytext=(i - 10, 600), arrowprops=dict(facecolor='red', shrink=0.05))
                res = 1
                database.insert(id, res)
            elif result[0] == '2':
                plt.annotate('normal', xy=(i, 600), xytext=(i - 10, 600), arrowprops=dict(facecolor='blue', shrink=0.05))
                res = 0
                database.insert(id, res)
            else:
                plt.annotate('sad', xy=(i, 600), xytext=(i - 10, 600),arrowprops=dict(facecolor='black', shrink=0.05))
                res = 2
                database.insert(id, res)
            print(result)
            id += 1
        plt.pause(0.001)