#457

Committed 24 Aug 2024 09:15PM UTC coverage: 0.0% (-59.9%) from 59.947%

Build # #457

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge 7c6dbce13 into c467d6f14

Run Details

0 of 4902 relevant lines covered (0.0%)

0.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/mbed_os_tools/detect/linux.py

# Copyright (c) 2018, Arm Limited and affiliates.
# SPDX-License-Identifier: Apache-2.0
#
# 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 re
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 os.path.abspath(os.path.join(os.path.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 = os.path.join("/dev", device_type, "by-id")
        if os.path.isdir(dir):
            to_ret = dict(
                self._hex_ids([os.path.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 is 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

1	# Copyright (c) 2018, Arm Limited and affiliates.
2	# SPDX-License-Identifier: Apache-2.0
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15
16	import re	×
17	import os	×
18
19	from .lstools_base import MbedLsToolsBase	×
20
21	import logging	×
22
23	logger = logging.getLogger("mbedls.lstools_linux")	×
24	logger.addHandler(logging.NullHandler())	×
25	del logging	×
26
27	SYSFS_BLOCK_DEVICE_PATH = "/sys/class/block"	×
28
29
30	def _readlink(link):	×
31	content = os.readlink(link)	×
32	if content.startswith(".."):	×
33	return os.path.abspath(os.path.join(os.path.dirname(link), content))	×
34	else:
35	return content	×
36
37
38	class MbedLsToolsLinuxGeneric(MbedLsToolsBase):	×
39	""" mbed-enabled platform for Linux with udev
40	"""
41
42	def __init__(self, **kwargs):	×
43	"""! ctor
44	"""
45	MbedLsToolsBase.__init__(self, **kwargs)	×
46	self.nlp = re.compile(r"(pci\|usb)-[0-9a-zA-Z:_-]_(?P<usbid>[0-9a-zA-Z])-.*$")	×
47	self.mmp = re.compile(r"(?P<dev>(/[^/ ])+) on (?P<dir>(/[^/ ])+) ")	×
48	self.udp = re.compile(r"^[0-9]+-[0-9]+[^:\s]*$")	×
49
50	def find_candidates(self):	×
51	disk_ids = self._dev_by_id("disk")	×
52	serial_ids = self._dev_by_id("serial")	×
53	mount_ids = dict(self._fat_mounts())	×
54	usb_info = self._sysfs_block_devices(disk_ids.values())	×
55	logger.debug("Mount mapping %r", mount_ids)	×
56
57	return [	×
58	{
59	"mount_point": mount_ids.get(disk_dev),
60	"serial_port": serial_ids.get(disk_uuid),
61	"target_id_usb_id": disk_uuid,
62	"vendor_id": usb_info.get(disk_dev, {}).get("vendor_id"),
63	"product_id": usb_info.get(disk_dev, {}).get("product_id"),
64	}
65	for disk_uuid, disk_dev in disk_ids.items()
66	]
67
68	def _dev_by_id(self, device_type):	×
69	"""! Get a dict, USBID -> device, for a device class
70	@param device_type The type of devices to search. For exmaple, "serial"
71	looks for all serial devices connected to this computer
72	@return A dict: Device USBID -> device file in /dev
73	"""
74	dir = os.path.join("/dev", device_type, "by-id")	×
75	if os.path.isdir(dir):	×
76	to_ret = dict(	×
77	self._hex_ids([os.path.join(dir, f) for f in os.listdir(dir)])
78	)
79	logger.debug("Found %s devices by id %r", device_type, to_ret)	×
80	return to_ret	×
81	else:
82	logger.error(	×
83	"Could not get %s devices by id. "
84	"This could be because your Linux distribution "
85	"does not use udev, or does not create /dev/%s/by-id "
86	"symlinks. Please submit an issue to github.com/"
87	"armmbed/mbed-ls.",
88	device_type,
89	device_type,
90	)
91	return {}	×
92
93	def _fat_mounts(self):	×
94	"""! Lists mounted devices with vfat file system (potential mbeds)
95	@result Returns list of all mounted vfat devices
96	@details Uses Linux shell command: 'mount'
97	"""
98	_stdout, _, retval = self._run_cli_process("mount")	×
99	if not retval:	×
100	for line in _stdout.splitlines():	×
101	if b"vfat" in line:	×
102	match = self.mmp.search(line.decode("utf-8"))	×
103	if match:	×
104	yield match.group("dev"), match.group("dir")	×
105
106	def _hex_ids(self, dev_list):	×
107	"""! Build a USBID map for a device list
108	@param disk_list List of disks in a system with USBID decoration
109	@return Returns map USBID -> device file in /dev
110	@details Uses regular expressions to get a USBID (TargeTIDs) a "by-id"
111	symbolic link
112	"""
113	logger.debug("Converting device list %r", dev_list)	×
114	for dl in dev_list:	×
115	match = self.nlp.search(dl)	×
116	if match:	×
117	yield match.group("usbid"), _readlink(dl)	×
118
119	def _sysfs_block_devices(self, block_devices):	×
120	device_names = {os.path.basename(d): d for d in block_devices}	×
121	sysfs_block_devices = set(os.listdir(SYSFS_BLOCK_DEVICE_PATH))	×
122	common_device_names = sysfs_block_devices.intersection(set(device_names.keys()))	×
123	result = {}	×
124
125	for common_device_name in common_device_names:	×
126	sysfs_path = os.path.join(SYSFS_BLOCK_DEVICE_PATH, common_device_name)	×
127	full_sysfs_path = os.readlink(sysfs_path)	×
128	path_parts = full_sysfs_path.split("/")	×
129
130	end_index = None	×
131	for index, part in enumerate(path_parts):	×
132	if self.udp.search(part):	×
133	end_index = index	×
134
135	if end_index is None:	×
136	logger.debug(	×
137	"Did not find suitable usb folder for usb info: %s", full_sysfs_path
138	)
139	continue	×
140
141	usb_info_rel_path = path_parts[: end_index + 1]	×
142	usb_info_path = os.path.join(	×
143	SYSFS_BLOCK_DEVICE_PATH, os.sep.join(usb_info_rel_path)
144	)
145
146	vendor_id = None	×
147	product_id = None	×
148
149	vendor_id_file_paths = os.path.join(usb_info_path, "idVendor")	×
150	product_id_file_paths = os.path.join(usb_info_path, "idProduct")	×
151
152	try:	×
153	with open(vendor_id_file_paths, "r") as vendor_file:	×
154	vendor_id = vendor_file.read().strip()	×
155	except OSError as e:	×
156	logger.debug(	×
157	"Failed to read vendor id file %s weith error:",
158	vendor_id_file_paths,
159	e,
160	)
161
162	try:	×
163	with open(product_id_file_paths, "r") as product_file:	×
164	product_id = product_file.read().strip()	×
165	except OSError as e:	×
166	logger.debug(	×
167	"Failed to read product id file %s weith error:",
168	product_id_file_paths,
169	e,
170	)
171
172	result[device_names[common_device_name]] = {	×
173	"vendor_id": vendor_id,
174	"product_id": product_id,
175	}
176
177	return result	×

ARMmbed / mbed-os-tools / #457

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous