Admin Control Panel v2.0

✎ Editing: _libudev.py ← Back

# -*- coding: utf-8 -*-
# Copyright (C) 2010, 2011, 2012 Sebastian Wiesner <lunaryorn@googlemail.com>

# This library is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation; either version 2.1 of the License, or (at your
# option) any later version.

# This library 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 Lesser General Public License
# for more details.

# You should have received a copy of the GNU Lesser General Public License
# along with this library; if not, write to the Free Software Foundation,
# Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA

"""
    _libudev
    ========

Wrapper types for libudev.  Use ``libudev`` attribute to access libudev
    functions.

.. moduleauthor::  Sebastian Wiesner  <lunaryorn@googlemail.com>
"""

from __future__ import (print_function, division, unicode_literals,
                        absolute_import)

import os
import errno
from ctypes import (CDLL, Structure, POINTER, get_errno,
                    c_char, c_char_p, c_int, c_ulonglong)
from ctypes.util import find_library

class udev(Structure):
    """
    Dummy for ``udev`` structure.
    """
    pass

udev_p = POINTER(udev)

class udev_enumerate(Structure):
    """
    Dummy for ``udev_enumerate`` structure.
    """

udev_enumerate_p = POINTER(udev_enumerate)

class udev_list_entry(Structure):
    """
    Dummy for ``udev_list_entry`` structure.
    """

udev_list_entry_p = POINTER(udev_list_entry)

class udev_device(Structure):
    """
    Dummy for ``udev_device`` structure.
    """

udev_device_p = POINTER(udev_device)

class udev_monitor(Structure):
    """
    Dummy for ``udev_device`` structure.
    """

udev_monitor_p = POINTER(udev_monitor)

dev_t = c_ulonglong

SIGNATURES = {
    # context
    'udev': dict(
        new=(None, udev_p),
        unref=([udev_p], None),
        ref=([udev_p], udev_p),
        get_sys_path=([udev_p], c_char_p),
        get_dev_path=([udev_p], c_char_p),
        get_run_path=([udev_p], c_char_p),
        get_log_priority=([udev_p], c_int),
        set_log_priority=([udev_p, c_int], None)),
    # enumeration
    'udev_enumerate': dict(
        new=([udev_p], udev_enumerate_p),
        ref=([udev_enumerate_p], udev_enumerate_p),
        unref=([udev_enumerate_p], None),
        add_match_subsystem=([udev_enumerate_p, c_char_p], c_int),
        add_nomatch_subsystem=([udev_enumerate_p, c_char_p], c_int),
        add_match_property=([udev_enumerate_p, c_char_p, c_char_p], c_int),
        add_match_sysattr=([udev_enumerate_p, c_char_p, c_char_p], c_int),
        add_nomatch_sysattr=([udev_enumerate_p, c_char_p, c_char_p], c_int),
        add_match_tag=([udev_enumerate_p, c_char_p], c_int),
        add_match_sysname=([udev_enumerate_p, c_char_p], c_int),
        add_match_parent=([udev_enumerate_p, udev_device_p], c_int),
        add_match_is_initialized=([udev_enumerate_p], c_int),
        scan_devices=([udev_enumerate_p], c_int),
        get_list_entry=([udev_enumerate_p], udev_list_entry_p)),
    # list entries
    'udev_list_entry': dict(
        get_next=([udev_list_entry_p], udev_list_entry_p),
        get_name=([udev_list_entry_p], c_char_p),
        get_value=([udev_list_entry_p], c_char_p)),
    # devices
    'udev_device': dict(
        ref=([udev_device_p], udev_device_p),
        unref=([udev_device_p], None),
        new_from_syspath=([udev_p, c_char_p], udev_device_p),
        new_from_subsystem_sysname=([udev_p, c_char_p, c_char_p],
                                    udev_device_p),
        new_from_devnum=([udev_p, c_char, dev_t], udev_device_p),
        new_from_environment=([udev_p], udev_device_p),
        get_parent=([udev_device_p], udev_device_p),
        get_parent_with_subsystem_devtype=([udev_device_p, c_char_p, c_char_p],
                                           udev_device_p),
        get_devpath=([udev_device_p], c_char_p),
        get_subsystem=([udev_device_p], c_char_p),
        get_syspath=([udev_device_p], c_char_p),
        get_sysnum=([udev_device_p], c_char_p),
        get_sysname=([udev_device_p], c_char_p),
        get_driver=([udev_device_p], c_char_p),
        get_devtype=([udev_device_p], c_char_p),
        get_devnode=([udev_device_p], c_char_p),
        get_property_value=([udev_device_p, c_char_p], c_char_p),
        get_sysattr_value=([udev_device_p, c_char_p], c_char_p),
        get_devnum=([udev_device_p], dev_t),
        get_action=([udev_device_p], c_char_p),
        get_is_initialized=([udev_device_p], c_int),
        get_usec_since_initialized=([udev_device_p], c_ulonglong),
        get_devlinks_list_entry=([udev_device_p], udev_list_entry_p),
        get_tags_list_entry=([udev_device_p], udev_list_entry_p),
        get_properties_list_entry=([udev_device_p], udev_list_entry_p),
        get_sysattr_list_entry=([udev_device_p], udev_list_entry_p),
        has_tag=([udev_device_p, c_char_p], c_int)),
    # monitoring
    'udev_monitor': dict(
        ref=([udev_monitor_p], udev_monitor_p),
        unref=([udev_monitor_p], None),
        new_from_netlink=([udev_p, c_char_p], udev_monitor_p),
        new_from_socket=([udev_p, c_char_p], udev_monitor_p),
        enable_receiving=([udev_monitor_p], c_int),
        set_receive_buffer_size=([udev_monitor_p], c_int),
        get_fd=([udev_monitor_p], c_int),
        receive_device=([udev_monitor_p], udev_device_p),
        filter_add_match_subsystem_devtype=(
            [udev_monitor_p, c_char_p, c_char_p], c_int),
        filter_add_match_tag=([udev_monitor_p, c_char_p], c_int),
        filter_update=([udev_monitor_p], c_int),
        filter_remove=([udev_monitor_p], c_int))
    }

ERRNO_EXCEPTIONS = {
    errno.ENOMEM: MemoryError,
    errno.EOVERFLOW: OverflowError,
    errno.EINVAL: ValueError
}

def exception_from_errno(errno):
    """
    Create an exception from ``errno``.

``errno`` is an integral error number.

Return an exception object appropriate to ``errno``.
    """
    exception = ERRNO_EXCEPTIONS.get(errno)
    if exception is not None:
        return exception()
    else:
        return EnvironmentError(errno, os.strerror(errno))

def check_negative_errorcode(result, func, *args):
    """
    Error checker for udev funtions, which return negative error codes.

If ``result`` is smaller than ``0``, it is interpreted as negative error
    code, and an appropriate exception is raised:

- ``-ENOMEM`` raises a :exc:`~exceptions.MemoryError`
    - ``-EOVERFLOW`` raises a :exc:`~exceptions.OverflowError`
    - all other error codes raise :exc:`~exceptions.EnvironmentError`

If result is greater or equal to ``0``, it is returned unchanged.
    """
    if result < 0:
        # udev returns the *negative* errno code at this point
        errno = -result
        raise exception_from_errno(errno)
    else:
        return result

def check_errno(result, func, *args):
    """
    Error checker to check the system ``errno`` as returned by
    :func:`ctypes.get_errno()`.

If ``result`` is not ``0``, an exception according to this errno is raised.
    Otherwise nothing happens.
    """
    if result != 0:
        errno = get_errno()
        if errno != 0:
            raise exception_from_errno(errno)
    return result

def check_errno_on_null_pointer(result, func, *args):
    """
    Error checker to check the system ``errno`` as returned by
    :func:`ctypes.get_errno()`.

If ``result`` is a null pointer, an exception according to this errno is
    raised.  Otherwise nothing happens.
    """
    if not result:
        errno = get_errno()
        if errno != 0:
            raise exception_from_errno(errno)
    return result

ERROR_CHECKERS = dict(
    udev_enumerate_add_match_parent=check_negative_errorcode,
    udev_enumerate_add_match_subsystem=check_negative_errorcode,
    udev_enumerate_add_nomatch_subsystem=check_negative_errorcode,
    udev_enumerate_add_match_property=check_negative_errorcode,
    udev_enumerate_add_match_sysattr=check_negative_errorcode,
    udev_enumerate_add_nomatch_sysattr=check_negative_errorcode,
    udev_enumerate_add_match_tag=check_negative_errorcode,
    udev_enumerate_add_match_sysname=check_negative_errorcode,
    udev_enumerate_add_match_is_initialized=check_negative_errorcode,
    udev_monitor_set_receive_buffer_size=check_errno,
    # libudev doc says, enable_receiving returns a negative errno, but tests
    # show that this is not reliable, so query the real error code
    udev_monitor_enable_receiving=check_errno,
    udev_monitor_receive_device=check_errno_on_null_pointer,
    udev_monitor_filter_add_match_subsystem_devtype=check_negative_errorcode,
    udev_monitor_filter_add_match_tag=check_negative_errorcode,
    udev_monitor_filter_update=check_errno,
    udev_monitor_filter_remove=check_errno,
)

def load_udev_library():
    """
    Load the ``udev`` library and return a :class:`ctypes.CDLL` object for
    it.  The library has errno handling enabled.

Important functions are given proper signatures and return types to
    support type checking and argument conversion.

Raise :exc:`~exceptions.ImportError`, if the udev library was not found.
    """
    udev_library_name = find_library('udev')
    if not udev_library_name:
        raise ImportError('No library named udev')
    libudev = CDLL(udev_library_name, use_errno=True)
    # context function signature
    for namespace, members in SIGNATURES.items():
        for funcname in members:
            fullname = '{0}_{1}'.format(namespace, funcname)
            func = getattr(libudev, fullname, None)
            if func:
                argtypes, restype = members[funcname]
                func.argtypes = argtypes
                func.restype = restype
                errorchecker = ERROR_CHECKERS.get(fullname)
                if errorchecker:
                    func.errcheck = errorchecker
    return libudev

Cancel