#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/darwin.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 subprocess
import platform

from bs4 import BeautifulSoup

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:
        try:
            # Try simple and fast first if this fails fall back to the slower but
            # more robust process.
            return loads(out)
        except ExpatError:
            # Beautiful soup ensures the XML is properly formed after it is parsed
            # so that it can be used by other less lenient commands without problems
            xml_representation = BeautifulSoup(out.decode('utf8'), 'xml')

            if not xml_representation.get_text():
                # The output is not in the XML format
                return loads(out)
            return loads(xml_representation.decode().encode('utf8'))
    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"

        usb_tree = []
        for usb_controller in usb_controllers:
            ioreg_usb = subprocess.Popen(
                ["ioreg", "-a", "-r", cmp_par, usb_controller, "-l"],
                stdout=subprocess.PIPE,
            )
            usb_tree.extend(_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

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 subprocess	×
18	import platform	×
19
20	from bs4 import BeautifulSoup	×
21
22	try:	×
23	from plistlib import loads	×
24	except ImportError:	×
25	from plistlib import readPlistFromString as loads	×
26	from xml.parsers.expat import ExpatError	×
27
28	from .lstools_base import MbedLsToolsBase	×
29
30	import logging	×
31
32	logger = logging.getLogger("mbedls.lstools_darwin")	×
33	logger.addHandler(logging.NullHandler())	×
34	DEBUG = logging.DEBUG	×
35	del logging	×
36
37	mbed_volume_name_match = re.compile(r"\b(mbed\|SEGGER MSD\|ATMEL EDBG Media)\b", re.I)	×
38
39
40	def _plist_from_popen(popen):	×
41	out, _ = popen.communicate()	×
42	if not out:	×
43	return []	×
44	try:	×
45	try:	×
46	# Try simple and fast first if this fails fall back to the slower but
47	# more robust process.
48	return loads(out)	×
49	except ExpatError:	×
50	# Beautiful soup ensures the XML is properly formed after it is parsed
51	# so that it can be used by other less lenient commands without problems
52	xml_representation = BeautifulSoup(out.decode('utf8'), 'xml')	×
53
54	if not xml_representation.get_text():	×
55	# The output is not in the XML format
56	return loads(out)	×
57	return loads(xml_representation.decode().encode('utf8'))	×
58	except ExpatError:	×
59	return []	×
60
61
62	def _find_TTY(obj):	×
63	""" Find the first tty (AKA IODialinDevice) that we can find in the
64	children of the specified object, or None if no tty is present.
65	"""
66	try:	×
67	return obj["IODialinDevice"]	×
68	except KeyError:	×
69	for child in obj.get("IORegistryEntryChildren", []):	×
70	found = _find_TTY(child)	×
71	if found:	×
72	return found	×
73	return None	×
74
75
76	def _prune(current, keys):	×
77	""" Reduce the amount of data we have to sift through to only
78	include the specified keys, and children that contain the
79	specified keys
80	"""
81	pruned_current = {k: current[k] for k in keys if k in current}	×
82	pruned_children = list(	×
83	filter(
84	None, [_prune(c, keys) for c in current.get("IORegistryEntryChildren", [])]
85	)
86	)
87	keep_current = any(k in current for k in keys) or pruned_children	×
88	if keep_current:	×
89	if pruned_children:	×
90	pruned_current["IORegistryEntryChildren"] = pruned_children	×
91	return pruned_current	×
92	else:
93	return {}	×
94
95
96	def _dfs_usb_info(obj, parents):	×
97	""" Find all of the usb info that we can from this particular IORegistry
98	tree with depth first search (and searching the parent stack....)
99	"""
100	output = {}	×
101	if (	×
102	"BSD Name" in obj
103	and obj["BSD Name"].startswith("disk")
104	and mbed_volume_name_match.search(obj["IORegistryEntryName"])
105	):
106	disk_id = obj["BSD Name"]	×
107	usb_info = {"serial": None, "vendor_id": None, "product_id": None, "tty": None}	×
108	for parent in [obj] + parents:	×
109	if "USB Serial Number" in parent:	×
110	usb_info["serial"] = parent["USB Serial Number"]	×
111	if "idVendor" in parent and "idProduct" in parent:	×
112	usb_info["vendor_id"] = format(parent["idVendor"], "04x")	×
113	usb_info["product_id"] = format(parent["idProduct"], "04x")	×
114	if usb_info["serial"]:	×
115	usb_info["tty"] = _find_TTY(parent)	×
116	if all(usb_info.values()):	×
117	break	×
118	logger.debug("found usb info %r", usb_info)	×
119	output[disk_id] = usb_info	×
120	for child in obj.get("IORegistryEntryChildren", []):	×
121	output.update(_dfs_usb_info(child, [obj] + parents))	×
122	return output	×
123
124
125	class MbedLsToolsDarwin(MbedLsToolsBase):	×
126	""" mbed-enabled platform detection on Mac OS X
127	"""
128
129	def __init__(self, **kwargs):	×
130	MbedLsToolsBase.__init__(self, **kwargs)	×
131	self.mac_version = float(".".join(platform.mac_ver()[0].split(".")[:2]))	×
132
133	def find_candidates(self):	×
134	# {volume_id: {serial:, vendor_id:, product_id:, tty:}}
135	volumes = self._volumes()	×
136
137	# {volume_id: mount_point}
138	mounts = self._mount_points()	×
139	return [	×
140	{
141	"mount_point": mounts[v],
142	"serial_port": volumes[v]["tty"],
143	"target_id_usb_id": volumes[v].get("serial"),
144	"vendor_id": volumes[v].get("vendor_id"),
145	"product_id": volumes[v].get("product_id"),
146	}
147	for v in set(volumes.keys()) and set(mounts.keys())
148	if v in mounts and v in volumes
149	]
150
151	def _mount_points(self):	×
152	""" Returns map {volume_id: mount_point} """
153	diskutil_ls = subprocess.Popen(	×
154	["diskutil", "list", "-plist"], stdout=subprocess.PIPE
155	)
156	disks = _plist_from_popen(diskutil_ls)	×
157
158	if logger.isEnabledFor(DEBUG):	×
159	import pprint	×
160
161	logger.debug(	×
162	"disks dict \n%s", pprint.PrettyPrinter(indent=2).pformat(disks)
163	)
164	return {	×
165	disk["DeviceIdentifier"]: disk.get("MountPoint", None)
166	for disk in disks["AllDisksAndPartitions"]
167	}
168
169	def _volumes(self):	×
170	""" returns a map {volume_id: {serial:, vendor_id:, product_id:, tty:}"""
171
172	# to find all the possible mbed volumes, we look for registry entries
173	# under all possible USB tree which have a "BSD Name" that starts with
174	# "disk" # (i.e. this is a USB disk), and have a IORegistryEntryName that
175	# matches /\cmbed/
176	# Once we've found a disk, we can search up for a parent with a valid
177	# serial number, and then search down again to find a tty that's part
178	# of the same composite device
179	# ioreg -a -r -n <usb_controller_name> -l
180	usb_controllers = [	×
181	"AppleUSBXHCI",
182	"AppleUSBUHCI",
183	"AppleUSBEHCI",
184	"AppleUSBOHCI",
185	"IOUSBHostDevice",
186	]
187
188	cmp_par = "-n"	×
189	# For El Captain we need to list all the instances of (-c) rather than
190	# compare names (-n)
191	if self.mac_version >= 10.11:	×
192	cmp_par = "-c"	×
193
194	usb_tree = []	×
195	for usb_controller in usb_controllers:	×
196	ioreg_usb = subprocess.Popen(	×
197	["ioreg", "-a", "-r", cmp_par, usb_controller, "-l"],
198	stdout=subprocess.PIPE,
199	)
200	usb_tree.extend(_plist_from_popen(ioreg_usb))	×
201
202	r = {}	×
203
204	for name, obj in enumerate(usb_tree):	×
205	pruned_obj = _prune(	×
206	obj,
207	[
208	"USB Serial Number",
209	"idVendor",
210	"BSD Name",
211	"IORegistryEntryName",
212	"idProduct",
213	"IODialinDevice",
214	],
215	)
216	if logger.isEnabledFor(DEBUG):	×
217	import pprint	×
218
219	logger.debug(	×
220	"finding in \n%s",
221	pprint.PrettyPrinter(indent=2).pformat(pruned_obj),
222	)
223	r.update(_dfs_usb_info(pruned_obj, []))	×
224
225	logger.debug("_volumes return %r", r)	×
226	return r	×

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