# Miro - an RSS based video player application# Copyright (C) 2010, 2011# Participatory Culture Foundation## This program is free software; you can redistribute it and/or modify# it under the terms of the GNU General Public License as published by# the Free Software Foundation; either version 2 of the License, or# (at your option) any later version.## This program is distributed in the hope that it will be useful,# but WITHOUT ANY WARRANTY; without even the implied warranty of# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the# GNU General Public License for more details.## You should have received a copy of the GNU General Public License# along with this program; if not, write to the Free Software# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA## In addition, as a special exception, the copyright holders give# permission to link the code of portions of this program with the OpenSSL# library.## You must obey the GNU General Public License in all respects for all of# the code used other than OpenSSL. If you modify file(s) with this# exception, you may extend this exception to your version of the file(s),# but you are not obligated to do so. If you do not wish to do so, delete# this exception statement from your version. If you delete this exception# statement from all source files in the program, then also delete it here.import loggingimport ctypes, ctypes.wintypesimport _winregLOTS_OF_DEBUGGING = Falseif __name__ == '__main__' and LOTS_OF_DEBUGGING: import sys logging.basicConfig(level=logging.DEBUG)def warn(what, code, message): logging.warn('error doing %s (%d): %s', what, code, message)INVALID_HANDLE_VALUE = -1DIGCF_PRESENT = 0x00000002DIGCF_DEVICEINTERFACE = 0x00000010ERROR_INSUFFICIENT_BUFFER = 122ERROR_NO_MORE_ITEMS = 259MAXIMUM_USB_STRING_LENGTH = 255kernel32 = ctypes.windll.kernel32setupapi = ctypes.windll.setupapiSetupDiGetClassDevs = setupapi.SetupDiGetClassDevsWSetupDiEnumDeviceInfo = setupapi.SetupDiEnumDeviceInfoSetupDiEnumDeviceInterfaces = setupapi.SetupDiEnumDeviceInterfacesSetupDiGetDeviceInterfaceDetail = setupapi.SetupDiGetDeviceInterfaceDetailWCM_Get_Parent = setupapi.CM_Get_ParentCM_Get_Device_ID = setupapi.CM_Get_Device_IDWCM_Request_Device_Eject = setupapi.CM_Request_Device_EjectWclass GUID(ctypes.Structure): _fields_ = [("Data1", ctypes.c_ulong), ("Data2", ctypes.c_ushort), ("Data3", ctypes.c_ushort), ("Data4", ctypes.c_ubyte * 8)] def __str__(self): return '{%08X-%04X-%04X-%04X-%012X}' % ( self.Data1, self.Data2, self.Data3, self.Data4[0] * 256 + self.Data4[1], self.Data4[2] * (256 ** 5) + self.Data4[3] * (256 ** 4) + self.Data4[4] * (256 ** 3) + self.Data4[5] * (256 ** 2) + self.Data4[6] * 256 + self.Data4[7])class SP_DEVINFO_DATA(ctypes.Structure): _fields_ = [("cbSize", ctypes.wintypes.DWORD), ("ClassGuid", GUID), ("DevInst", ctypes.wintypes.DWORD), ("Reserved", ctypes.c_void_p) ]class SP_DEVICE_INTERFACE_DATA(ctypes.Structure): _fields_ = [("cbSize", ctypes.wintypes.DWORD), ("InterfaceClassGuid", GUID), ("Flags", ctypes.wintypes.DWORD), ("Reserved", ctypes.c_void_p) ]class SP_DEVICE_INTERFACE_DETAIL_DATA(ctypes.Structure): _fields_ = [("cbSize", ctypes.wintypes.DWORD), ("DevicePath", ctypes.c_wchar*255)]GUID_DEVINTERFACE_VOLUME = GUID(0x53F5630D, 0xB6BF, 0x11D0, (ctypes.c_ubyte*8)(0x94, 0xF2, 0x00, 0xA0, 0xC9, 0x1E, 0xFB, 0x8B))GUID_DEVINTERFACE_USB_DEVICE = GUID(0xA5DCBF10, 0x6530, 0x11D2, (ctypes.c_ubyte*8)(0x90, 0x1F, 0x00, 0xC0, 0x4F, 0xB9, 0x51, 0xED))GUID_USB_STORAGE = GUID(0x4D36E967, 0xE325, 0x11CE, (ctypes.c_ubyte*8)(0xBF, 0xC1, 0x08, 0x00, 0x2B, 0xE1, 0x03, 0x18))GUID_STORAGE_VOLUME = GUID(0x71A27CDD, 0x812a, 0x11D0, (ctypes.c_ubyte*8)(0xBE, 0xC7, 0x08, 0x00, 0x2B, 0xE2, 0x09, 0x2F))def get_class_devs(guid=None, enum=0, flags=0): hDevInfo = SetupDiGetClassDevs(guid and ctypes.byref(guid) or None, enum, 0, flags) if hDevInfo == INVALID_HANDLE_VALUE: warn('get_class_devs', ctypes.GetLastError(), ctypes.FormatError()) return None return hDevInfodef get_device_info(handle, i): info_data = SP_DEVINFO_DATA() info_data.cbSize = ctypes.sizeof(SP_DEVINFO_DATA) if SetupDiEnumDeviceInfo( handle, i, ctypes.byref(info_data)): return info_data elif ctypes.GetLastError() == ERROR_NO_MORE_ITEMS: return else: warn('get_device_info', ctypes.GetLastError(), ctypes.FormatError())def device_info_iter(handle): if not handle: return index = 0 while True: info = get_device_info(handle, index) if info: yield info else: break index += 1def get_device_interface(handle, i, guid, device=None): interfaceData = SP_DEVICE_INTERFACE_DATA() interfaceData.cbSize = ctypes.sizeof(SP_DEVICE_INTERFACE_DATA) if SetupDiEnumDeviceInterfaces( handle, device and ctypes.byref(device) or None, ctypes.byref(guid), i, ctypes.byref(interfaceData)): return interfaceData elif ctypes.GetLastError() == ERROR_NO_MORE_ITEMS: return else: warn('get_device_interface', ctypes.GetLastError(), ctypes.FormatError())def device_interface_iter(handle, guid, device=None): if not handle: return index = 0 while True: info = get_device_interface(handle, index, guid, device) if info: yield info else: break index += 1def get_device_interface_detail(handle, interface): detail = None size = 0 length = ctypes.wintypes.DWORD(0) device = SP_DEVINFO_DATA(cbSize=ctypes.sizeof(SP_DEVINFO_DATA)) while not SetupDiGetDeviceInterfaceDetail( handle, ctypes.byref(interface), detail and ctypes.byref(detail) or None, size, ctypes.byref(length), ctypes.byref(device) ): if ctypes.GetLastError() == ERROR_INSUFFICIENT_BUFFER: size = length.value detail = SP_DEVICE_INTERFACE_DETAIL_DATA( cbSize=6) else: warn('get_device_interface_detail', ctypes.windll.GetLastError(), ctypes.FormatError()) return return detail.DevicePath, devicedef device_eject(devInst): CM_Request_Device_Eject(devInst, None, None, 0, 0)def get_parent(devInst): parent = ctypes.wintypes.DWORD(0) CM_Get_Parent(ctypes.byref(parent), devInst, 0) return parent.valuedef get_device_id(devInst): buffer = ctypes.create_unicode_buffer(255) CM_Get_Device_ID(devInst, ctypes.byref(buffer), 255, 0) return buffer.valuedef get_volume_name(mount_point): buffer = ctypes.create_unicode_buffer(50) kernel32.GetVolumeNameForVolumeMountPointW(mount_point, ctypes.byref(buffer), 50) return buffer.valuedef get_path_name(volume): buffer = ctypes.create_unicode_buffer(255) length = ctypes.wintypes.DWORD(0) kernel32.GetVolumePathNamesForVolumeNameW(volume, ctypes.byref(buffer), 255, ctypes.byref(length)) return buffer.valuedef registry_key_to_dict(section, key): """ Returns a dictionary of data for a given registry key. """ data = {} with _winreg.OpenKey(section, key) as k: index = 0 friendly_name = None while True: try: name, value, type_ = _winreg.EnumValue(k, index) except WindowsError: break data[name] = value index += 1 return data def mounted_drives(): drives = {} for k, v in registry_key_to_dict(_winreg.HKEY_LOCAL_MACHINE, 'SYSTEM\\MountedDevices').items(): if k.startswith('\\DosDevices'): drives[k[-2:]] = v.decode('utf-16-le') return drivesdef connected_devices(): """ Returns a generator which returns small dictionaries of data representing the connected USB storage devices. """ handle = get_class_devs(GUID_USB_STORAGE, flags=DIGCF_PRESENT) drives = mounted_drives() for info in device_info_iter(handle): device_id = get_device_id(info.DevInst) if LOTS_OF_DEBUGGING: logging.debug('connected_devices(): %r', device_id) reg_data = registry_key_to_dict(_winreg.HKEY_LOCAL_MACHINE, 'SYSTEM\\CurrentControlSet\\Enum\\%s' % device_id) pip = reg_data.get('ParentIdPrefix') if pip: for drive, volume in drives.items(): if pip in volume: if LOTS_OF_DEBUGGING: logging.debug('volume/drive name: %r/%r', volume, drive) yield { 'volume': volume, 'mount': drive + '\\', 'name': reg_data['FriendlyName'][:-len(' USB Device')] }if __name__ == '__main__': for d in connected_devices(): print d