Use moved implementation inside mbed-os-tools

Author: bridadan

legacy/mbed-ls/mbed_lstools/darwin.py

@@ -15,167 +15,4 @@
 limitations under the License.
 """
 
-import re
-import subprocess
-import platform
-try:
-    from plistlib import loads
-except ImportError:
-    from plistlib import readPlistFromString as loads
-from xml.parsers.expat import ExpatError
-
-from .lstools_base import MbedLsToolsBase
-
-import logging
-logger = logging.getLogger("mbedls.lstools_darwin")
-logger.addHandler(logging.NullHandler())
-DEBUG = logging.DEBUG
-del logging
-
-mbed_volume_name_match = re.compile(r'\b(mbed|SEGGER MSD|ATMEL EDBG Media)\b', re.I)
-
-
-def _plist_from_popen(popen):
-    out, _ = popen.communicate()
-    if not out:
-        return []
-    try:
-        return loads(out)
-    except ExpatError:
-        return []
-
-def _find_TTY(obj):
-    ''' Find the first tty (AKA IODialinDevice) that we can find in the
-        children of the specified object, or None if no tty is present.
-    '''
-    try:
-        return obj['IODialinDevice']
-    except KeyError:
-        for child in obj.get('IORegistryEntryChildren', []):
-            found = _find_TTY(child)
-            if found:
-                return found
-        return None
-
-
-def _prune(current, keys):
-    """ Reduce the amount of data we have to sift through to only
-        include the specified keys, and children that contain the
-        specified keys
-    """
-    pruned_current = {k: current[k] for k in keys if k in current}
-    pruned_children = list(
-      filter(None, [_prune(c, keys) for c in
-                    current.get('IORegistryEntryChildren', [])]))
-    keep_current = any(k in current for k in keys) or pruned_children
-    if keep_current:
-        if pruned_children:
-            pruned_current['IORegistryEntryChildren'] = pruned_children
-        return pruned_current
-    else:
-        return {}
-
-
-def _dfs_usb_info(obj, parents):
-    """ Find all of the usb info that we can from this particular IORegistry
-        tree with depth first search (and searching the parent stack....)
-    """
-    output = {}
-    if  ('BSD Name' in obj and obj['BSD Name'].startswith('disk') and
-         mbed_volume_name_match.search(obj['IORegistryEntryName'])):
-        disk_id = obj['BSD Name']
-        usb_info = {
-                'serial':None,
-             'vendor_id':None,
-            'product_id':None,
-                   'tty':None,
-        }
-        for parent in [obj] + parents:
-            if 'USB Serial Number' in parent:
-                usb_info['serial'] = parent['USB Serial Number']
-            if 'idVendor' in parent and 'idProduct' in parent:
-                usb_info['vendor_id'] = format(parent['idVendor'], '04x')
-                usb_info['product_id'] = format(parent['idProduct'], '04x')
-            if usb_info['serial']:
-                usb_info['tty'] = _find_TTY(parent)
-            if all(usb_info.values()):
-                break
-        logger.debug("found usb info %r", usb_info)
-        output[disk_id] = usb_info
-    for child in obj.get('IORegistryEntryChildren', []):
-        output.update(_dfs_usb_info(child, [obj] + parents))
-    return output
-
-
-class MbedLsToolsDarwin(MbedLsToolsBase):
-    """ mbed-enabled platform detection on Mac OS X
-    """
-
-    def __init__(self, **kwargs):
-        MbedLsToolsBase.__init__(self, **kwargs)
-        self.mac_version = float('.'.join(platform.mac_ver()[0].split('.')[:2]))
-
-    def find_candidates(self):
-        # {volume_id: {serial:, vendor_id:, product_id:, tty:}}
-        volumes = self._volumes()
-
-        # {volume_id: mount_point}
-        mounts = self._mount_points()
-        return [
-            {
-                'mount_point': mounts[v],
-                'serial_port': volumes[v]['tty'],
-                'target_id_usb_id': volumes[v].get('serial'),
-                'vendor_id': volumes[v].get('vendor_id'),
-                'product_id': volumes[v].get('product_id')
-            } for v in set(volumes.keys()) and set(mounts.keys())
-            if v in mounts and v in volumes
-        ]
-
-    def _mount_points(self):
-        ''' Returns map {volume_id: mount_point} '''
-        diskutil_ls = subprocess.Popen(['diskutil', 'list', '-plist'], stdout=subprocess.PIPE)
-        disks = _plist_from_popen(diskutil_ls)
-
-        if logger.isEnabledFor(DEBUG):
-            import pprint
-            logger.debug("disks dict \n%s", pprint.PrettyPrinter(indent=2).pformat(disks))
-        return {disk['DeviceIdentifier']: disk.get('MountPoint', None)
-                for disk in disks['AllDisksAndPartitions']}
-
-    def _volumes(self):
-        ''' returns a map {volume_id: {serial:, vendor_id:, product_id:, tty:}'''
-
-        # to find all the possible mbed volumes, we look for registry entries
-        # under all possible USB tree which have a "BSD Name" that starts with
-        # "disk" # (i.e. this is a USB disk), and have a IORegistryEntryName that
-        # matches /\cmbed/
-        # Once we've found a disk, we can search up for a parent with a valid
-        # serial number, and then search down again to find a tty that's part
-        # of the same composite device
-        # ioreg -a -r -n <usb_controller_name> -l
-        usb_controllers = ['AppleUSBXHCI', 'AppleUSBUHCI', 'AppleUSBEHCI',
-                           'AppleUSBOHCI', 'IOUSBHostDevice']
-
-        cmp_par = '-n'
-        # For El Captain we need to list all the instances of (-c) rather than
-        # compare names (-n)
-        if self.mac_version >= 10.11:
-            cmp_par = '-c'
-
-        for usb_controller in usb_controllers:
-            ioreg_usb = subprocess.Popen(['ioreg', '-a', '-r', cmp_par, usb_controller, '-l'], stdout=subprocess.PIPE)
-            usb_tree = _plist_from_popen(ioreg_usb)
-
-        r = {}
-
-        for name, obj in enumerate(usb_tree):
-            pruned_obj = _prune(obj, ['USB Serial Number', 'idVendor', 'BSD Name',
-                                      'IORegistryEntryName', 'idProduct', 'IODialinDevice'])
-            if logger.isEnabledFor(DEBUG):
-                import pprint
-                logger.debug("finding in \n%s", pprint.PrettyPrinter(indent=2).pformat(pruned_obj))
-            r.update(_dfs_usb_info(pruned_obj, []))
-
-        logger.debug("_volumes return %r", r)
-        return r
+from mbed_os_tools.detect.darwin import MbedLsToolsDarwin

legacy/mbed-ls/mbed_lstools/linux.py

@@ -15,146 +15,4 @@
 limitations under the License.
 """
 
-import re
-from os.path import join, isdir, dirname, abspath
-import os
-
-from .lstools_base import MbedLsToolsBase
-
-import logging
-logger = logging.getLogger("mbedls.lstools_linux")
-logger.addHandler(logging.NullHandler())
-del logging
-
-SYSFS_BLOCK_DEVICE_PATH = '/sys/class/block'
-
-def _readlink(link):
-    content = os.readlink(link)
-    if content.startswith(".."):
-        return abspath(join(dirname(link), content))
-    else:
-        return content
-
-
-class MbedLsToolsLinuxGeneric(MbedLsToolsBase):
-    """ mbed-enabled platform for Linux with udev
-    """
-    def __init__(self, **kwargs):
-        """! ctor
-        """
-        MbedLsToolsBase.__init__(self, **kwargs)
-        self.nlp = re.compile(
-            r'(pci|usb)-[0-9a-zA-Z:_-]*_(?P<usbid>[0-9a-zA-Z]*)-.*$')
-        self.mmp = re.compile(
-            r'(?P<dev>(/[^/ ]*)+) on (?P<dir>(/[^/ ]*)+) ')
-        self.udp = re.compile(r'^[0-9]+-[0-9]+[^:\s]*$')
-
-    def find_candidates(self):
-        disk_ids = self._dev_by_id('disk')
-        serial_ids = self._dev_by_id('serial')
-        mount_ids = dict(self._fat_mounts())
-        usb_info = self._sysfs_block_devices(disk_ids.values())
-        logger.debug("Mount mapping %r", mount_ids)
-
-        return [
-            {
-                'mount_point' : mount_ids.get(disk_dev),
-                'serial_port' : serial_ids.get(disk_uuid),
-                'target_id_usb_id' : disk_uuid,
-                'vendor_id': usb_info.get(disk_dev, {}).get('vendor_id'),
-                'product_id': usb_info.get(disk_dev, {}).get('product_id')
-            } for disk_uuid, disk_dev in disk_ids.items()
-        ]
-
-    def _dev_by_id(self, device_type):
-        """! Get a dict, USBID -> device, for a device class
-        @param device_type The type of devices to search. For exmaple, "serial"
-          looks for all serial devices connected to this computer
-        @return A dict: Device USBID -> device file in /dev
-        """
-        dir = join("/dev", device_type, "by-id")
-        if isdir(dir):
-            to_ret = dict(self._hex_ids([join(dir, f) for f in os.listdir(dir)]))
-            logger.debug("Found %s devices by id %r", device_type, to_ret)
-            return to_ret
-        else:
-            logger.error("Could not get %s devices by id. "
-                         "This could be because your Linux distribution "
-                         "does not use udev, or does not create /dev/%s/by-id "
-                         "symlinks. Please submit an issue to github.com/"
-                         "armmbed/mbed-ls.", device_type, device_type)
-            return {}
-
-    def _fat_mounts(self):
-        """! Lists mounted devices with vfat file system (potential mbeds)
-        @result Returns list of all mounted vfat devices
-        @details Uses Linux shell command: 'mount'
-        """
-        _stdout, _, retval = self._run_cli_process('mount')
-        if not retval:
-            for line in _stdout.splitlines():
-                if b'vfat' in line:
-                    match = self.mmp.search(line.decode('utf-8'))
-                    if match:
-                        yield match.group("dev"), match.group("dir")
-
-    def _hex_ids(self, dev_list):
-        """! Build a USBID map for a device list
-        @param disk_list List of disks in a system with USBID decoration
-        @return Returns map USBID -> device file in /dev
-        @details Uses regular expressions to get a USBID (TargeTIDs) a "by-id"
-          symbolic link
-        """
-        logger.debug("Converting device list %r", dev_list)
-        for dl in dev_list:
-            match = self.nlp.search(dl)
-            if match:
-                yield match.group("usbid"), _readlink(dl)
-
-    def _sysfs_block_devices(self, block_devices):
-        device_names = { os.path.basename(d): d  for d in block_devices }
-        sysfs_block_devices = set(os.listdir(SYSFS_BLOCK_DEVICE_PATH))
-        common_device_names = sysfs_block_devices.intersection(set(device_names.keys()))
-        result = {}
-
-        for common_device_name in common_device_names:
-            sysfs_path = os.path.join(SYSFS_BLOCK_DEVICE_PATH, common_device_name)
-            full_sysfs_path = os.readlink(sysfs_path)
-            path_parts = full_sysfs_path.split('/')
-
-            end_index = None
-            for index, part in enumerate(path_parts):
-                if self.udp.search(part):
-                    end_index = index
-
-            if end_index == None:
-                logger.debug('Did not find suitable usb folder for usb info: %s', full_sysfs_path)
-                continue
-
-            usb_info_rel_path = path_parts[:end_index + 1]
-            usb_info_path = os.path.join(SYSFS_BLOCK_DEVICE_PATH, os.sep.join(usb_info_rel_path))
-
-            vendor_id = None
-            product_id = None
-
-            vendor_id_file_paths = os.path.join(usb_info_path, 'idVendor')
-            product_id_file_paths = os.path.join(usb_info_path, 'idProduct')
-
-            try:
-                with open(vendor_id_file_paths, 'r') as vendor_file:
-                    vendor_id = vendor_file.read().strip()
-            except OSError as e:
-                logger.debug('Failed to read vendor id file %s weith error:', vendor_id_file_paths, e)
-
-            try:
-                with open(product_id_file_paths, 'r') as product_file:
-                    product_id = product_file.read().strip()
-            except OSError as e:
-                logger.debug('Failed to read product id file %s weith error:', product_id_file_paths, e)
-
-            result[device_names[common_device_name]] = {
-                'vendor_id': vendor_id,
-                'product_id': product_id
-            }
-
-        return result
+from mbed_os_tools.detect.linux import MbedLsToolsLinuxGeneric