| #!/usr/bin/env python3 |
| # Copyright 2016 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Allow creation of uart/console interface via usb google serial endpoint.""" |
| |
| import argparse |
| import array |
| import os |
| import sys |
| import termios |
| import threading |
| import time |
| import tty |
| |
| |
| try: |
| import usb # pylint:disable=import-error |
| except ModuleNotFoundError: |
| print("import usb failed") |
| print("try running these commands:") |
| print(" sudo apt-get install python-pip") |
| print(" sudo pip install --pre pyusb") |
| print() |
| sys.exit(-1) |
| |
| |
| class SusbError(Exception): |
| """Class for exceptions of Susb.""" |
| |
| def __init__(self, msg, value=0): |
| """SusbError constructor. |
| |
| Args: |
| msg: string, message describing error in detail |
| value: integer, value of error when non-zero status returned. Default=0 |
| """ |
| super().__init__(msg, value) |
| self.msg = msg |
| self.value = value |
| |
| |
| class Susb: |
| """Provide USB functionality. |
| |
| Instance Variables: |
| _read_ep: pyUSB read endpoint for this interface |
| _write_ep: pyUSB write endpoint for this interface |
| """ |
| |
| READ_ENDPOINT = 0x81 |
| WRITE_ENDPOINT = 0x1 |
| TIMEOUT_MS = 100 |
| |
| def __init__( |
| self, vendor=0x18D1, product=0x500F, interface=1, serialname=None |
| ): |
| """Susb constructor. |
| |
| Discovers and connects to USB endpoints. |
| |
| Args: |
| vendor: usb vendor id of device |
| product: usb product id of device |
| interface: interface number ( 1 - 8 ) of device to use |
| serialname: string of device serialnumber. |
| |
| Raises: |
| SusbError: An error accessing Susb object |
| """ |
| # Find the device. |
| dev_g = usb.core.find(idVendor=vendor, idProduct=product, find_all=True) |
| dev_list = list(dev_g) |
| if dev_list is None: |
| raise SusbError("USB device not found") |
| |
| # Check if we have multiple devices. |
| dev = None |
| if serialname: |
| for device in dev_list: |
| dev_serial = usb.util.get_string(device, device.iSerialNumber) |
| if dev_serial == serialname: |
| dev = device |
| break |
| if dev is None: |
| raise SusbError(f"USB device({serialname}) not found") |
| else: |
| try: |
| dev = dev_list[0] |
| except IndexError as err: |
| raise SusbError( |
| f"USB device {vendor:04x}:{product:04x} not found" |
| ) from err |
| |
| # If we can't set configuration, it's already been set. |
| try: |
| dev.set_configuration() |
| except usb.core.USBError: |
| pass |
| |
| # Get an endpoint instance. |
| cfg = dev.get_active_configuration() |
| intf = usb.util.find_descriptor(cfg, bInterfaceNumber=interface) |
| self._intf = intf |
| |
| if not intf: |
| raise SusbError("Interface not found") |
| |
| # Detach raiden.ko if it is loaded. |
| if dev.is_kernel_driver_active(intf.bInterfaceNumber): |
| dev.detach_kernel_driver(intf.bInterfaceNumber) |
| |
| read_ep_number = intf.bInterfaceNumber + self.READ_ENDPOINT |
| read_ep = usb.util.find_descriptor( |
| intf, bEndpointAddress=read_ep_number |
| ) |
| self._read_ep = read_ep |
| |
| write_ep_number = intf.bInterfaceNumber + self.WRITE_ENDPOINT |
| write_ep = usb.util.find_descriptor( |
| intf, bEndpointAddress=write_ep_number |
| ) |
| self._write_ep = write_ep |
| |
| |
| class SuartError(Exception): |
| """Class for exceptions of Suart.""" |
| |
| def __init__(self, msg, value=0): |
| """SuartError constructor. |
| |
| Args: |
| msg: string, message describing error in detail |
| value: integer, value of error when non-zero status returned. Default=0 |
| """ |
| super().__init__(msg, value) |
| self.msg = msg |
| self.value = value |
| |
| |
| class Suart: |
| """Provide interface to serial usb endpoint.""" |
| |
| def __init__( |
| self, vendor=0x18D1, product=0x501C, interface=0, serialname=None |
| ): |
| """Suart constructor. |
| |
| Initializes USB stream interface. |
| |
| Args: |
| vendor: usb vendor id of device |
| product: usb product id of device |
| interface: interface number of device to use |
| serialname: Defaults to None. |
| |
| Raises: |
| SuartError: If init fails |
| """ |
| self._done = threading.Event() |
| self._susb = Susb( |
| vendor=vendor, |
| product=product, |
| interface=interface, |
| serialname=serialname, |
| ) |
| self._exit = False |
| self._rx_thread = None |
| self._tx_thread = None |
| |
| def wait_until_done(self, timeout=None): |
| """Wait until the background threads are done.""" |
| return self._done.wait(timeout=timeout) |
| |
| def run_rx_thread(self): |
| """Runs the reading background thread.""" |
| try: |
| while True: |
| try: |
| data = self._susb._read_ep.read( # pylint:disable=protected-access |
| 64, self._susb.TIMEOUT_MS |
| ) |
| if data: |
| sys.stdout.buffer.write(data.tobytes()) |
| sys.stdout.buffer.flush() |
| |
| except Exception as err: # pylint:disable=broad-except |
| # If we miss some characters on pty disconnect, that's fine. |
| # ep.read() also throws USBError on timeout, which we discard. |
| if not isinstance(err, (OSError, usb.core.USBError)): |
| print(f"rx {err}") |
| finally: |
| self._done.set() |
| |
| def run_tx_thread(self): |
| """Runs the writing background thread.""" |
| try: |
| while True: |
| try: |
| data = sys.stdin.buffer.read(1) |
| if not data or data == b"\x03": |
| break |
| if data: |
| self._susb._write_ep.write( # pylint:disable=protected-access |
| array.array("B", data), self._susb.TIMEOUT_MS |
| ) |
| except Exception as err: # pylint:disable=broad-except |
| print(f"tx {err}") |
| finally: |
| self._done.set() |
| |
| def run(self): |
| """Creates pthreads to poll USB & PTY for data.""" |
| self._exit = False |
| |
| self._rx_thread = threading.Thread(target=self.run_rx_thread) |
| self._rx_thread.daemon = True |
| self._rx_thread.start() |
| |
| self._tx_thread = threading.Thread(target=self.run_tx_thread) |
| self._tx_thread.daemon = True |
| self._tx_thread.start() |
| |
| |
| parser = argparse.ArgumentParser( |
| description="Open a console to a USB device", |
| formatter_class=argparse.ArgumentDefaultsHelpFormatter, |
| ) |
| parser.add_argument( |
| "-d", |
| "--device", |
| type=str, |
| help="vid:pid of target device", |
| default="18d1:501c", |
| ) |
| parser.add_argument( |
| "-i", "--interface", type=int, help="interface number of console", default=0 |
| ) |
| parser.add_argument( |
| "-s", "--serialno", type=str, help="serial number of device", default="" |
| ) |
| parser.add_argument( |
| "-S", |
| "--notty-exit-sleep", |
| type=float, |
| default=0.2, |
| help="When stdin is *not* a TTY, wait this many seconds " |
| "after EOF from stdin before exiting, to give time for " |
| "receiving a reply from the USB device.", |
| ) |
| |
| |
| def runconsole(): |
| """Run the usb console code |
| |
| Starts the pty thread, and idles until a ^C is caught. |
| """ |
| args = parser.parse_args() |
| |
| vidstr, pidstr = args.device.split(":") |
| vid = int(vidstr, 16) |
| pid = int(pidstr, 16) |
| |
| serialno = args.serialno |
| interface = args.interface |
| |
| sobj = Suart( |
| vendor=vid, product=pid, interface=interface, serialname=serialno |
| ) |
| if sys.stdin.isatty(): |
| tty.setraw(sys.stdin.fileno()) |
| sobj.run() |
| sobj.wait_until_done() |
| if not sys.stdin.isatty() and args.notty_exit_sleep > 0: |
| time.sleep(args.notty_exit_sleep) |
| |
| |
| def main(): |
| """The main function.""" |
| stdin_isatty = sys.stdin.isatty() |
| if stdin_isatty: |
| ffd = sys.stdin.fileno() |
| os.system("stty -echo") |
| old_settings = termios.tcgetattr(ffd) |
| |
| try: |
| runconsole() |
| finally: |
| if stdin_isatty: |
| termios.tcsetattr(ffd, termios.TCSADRAIN, old_settings) |
| os.system("stty echo") |
| # Avoid having the user's shell prompt start mid-line after the final output |
| # from this program. |
| print() |
| |
| |
| if __name__ == "__main__": |
| main() |