xpra icon
Bug tracker and wiki

This bug tracker and wiki are being discontinued
please use https://github.com/Xpra-org/xpra instead.


Ticket #1113: osx-device-notifications.patch

File osx-device-notifications.patch, 14.3 KB (added by Antoine Martin, 5 years ago)

try to get notifications when USB devices are added or removed

  • xpra/platform/darwin/IOKit.py

     
     1#!/usr/bin/env python
     2
     3# Unknown license:
     4#https://github.com/jburgy/hackbug/blob/master/source/IOKit.py
     5
     6'''Python bindings for the `I/O Kit`_ framework
     7
     8The `I/O Kit`_ is a collection of system frameworks, libraries, tools, and other resources for creating device drivers in OS X.
     9It is based on an object-oriented programming model implemented in a restricted form of C++ that omits features unsuitable for
     10use within a multithreaded kernel.  Many of its entry points are wrappers for corresponding mach_ functions.
     11
     12
     13.. _`I/O Kit`: https://developer.apple.com/library/mac/#documentation/devicedrivers/conceptual/IOKitFundamentals/Introduction/Introduction.html
     14.. _mach: http://en.wikipedia.org/wiki/Mach_(kernel)
     15'''
     16
     17#@PydevCodeAnalysisIgnore
     18
     19from CoreFoundation import *
     20from ctypes import *
     21from objc import pyobjc_id, objc_object
     22
     23_iokit = CDLL('/System/Library/Frameworks/iokit.framework/iokit')
     24
     25kUSBVendorID                    = 'idVendor'
     26kUSBProductID                   = 'idProduct'
     27kIOUSBPlane                 = 'kIOUSBPlane'
     28
     29kIOUSBDeviceClassName           = 'IOUSBDevice'
     30kIOUSBInterfaceClassName        = 'IOUSBInterface'
     31
     32kIOPublishNotification          = 'IOServicePublish'
     33kIOFirstPublishNotification     = 'IOServiceFirstPublish'
     34kIOMatchedNotification          = 'IOServiceMatched'
     35kIOFirstMatchNotification       = 'IOServiceFirstMatch'
     36kIOTerminatedNotification       = 'IOServiceTerminate'
     37
     38kIOMasterPortDefault            = c_void_p.in_dll(_iokit, 'kIOMasterPortDefault')
     39kIOProviderClassKey             = 'IOProviderClass'
     40
     41kIOServicePlane                 = 'IOService'
     42kIOBSDNameKey                   = 'BSD Name'
     43kIOCalloutDeviceKey             = 'IOCalloutDevice'
     44kIORegistryIterateRecursively   = 0x00000001
     45kIORegistryIterateParents       = 0x00000002
     46
     47_iokit.IONotificationPortCreate.restype = c_void_p
     48_iokit.IONotificationPortCreate.argtypes = [c_void_p]
     49
     50_iokit.IONotificationPortDestroy.argtypes = [c_void_p]
     51
     52NOTIFICATION = CFUNCTYPE(None, py_object, c_void_p)
     53
     54_iokit.IOServiceAddMatchingNotification.restype = c_int32
     55_iokit.IOServiceAddMatchingNotification.argtypes = [c_void_p, c_char_p, c_void_p, NOTIFICATION, py_object, c_void_p]
     56
     57_iokit.IOObjectRelease.argtypes = [c_void_p]
     58
     59_iokit.IONotificationPortGetRunLoopSource.restype = c_void_p
     60_iokit.IONotificationPortGetRunLoopSource.argtypes = [c_void_p]
     61
     62_iokit.IORegistryEntrySearchCFProperty.restype = c_void_p
     63_iokit.IORegistryEntrySearchCFProperty.argtypes = [c_void_p, c_void_p, c_void_p, c_void_p, c_uint32]
     64
     65def IOServiceMatching(name = kIOUSBDeviceClassName):
     66    '''Create a matching dictionary that specifies an IOService class match.
     67
     68    Reimplements the `simple C function`_ in pure python using pyobjc_ CoreFoundation classes
     69    instead of using ctypes to wrap it and having to promote its result to an `objc_object`.
     70
     71    Returns a CFMutableDictionaryRef_ with a special `kIOProviderClassKey` entry.
     72    A CFMutableDictionaryRef_ behaves very much like a `dict`.
     73
     74    .. _`simple C function`: https://developer.apple.com/library/mac/#documentation/IOKit/Reference/IOKitLib_header_reference/Reference/reference.html#//apple_ref/doc/c_ref/IOServiceMatching
     75    .. _pyobjc: http://pythonhosted.org/pyobjc/
     76    .. _CFMutableDictionaryRef: https://developer.apple.com/library/mac/#documentation/CoreFoundation/Reference/CFMutableDictionaryRef/Reference/reference.html   
     77    '''
     78    dct = CFDictionaryCreateMutable(None, 0, kCFTypeDictionaryKeyCallBacks, kCFTypeDictionaryValueCallBacks)
     79    dct[kIOProviderClassKey] = name
     80    return dct
     81
     82def raw_ptr(obj):
     83    '''Get pyobjc CFString raw pointer
     84
     85    to pass it to IOKit functions via ctypes'''
     86    return pyobjc_id(obj.nsstring())
     87
     88def IORegistryEntrySearchCFProperty(entry, plane, key, allocator=kCFAllocatorDefault, options=kIORegistryIterateRecursively):
     89    '''IORegistryEntrySearchCFProperty_ wrapper
     90
     91    .. _IORegistryEntrySearchCFProperty: https://developer.apple.com/library/mac/#documentation/IOKit/Reference/IOKitLib_header_reference/Reference/reference.html#//apple_ref/doc/c_ref/IORegistryEntrySearchCFProperty
     92    '''
     93    prop = _iokit.IORegistryEntrySearchCFProperty(entry, plane, raw_ptr(CFSTR(key)), allocator, options)
     94    #print("IORegistryEntrySearchCFProperty: %s=%s (%s, %s)" % (key, prop, entry, plane))
     95    if not prop:
     96        return None
     97    return objc_object(c_void_p = prop)
     98
     99class IOIterator(object):
     100    '''Abstraction to iterate over a collection of `I/O Kit`_ objects
     101
     102    Returned by IOServiceGetMatchingServices_ and IOServiceAddMatchingNotification_
     103
     104    .. _IOServiceGetMatchingServices: https://developer.apple.com/library/mac/#documentation/IOKit/Reference/IOKitLib_header_reference/Reference/reference.html#//apple_ref/doc/c_ref/IOServiceGetMatchingServices
     105    '''
     106    def __init__(self, iterator):
     107        self._obj = iterator
     108    def release(self):
     109        _iokit.IOObjectRelease(self._obj)
     110    def __iter__(self):
     111        return self
     112    def next(self):
     113        '''IOIteratorNext_ wrapper
     114
     115        Returns the next object in an iteration.
     116       
     117        .. _IOIteratorNext: https://developer.apple.com/library/mac/#documentation/IOKit/Reference/IOKitLib_header_reference/Reference/reference.html#//apple_ref/doc/c_ref/IOIteratorNext
     118        '''
     119        service = _iokit.IOIteratorNext(self._obj)
     120        if not service:
     121            raise StopIteration
     122        return service
     123
     124def _callback(context, iterator):
     125    context(IOIterator(iterator))
     126
     127def _path_callback(context, iterator):
     128    generator = (IORegistryEntrySearchCFProperty(service, kIOServicePlane, kIOCalloutDeviceKey) for service in IOIterator(iterator) if service is not None)
     129    context(generator)
     130
     131_notification      = NOTIFICATION(_callback)
     132_path_notification = NOTIFICATION(_path_callback)
     133
     134class IONotificationPort(object):
     135    '''IONotificationPortCreate_ wrapper
     136
     137    Creates a notification object for receiving IOKit notifications of new devices or state changes.
     138
     139    .. _IONotificationPortCreate: https://developer.apple.com/library/mac/#documentation/IOKit/Reference/IOKitLib_header_reference/Reference/reference.html#//apple_ref/doc/c_ref/IONotificationPortCreate
     140    '''
     141    def __init__(self, master_port = kIOMasterPortDefault):
     142        self._obj = _iokit.IONotificationPortCreate(master_port)
     143
     144    def __del__(self):
     145        _iokit.IONotificationPortDestroy(self._obj)
     146
     147    def addMatchingNotifications(self, matching, receiver):
     148        '''Rich IOServiceAddMatchingNotification_ wrapper
     149
     150        Look up registered IOService objects that match a matching dictionary, and
     151        hooks notification requests to specially named methods on receiver.
     152
     153        +------------------+-------------------------------+
     154        |Method Name       |Notification Type              |
     155        +==================+===============================+
     156        |`on_publish`      |``kIOPublishNotification``     |
     157        +------------------+-------------------------------+
     158        |`on_first_publish`|``kIOFirstPublishNotification``|
     159        +------------------+-------------------------------+
     160        |`on_match`        |``kIOMatchedNotification``     |
     161        +------------------+-------------------------------+
     162        |`on_first_match`  |``kIOFirstMatchNotification``  |
     163        +------------------+-------------------------------+
     164        |`on_terminate`    |``kIOTerminateNotification``   |
     165        +------------------+-------------------------------+
     166
     167        The method should expect a single argument, which will be an :class:`.IOIterator`.  As a
     168        convenience, receiver can also implement methods whose names start with `on_path_`
     169        instead of `on_`.  These methods will receiver a generator which yields path names for
     170        the relevant devices.
     171
     172
     173        Sample_ receiver implementation::
     174
     175            from CoreFoundation import *
     176            from IOKit import *
     177           
     178            port = IONotificationPort()
     179
     180            matching = IOServiceMatching()
     181            matching[kUSBVendorID] = 0x0451
     182            matching[kUSBProductID] = 0xf432
     183
     184            class Receiver(object):
     185                def on_path_match(self, paths):
     186                    for path in paths:
     187                        print '%s matched' % path
     188                def on_path_terminate(self, paths):
     189                    for path in paths:
     190                        print '%s terminated' % path
     191
     192            receiver = Receiver()
     193            iterator = port.addMatchingNotifications(matching, receiver)
     194
     195            source = port.getRunLoopSource()
     196            CFRunLoopAddSource(CFRunLoopGetCurrent(), source, kCFRunLoopDefaultMode)
     197            CFRunLoopRun()
     198       
     199        .. _IOServiceAddMatchingNotification: https://developer.apple.com/library/mac/#documentation/IOKit/Reference/IOKitLib_header_reference/Reference/reference.html#//apple_ref/doc/c_ref/IOServiceMatchingAddNotification
     200        .. _Sample: https://developer.apple.com/library/mac/#documentation/DeviceDrivers/Conceptual/USBBook/USBDeviceInterfaces/USBDevInterfaces.html
     201        '''
     202        map = (
     203            (kIOPublishNotification,        'on_publish',               _notification),
     204            (kIOFirstPublishNotification,   'on_first_publish',         _notification),
     205            (kIOMatchedNotification,        'on_match',                 _notification),
     206            (kIOFirstMatchNotification,     'on_first_match',           _notification),
     207            (kIOTerminatedNotification,     'on_terminate',             _notification),
     208            (kIOPublishNotification,        'on_path_publish',          _path_notification),
     209            (kIOFirstPublishNotification,   'on_path_first_publish',    _path_notification),
     210            (kIOMatchedNotification,        'on_path_match',            _path_notification),
     211            (kIOFirstMatchNotification,     'on_path_first_match',      _path_notification),
     212            (kIOTerminatedNotification,     'on_path_terminate',        _path_notification),
     213        )
     214
     215        obj = self._obj
     216        matching_id = pyobjc_id(matching)
     217
     218        iterators = {}
     219        for k, v, n in map:
     220            attr = getattr(receiver, v, None)
     221            if not attr:
     222                continue
     223            wrap = py_object(attr)
     224            setattr(receiver, k + '_ctype', wrap) # keep reference
     225            it = c_void_p()
     226            ret = _iokit.IOServiceAddMatchingNotification(obj, k, matching_id, n, wrap, byref(it))
     227            if ret:
     228                raise IOException(ret)
     229            n(wrap, it)
     230            iterators[v] = it
     231
     232        return iterators
     233
     234    def getRunLoopSource(self):
     235        '''IONotificationPortGetRunLoopSource_ wrapper
     236
     237        .. _IONotificationPortGetRunLoopSource: https://developer.apple.com/library/mac/#documentation/IOKit/Reference/IOKitLib_header_reference/Reference/reference.html#//apple_ref/doc/c_ref/IONotificationPortGetRunLoopSource
     238        '''
     239        source = _iokit.IONotificationPortGetRunLoopSource(self._obj)
     240        return CFRunLoopSourceRef(c_void_p = source)
     241
     242if __name__ == '__main__':
     243
     244    port = IONotificationPort()
     245
     246    matching = IOServiceMatching()
     247    #matching[kUSBVendorID] = 0x0451
     248    #matching[kUSBProductID] = 0xf432
     249
     250    class Receiver(object):
     251        def on_path_match(self, paths):
     252            for path in paths:
     253                print '%s matched' % path
     254#                self.serial = serial.serial_for_url(port, baudrate=9600, parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_TWO, timeout=1)
     255
     256        def on_path_terminate(self, paths):
     257            for path in paths:
     258                print '%s terminated' % path
     259
     260    receiver = Receiver()
     261    iterator = port.addMatchingNotifications(matching, receiver)
     262
     263    source = port.getRunLoopSource()
     264   
     265    assert not CFRunLoopAddSource(CFRunLoopGetCurrent(), source, kCFRunLoopDefaultMode)
     266    CFRunLoopRun()
  • xpra/platform/darwin/webcam.py

     
     1# This file is part of Xpra.
     2# Copyright (C) 2016 Antoine Martin <antoine@devloop.org.uk>
     3# Xpra is released under the terms of the GNU GPL v2, or, at your option, any
     4# later version. See the file COPYING for details.
     5
     6from xpra.log import Logger
     7log = Logger("webcam")
     8
     9
     10
     11_listener = None
     12def _init_notification_listener():
     13    global _listener
     14    if _listener is not None:
     15        return
     16    #run in its own thread
     17    import threading
     18    t = threading.Thread(target=_setup_notification_listener)
     19    t.setDaemon(True)
     20    t.start()
     21
     22def _setup_notification_listener():
     23    global _listener
     24    if _listener is not None:
     25        return
     26    try:
     27        _listener = False
     28        from xpra.platform.darwin.IOKit import IONotificationPort, IOServiceMatching
     29        from CoreFoundation import CFRunLoopAddSource, CFRunLoopGetCurrent, kCFRunLoopDefaultMode, CFRunLoopRun
     30
     31        from xpra.platform.webcam import _fire_video_device_change
     32        from xpra.gtk_common.gobject_compat import import_glib
     33        glib = import_glib()
     34        class Receiver(object):
     35            def on_publish(self, paths):
     36                log("Receiver.on_publish(%s)", list(paths))
     37                glib.idle_add(_fire_video_device_change, True, "unknown")
     38
     39            def on_terminate(self, paths):
     40                log("Receiver.on_terminate(%s)", list(paths))
     41                glib.idle_add(_fire_video_device_change, False, "unknown")
     42
     43        port = IONotificationPort()
     44        matching = IOServiceMatching()
     45        receiver = Receiver()
     46        iterator = port.addMatchingNotifications(matching, receiver)
     47        source = port.getRunLoopSource()
     48        CFRunLoopAddSource(CFRunLoopGetCurrent(), source, kCFRunLoopDefaultMode)
     49        _listener = (port, matching, receiver, iterator, source)
     50        CFRunLoopRun()
     51    except ImportError as e:
     52        log("cannot watch for device changes: %s", e)
     53
     54
     55def add_video_device_change_callback(callback):
     56    from xpra.platform.webcam import _video_device_change_callbacks
     57    _init_notification_listener()
     58    _video_device_change_callbacks.append(callback)
     59
     60def remove_video_device_change_callback(callback):
     61    #TODO!
     62    pass