__init__.py 文件源码

def _setup_connection(self, device: str, speed: int):
        ''' todo: doc '''
        try:
            self._conn = serial.Serial(device, speed, timeout=2)
        except serial.SerialException as ex:
            if ex.errno == 2:
                raise device_error("there's no device '%s'" % device)
            raise

        self.disconnect()
        self._conn.flush()

        # todo: refactor
        _conn_settings = self._conn.getSettingsDict()
        for _ in range(2):
            _conn_settings['rtscts'] = not _conn_settings['rtscts']
            self._conn.applySettingsDict(_conn_settings)

#        connection._assert_token(connection._read_byte(self._conn), _token.SYNC)
        self._handler_thread = threading.Thread(target=self._handler_thread_fn, daemon=True)
        self._handler_thread.start()
        LOG.info('connection initialized')