57 logger.debug(
"Mount mapping %r", mount_ids)
61 'mount_point' : mount_ids.get(disk_dev),
62 'serial_port' : serial_ids.get(disk_uuid),
63 'target_id_usb_id' : disk_uuid,
64 'vendor_id': usb_info.get(disk_dev, {}).get(
'vendor_id'),
65 'product_id': usb_info.get(disk_dev, {}).get(
'product_id')
66 }
for disk_uuid, disk_dev
in disk_ids.items()
115 device_names = { os.path.basename(d): d
for d
in block_devices }
116 sysfs_block_devices = set(os.listdir(SYSFS_BLOCK_DEVICE_PATH))
117 common_device_names = sysfs_block_devices.intersection(set(device_names.keys()))
120 for common_device_name
in common_device_names:
121 sysfs_path = os.path.join(SYSFS_BLOCK_DEVICE_PATH, common_device_name)
122 full_sysfs_path = os.readlink(sysfs_path)
123 path_parts = full_sysfs_path.split(
'/')
126 for index, part
in enumerate(path_parts):
127 if self.
udp.search(part):
130 if end_index ==
None:
131 logger.debug(
'Did not find suitable usb folder for usb info: %s', full_sysfs_path)
134 usb_info_rel_path = path_parts[:end_index + 1]
135 usb_info_path = os.path.join(SYSFS_BLOCK_DEVICE_PATH, os.sep.join(usb_info_rel_path))
140 vendor_id_file_paths = os.path.join(usb_info_path,
'idVendor')
141 product_id_file_paths = os.path.join(usb_info_path,
'idProduct')
144 with open(vendor_id_file_paths,
'r')
as vendor_file:
145 vendor_id = vendor_file.read().strip()
147 logger.debug(
'Failed to read vendor id file %s weith error:', vendor_id_file_paths, e)
150 with open(product_id_file_paths,
'r')
as product_file:
151 product_id = product_file.read().strip()
153 logger.debug(
'Failed to read product id file %s weith error:', product_id_file_paths, e)
155 result[device_names[common_device_name]] = {
156 'vendor_id': vendor_id,
157 'product_id': product_id