# 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