| # Copyright 2016 The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import os |
| import re |
| |
| |
| class Device(object): |
| """Create dict object for relay usb connection. |
| |
| This class provides an interface to locate lab equipment without encoding |
| knowledge of the USB bus topology in the lab equipment device drivers. |
| """ |
| |
| KEY_VID = 'vendor_id' |
| KEY_PID = 'product_id' |
| KEY_SN = 'serial_no' |
| KEY_INF = 'inf' |
| KEY_CFG = 'config' |
| KEY_NAME = 'name' |
| KEY_TTY = 'tty_path' |
| KEY_MFG = 'mfg' |
| KEY_PRD = 'product' |
| KEY_VER = 'version' |
| |
| _instance = None |
| |
| _USB_DEVICE_SYS_ROOT = '/sys/bus/usb/devices' |
| _DEV_ROOT = '/dev' |
| |
| _SYS_VENDOR_ID = 'idVendor' |
| _SYS_PRODUCT_ID = 'idProduct' |
| _SYS_SERIAL_NO = 'serial' |
| _INF_CLASS = 'bInterfaceClass' |
| _INF_SUB_CLASS = 'bInterfaceSubClass' |
| _INF_PROTOCOL = 'bInterfaceProtocol' |
| _MFG_STRING = 'manufacturer' |
| _PRODUCT_STRING = 'product' |
| _VERSION_STRING = 'version' |
| |
| _USB_CDC_ACM_CLASS = 0x02 |
| _USB_CDC_ACM_SUB_CLASS = 0x02 |
| _USB_CDC_ACM_PROTOCOL = 0x01 |
| |
| def __init__(self, name, vid, pid, cfg, inf): |
| self._device_list = [] |
| |
| self._build_device(name, vid, pid, cfg, inf) |
| |
| self._walk_usb_tree(self._init_device_list_callback, None) |
| |
| def __new__(cls, *args, **kwargs): |
| # The Device class should be a singleton. A lab test procedure may |
| # use multiple pieces of lab equipment and we do not want to have to |
| # create a new instance of the Device for each device. |
| if not cls._instance: |
| cls._instance = super(Device, cls).__new__(cls, *args, **kwargs) |
| return cls._instance |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, exception_type, exception_value, traceback): |
| pass |
| |
| def _build_device(self, name, vid, pid, cfg, inf): |
| """Build relay device information. |
| |
| Args: |
| name: device |
| vid: vendor ID |
| pid: product ID |
| cfg: configuration |
| inf: interface |
| |
| Returns: |
| Nothing |
| """ |
| entry = {} |
| entry[self.KEY_NAME] = name |
| entry[self.KEY_VID] = int(vid, 16) |
| entry[self.KEY_PID] = int(pid, 16) |
| |
| # The serial number string is optional in USB and not all devices |
| # use it. The relay devices do not use it then we specify 'None' in |
| # the lab configuration file. |
| entry[self.KEY_SN] = None |
| entry[self.KEY_CFG] = int(cfg) |
| entry[self.KEY_INF] = int(inf) |
| entry[self.KEY_TTY] = None |
| |
| self._device_list.append(entry) |
| |
| def _find_lab_device_entry(self, vendor_id, product_id, serial_no): |
| """find a device in the lab device list. |
| |
| Args: |
| vendor_id: unique vendor id for device |
| product_id: unique product id for device |
| serial_no: serial string for the device (may be None) |
| |
| Returns: |
| device entry or None |
| """ |
| for device in self._device_list: |
| if device[self.KEY_VID] != vendor_id: |
| continue |
| if device[self.KEY_PID] != product_id: |
| continue |
| if device[self.KEY_SN] == serial_no: |
| return device |
| |
| return None |
| |
| def _read_sys_attr(self, root, attr): |
| """read a sysfs attribute. |
| |
| Args: |
| root: path of the sysfs directory |
| attr: attribute to read |
| |
| Returns: |
| attribute value or None |
| """ |
| try: |
| path = os.path.join(root, attr) |
| with open(path) as f: |
| return f.readline().rstrip() |
| except IOError: |
| return None |
| |
| def _read_sys_hex_attr(self, root, attr): |
| """read a sysfs hexadecimal integer attribute. |
| |
| Args: |
| root: path of the sysfs directory |
| attr: attribute to read |
| |
| Returns: |
| attribute value or None |
| """ |
| try: |
| path = os.path.join(root, attr) |
| with open(path) as f: |
| return int(f.readline(), 16) |
| except IOError: |
| return None |
| |
| def _is_cdc_acm(self, inf_path): |
| """determine if the interface implements the CDC ACM class. |
| |
| Args: |
| inf_path: directory entry for the inf under /sys/bus/usb/devices |
| |
| Returns: |
| True if the inf is CDC ACM, false otherwise |
| """ |
| cls = self._read_sys_hex_attr(inf_path, self._INF_CLASS) |
| sub_cls = self._read_sys_hex_attr(inf_path, self._INF_SUB_CLASS) |
| proto = self._read_sys_hex_attr(inf_path, self._INF_PROTOCOL) |
| if self._USB_CDC_ACM_CLASS != cls: |
| return False |
| if self._USB_CDC_ACM_SUB_CLASS != sub_cls: |
| return False |
| if self._USB_CDC_ACM_PROTOCOL != proto: |
| return False |
| |
| return True |
| |
| def _read_tty_name(self, dir_entry, inf, cfg): |
| """Get the path to the associated tty device. |
| |
| Args: |
| dir_entry: directory entry for the device under /sys/bus/usb/devices |
| inf: Interface number of the device |
| cfg: Configuration number of the device |
| |
| Returns: |
| Path to a tty device or None |
| """ |
| inf_path = os.path.join(self._USB_DEVICE_SYS_ROOT, |
| '%s:%d.%d' % (dir_entry, cfg, inf)) |
| |
| # first determine if this is a CDC-ACM or USB Serial device. |
| if self._is_cdc_acm(inf_path): |
| tty_list = os.listdir(os.path.join(inf_path, 'tty')) |
| |
| # Each CDC-ACM interface should only have one tty device associated |
| # with it so just return the first item in the list. |
| return os.path.join(self._DEV_ROOT, tty_list[0]) |
| else: |
| # USB Serial devices have a link to their ttyUSB* device in the inf |
| # directory |
| tty_re = re.compile(r'ttyUSB\d+$') |
| |
| dir_list = os.listdir(inf_path) |
| for entry in dir_list: |
| if tty_re.match(entry): |
| return os.path.join(self._DEV_ROOT, entry) |
| |
| return None |
| |
| def _init_device_list_callback(self, _, dir_entry): |
| """Callback function used with _walk_usb_tree for device list init. |
| |
| Args: |
| _: Callback context (unused) |
| dir_entry: Directory entry reported by _walk_usb_tree |
| |
| """ |
| path = os.path.join(self._USB_DEVICE_SYS_ROOT, dir_entry) |
| |
| # The combination of vendor id, product id, and serial number |
| # should be sufficient to uniquely identify each piece of lab |
| # equipment. |
| vendor_id = self._read_sys_hex_attr(path, self._SYS_VENDOR_ID) |
| product_id = self._read_sys_hex_attr(path, self._SYS_PRODUCT_ID) |
| serial_no = self._read_sys_attr(path, self._SYS_SERIAL_NO) |
| |
| # For each device try to match it with a device entry in the lab |
| # configuration. |
| device = self._find_lab_device_entry(vendor_id, product_id, serial_no) |
| if device: |
| # If the device is in the lab configuration then determine |
| # which tty device it associated with. |
| device[self.KEY_TTY] = self._read_tty_name(dir_entry, |
| device[self.KEY_INF], |
| device[self.KEY_CFG]) |
| |
| def _list_all_tty_devices_callback(self, dev_list, dir_entry): |
| """Callback for _walk_usb_tree when listing all USB serial devices. |
| |
| Args: |
| dev_list: Device list to fill |
| dir_entry: Directory entry reported by _walk_usb_tree |
| |
| """ |
| dev_path = os.path.join(self._USB_DEVICE_SYS_ROOT, dir_entry) |
| |
| # Determine if there are any interfaces in the sys directory for the |
| # USB Device. |
| inf_re = re.compile(r'\d+-\d+(\.\d+){0,}:(?P<cfg>\d+)\.(?P<inf>\d+)$') |
| inf_dir_list = os.listdir(dev_path) |
| |
| for inf_entry in inf_dir_list: |
| inf_match = inf_re.match(inf_entry) |
| if inf_match is None: |
| continue |
| |
| inf_dict = inf_match.groupdict() |
| inf = int(inf_dict['inf']) |
| cfg = int(inf_dict['cfg']) |
| |
| # Check to see if there is a tty device associated with this |
| # interface. |
| tty_path = self._read_tty_name(dir_entry, inf, cfg) |
| if tty_path is None: |
| continue |
| |
| # This is a TTY interface, create a dictionary of the relevant |
| # sysfs attributes for this device. |
| entry = {} |
| entry[self.KEY_TTY] = tty_path |
| entry[self.KEY_INF] = inf |
| entry[self.KEY_CFG] = cfg |
| entry[self.KEY_VID] = self._read_sys_hex_attr(dev_path, |
| self._SYS_VENDOR_ID) |
| entry[self.KEY_PID] = self._read_sys_hex_attr(dev_path, |
| self._SYS_PRODUCT_ID) |
| entry[self.KEY_SN] = self._read_sys_attr(dev_path, |
| self._SYS_SERIAL_NO) |
| entry[self.KEY_MFG] = self._read_sys_attr(dev_path, |
| self._MFG_STRING) |
| entry[self.KEY_PRD] = self._read_sys_attr(dev_path, |
| self._PRODUCT_STRING) |
| entry[self.KEY_VER] = self._read_sys_attr(dev_path, |
| self._VERSION_STRING) |
| |
| # If this device is also in the lab device list then add the |
| # friendly name for it. |
| lab_device = self._find_lab_device_entry(entry[self.KEY_VID], |
| entry[self.KEY_PID], |
| entry[self.KEY_SN]) |
| if lab_device is not None: |
| entry[self.KEY_NAME] = lab_device[self.KEY_NAME] |
| |
| dev_list.append(entry) |
| |
| def _walk_usb_tree(self, callback, context): |
| """Walk the USB device and locate lab devices. |
| |
| Traverse the USB device tree in /sys/bus/usb/devices and inspect each |
| device and see if it matches a device in the lab configuration. If |
| it does then get the path to the associated tty device. |
| |
| Args: |
| callback: Callback to invoke when a USB device is found. |
| context: Context variable for callback. |
| |
| Returns: |
| Nothing |
| """ |
| # Match only devices, exclude interfaces and root hubs |
| file_re = re.compile(r'\d+-\d+(\.\d+){0,}$') |
| dir_list = os.listdir(self._USB_DEVICE_SYS_ROOT) |
| |
| for dir_entry in dir_list: |
| if file_re.match(dir_entry): |
| callback(context, dir_entry) |
| |
| def get_tty_path(self, name): |
| """Get the path to the tty device for a given lab device. |
| |
| Args: |
| name: lab device identifier, e.g. 'rail', or 'bt_trigger' |
| |
| Returns: |
| Path to the tty device otherwise None |
| """ |
| for dev in self._device_list: |
| if dev[self.KEY_NAME] == name and dev[self.KEY_NAME] is not None: |
| return dev[self.KEY_TTY] |
| |
| return None |
| |
| def get_tty_devices(self): |
| """Get a list of all USB based tty devices attached to the machine. |
| |
| Returns: |
| List of dictionaries where each dictionary contains a description of |
| the USB TTY device. |
| """ |
| all_dev_list = [] |
| self._walk_usb_tree(self._list_all_tty_devices_callback, all_dev_list) |
| |
| return all_dev_list |
| |
