79 def __init__(self, list_unmounted=False, **kwargs):
82 self.retarget_data = {} # Used to retarget mbed-enabled platform properties
85 if isfile(self.MOCK_FILE_NAME) or ("force_mock" in kwargs and kwargs['force_mock']):
86 platform_dbs.append(self.MOCK_FILE_NAME)
87 elif isfile(LOCAL_MOCKS_DATABASE):
88 platform_dbs.append(LOCAL_MOCKS_DATABASE)
89 platform_dbs.append(LOCAL_PLATFORM_DATABASE)
90 self.plat_db = PlatformDatabase(platform_dbs,
91 primary_database=platform_dbs[0])
92 self.list_unmounted = list_unmounted
94 if 'skip_retarget' not in kwargs or not kwargs['skip_retarget']:
119 self, fs_interaction=FSInteraction.BeforeFilter,
120 filter_function=None, unique_names=False,
121 read_details_txt=False):
122 """ List details of connected devices
123 @return Returns list of structures with detailed info about each mbed
124 @param fs_interaction A member of the FSInteraction class that picks the
125 trade of between quality of service and speed
126 @param filter_function Function that is passed each mbed candidate,
127 should return True if it should be included in the result
128 Ex. mbeds = list_mbeds(filter_function=lambda m: m['platform_name'] == 'K64F')
129 @param unique_names A boolean controlling the presence of the
130 'platform_unique_name' member of the output dict
131 @param read_details_txt A boolean controlling the presense of the
132 output dict attributes read from other files present on the 'mount_point'
133 @details Function returns list of dictionaries with mbed attributes 'mount_point', TargetID name etc.
134 Function returns mbed list with platform names if possible
137 candidates = list(self.find_candidates())
138 logger.debug("Candidates for display %r", candidates)
140 for device in candidates:
141 device['device_type'] = self._detect_device_type(device)
142 if ((not device['mount_point'] or
143 not self.mount_point_ready(device['mount_point'])) and
144 not self.list_unmounted):
145 if (device['target_id_usb_id'] and device['serial_port']):
147 "MBED with target id '%s' is connected, but not mounted. "
148 "Use the '-u' flag to include it in the list.",
149 device['target_id_usb_id'])
151 platform_data = self.plat_db.get(device['target_id_usb_id'][0:4],
152 device_type=device['device_type'] or 'daplink', verbose_data=True)
153 device.update(platform_data or {"platform_name": None})
155 FSInteraction.BeforeFilter: self._fs_before_id_check,
156 FSInteraction.AfterFilter: self._fs_after_id_check,
157 FSInteraction.Never: self._fs_never
158 }[fs_interaction](device, filter_function, read_details_txt)
159 if maybe_device and (maybe_device['mount_point'] or self.list_unmounted):
161 name = device['platform_name']
162 platform_count.setdefault(name, -1)
163 platform_count[name] += 1
164 device['platform_name_unique'] = (
165 "%s[%d]" % (name, platform_count[name]))
167 device.update(self.retarget_data[device['target_id']])
168 logger.debug("retargeting %s with %r",
170 self.retarget_data[device['target_id']])
174 # This is done for API compatibility, would prefer for this to just be None
175 device['device_type'] = device['device_type'] if device['device_type'] else 'unknown'
176 result.append(maybe_device)
189 def _fs_before_id_check(self, device, filter_function, read_details_txt):
190 """Filter device after touching the file system of the device.
191 Said another way: Touch the file system before filtering
194 device['target_id'] = device['target_id_usb_id']
195 self._update_device_from_fs(device, read_details_txt)
196 if not filter_function or filter_function(device):
201 def _fs_after_id_check(self, device, filter_function, read_details_txt):
202 """Filter device before touching the file system of the device.
203 Said another way: Touch the file system after filtering
205 device['target_id'] = device['target_id_usb_id']
206 device['target_id_mbed_htm'] = None
207 if not filter_function or filter_function(device):
208 self._update_device_from_fs(device, read_details_txt)
213 def _update_device_from_fs(self, device, read_details_txt):
214 """ Updates the device information based on files from its 'mount_point'
215 @param device Dictionary containing device information
216 @param read_details_txt A boolean controlling the presense of the
217 output dict attributes read from other files present on the 'mount_point'
219 if not device.get('mount_point', None):
223 directory_entries = os.listdir(device['mount_point'])
224 device['directory_entries'] = directory_entries
225 device['target_id'] = device['target_id_usb_id']
227 # Always try to update using daplink compatible boards processself.
228 # This is done for backwards compatibility.
229 self._update_device_details_daplink_compatible(device, read_details_txt)
231 if device.get('device_type') == 'jlink':
232 self._update_device_details_jlink(device, read_details_txt)
234 if device.get('device_type') == 'atmel':
235 self._update_device_details_atmel(device, read_details_txt)
237 except (OSError, IOError) as e:
239 'Marking device with mount point "%s" as unmounted due to the '
240 'following error: %s', device['mount_point'], e)
241 device['mount_point'] = None
253 def _update_device_details_daplink_compatible(self, device, read_details_txt):
254 """ Updates the daplink-specific device information based on files from its 'mount_point'
255 @param device Dictionary containing device information
256 @param read_details_txt A boolean controlling the presense of the
257 output dict attributes read from other files present on the 'mount_point'
259 lowercase_directory_entries = [e.lower() for e in device['directory_entries']]
260 if self.MBED_HTM_NAME.lower() in lowercase_directory_entries:
261 self._update_device_from_htm(device)
262 elif not read_details_txt:
263 logger.debug('Since mbed.htm is not present, attempting to use '
264 'details.txt for the target id')
265 read_details_txt = True
267 if read_details_txt and self.DETAILS_TXT_NAME.lower() in lowercase_directory_entries:
268 details_txt = self._details_txt(device['mount_point']) or {}
269 device.update({"daplink_%s" % f.lower().replace(' ', '_'): v
270 for f, v in details_txt.items()})
272 # If details.txt contains the target id, this is the most trusted source
273 if device.get('daplink_unique_id', None):
274 device['target_id'] = device['daplink_unique_id']
276 if device['target_id']:
277 identifier = device['target_id'][0:4]
278 platform_data = self.plat_db.get(identifier,
279 device_type='daplink',
281 if not platform_data:
282 logger.warning('daplink entry: "%s" not found in platform database', identifier)
284 device.update(platform_data)
286 device['platform_name'] = None
288 def _update_device_details_jlink(self, device, _):
289 """ Updates the jlink-specific device information based on files from its 'mount_point'
290 @param device Dictionary containing device information
292 lower_case_map = {e.lower(): e for e in device['directory_entries']}
294 if 'board.html' in lower_case_map:
295 board_file_key = 'board.html'
296 elif 'user guide.html' in lower_case_map:
297 board_file_key = 'user guide.html'
299 logger.warning('No valid file found to update JLink device details')
302 board_file_path = os.path.join(device['mount_point'], lower_case_map[board_file_key])
303 with open(board_file_path, 'r') as board_file:
304 board_file_lines = board_file.readlines()
306 for line in board_file_lines:
307 m = re.search(r'url=([\w\d\:\-/\\\?\.=-_]+)', line)
309 device['url'] = m.group(1).strip()
310 identifier = device['url'].split('/')[-1]
311 platform_data = self.plat_db.get(identifier,
314 if not platform_data:
315 logger.warning('jlink entry: "%s", not found in platform database', identifier)
317 device.update(platform_data)
320 def _update_device_from_htm(self, device):
321 """Set the 'target_id', 'target_id_mbed_htm', 'platform_name' and
322 'daplink_*' attributes by reading from mbed.htm on the device
324 htm_target_id, daplink_info = self._read_htm_ids(device['mount_point'])
326 device.update({"daplink_%s" % f.lower().replace(' ', '_'): v
327 for f, v in daplink_info.items()})
329 logger.debug("Found htm target id, %s, for usb target id %s",
330 htm_target_id, device['target_id_usb_id'])
331 device['target_id'] = htm_target_id
333 logger.debug("Could not read htm on from usb id %s. "
334 "Falling back to usb id",
335 device['target_id_usb_id'])
336 device['target_id'] = device['target_id_usb_id']
337 device['target_id_mbed_htm'] = htm_target_id
339 def _update_device_details_atmel(self, device, _):
340 """ Updates the Atmel device information based on files from its 'mount_point'
341 @param device Dictionary containing device information
342 @param read_details_txt A boolean controlling the presense of the
343 output dict attributes read from other files present on the 'mount_point'
346 # Atmel uses a system similar to DAPLink, but there's no details.txt with a target ID
347 # to identify device we can use the serial, which is ATMLXXXXYYYYYYY
348 # where XXXX is the board identifier.
349 # This can be verified by looking at readme.htm, which also uses the board ID to redirect to platform page
351 device['target_id'] = device['target_id_usb_id'][4:8]
352 platform_data = self.plat_db.get(device['target_id'],
356 device.update(platform_data or {"platform_name": None})
358 def mock_manufacture_id(self, mid, platform_name, oper='+'):
359 """! Replace (or add if manufacture id doesn't exist) entry in self.manufacture_ids
360 @param oper '+' add new mock / override existing entry
361 '-' remove mid from mocking entry
362 @return Mocked structure (json format)
365 self.plat_db.add(mid, platform_name, permanent=True)
367 self.plat_db.remove(mid, permanent=True)
369 raise ValueError("oper can only be [+-]")
373 def list_manufacture_ids(self):
374 """! Creates list of all available mappings for target_id -> Platform
375 @return String with table formatted output
377 from prettytable import PrettyTable, HEADER
379 columns = ['target_id_prefix', 'platform_name']
380 pt = PrettyTable(columns, junction_char="|", hrules=HEADER)
384 for target_id_prefix, platform_name in sorted(self.plat_db.items()):
385 pt.add_row([target_id_prefix, platform_name])
387 return pt.get_string()
412 def get_dummy_platform(self, platform_name):
413 """! Returns simple dummy platform """
414 if not hasattr(self, "dummy_counter"):
415 self.dummy_counter = {} # platform<str>: counter<int>
417 if platform_name not in self.dummy_counter:
418 self.dummy_counter[platform_name] = 0
421 "platform_name": platform_name,
422 "platform_name_unique": "%s[%d]"% (platform_name, self.dummy_counter[platform_name]),
423 "mount_point": "DUMMY",
424 "serial_port": "DUMMY",
425 "target_id": "DUMMY",
426 "target_id_mbed_htm": "DUMMY",
427 "target_id_usb_id": "DUMMY",
428 "daplink_version": "DUMMY"
430 self.dummy_counter[platform_name] += 1
433 def get_supported_platforms(self, device_type=None):
434 """! Return a dictionary of supported target ids and the corresponding platform name
435 @param device_type Filter which device entries are returned from the platform database
436 @return Dictionary of { 'target_id': 'platform_name', ... }
439 if device_type is not None:
440 kwargs['device_type'] = device_type
442 items = self.plat_db.items(**kwargs)
443 return {i[0]: i[1] for i in items}
447 def list_platforms(self):
448 """! Useful if you just want to know which platforms are currently available on the system
449 @return List of (unique values) available platforms
452 mbeds = self.list_mbeds()
453 for i, val in enumerate(mbeds):
454 platform_name = str(val['platform_name'])
455 if platform_name not in result:
456 result.append(platform_name)
461 def list_platforms_ext(self):
462 """! Useful if you just want to know how many platforms of each type are currently available on the system
463 @return Dict of platform: platform_count
466 mbeds = self.list_mbeds()
467 for i, val in enumerate(mbeds):
468 platform_name = str(val['platform_name'])
469 if platform_name not in result:
470 result[platform_name] = 1
472 result[platform_name] += 1
498 def get_string(self, border=False, header=True, padding_width=1, sortby='platform_name'):
499 """! Printing with some sql table like decorators
500 @param border Table border visibility
501 @param header Table header visibility
502 @param padding_width Table padding
503 @param sortby Column used to sort results
504 @return Returns string which can be printed on console
506 from prettytable import PrettyTable, HEADER
508 mbeds = self.list_mbeds(unique_names=True, read_details_txt=True)
510 """ ['platform_name', 'mount_point', 'serial_port', 'target_id'] - columns generated from USB auto-detection
511 ['platform_name_unique', ...] - columns generated outside detection subsystem (OS dependent detection)
513 columns = ['platform_name', 'platform_name_unique', 'mount_point', 'serial_port', 'target_id', 'daplink_version']
514 pt = PrettyTable(columns, junction_char="|", hrules=HEADER)
521 row.append(mbed[col] if col in mbed and mbed[col] else 'unknown')
523 result = pt.get_string(border=border, header=header, padding_width=padding_width, sortby=sortby)
530 def get_json_data_from_file(self, json_spec_filename, verbose=False):
531 """! Loads from file JSON formatted string to data structure
532 @return None if JSON can be loaded
535 with open(json_spec_filename) as data_file:
537 return json.load(data_file)
538 except ValueError as json_error_msg:
539 logger.error("Parsing file(%s): %s", json_spec_filename, json_error_msg)
541 except IOError as fileopen_error_msg:
542 logger.warning(fileopen_error_msg)
557 def _read_htm_ids(self, mount_point):
558 """! Function scans mbed.htm to get information about TargetID.
559 @param mount_point mbed mount point (disk / drive letter)
560 @return Function returns targetID, in case of failure returns None.
561 @details Note: This function should be improved to scan variety of boards' mbed.htm files
565 for line in self._htm_lines(mount_point):
566 target_id = target_id or self._target_id_from_htm(line)
567 ver_bld = self._mbed_htm_comment_section_ver_build(line)
569 result['version'], result['build'] = ver_bld
571 m = re.search(r'url=([\w\d\:/\\\?\.=-_]+)', line)
573 result['url'] = m.group(1).strip()
574 return target_id, result
581 def _mbed_htm_comment_section_ver_build(self, line):
582 """! Check for Version and Build date of interface chip firmware im mbed.htm file
583 @return (version, build) tuple if successful, None if no info found
585 # <!-- Version: 0200 Build: Mar 26 2014 13:22:20 -->
586 m = re.search(r'^<!-- Version: (\d+) Build: ([\d\w: ]+) -->', line)
588 version_str, build_str = m.groups()
589 return (version_str.strip(), build_str.strip())
591 # <!-- Version: 0219 Build: Feb 2 2016 15:20:54 Git Commit SHA: 0853ba0cdeae2436c52efcba0ba76a6434c200ff Git local mods:No-->
592 m = re.search(r'^<!-- Version: (\d+) Build: ([\d\w: ]+) Git Commit SHA', line)
594 version_str, build_str = m.groups()
595 return (version_str.strip(), build_str.strip())
597 # <!-- Version: 0.14.3. build 471 -->
598 m = re.search(r'^<!-- Version: ([\d+\.]+)\. build (\d+) -->', line)
600 version_str, build_str = m.groups()
601 return (version_str.strip(), build_str.strip())
620 def _details_txt(self, mount_point):
621 """! Load DETAILS.TXT to dictionary:
624 Build: Aug 24 2015 17:06:30
625 Git Commit SHA: 27a236b9fe39c674a703c5c89655fbd26b8e27e1
630 # DAPLink Firmware - see https://mbed.com/daplink
631 Unique ID: 0240000029164e45002f0012706e0006f301000097969900
634 Automation allowed: 0
635 Daplink Mode: Interface
636 Interface Version: 0240
637 Git SHA: c765cbb590f57598756683254ca38b211693ae5e
639 USB Interfaces: MSD, CDC, HID
640 Interface CRC: 0x26764ebf
644 path_to_details_txt = os.path.join(mount_point, self.DETAILS_TXT_NAME)
645 with open(path_to_details_txt, 'r') as f:
646 return self._parse_details(f.readlines())
670 def _target_id_from_htm(self, line):
671 """! Extract Target id from htm line.
672 @return Target id or None
674 # Detecting modern mbed.htm file format
675 m = re.search('\?code=([a-fA-F0-9]+)', line)
677 result = m.groups()[0]
678 logger.debug("Found target id %s in htm line %s", result, line)
680 # Last resort, we can try to see if old mbed.htm format is there
681 m = re.search('\?auth=([a-fA-F0-9]+)', line)
683 result = m.groups()[0]
684 logger.debug("Found target id %s in htm line %s", result, line)
701 def _run_cli_process(cmd, shell=True):
702 """! Runs command as a process and return stdout, stderr and ret code
703 @param cmd Command to execute
704 @return Tuple of (stdout, stderr, returncode)
706 from subprocess import Popen, PIPE
708 p = Popen(cmd, shell=shell, stdout=PIPE, stderr=PIPE)
709 _stdout, _stderr = p.communicate()
710 return _stdout, _stderr, p.returncode