Da

✏️ Editing: protocol_cp2110.py

#! python
#
# Backend for Silicon Labs CP2110/4 HID-to-UART devices.
#
# This file is part of pySerial. https://github.com/pyserial/pyserial
# (C) 2001-2015 Chris Liechti <cliechti@gmx.net>
# (C) 2019 Google LLC
#
# SPDX-License-Identifier:    BSD-3-Clause

# This backend implements support for HID-to-UART devices manufactured
# by Silicon Labs and marketed as CP2110 and CP2114. The
# implementation is (mostly) OS-independent and in userland. It relies
# on cython-hidapi (https://github.com/trezor/cython-hidapi).

# The HID-to-UART protocol implemented by CP2110/4 is described in the
# AN434 document from Silicon Labs:
# https://www.silabs.com/documents/public/application-notes/AN434-CP2110-4-Interface-Specification.pdf

# TODO items:

# - rtscts support is configured for hardware flow control, but the
#   signaling is missing (AN434 suggests this is done through GPIO).
# - Cancelling reads and writes is not supported.
# - Baudrate validation is not implemented, as it depends on model and configuration.

import struct
import threading

try:
    import urlparse
except ImportError:
    import urllib.parse as urlparse

try:
    import Queue
except ImportError:
    import queue as Queue

import hid  # hidapi

import serial
from serial.serialutil import SerialBase, SerialException, PortNotOpenError, to_bytes, Timeout

# Report IDs and related constant
_REPORT_GETSET_UART_ENABLE = 0x41
_DISABLE_UART = 0x00
_ENABLE_UART = 0x01

_REPORT_SET_PURGE_FIFOS = 0x43
_PURGE_TX_FIFO = 0x01
_PURGE_RX_FIFO = 0x02

_REPORT_GETSET_UART_CONFIG = 0x50

_REPORT_SET_TRANSMIT_LINE_BREAK = 0x51
_REPORT_SET_STOP_LINE_BREAK = 0x52

class Serial(SerialBase):
    # This is not quite correct. AN343 specifies that the minimum
    # baudrate is different between CP2110 and CP2114, and it's halved
    # when using non-8-bit symbols.
    BAUDRATES = (300, 375, 600, 1200, 1800, 2400, 4800, 9600, 19200,
                 38400, 57600, 115200, 230400, 460800, 500000, 576000,
                 921600, 1000000)

def __init__(self, *args, **kwargs):
        self._hid_handle = None
        self._read_buffer = None
        self._thread = None
        super(Serial, self).__init__(*args, **kwargs)

def open(self):
        if self._port is None:
            raise SerialException("Port must be configured before it can be used.")
        if self.is_open:
            raise SerialException("Port is already open.")

self._read_buffer = Queue.Queue()

self._hid_handle = hid.device()
        try:
            portpath = self.from_url(self.portstr)
            self._hid_handle.open_path(portpath)
        except OSError as msg:
            raise SerialException(msg.errno, "could not open port {}: {}".format(self._port, msg))

try:
            self._reconfigure_port()
        except:
            try:
                self._hid_handle.close()
            except:
                pass
            self._hid_handle = None
            raise
        else:
            self.is_open = True
            self._thread = threading.Thread(target=self._hid_read_loop)
            self._thread.setDaemon(True)
            self._thread.setName('pySerial CP2110 reader thread for {}'.format(self._port))
            self._thread.start()

def from_url(self, url):
        parts = urlparse.urlsplit(url)
        if parts.scheme != "cp2110":
            raise SerialException(
                'expected a string in the forms '
                '"cp2110:///dev/hidraw9" or "cp2110://0001:0023:00": '
                'not starting with cp2110:// {{!r}}'.format(parts.scheme))
        if parts.netloc:  # cp2100://BUS:DEVICE:ENDPOINT, for libusb
            return parts.netloc.encode('utf-8')
        return parts.path.encode('utf-8')

def close(self):
        self.is_open = False
        if self._thread:
            self._thread.join(1)  # read timeout is 0.1
            self._thread = None
        self._hid_handle.close()
        self._hid_handle = None

def _reconfigure_port(self):
        parity_value = None
        if self._parity == serial.PARITY_NONE:
            parity_value = 0x00
        elif self._parity == serial.PARITY_ODD:
            parity_value = 0x01
        elif self._parity == serial.PARITY_EVEN:
            parity_value = 0x02
        elif self._parity == serial.PARITY_MARK:
            parity_value = 0x03
        elif self._parity == serial.PARITY_SPACE:
            parity_value = 0x04
        else:
            raise ValueError('Invalid parity: {!r}'.format(self._parity))

if self.rtscts:
            flow_control_value = 0x01
        else:
            flow_control_value = 0x00

data_bits_value = None
        if self._bytesize == 5:
            data_bits_value = 0x00
        elif self._bytesize == 6:
            data_bits_value = 0x01
        elif self._bytesize == 7:
            data_bits_value = 0x02
        elif self._bytesize == 8:
            data_bits_value = 0x03
        else:
            raise ValueError('Invalid char len: {!r}'.format(self._bytesize))

stop_bits_value = None
        if self._stopbits == serial.STOPBITS_ONE:
            stop_bits_value = 0x00
        elif self._stopbits == serial.STOPBITS_ONE_POINT_FIVE:
            stop_bits_value = 0x01
        elif self._stopbits == serial.STOPBITS_TWO:
            stop_bits_value = 0x01
        else:
            raise ValueError('Invalid stop bit specification: {!r}'.format(self._stopbits))

configuration_report = struct.pack(
            '>BLBBBB',
            _REPORT_GETSET_UART_CONFIG,
            self._baudrate,
            parity_value,
            flow_control_value,
            data_bits_value,
            stop_bits_value)

self._hid_handle.send_feature_report(configuration_report)

self._hid_handle.send_feature_report(
            bytes((_REPORT_GETSET_UART_ENABLE, _ENABLE_UART)))
        self._update_break_state()

@property
    def in_waiting(self):
        return self._read_buffer.qsize()

def reset_input_buffer(self):
        if not self.is_open:
            raise PortNotOpenError()
        self._hid_handle.send_feature_report(
            bytes((_REPORT_SET_PURGE_FIFOS, _PURGE_RX_FIFO)))
        # empty read buffer
        while self._read_buffer.qsize():
            self._read_buffer.get(False)

def reset_output_buffer(self):
        if not self.is_open:
            raise PortNotOpenError()
        self._hid_handle.send_feature_report(
            bytes((_REPORT_SET_PURGE_FIFOS, _PURGE_TX_FIFO)))

def _update_break_state(self):
        if not self._hid_handle:
            raise PortNotOpenError()

if self._break_state:
            self._hid_handle.send_feature_report(
                bytes((_REPORT_SET_TRANSMIT_LINE_BREAK, 0)))
        else:
            # Note that while AN434 states "There are no data bytes in
            # the payload other than the Report ID", either hidapi or
            # Linux does not seem to send the report otherwise.
            self._hid_handle.send_feature_report(
                bytes((_REPORT_SET_STOP_LINE_BREAK, 0)))

def read(self, size=1):
        if not self.is_open:
            raise PortNotOpenError()

data = bytearray()
        try:
            timeout = Timeout(self._timeout)
            while len(data) < size:
                if self._thread is None:
                    raise SerialException('connection failed (reader thread died)')
                buf = self._read_buffer.get(True, timeout.time_left())
                if buf is None:
                    return bytes(data)
                data += buf
                if timeout.expired():
                    break
        except Queue.Empty:  # -> timeout
            pass
        return bytes(data)

def write(self, data):
        if not self.is_open:
            raise PortNotOpenError()
        data = to_bytes(data)
        tx_len = len(data)
        while tx_len > 0:
            to_be_sent = min(tx_len, 0x3F)
            report = to_bytes([to_be_sent]) + data[:to_be_sent]
            self._hid_handle.write(report)

data = data[to_be_sent:]
            tx_len = len(data)

def _hid_read_loop(self):
        try:
            while self.is_open:
                data = self._hid_handle.read(64, timeout_ms=100)
                if not data:
                    continue
                data_len = data.pop(0)
                assert data_len == len(data)
                self._read_buffer.put(bytearray(data))
        finally:
            self._thread = None

📁 Sid Gifari File Manager