Python serial 模块，SEVENBITS 实例源码

def __init__(self,port=None,logFileName=False):
        print("\n### TENMA MULTIMETER INTERFACE ###")
        devList=self.device_list()
        self.port=port
        self.logFileName=logFileName
        if self.logFileName:
            self.log_init()
        if port is None:
            self.port=self.guessPort
        if not self.port in devList:
            print("!!!  %s isn't in the list of serial ports !!!"%self.port)
        print("Preparing to use",self.port)
        self.ser = serial.Serial()
        self.ser.port = self.port
        self.ser.baudrate = 19230
        self.ser.bytesize = serial.SEVENBITS
        self.ser.parity = serial.PARITY_ODD
        self.ser.stopbits = serial.STOPBITS_ONE
        self.ser.timeout = 3
        self.ser.xonxoff = False
        self.ser.rtscts = False
        self.ser.dsrdtr = False
        self.ser.writeTimeout = 3

def get_dsmr_connection_parameters():
    """ Returns the communication settings required for the DSMR version set. """
    DSMR_VERSION_MAPPING = {
        DataloggerSettings.DSMR_VERSION_2: {
            'baudrate': 9600,
            'bytesize': serial.SEVENBITS,
            'parity': serial.PARITY_EVEN,
            'crc': False,
        },
        DataloggerSettings.DSMR_VERSION_4_PLUS: {
            'baudrate': 115200,
            'bytesize': serial.EIGHTBITS,
            'parity': serial.PARITY_NONE,
            'crc': True,
        },
    }

    datalogger_settings = DataloggerSettings.get_solo()
    connection_parameters = DSMR_VERSION_MAPPING[datalogger_settings.dsmr_version]
    connection_parameters['com_port'] = datalogger_settings.com_port
    return connection_parameters

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 __init__(self,
                 path,
                 baudrate=9600,
                 timeout=1,
                 open_on_create=True,
                 debugging=False):
        super(TeliumNativeSerial, self).__init__(
            path,
            baudrate=baudrate,
            bytesize=SEVENBITS,
            parity=PARITY_EVEN,
            stopbits=STOPBITS_ONE,
            timeout=timeout,
            open_on_create=open_on_create,
            debugging=debugging)

def connect(
            self, baudrate=None, parity=None, databits=None, stopbits=None):
        """Open a serial port for communication with barcode reader device

        Connection settings are taken from three possible locations in the
        following order:

        - method arguments
        - object properties set directly or in constructor or
          detect_connection_settings()
        - default device settings according to device documentation

        If connection settings (baud rate, parity, etc.) are neither provided
        as arguments nor previously set as object properties, for example by
        a search with the detect_connection_settings() method, the default
        values specified in the device documentation are used. For example,
        page 3-1 of the MS3 user manual specifies:
            - baud rate: 9600
            - parity: even
            - data bits: 7
            - stop bits: 1
            - flow control: none
        """
        baudrate = baudrate or self.baudrate or 9600
        parity = parity or self.parity or serial.PARITY_EVEN
        bytesize = databits or self.databits or serial.SEVENBITS
        stopbits = stopbits or self.stopbits or serial.STOPBITS_ONE

        self.port = serial.Serial(
            self.portname,
            baudrate=baudrate,
            parity=parity,
            bytesize=bytesize,
            stopbits=stopbits,
            timeout=1,
            xonxoff=False,
            rtscts=False,
            dsrdtr=False,
        )

        self._config = self.read_config()

def __init__(self, tty, timeout=2):
        self.conn = serial.Serial(tty,
                       baudrate=19200,
                       bytesize=serial.SEVENBITS,
                       parity=serial.PARITY_ODD,
                       stopbits=serial.STOPBITS_ONE,
                       timeout=timeout,
                       xonxoff=1,
                       rtscts=0,
                       dsrdtr=0
                       )
        # switch on power for hardware RS232
        # transmitter via handshake-signals
        self.conn.dtr = True
        self.conn.rts = False

def test_dsmr_version_2(self):
        """ Test connection parameters for DSMR v2. """
        datalogger_settings = DataloggerSettings.get_solo()
        datalogger_settings.dsmr_version = DataloggerSettings.DSMR_VERSION_2
        datalogger_settings.save()

        self.assertEqual(DataloggerSettings.get_solo().dsmr_version, DataloggerSettings.DSMR_VERSION_2)

        connection_parameters = dsmr_datalogger.services.get_dsmr_connection_parameters()
        self.assertEqual(connection_parameters['baudrate'], 9600)
        self.assertEqual(connection_parameters['bytesize'], serial.SEVENBITS)
        self.assertEqual(connection_parameters['parity'], serial.PARITY_EVEN)

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 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)