Catch Error Python Reading From Serial Device Failed
Python serial.SerialException() Examples
The following are 30 code examples for showing how to utilise serial.SerialException() . These examples are extracted from open up source projects. You can vote up the ones you like or vote downwards the ones you lot don't like, and become to the original project or source file past following the links above each case.
You may cheque out the related API usage on the sidebar.
You may also want to cheque out all available functions/classes of the module serial , or try the search function 
.
Example 1
def serial_ports(): """ Lists serial port names :raises EnvironmentError: On unsupported or unknown platforms :returns: A list of the serial ports available on the organisation """ if sys.platform.startswith('win'): ports = ['COM%s' % (i + 1) for i in range(256)] elif sys.platform.startswith('linux') or sys.platform.startswith('cygwin'): # this excludes your current terminal "/dev/tty" ports = glob.glob('/dev/ttyUSB*') # ubuntu is /dev/ttyUSB0 elif sys.platform.startswith('darwin'): # ports = glob.glob('/dev/tty.*') ports = glob.glob('/dev/tty.SLAB_USBtoUART*') else: raise EnvironmentError('Unsupported platform') result = [] for port in ports: try: s = serial.Serial(port) s.close() consequence.append(port) except serial.SerialException equally due east: if e.errno == 13: enhance e pass except OSError: pass return result # if __name__ == '__main__': # impress(serial_ports()) Case 2
def _call_connection_lost(cocky, exc): """Close the connection. Informs the protocol through connection_lost() and clears pending buffers and closes the serial connection. """ affirm self._closing assert not self._has_writer assert not self._has_reader try: self._serial.flush() except (series.SerialException if os.proper noun == "nt" else termios.error): # ignore serial errors which may happen if the serial device was # hot-unplugged. laissez passer try: self._protocol.connection_lost(exc) finally: cocky._write_buffer.articulate() self._serial.close() self._serial = None cocky._protocol = None self._loop = None
Instance 3
def run(cocky): while self.enabled: #Filibuster to let threaded CPU utilization relaxing time.sleep(0.01) if(self.enabled): #Check for bytes to transmit over serial if(not cocky.serial_tx_queue.empty()): while(not cocky.serial_tx_queue.empty()): self.ser.write(self.serial_tx_queue.become()) #Bank check for bytes to receive from serial endeavor: if cocky.ser.inWaiting() > 0: rx_buffer_inwaiting = self.ser.inWaiting() self.serial_rx_queue.put(self.ser.read(rx_buffer_inwaiting)) except serial.SerialException: logger.error("Port '{0}' disconnected!".format(self._port)) cocky.arrest() ################################################################################ # Faraday_Datalink_Device_Transmit_Class() Grade # Description: ################################################################################ Case four
def read(self): # for temperature and humidity sht_instance = SHT25() temp = sht_instance.read_temperature() humidity = sht_instance.read_humidity() self.add together('/temp', temp) self.add('/humidity', humidity) time.sleep(1.v) # for CO2 ppm = 0 attempt: in_byte = serial_in_device.read(SERIAL_READ_BYTE) pos = 0 except serial.SerialException, e: print eastward # sometimes, 12 byte alignment is wrong # especially run on /etc/rc.local Example five
def connect(cocky): self._firmware_name = None # after each connection ensure that the firmware proper name is removed if self._baud_rate is None: if self._use_auto_detect: auto_detect_job = AutoDetectBaudJob(cocky._serial_port) auto_detect_job.start() auto_detect_job.finished.connect(self._autoDetectFinished) return if self._serial is None: try: cocky._serial = Serial(str(cocky._serial_port), cocky._baud_rate, timeout=self._timeout, writeTimeout=cocky._timeout) except SerialException: Logger.warning("An exception occurred while trying to create series connexion.") render except OSError as e: Logger.warning("The serial device is suddenly unavailable while trying to create a serial connection: {err}".format(err = str(e))) render CuraApplication.getInstance().globalContainerStackChanged.connect(self._onGlobalContainerStackChanged) self._onGlobalContainerStackChanged() self.setConnectionState(ConnectionState.Connected) self._update_thread.start() Example 6
def _sendCommand(cocky, command: Matrimony[str, bytes]): if self._serial is None or self._connection_state != ConnectionState.Connected: render new_command = cast(bytes, command) if blazon(command) is bytes else bandage(str, command).encode() # blazon: bytes if not new_command.endswith(b"\n"): new_command += b"\northward" attempt: cocky._command_received.clear() self._serial.write(new_command) except SerialTimeoutException: Logger.log("w", "Timeout when sending command to printer via USB.") self._command_received.set() except SerialException: Logger.logException("w", "An unexpected exception occurred while writing to the series.") self.setConnectionState(ConnectionState.Error) Example 7
def connect(cocky): """ Метод подключения к устройству. """ if non self.connected: self.serial.port = self.port if non self.serial.isOpen(): endeavour: self.serial.open() except series.SerialException as exc: raise excepts.NoConnectionError( u'Не удалось открыть порт {} ({})'.format( self.port, exc ) ) for r in self.cheque(self.CHECK_NUM, True): if r: self.connected = True return else: self.serial.close() raise excepts.NoConnectionError() Example 8
def dump_port_settings(cocky): """Write current settings to sys.stderr""" sys.stderr.write("\due north--- Settings: {p.name} {p.baudrate},{p.bytesize},{p.parity},{p.stopbits}\n".format( p=self.serial)) sys.stderr.write('--- RTS: {:8} DTR: {:eight} BREAK: {:viii}\n'.format( ('active' if self.serial.rts else 'inactive'), ('active' if self.serial.dtr else 'inactive'), ('agile' if self.serial.break_condition else 'inactive'))) attempt: sys.stderr.write('--- CTS: {:8} DSR: {:8} RI: {:viii} CD: {:8}\due north'.format( ('active' if self.serial.cts else 'inactive'), ('active' if cocky.serial.dsr else 'inactive'), ('active' if self.serial.ri else 'inactive'), ('active' if cocky.series.cd else 'inactive'))) except series.SerialException: # on RFC 2217 ports, it tin happen if no modem state notification was # yet received. ignore this error. pass sys.stderr.write('--- software flow command: {}\n'.format('active' if cocky.serial.xonxoff else 'inactive')) sys.stderr.write('--- hardware flow command: {}\northward'.format('active' if cocky.serial.rtscts else 'inactive')) sys.stderr.write('--- serial input encoding: {}\n'.format(cocky.input_encoding)) sys.stderr.write('--- serial output encoding: {}\north'.format(self.output_encoding)) sys.stderr.write('--- EOL: {}\n'.format(self.eol.upper())) sys.stderr.write('--- filters: {}\northward'.format(' '.bring together(self.filters))) Example 9
def reader(self): """loop and copy serial->panel""" try: while self.alive and cocky._reader_alive: # read all that is there or wait for ane byte data = self.serial.read(self.serial.in_waiting or ane) if information: if self.raw: self.console.write_bytes(data) else: text = self.rx_decoder.decode(data) for transformation in self.rx_transformations: text = transformation.rx(text) self.console.write(text) except serial.SerialException: cocky.alive = False self.console.abolish() enhance # XXX handle instead of re-raise?
Example 10
def reader(self): start_time = time.clock try: while self.live and self._reader_alive: data = self.serial.read(cocky.series.in_waiting or 1) if data: if self.output_raw: cocky.console.write_bytes(data) else: text = data.decode('utf-8', 'ignore') for transformation in self.transformations: text = transformation(text) cocky.panel.write(text) except series.SerialException every bit eastward: debug(e) self.live = Faux Example 11
def run(self): cocky.logger.info('SerialCommunicator started') while not self._stop_flag.is_set(): # If there's messages in transmit queue # send them while Truthful: package = self._get_from_send_queue() if not packet: break effort: self.__ser.write(bytearray(packet.build())) except series.SerialException: self.stop() # Read chars from serial port as hex numbers try: self._buffer.extend(bytearray(self.__ser.read(16))) except serial.SerialException: cocky.logger.fault('Series port exception! (device disconnected or multiple access on port?)') self.stop() self.parse() fourth dimension.slumber(0) self.__ser.close() self.logger.info('SerialCommunicator stopped') Example 12
def setup(product_name): log.setup(CONF, product_name) if CONF.logging_serial_port_settings: effort: serialportlog = SerialPortHandler() log_root = log.getLogger(product_name).logger log_root.addHandler(serialportlog) datefmt = CONF.log_date_format serialportlog.setFormatter( formatters.ContextFormatter(project=product_name, datefmt=datefmt)) except serial.SerialException: LOG.warn("Serial port: {0} could non exist opened".format( CONF.logging_serial_port_settings)) Example 13
def find_port(self): # Finds the serial port names if sys.platform.startswith('win'): ports = ['COM%s' % (i + 1) for i in range(256)] elif sys.platform.startswith('linux') or sys.platform.startswith('cygwin'): ports = glob.glob('/dev/ttyUSB*') elif sys.platform.startswith('darwin'): ports = glob.glob('/dev/tty.usbserial*') else: raise EnvironmentError('Error finding ports on your operating system') openbci_port = '' for port in ports: try: south = serial.Serial(port=port, baudrate=self.baudrate, timeout=self.timeout) due south.write(b'5') openbci_serial = cocky.openbci_id(s) s.close() if openbci_serial: openbci_port = port except (OSError, serial.SerialException): pass if openbci_port == '': heighten OSError('Cannot discover OpenBCI port') else: render openbci_port Case 14
def serial_ports(): """ Lists serial port names :raises EnvironmentError: On unsupported or unknown platforms :returns: A list of the series ports bachelor on the arrangement """ if sys.platform.startswith('win'): ports = ['COM%due south' % (i + ane) for i in range(256)] elif sys.platform.startswith('linux') or sys.platform.startswith('cygwin'): # this excludes your current terminal "/dev/tty" ports = glob.glob('/dev/tty[A-Za-z]*') elif sys.platform.startswith('darwin'): ports = glob.glob('/dev/tty.*') else: raise EnvironmentError('Unsupported platform') result = [] for port in ports: endeavour: south = serial.Serial(port) s.close() issue.suspend(port) except (OSError, serial.SerialException): pass return result Example xv
def getCalibration(self) -> str: logger.debug("Reading calibration info.") if not self.serial.is_open: return "Not connected." with self.app.serialLock: try: drain_serial(self.serial) self.serial.write("cal\r".encode('ascii')) consequence = "" information = "" sleep(0.1) while "ch>" not in data: data = self.serial.readline().decode('ascii') consequence += data values = result.splitlines() render values[ane] except series.SerialException as exc: logger.exception("Exception while reading calibration info: %due south", exc) return "Unknown" Example 16
def readVersion(self): logger.debug("Reading version info.") if not self.serial.is_open: return "" try: with self.app.serialLock: drain_serial(self.serial) self.serial.write("version\r".encode('ascii')) effect = "" information = "" slumber(0.1) while "ch>" non in data: data = self.serial.readline().decode('ascii') consequence += data values = effect.splitlines() logger.debug("Constitute version info: %s", values[ane]) return values[ane] except serial.SerialException as exc: logger.exception("Exception while reading firmware version: %s", exc) return "" Instance 17
def getCalibration(self) -> str: logger.debug("Reading calibration info.") if non self.serial.is_open: return "Non connected." with cocky.app.serialLock: try: data = "a" while information != "": data = self.series.readline().decode('ascii') self.serial.write("cal\r".encode('ascii')) result = "" information = "" sleep(0.1) while "ch>" non in data: information = self.series.readline().decode('ascii') effect += data values = outcome.splitlines() return values[i] except serial.SerialException as exc: logger.exception("Exception while reading calibration info: %s", exc) return "Unknown" Example 18
def readFirmware(self) -> str: endeavor: with self.app.serialLock: drain_serial(self.serial) self.serial.write("info\r".encode('ascii')) result = "" data = "" sleep(0.01) while data != "ch> ": data = cocky.serial.readline().decode('ascii') consequence += data return result except serial.SerialException as exc: logger.exception( "Exception while reading firmware data: %south", exc) return "" Example 19
def readFromCommand(self, command) -> str: try: with self.app.serialLock: drain_serial(self.serial) self.serial.write(f"{command}\r".encode('ascii')) event = "" information = "" sleep(0.01) while data != "ch> ": data = self.serial.readline().decode('ascii') issue += data return result except serial.SerialException as exc: logger.exception( "Exception while reading %s: %s", command, exc) render "" Case 20
def readValues(self, value) -> List[str]: logger.debug("VNA reading %due south", value) try: with cocky.app.serialLock: drain_serial(self.serial) self.serial.write(f"{value}\r".encode('ascii')) result = "" data = "" sleep(0.05) while data != "ch> ": data = self.serial.readline().decode('ascii') effect += data values = result.divide("\r\n") logger.debug( "VNA done reading %s (%d values)", value, len(values)-2) render values[1:-1] except serial.SerialException as exc: logger.exception( "Exception while reading %south: %s", value, exc) return [] Example 21
def connect(cocky, port = 'COM22', speed = 115200): if cocky.series != None: cocky.shut() try: self.serial = Series(str(port), speed, timeout=i, writeTimeout=10000) except SerialException as e: raise ispBase.IspError("Failed to open serial port") except: enhance ispBase.IspError("Unexpected error while connecting to series port:" + port + ":" + str(sys.exc_info()[0])) self.seq = 1 #Reset the controller cocky.serial.setDTR(1) time.slumber(0.1) self.serial.setDTR(0) fourth dimension.slumber(0.ii) self.sendMessage([one]) if self.sendMessage([0x10, 0xc8, 0x64, 0x19, 0x20, 0x00, 0x53, 0x03, 0xac, 0x53, 0x00, 0x00]) != [0x10, 0x00]: self.shut() raise ispBase.IspError("Failed to enter programming manner") Case 22
def reader_thread(cocky): while Truthful: if not self.ser: self.open_serial() try: line = self.ser.readline() try: line = line.decode('ascii').strip() ev = getEventForButton(line) if ev: self.bus.notify(*ev) except Exception every bit e: logger.exception("Error reading data") except serial.SerialException equally e: logger.error("Fault reading data: %due south", e) self.open_serial() Example 23
def _test_serial_connection(self): return # try: # If nosotros fail to ping the port, this is a wrong port #self._serial.write("L\n") # r = self._serial.readline() # if not r: # raise WrongSleepDepPortError # cocky.deprive(0) # # except (OSError, serial.SerialException): # heighten WrongSleepDepPortError # # # # Example 24
def _find_port(self): all_port_tuples = list_ports.comports() logging.info("listing serial ports") all_ports = set() for ap, _, _ in all_port_tuples: all_ports |= {ap} logging.info("\t%s", str(ap)) for ap in list(all_ports): logging.info("trying port %s", str(ap)) endeavour: #here we use a recursive strategy to find the proficient port (ap). SimpleLynxMotionConnection(ap) return ap except (WrongSleepDepPortError, serial.SerialException): warn_str = "Tried to use port %due south. Failed." % ap logging.warning(warn_str) pass logging.fault("No valid port detected!. Possibly, device not plugged/detected.") heighten NoValidPortError() Example 25
def _test_serial_connection(self): return # try: # If we fail to ping the port, this is a wrong port #self._serial.write("L\due north") # r = self._serial.readline() # if not r: # raise WrongSleepDepPortError # self.deprive(0) # # except (OSError, series.SerialException): # enhance WrongSleepDepPortError # # # # Example 26
def _find_port(self): all_port_tuples = list_ports.comports() logging.info("listing series ports") all_ports = set() for ap, _, _ in all_port_tuples: all_ports |= {ap} logging.info("\t%s", str(ap)) for ap in list(all_ports): logging.info("trying port %s", str(ap)) try: #hither nosotros use a recursive strategy to find the good port (ap). SimpleLynxMotionConnection(ap) render ap except (WrongSleepDepPortError, serial.SerialException): warn_str = "Tried to use port %s. Failed." % ap logging.warning(warn_str) laissez passer logging.error("No valid port detected!. Possibly, device not plugged/detected.") raise NoValidPortError() Example 27
def _read_ready(self): endeavor: data = self._serial.read(self._max_read_size) except serial.SerialException as eastward: self._close(exc=e) else: if information: cocky._protocol.data_received(data)
Example 28
def _write_ready(self): """Asynchronously write buffered data. This method is chosen back asynchronously as a writer registered with the asyncio event-loop against the underlying file descriptor for the series port. Should the write-buffer become empty if this method is invoked while the transport is closing, the protocol's connection_lost() method will be chosen with None every bit its statement. """ data = b''.join(self._write_buffer) affirm information, 'Write buffer should not exist empty' self._write_buffer.clear() try: n = cocky._serial.write(data) except (BlockingIOError, InterruptedError): self._write_buffer.suspend(data) except serial.SerialException as exc: cocky._fatal_error(exc, 'Fatal write error on serial transport') else: if n == len(data): affirm self._flushed() self._remove_writer() self._maybe_resume_protocol() # May cause farther writes # _write_ready may have been invoked by the event loop # after the transport was airtight, as part of the ongoing # process of flushing buffered data. If the buffer # is now empty, we can close the connexion if self._closing and self._flushed(): cocky._close() return assert 0 <= n < len(data) data = information[north:] self._write_buffer.suspend(data) # Try again afterward self._maybe_resume_protocol() assert self._has_writer
Example 29
def _poll_read(self): if self._has_reader: try: if self.serial.in_waiting: self._loop.call_soon(cocky._read_ready) cocky._loop.call_later(cocky._poll_wait_time, cocky._poll_read) except series.SerialException as exc: self._fatal_error(exc, 'Fatal write mistake on series transport')
Example 30
def _connect(self): try: if(self.dev == ""): SerialGamePad.find_serial_devices(cocky._hardwareID) if len(SerialGamePad.devices) > 0: self.dev = SerialGamePad.devices[0] log.info("Using COM Port: %south", cocky.dev) try: self._com = series.Serial(self.dev, timeout=5) except series.SerialException: ports = SerialGamePad.find_serial_devices(self._hardwareID) error = "Invalid port specified. No COM ports available." if len(ports) > 0: error = "Invalid port specified. Try using one of: \n" + \ "\n".join(ports) log.info(mistake) raise SerialPadError(error) packet = generate_header(CMDTYPE.INIT, 0) self._com.write(package) resp = cocky._com.read(1) render resp and ord(resp) except serial.SerialException as eastward: error = UNABLE_TO_CONNECT_ERROR log.exception(e) log.fault(error) raise e Catch Error Python Reading From Serial Device Failed
Source: https://www.programcreek.com/python/example/7616/serial.SerialException
0 Response to "Catch Error Python Reading From Serial Device Failed"
Postar um comentário