Python >> Tutoriel Python >  >> Python Tag >> Windows

Détection de l'insertion/retrait de périphériques d'entrée USB sous Windows 10

Il existe plusieurs façons de détecter ce qui se passe avec les changements d'appareil

Détection des événements d'insertion/suppression USB dans Windows à l'aide de C++

Une méthode suggérée consiste à utiliser le message WM_DEVICECHANGE. https://stackoverflow.com/a/4078996/2830850

Un tel exemple se trouve ci-dessous

#Modified from: http://wiki.wxpython.org/HookingTheWndProc
##########################################################################
##
##  This is a modification of the original WndProcHookMixin by Kevin Moore,
##  modified to use ctypes only instead of pywin32, so it can be used
##  with no additional dependencies in Python 2.5
##
##########################################################################


import sys
import ctypes
#import GUID
from ctypes import c_long, c_int, wintypes

import wx


GWL_WNDPROC = -4
WM_DESTROY  = 2
DBT_DEVTYP_DEVICEINTERFACE = 0x00000005  # device interface class
DBT_DEVICEREMOVECOMPLETE = 0x8004  # device is gone
DBT_DEVICEARRIVAL = 0x8000  # system detected a new device
WM_DEVICECHANGE = 0x0219

class GUID(ctypes.Structure):
    _pack_ = 1
    _fields_ = [("Data1", ctypes.c_ulong),
                ("Data2", ctypes.c_ushort),
                ("Data3", ctypes.c_ushort),
                ("Data4", ctypes.c_ubyte * 8)]

## It's probably not neccesary to make this distinction, but it never hurts to be safe
if 'unicode' in wx.PlatformInfo:
    SetWindowLong = ctypes.windll.user32.SetWindowLongW
    CallWindowProc = ctypes.windll.user32.CallWindowProcW
else:
    SetWindowLong = ctypes.windll.user32.SetWindowLongA
    CallWindowProc = ctypes.windll.user32.CallWindowProcA

## Create a type that will be used to cast a python callable to a c callback function
## first arg is return type, the rest are the arguments
#WndProcType = ctypes.WINFUNCTYPE(c_int, wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM)
WndProcType = ctypes.WINFUNCTYPE(ctypes.c_long, ctypes.c_int, ctypes.c_uint, ctypes.c_int, ctypes.c_int)

if 'unicode' in wx.PlatformInfo:
    RegisterDeviceNotification = ctypes.windll.user32.RegisterDeviceNotificationW
else:
    RegisterDeviceNotification = ctypes.windll.user32.RegisterDeviceNotificationA
RegisterDeviceNotification.restype = wintypes.HANDLE
RegisterDeviceNotification.argtypes = [wintypes.HANDLE, wintypes.c_void_p, wintypes.DWORD]

UnregisterDeviceNotification = ctypes.windll.user32.UnregisterDeviceNotification
UnregisterDeviceNotification.restype = wintypes.BOOL
UnregisterDeviceNotification.argtypes = [wintypes.HANDLE]

class DEV_BROADCAST_DEVICEINTERFACE(ctypes.Structure):
    _fields_ = [("dbcc_size", ctypes.c_ulong),
                ("dbcc_devicetype", ctypes.c_ulong),
                ("dbcc_reserved", ctypes.c_ulong),
                ("dbcc_classguid", GUID),
                ("dbcc_name", ctypes.c_wchar * 256)]

class DEV_BROADCAST_HDR(ctypes.Structure):
    _fields_ = [("dbch_size", wintypes.DWORD),
              ("dbch_devicetype", wintypes.DWORD),
              ("dbch_reserved", wintypes.DWORD)]

GUID_DEVCLASS_PORTS = GUID(0x4D36E978, 0xE325, 0x11CE,
        (ctypes.c_ubyte*8)(0xBF, 0xC1, 0x08, 0x00, 0x2B, 0xE1, 0x03, 0x18))
GUID_DEVINTERFACE_USB_DEVICE = GUID(0xA5DCBF10L, 0x6530,0x11D2,
        (ctypes.c_ubyte*8)(0x90, 0x1F, 0x00,0xC0, 0x4F, 0xB9, 0x51, 0xED))

class WndProcHookMixin:
    """
    This class can be mixed in with any wxWindows window class in order to hook it's WndProc function.
    You supply a set of message handler functions with the function addMsgHandler. When the window receives that
    message, the specified handler function is invoked. If the handler explicitly returns False then the standard
    WindowProc will not be invoked with the message. You can really screw things up this way, so be careful.
    This is not the correct way to deal with standard windows messages in wxPython (i.e. button click, paint, etc)
    use the standard wxWindows method of binding events for that. This is really for capturing custom windows messages
    or windows messages that are outside of the wxWindows world.
    """
    def __init__(self):
        self.__msgDict = {}
        ## We need to maintain a reference to the WndProcType wrapper
        ## because ctypes doesn't
        self.__localWndProcWrapped = None
        self.rtnHandles = []

    def hookWndProc(self):
        self.__localWndProcWrapped = WndProcType(self.localWndProc)
        self.__oldWndProc = SetWindowLong(self.GetHandle(), GWL_WNDPROC, self.__localWndProcWrapped)

    def unhookWndProc(self):
        SetWindowLong(self.GetHandle(), GWL_WNDPROC, self.__oldWndProc)
        ## Allow the ctypes wrapper to be garbage collected
        self.__localWndProcWrapped = None

    def addMsgHandler(self,messageNumber,handler):
        self.__msgDict[messageNumber] = handler

    def localWndProc(self, hWnd, msg, wParam, lParam):
        # call the handler if one exists
        # performance note: "in" is the fastest way to check for a key
        # when the key is unlikely to be found
        # (which is the case here, since most messages will not have handlers).
        # This is called via a ctypes shim for every single windows message
        # so dispatch speed is important
        if msg in self.__msgDict:
            # if the handler returns false, we terminate the message here
            # Note that we don't pass the hwnd or the message along
            # Handlers should be really, really careful about returning false here
            if self.__msgDict[msg](wParam,lParam) == False:
                return

        # Restore the old WndProc on Destroy.
        if msg == WM_DESTROY: self.unhookWndProc()

        return CallWindowProc(self.__oldWndProc, hWnd, msg, wParam, lParam)

    def registerDeviceNotification(self, guid, devicetype=DBT_DEVTYP_DEVICEINTERFACE):
        devIF = DEV_BROADCAST_DEVICEINTERFACE()
        devIF.dbcc_size = ctypes.sizeof(DEV_BROADCAST_DEVICEINTERFACE)
        devIF.dbcc_devicetype = DBT_DEVTYP_DEVICEINTERFACE

        if guid:
            devIF.dbcc_classguid = GUID.GUID(guid)
        return RegisterDeviceNotification(self.GetHandle(), ctypes.byref(devIF), 0)

    def unregisterDeviceNotification(self, handle):
        if UnregisterDeviceNotification(handle) == 0:
            raise Exception("Unable to unregister device notification messages")


# a simple example
if __name__ == "__main__":

    class MyFrame(wx.Frame,WndProcHookMixin):
        def __init__(self,parent):
            WndProcHookMixin.__init__(self)
            wx.Frame.__init__(self,parent,-1,"Insert and Remove USE Device and Watch STDOUT",size=(640,480))

            self.Bind(wx.EVT_CLOSE, self.onClose)

            #Change the following guid to the GUID of the device you want notifications for
            #self.devNotifyHandle = self.registerDeviceNotification(guid="{3c5e1462-5695-4e18-876b-f3f3d08aaf18}")
            dbh = DEV_BROADCAST_DEVICEINTERFACE()
            dbh.dbcc_size = ctypes.sizeof(DEV_BROADCAST_DEVICEINTERFACE)
            dbh.dbcc_devicetype = DBT_DEVTYP_DEVICEINTERFACE
            dbh.dbcc_classguid = GUID_DEVINTERFACE_USB_DEVICE
            self.devNotifyHandle = RegisterDeviceNotification(self.GetHandle(), ctypes.byref(dbh), 0)
            self.addMsgHandler(WM_DEVICECHANGE, self.onDeviceChange)
            self.hookWndProc()

        def onDeviceChange(self,wParam,lParam):
            print "WM_DEVICECHANGE [WPARAM:%i][LPARAM:%i]"%(wParam,lParam)

            if wParam == DBT_DEVICEARRIVAL:
                print "Device Arrival"
            elif wParam == DBT_DEVICEREMOVECOMPLETE:
                print "Device Remvoed"

            if lParam:
                dbh = DEV_BROADCAST_HDR.from_address(lParam)
                if dbh.dbch_devicetype == DBT_DEVTYP_DEVICEINTERFACE:
                    dbd = DEV_BROADCAST_DEVICEINTERFACE.from_address(lParam)
                    #Verify that the USB VID and PID match our assigned VID and PID
                    if 'Vid_10c4&Pid_8382' in dbd.dbcc_name:
                        print "Was Our USB Device"

            return True

        def onClose(self, event):
            self.unregisterDeviceNotification(self.devNotifyHandle)
            event.Skip()

    app = wx.App(False)
    frame = MyFrame(None)
    frame.Show()
    app.MainLoop()

Ci-dessus est tiré de https://github.com/weiwei22844/UsefullPython/blob/fa8603b92cb0b3f6ce00c876f24138211f47e906/HookUsbMsg.py

Revenons maintenant à votre code

import wmi

device_connected_wql = "SELECT * FROM __InstanceCreationEvent WITHIN 2 WHERE TargetInstance ISA \'Win32_Keyboard\'"
device_disconnected_wql = "SELECT * FROM __InstanceDeletionEvent WITHIN 2 WHERE TargetInstance ISA \'Win32_Keyboard\'"

c = wmi.WMI()
connected_watcher = c.watch_for(raw_wql=device_connected_wql)
disconnected_watcher = c.watch_for(raw_wql=device_disconnected_wql)

while 1:
    connected = connected_watcher()
    disconnected = disconnected_watcher()
    if connected:
        print("Keyboard connected")
    if disconnected:
        print("Keyboard disconnected")

La façon dont vous appelez est connected_watcher et disconnected_watcher en série, les deux bloquent l'appel. Ainsi, lorsque vous l'appelez pour la première fois connected devient vrai puis le disconnected_watcher est appelé et bloqué jusqu'à ce qu'un appareil soit déconnecté. C'est pourquoi lorsque vous vous déconnectez, vous voyez les deux messages ensemble.

La façon dont vous pouvez résoudre ce problème est de vous assurer que vous expirez ces requêtes

while 1:
    try:
        connected = connected_watcher(timeout_ms=10)
    except wmi.x_wmi_timed_out:
      pass
    else:
        if connected:
            print("Keyboard connected")

    try:
        disconnected = disconnected_watcher(timeout_ms=10)
    except wmi.x_wmi_timed_out:
      pass
    else:
        if disconnected:
            print("Keyboard disconnected")

Une autre approche consiste à utiliser des threads. Mais pour rendre votre fil de code compatible, vous devez le faire comme ci-dessous

class VolumeRemovalWatcher:
    def __init__(self, callback=None):
        self.stop_wanted=False
        self.callback=callback
    def stop(self):
        self.stop_wanted = True
    def watch_for_events(self):
        if not threading.current_thread() is threading.main_thread():
            pythoncom.CoInitialize()

        try:
            w = WMI()
            watcher = w.Win32_VolumeChangeEvent.watch_for(EventType=3)
            while not self.stop_wanted:
                try:
                    event = watcher(timeout_ms=1000)
                except x_wmi_timed_out:
                    pass
                else:
                    print(event.DriveName)
                    if self.callback is not None:
                        self.callback(event.DriveName)
        except Exception as e:
            print(e)
            return None
        finally:
            if not threading.current_thread() is threading.main_thread():
                pythoncom.CoUninitialize()

Crédits pour le code ci-dessus à https://github.com/utytlanyjoe/pyWmiHandler/blob/89b934301990a4a955ec13db21caaf81d9a94f63/wmi_wrapper.py