Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:56
📄 __init__.py # -- coding: utf-8 -- # Copyright (C) 2010, 2011, 2012 Sebastian Wiesner <lunaryorn@gmail.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 """ pyudev ====== A binding to libudev. The :class:`Context` provides the connection to the udev device database and enumerates devices. Individual devices are represented by the :class:`Device` class. Device monitoring is provided by :class:`Monitor` and :class:`MonitorObserver`. With :mod:`pyudev.pyqt4`, :mod:`pyudev.pyside`, :mod:`pyudev.glib` and :mod:`pyudev.wx` device monitoring can be integrated into the event loop of various GUI toolkits. .. moduleauthor:: Sebastian Wiesner <lunaryorn@gmail.com> """ from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals from pyudev._errors import DeviceNotFoundAtPathError from pyudev._errors import DeviceNotFoundByFileError from pyudev._errors import DeviceNotFoundByNameError from pyudev._errors import DeviceNotFoundByNumberError from pyudev._errors import DeviceNotFoundError from pyudev._errors import DeviceNotFoundInEnvironmentError from pyudev.device import Attributes from pyudev.device import Device from pyudev.device import Devices from pyudev.device import Tags from pyudev.discover import DeviceFileHypothesis from pyudev.discover import DeviceNameHypothesis from pyudev.discover import DeviceNumberHypothesis from pyudev.discover import DevicePathHypothesis from pyudev.discover import Discovery from pyudev.core import Context from pyudev.core import Enumerator from pyudev.monitor import Monitor from pyudev.monitor import MonitorObserver from pyudev.version import __version__ from pyudev.version import __version_info__ from pyudev._util import udev_version	2.44 KB	2020-01-24 21:52
📁 __pycache__	-	2025-02-27 15:06
📄 _compat.py # -- coding: utf-8 -- # Copyright (C) 2011 Sebastian Wiesner <lunaryorn@gmail.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 """ pyudev._compat ============== Compatibility for Python versions, that lack certain functions. .. moduleauthor:: Sebastian Wiesner <lunaryorn@gmail.com> """ from __future__ import (print_function, division, unicode_literals, absolute_import) from subprocess import Popen, CalledProcessError, PIPE def check_output(command): """ Compatibility with :func:`subprocess.check_output` from Python 2.7 and upwards. """ proc = Popen(command, stdout=PIPE) output = proc.communicate()[0] if proc.returncode != 0: raise CalledProcessError(proc.returncode, command) return output	1.43 KB	2020-01-24 21:52
📁 _ctypeslib	-	2025-02-27 15:06
📄 _errors.py # -- coding: utf-8 -- # Copyright (C) 2015 mulhern <amulhern@redhat.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 """ pyudev.device._errors ===================== Errors raised by Device methods. .. moduleauthor:: Sebastian Wiesner <lunaryorn@gmail.com> """ from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals import abc from six import add_metaclass @add_metaclass(abc.ABCMeta) class DeviceError(Exception): """ Any error raised when messing around w/ or trying to discover devices. """ @add_metaclass(abc.ABCMeta) class DeviceNotFoundError(DeviceError): """ An exception indicating that no :class:`Device` was found. .. versionchanged:: 0.5 Rename from ``NoSuchDeviceError`` to its current name. """ class DeviceNotFoundAtPathError(DeviceNotFoundError): """ A :exc:`DeviceNotFoundError` indicating that no :class:`Device` was found at a given path. """ def __init__(self, sys_path): DeviceNotFoundError.__init__(self, sys_path) @property def sys_path(self): """ The path that caused this error as string. """ return self.args[0] def __str__(self): return 'No device at {0!r}'.format(self.sys_path) class DeviceNotFoundByFileError(DeviceNotFoundError): """ A :exc:`DeviceNotFoundError` indicating that no :class:`Device` was found from the given filename. """ class DeviceNotFoundByInterfaceIndexError(DeviceNotFoundError): """ A :exc:`DeviceNotFoundError` indicating that no :class:`Device` was found from the given interface index. """ class DeviceNotFoundByKernelDeviceError(DeviceNotFoundError): """ A :exc:`DeviceNotFoundError` indicating that no :class:`Device` was found from the given kernel device string. The format of the kernel device string is defined in the systemd.journal-fields man pages. """ class DeviceNotFoundByNameError(DeviceNotFoundError): """ A :exc:`DeviceNotFoundError` indicating that no :class:`Device` was found with a given name. """ def __init__(self, subsystem, sys_name): DeviceNotFoundError.__init__(self, subsystem, sys_name) @property def subsystem(self): """ The subsystem that caused this error as string. """ return self.args[0] @property def sys_name(self): """ The sys name that caused this error as string. """ return self.args[1] def __str__(self): return 'No device {0.sys_name!r} in {0.subsystem!r}'.format(self) class DeviceNotFoundByNumberError(DeviceNotFoundError): """ A :exc:`DeviceNotFoundError` indicating, that no :class:`Device` was found for a given device number. """ def __init__(self, typ, number): DeviceNotFoundError.__init__(self, typ, number) @property def device_type(self): """ The device type causing this error as string. Either ``'char'`` or ``'block'``. """ return self.args[0] @property def device_number(self): """ The device number causing this error as integer. """ return self.args[1] def __str__(self): return ('No {0.device_type} device with number ' '{0.device_number}'.format(self)) class DeviceNotFoundInEnvironmentError(DeviceNotFoundError): """ A :exc:`DeviceNotFoundError` indicating, that no :class:`Device` could be constructed from the process environment. """ def __str__(self): return 'No device found in environment' class DeviceValueError(DeviceError): """ Raised when a parameter has an unacceptable value. May also be raised when the parameter has an unacceptable type. """ _FMT_STR = "value '%s' for parameter %s is unacceptable" def __init__(self, value, param, msg=None): """ Initializer. :param object value: the value :param str param: the parameter :param str msg: an explanatory message """ # pylint: disable=super-init-not-called self._value = value self._param = param self._msg = msg def __str__(self): if self._msg: fmt_str = self._FMT_STR + ": %s" return fmt_str % (self._value, self._param, self._msg) return self._FMT_STR % (self._value, self._param)	5.09 KB	2020-01-24 21:52
📁 _os	-	2025-02-27 15:06
📄 _qt_base.py # -- coding: utf-8 -- # Copyright (C) 2010, 2011, 2012, 2013 Sebastian Wiesner <lunaryorn@gmail.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 """ pyudev._qt_base =============== Base mixin class for Qt4,Qt5 support. .. moduleauthor:: Sebastian Wiesner <lunaryorn@gmail.com> """ from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals import six from pyudev.device import Device class MonitorObserverMixin(object): """ Base mixin for pyqt monitor observers. """ # pylint: disable=too-few-public-methods def _setup_notifier(self, monitor, notifier_class): self.monitor = monitor self.notifier = notifier_class(monitor.fileno(), notifier_class.Read, self) self.notifier.activated[int].connect(self._process_udev_event) @property def enabled(self): """ Whether this observer is enabled or not. If ``True`` (the default), this observer is enabled, and emits events. Otherwise it is disabled and does not emit any events. This merely reflects the state of the ``enabled`` property of the underlying :attr:`notifier`. .. versionadded:: 0.14 """ return self.notifier.isEnabled() @enabled.setter def enabled(self, value): self.notifier.setEnabled(value) def _process_udev_event(self): """ Attempt to receive a single device event from the monitor, process the event and emit corresponding signals. Called by ``QSocketNotifier``, if data is available on the udev monitoring socket. """ device = self.monitor.poll(timeout=0) if device is not None: self._emit_event(device) def _emit_event(self, device): self.deviceEvent.emit(device) class QUDevMonitorObserverMixin(MonitorObserverMixin): """ Obsolete monitor observer mixin. """ # pylint: disable=too-few-public-methods def _setup_notifier(self, monitor, notifier_class): MonitorObserverMixin._setup_notifier(self, monitor, notifier_class) self._action_signal_map = { 'add': self.deviceAdded, 'remove': self.deviceRemoved, 'change': self.deviceChanged, 'move': self.deviceMoved, } import warnings warnings.warn('Will be removed in 1.0. ' 'Use pyudev.pyqt4.MonitorObserver instead.', DeprecationWarning) def _emit_event(self, device): self.deviceEvent.emit(device.action, device) signal = self._action_signal_map.get(device.action) if signal is not None: signal.emit(device) def make_init(qobject, socket_notifier): """ Generates an initializer to observer the given ``monitor`` (a :class:`~pyudev.Monitor`): ``parent`` is the parent :class:`~PyQt{4,5}.QtCore.QObject` of this object. It is passed unchanged to the inherited constructor of :class:`~PyQt{4,5}.QtCore.QObject`. """ def __init__(self, monitor, parent=None): qobject.__init__(self, parent) # pylint: disable=protected-access self._setup_notifier(monitor, socket_notifier) return __init__ class MonitorObserverGenerator(object): """ Class to generate a MonitorObserver class. """ # pylint: disable=too-few-public-methods @staticmethod def make_monitor_observer(qobject, signal, socket_notifier): """Generates an observer for device events integrating into the PyQt{4,5} mainloop. This class inherits :class:`~PyQt{4,5}.QtCore.QObject` to turn device events into Qt signals: >>> from pyudev import Context, Monitor >>> from pyudev.pyqt4 import MonitorObserver >>> context = Context() >>> monitor = Monitor.from_netlink(context) >>> monitor.filter_by(subsystem='input') >>> observer = MonitorObserver(monitor) >>> def device_event(device): ... print('event {0} on device {1}'.format(device.action, device)) >>> observer.deviceEvent.connect(device_event) >>> monitor.start() This class is a child of :class:`~{PySide, PyQt{4,5}}.QtCore.QObject`. """ return type( str("MonitorObserver"), (qobject, MonitorObserverMixin), { str("__init__"): make_init(qobject, socket_notifier), str("deviceEvent"): signal(Device) }) class QUDevMonitorObserverGenerator(object): """ Class to generate a MonitorObserver class. """ # pylint: disable=too-few-public-methods @staticmethod def make_monitor_observer(qobject, signal, socket_notifier): """Generates an observer for device events integrating into the PyQt{4,5} mainloop. This class inherits :class:`~PyQt{4,5}.QtCore.QObject` to turn device events into Qt signals: >>> from pyudev import Context, Monitor >>> from pyudev.pyqt4 import MonitorObserver >>> context = Context() >>> monitor = Monitor.from_netlink(context) >>> monitor.filter_by(subsystem='input') >>> observer = MonitorObserver(monitor) >>> def device_event(device): ... print('event {0} on device {1}'.format(device.action, device)) >>> observer.deviceEvent.connect(device_event) >>> monitor.start() This class is a child of :class:`~{PyQt{4,5}, PySide}.QtCore.QObject`. """ return type( str("QUDevMonitorObserver"), (qobject, QUDevMonitorObserverMixin), { str("__init__"): make_init(qobject, socket_notifier), #: emitted upon arbitrary device events str("deviceEvent"): signal(six.text_type, Device), #: emitted if a device was added str("deviceAdded"): signal(Device), #: emitted if a device was removed str("deviceRemoved"): signal(Device), #: emitted if a device was changed str("deviceChanged"): signal(Device), #: emitted if a device was moved str("deviceMoved"): signal(Device) })	6.98 KB	2020-01-24 21:52
📄 _util.py # -- coding: utf-8 -- # Copyright (C) 2010, 2011, 2012 Sebastian Wiesner <lunaryorn@gmail.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 """ pyudev._util ============ Internal utilities .. moduleauthor:: Sebastian Wiesner <lunaryorn@gmail.com> """ from __future__ import (print_function, division, unicode_literals, absolute_import) try: from subprocess import check_output except ImportError: from pyudev._compat import check_output import os import sys import stat import errno import six def ensure_byte_string(value): """ Return the given ``value`` as bytestring. If the given ``value`` is not a byte string, but a real unicode string, it is encoded with the filesystem encoding (as in :func:`sys.getfilesystemencoding()`). """ if not isinstance(value, bytes): value = value.encode(sys.getfilesystemencoding()) return value def ensure_unicode_string(value): """ Return the given ``value`` as unicode string. If the given ``value`` is not a unicode string, but a byte string, it is decoded with the filesystem encoding (as in :func:`sys.getfilesystemencoding()`). """ if not isinstance(value, six.text_type): value = value.decode(sys.getfilesystemencoding()) return value def property_value_to_bytes(value): """ Return a byte string, which represents the given ``value`` in a way suitable as raw value of an udev property. If ``value`` is a boolean object, it is converted to ``'1'`` or ``'0'``, depending on whether ``value`` is ``True`` or ``False``. If ``value`` is a byte string already, it is returned unchanged. Anything else is simply converted to a unicode string, and then passed to :func:`ensure_byte_string`. """ # udev represents boolean values as 1 or 0, therefore an explicit # conversion to int is required for boolean values if isinstance(value, bool): value = int(value) if isinstance(value, bytes): return value return ensure_byte_string(six.text_type(value)) def string_to_bool(value): """ Convert the given unicode string ``value`` to a boolean object. If ``value`` is ``'1'``, ``True`` is returned. If ``value`` is ``'0'``, ``False`` is returned. Any other value raises a :exc:`~exceptions.ValueError`. """ if value not in ('1', '0'): raise ValueError('Not a boolean value: {0!r}'.format(value)) return value == '1' def udev_list_iterate(libudev, entry): """ Iteration helper for udev list entry objects. Yield a tuple ``(name, value)``. ``name`` and ``value`` are bytestrings containing the name and the value of the list entry. The exact contents depend on the list iterated over. """ while entry: name = libudev.udev_list_entry_get_name(entry) value = libudev.udev_list_entry_get_value(entry) yield (name, value) entry = libudev.udev_list_entry_get_next(entry) def get_device_type(filename): """ Get the device type of a device file. ``filename`` is a string containing the path of a device file. Return ``'char'`` if ``filename`` is a character device, or ``'block'`` if ``filename`` is a block device. Raise :exc:`~exceptions.ValueError` if ``filename`` is no device file at all. Raise :exc:`~exceptions.EnvironmentError` if ``filename`` does not exist or if its metadata was inaccessible. .. versionadded:: 0.15 """ mode = os.stat(filename).st_mode if stat.S_ISCHR(mode): return 'char' elif stat.S_ISBLK(mode): return 'block' else: raise ValueError('not a device file: {0!r}'.format(filename)) def eintr_retry_call(func, args, kwargs): """ Handle interruptions to an interruptible system call. Run an interruptible system call in a loop and retry if it raises EINTR. The signal calls that may raise EINTR prior to Python 3.5 are listed in PEP 0475. Any calls to these functions must be wrapped in eintr_retry_call in order to handle EINTR returns in older versions of Python. This function is safe to use under Python 3.5 and newer since the wrapped function will simply return without raising EINTR. This function is based on _eintr_retry_call in python's subprocess.py. """ # select.error inherits from Exception instead of OSError in Python 2 import select while True: try: return func(args, **kwargs) except (OSError, IOError, select.error) as err: # If this is not an IOError or OSError, it's the old select.error # type, which means that the errno is only accessible via subscript if isinstance(err, (OSError, IOError)): error_code = err.errno else: error_code = err.args[0] if error_code == errno.EINTR: continue raise def udev_version(): """ Get the version of the underlying udev library. udev doesn't use a standard major-minor versioning scheme, but instead labels releases with a single consecutive number. Consequently, the version number returned by this function is a single integer, and not a tuple (like for instance the interpreter version in :data:`sys.version_info`). As libudev itself does not provide a function to query the version number, this function calls the ``udevadm`` utility, so be prepared to catch :exc:`~exceptions.EnvironmentError` and :exc:`~subprocess.CalledProcessError` if you call this function. Return the version number as single integer. Raise :exc:`~exceptions.ValueError`, if the version number retrieved from udev could not be converted to an integer. Raise :exc:`~exceptions.EnvironmentError`, if ``udevadm`` was not found, or could not be executed. Raise :exc:`subprocess.CalledProcessError`, if ``udevadm`` returned a non-zero exit code. On Python 2.7 or newer, the ``output`` attribute of this exception is correctly set. .. versionadded:: 0.8 """ output = ensure_unicode_string(check_output(['udevadm', '--version'])) return int(output.strip())	6.77 KB	2020-01-24 21:52
📄 core.py # -- coding: utf-8 -- # Copyright (C) 2010, 2011 Sebastian Wiesner <lunaryorn@gmail.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 """ pyudev.core =========== Core types and functions of :mod:`pyudev`. .. moduleauthor:: Sebastian Wiesner <lunaryorn@gmail.com> """ from __future__ import (print_function, division, unicode_literals, absolute_import) from pyudev.device import Devices from pyudev._errors import DeviceNotFoundAtPathError from pyudev._ctypeslib.libudev import ERROR_CHECKERS from pyudev._ctypeslib.libudev import SIGNATURES from pyudev._ctypeslib.utils import load_ctypes_library from pyudev._util import ensure_byte_string from pyudev._util import ensure_unicode_string from pyudev._util import property_value_to_bytes from pyudev._util import udev_list_iterate class Context(object): """ A device database connection. This class represents a connection to the udev device database, and is really the central object to access udev. You need an instance of this class for almost anything else in pyudev. This class itself gives access to various udev configuration data (e.g. :attr:`sys_path`, :attr:`device_path`), and provides device enumeration (:meth:`list_devices()`). Instances of this class can directly be given as ``udev `` to functions wrapped through :mod:`ctypes`. """ def __init__(self): """ Create a new context. """ self._libudev = load_ctypes_library('udev', SIGNATURES, ERROR_CHECKERS) self._as_parameter_ = self._libudev.udev_new() def __del__(self): self._libudev.udev_unref(self) @property def sys_path(self): """ The ``sysfs`` mount point defaulting to ``/sys'`` as unicode string. """ if hasattr(self._libudev, 'udev_get_sys_path'): return ensure_unicode_string(self._libudev.udev_get_sys_path(self)) return '/sys' # Fixed path since udev 183 @property def device_path(self): """ The device directory path defaulting to ``/dev`` as unicode string. """ if hasattr(self._libudev, 'udev_get_dev_path'): return ensure_unicode_string(self._libudev.udev_get_dev_path(self)) return '/dev' # Fixed path since udev 183 @property def run_path(self): """ The run runtime directory path defaulting to ``/run`` as unicode string. .. udevversion:: 167 .. versionadded:: 0.10 """ if hasattr(self._libudev, 'udev_get_run_path'): return ensure_unicode_string(self._libudev.udev_get_run_path(self)) return '/run/udev' @property def log_priority(self): """ The logging priority of the interal logging facitility of udev as integer with a standard :mod:`syslog` priority. Assign to this property to change the logging priority. UDev uses the standard :mod:`syslog` priorities. Constants for these priorities are defined in the :mod:`syslog` module in the standard library: >>> import syslog >>> context = pyudev.Context() >>> context.log_priority = syslog.LOG_DEBUG .. versionadded:: 0.9 """ return self._libudev.udev_get_log_priority(self) @log_priority.setter def log_priority(self, value): """ Set the log priority. :param int value: the log priority. """ self._libudev.udev_set_log_priority(self, value) def list_devices(self, kwargs): """ List all available devices. The arguments of this method are the same as for :meth:`Enumerator.match()`. In fact, the arguments are simply passed straight to method :meth:`~Enumerator.match()`. This function creates and returns an :class:`Enumerator` object, that can be used to filter the list of devices, and eventually retrieve :class:`Device` objects representing matching devices. .. versionchanged:: 0.8 Accept keyword arguments now for easy matching. """ return Enumerator(self).match(kwargs) class Enumerator(object): """ A filtered iterable of devices. To retrieve devices, simply iterate over an instance of this class. This operation yields :class:`Device` objects representing the available devices. Before iteration the device list can be filtered by subsystem or by property values using :meth:`match_subsystem` and :meth:`match_property`. Multiple subsystem (property) filters are combined using a logical OR, filters of different types are combined using a logical AND. The following filter for instance:: devices.match_subsystem('block').match_property( 'ID_TYPE', 'disk').match_property('DEVTYPE', 'disk') means the following:: subsystem == 'block' and (ID_TYPE == 'disk' or DEVTYPE == 'disk') Once added, a filter cannot be removed anymore. Create a new object instead. Instances of this class can directly be given as given ``udev_enumerate `` to functions wrapped through :mod:`ctypes`. """ def __init__(self, context): """ Create a new enumerator with the given ``context`` (a :class:`Context` instance). While you can create objects of this class directly, this is not recommended. Call :method:`Context.list_devices()` instead. """ if not isinstance(context, Context): raise TypeError('Invalid context object') self.context = context self._as_parameter_ = context._libudev.udev_enumerate_new(context) self._libudev = context._libudev def __del__(self): self._libudev.udev_enumerate_unref(self) def match(self, *kwargs): """ Include devices according to the rules defined by the keyword arguments. These keyword arguments are interpreted as follows: - The value for the keyword argument ``subsystem`` is forwarded to :meth:`match_subsystem()`. - The value for the keyword argument ``sys_name`` is forwared to :meth:`match_sys_name()`. - The value for the keyword argument ``tag`` is forwared to :meth:`match_tag()`. - The value for the keyword argument ``parent`` is forwared to :meth:`match_parent()`. - All other keyword arguments are forwareded one by one to :meth:`match_property()`. The keyword argument itself is interpreted as property name, the value of the keyword argument as the property value. All keyword arguments are optional, calling this method without no arguments at all is simply a noop. Return the instance again. .. versionadded:: 0.8 .. versionchanged:: 0.13 Add ``parent`` keyword. """ subsystem = kwargs.pop('subsystem', None) if subsystem is not None: self.match_subsystem(subsystem) sys_name = kwargs.pop('sys_name', None) if sys_name is not None: self.match_sys_name(sys_name) tag = kwargs.pop('tag', None) if tag is not None: self.match_tag(tag) parent = kwargs.pop('parent', None) if parent is not None: self.match_parent(parent) for prop, value in kwargs.items(): self.match_property(prop, value) return self def match_subsystem(self, subsystem, nomatch=False): """ Include all devices, which are part of the given ``subsystem``. ``subsystem`` is either a unicode string or a byte string, containing the name of the subsystem. If ``nomatch`` is ``True`` (default is ``False``), the match is inverted: A device is only included if it is not* part of the given ``subsystem``. Note that, if a device has no subsystem, it is not included either with value of ``nomatch`` True or with value of ``nomatch`` False. Return the instance again. """ match = self._libudev.udev_enumerate_add_nomatch_subsystem \ if nomatch else self._libudev.udev_enumerate_add_match_subsystem match(self, ensure_byte_string(subsystem)) return self def match_sys_name(self, sys_name): """ Include all devices with the given name. ``sys_name`` is a byte or unicode string containing the device name. Return the instance again. .. versionadded:: 0.8 """ self._libudev.udev_enumerate_add_match_sysname( self, ensure_byte_string(sys_name)) return self def match_property(self, prop, value): """ Include all devices, whose ``prop`` has the given ``value``. ``prop`` is either a unicode string or a byte string, containing the name of the property to match. ``value`` is a property value, being one of the following types: - :func:`int` - :func:`bool` - A byte string - Anything convertable to a unicode string (including a unicode string itself) Return the instance again. """ self._libudev.udev_enumerate_add_match_property( self, ensure_byte_string(prop), property_value_to_bytes(value)) return self def match_attribute(self, attribute, value, nomatch=False): """ Include all devices, whose ``attribute`` has the given ``value``. ``attribute`` is either a unicode string or a byte string, containing the name of a sys attribute to match. ``value`` is an attribute value, being one of the following types: - :func:`int`, - :func:`bool` - A byte string - Anything convertable to a unicode string (including a unicode string itself) If ``nomatch`` is ``True`` (default is ``False``), the match is inverted: A device is include if the ``attribute`` does not match the given ``value``. .. note:: If ``nomatch`` is ``True``, devices which do not have the given ``attribute`` at all are also included. In other words, with ``nomatch=True`` the given ``attribute`` is not guaranteed to exist on all returned devices. Return the instance again. """ match = (self._libudev.udev_enumerate_add_match_sysattr if not nomatch else self._libudev.udev_enumerate_add_nomatch_sysattr) match(self, ensure_byte_string(attribute), property_value_to_bytes(value)) return self def match_tag(self, tag): """ Include all devices, which have the given ``tag`` attached. ``tag`` is a byte or unicode string containing the tag name. Return the instance again. .. udevversion:: 154 .. versionadded:: 0.6 """ self._libudev.udev_enumerate_add_match_tag(self, ensure_byte_string(tag)) return self def match_is_initialized(self): """ Include only devices, which are initialized. Initialized devices have properly set device node permissions and context, and are (in case of network devices) fully renamed. Currently this will not affect devices which do not have device nodes and are not network interfaces. Return the instance again. .. seealso:: :attr:`Device.is_initialized` .. udevversion:: 165 .. versionadded:: 0.8 """ self._libudev.udev_enumerate_add_match_is_initialized(self) return self def match_parent(self, parent): """ Include all devices on the subtree of the given ``parent`` device. The ``parent`` device itself is also included. ``parent`` is a :class:`~pyudev.Device`. Return the instance again. .. udevversion:: 172 .. versionadded:: 0.13 """ self._libudev.udev_enumerate_add_match_parent(self, parent) return self def __iter__(self): """ Iterate over all matching devices. Yield :class:`Device` objects. """ self._libudev.udev_enumerate_scan_devices(self) entry = self._libudev.udev_enumerate_get_list_entry(self) for name, _ in udev_list_iterate(self._libudev, entry): try: yield Devices.from_sys_path(self.context, name) except DeviceNotFoundAtPathError: continue	13.06 KB	2020-01-24 21:52
📁 device	-	2025-02-27 15:06
📄 discover.py # -- coding: utf-8 -- # Copyright (C) 2015 mulhern <amulhern@redhat.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 """ pyudev.discover =============== Tools to discover a device given limited information. .. moduleauthor:: mulhern <amulhern@redhat.com> """ from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals import abc import functools import os import re import six from pyudev._errors import DeviceNotFoundError from pyudev.device import Devices def wrap_exception(func): """ Allow Device discovery methods to return None instead of raising an exception. """ @functools.wraps(func) def the_func(args, kwargs): """ Returns result of calling ``func`` on ``args``, ``kwargs``. Returns None if ``func`` raises :exc:`DeviceNotFoundError`. """ try: return func(args, **kwargs) except DeviceNotFoundError: return None return the_func @six.add_metaclass(abc.ABCMeta) class Hypothesis(object): """ Represents a hypothesis about the meaning of the device identifier. """ @classmethod @abc.abstractmethod def match(cls, value): # pragma: no cover """ Match the given string according to the hypothesis. The purpose of this method is to obtain a value corresponding to ``value`` if that is possible. It may use a regular expression, but in general it should just return ``value`` and let the lookup method sort out the rest. :param str value: the string to inspect :returns: the matched thing or None if unmatched :rtype: the type of lookup's key parameter or NoneType """ raise NotImplementedError() @classmethod @abc.abstractmethod def lookup(cls, context, key): # pragma: no cover """ Lookup the given string according to the hypothesis. :param Context context: the pyudev context :param key: a key with which to lookup the device :type key: the type of match's return value if not None :returns: a list of Devices obtained :rtype: frozenset of :class:`Device` """ raise NotImplementedError() @classmethod def setup(cls, context): """ A potentially expensive method that may allow an :class:`Hypothesis` to find devices more rapidly or to find a device that it would otherwise miss. :param Context context: the pyudev context """ pass @classmethod def get_devices(cls, context, value): """ Get any devices that may correspond to the given string. :param Context context: the pyudev context :param str value: the value to look for :returns: a list of devices obtained :rtype: set of :class:`Device` """ key = cls.match(value) return cls.lookup(context, key) if key is not None else frozenset() class DeviceNumberHypothesis(Hypothesis): """ Represents the hypothesis that the device is a device number. The device may be separated into major/minor number or a composite number. """ @classmethod def _match_major_minor(cls, value): """ Match the number under the assumption that it is a major,minor pair. :param str value: value to match :returns: the device number or None :rtype: int or NoneType """ major_minor_re = re.compile(r'^(?P<major>\d+)(\D+)(?P<minor>\d+)$') match = major_minor_re.match(value) return match and os.makedev( int(match.group('major')), int(match.group('minor'))) @classmethod def _match_number(cls, value): """ Match the number under the assumption that it is a single number. :param str value: value to match :returns: the device number or None :rtype: int or NoneType """ number_re = re.compile(r'^(?P<number>\d+)$') match = number_re.match(value) return match and int(match.group('number')) @classmethod def match(cls, value): """ Match the number under the assumption that it is a device number. :returns: the device number or None :rtype: int or NoneType """ return cls._match_major_minor(value) or cls._match_number(value) @classmethod def find_subsystems(cls, context): """ Find subsystems in /sys/dev. :param Context context: the context :returns: a lis of available subsystems :rtype: list of str """ sys_path = context.sys_path return os.listdir(os.path.join(sys_path, 'dev')) @classmethod def lookup(cls, context, key): """ Lookup by the device number. :param Context context: the context :param int key: the device number :returns: a list of matching devices :rtype: frozenset of :class:`Device` """ func = wrap_exception(Devices.from_device_number) res = (func(context, s, key) for s in cls.find_subsystems(context)) return frozenset(r for r in res if r is not None) class DevicePathHypothesis(Hypothesis): """ Discover the device assuming the identifier is a device path. """ @classmethod def match(cls, value): """ Match ``value`` under the assumption that it is a device path. :returns: the device path or None :rtype: str or NoneType """ return value @classmethod def lookup(cls, context, key): """ Lookup by the path. :param Context context: the context :param str key: the device path :returns: a list of matching devices :rtype: frozenset of :class:`Device` """ res = wrap_exception(Devices.from_path)(context, key) return frozenset((res, )) if res is not None else frozenset() class DeviceNameHypothesis(Hypothesis): """ Discover the device assuming the input is a device name. Try every available subsystem. """ @classmethod def find_subsystems(cls, context): """ Find all subsystems in sysfs. :param Context context: the context :rtype: frozenset :returns: subsystems in sysfs """ sys_path = context.sys_path dirnames = ('bus', 'class', 'subsystem') absnames = (os.path.join(sys_path, name) for name in dirnames) realnames = (d for d in absnames if os.path.isdir(d)) return frozenset(n for d in realnames for n in os.listdir(d)) @classmethod def match(cls, value): """ Match ``value`` under the assumption that it is a device name. :returns: the device path or None :rtype: str or NoneType """ return value @classmethod def lookup(cls, context, key): """ Lookup by the path. :param Context context: the context :param str key: the device path :returns: a list of matching devices :rtype: frozenset of :class:`Device` """ func = wrap_exception(Devices.from_name) res = (func(context, s, key) for s in cls.find_subsystems(context)) return frozenset(r for r in res if r is not None) class DeviceFileHypothesis(Hypothesis): """ Discover the device assuming the value is some portion of a device file. The device file may be a link to a device node. """ _LINK_DIRS = [ '/dev', '/dev/disk/by-id', '/dev/disk/by-label', '/dev/disk/by-partlabel', '/dev/disk/by-partuuid', '/dev/disk/by-path', '/dev/disk/by-uuid', '/dev/input/by-path', '/dev/mapper', '/dev/md', '/dev/vg' ] @classmethod def get_link_dirs(cls, context): """ Get all directories that may contain links to device nodes. This method checks the device links of every device, so it is very expensive. :param Context context: the context :returns: a sorted list of directories that contain device links :rtype: list """ devices = context.list_devices() devices_with_links = (d for d in devices if list(d.device_links)) links = (l for d in devices_with_links for l in d.device_links) return sorted(set(os.path.dirname(l) for l in links)) @classmethod def setup(cls, context): """ Set the link directories to be used when discovering by file. Uses `get_link_dirs`, so is as expensive as it is. :param Context context: the context """ cls._LINK_DIRS = cls.get_link_dirs(context) @classmethod def match(cls, value): return value @classmethod def lookup(cls, context, key): """ Lookup the device under the assumption that the key is part of the name of a device file. :param Context context: the context :param str key: a portion of the device file name It is assumed that either it is the whole name of the device file or it is the basename. A device file may be a device node or a device link. """ func = wrap_exception(Devices.from_device_file) if '/' in key: device = func(context, key) return frozenset((device, )) if device is not None else frozenset() files = (os.path.join(ld, key) for ld in cls._LINK_DIRS) devices = (func(context, f) for f in files) return frozenset(d for d in devices if d is not None) class Discovery(object): # pylint: disable=too-few-public-methods """ Provides discovery methods for devices. """ _HYPOTHESES = [ DeviceFileHypothesis, DeviceNameHypothesis, DeviceNumberHypothesis, DevicePathHypothesis ] def __init__(self): self._hypotheses = self._HYPOTHESES def setup(self, context): """ Set up individual hypotheses. May be an expensive call. :param Context context: the context """ for hyp in self._hypotheses: hyp.setup(context) def get_devices(self, context, value): """ Get the devices corresponding to value. :param Context context: the context :param str value: some identifier of the device :returns: a list of corresponding devices :rtype: frozenset of :class:`Device` """ return frozenset( d for h in self._hypotheses for d in h.get_devices(context, value))	11.12 KB	2020-01-24 21:52
📄 monitor.py # -- coding: utf-8 -- # Copyright (C) 2010, 2011, 2012, 2013 Sebastian Wiesner <lunaryorn@gmail.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 """ pyudev.monitor ============== Monitor implementation. .. moduleauthor:: Sebastian Wiesner <lunaryorn@gmail.com> """ from __future__ import (print_function, division, unicode_literals, absolute_import) import os import errno from threading import Thread from functools import partial from pyudev.device import Device from pyudev._util import eintr_retry_call from pyudev._util import ensure_byte_string from pyudev._os import pipe from pyudev._os import poll class Monitor(object): """ A synchronous device event monitor. A :class:`Monitor` objects connects to the udev daemon and listens for changes to the device list. A monitor is created by connecting to the kernel daemon through netlink (see :meth:`from_netlink`): >>> from pyudev import Context, Monitor >>> context = Context() >>> monitor = Monitor.from_netlink(context) Once the monitor is created, you can add a filter using :meth:`filter_by()` or :meth:`filter_by_tag()` to drop incoming events in subsystems, which are not of interest to the application: >>> monitor.filter_by('input') When the monitor is eventually set up, you can either poll for events synchronously: >>> device = monitor.poll(timeout=3) >>> if device: ... print('{0.action}: {0}'.format(device)) ... Or you can monitor events asynchronously with :class:`MonitorObserver`. To integrate into various event processing frameworks, the monitor provides a :func:`selectable <select.select>` file description by :meth:`fileno()`. However, do not read or write directly on this file descriptor. Instances of this class can directly be given as ``udev_monitor `` to functions wrapped through :mod:`ctypes`. .. versionchanged:: 0.16 Remove :meth:`from_socket()` which is deprecated, and even removed in recent udev versions. """ def __init__(self, context, monitor_p): self.context = context self._as_parameter_ = monitor_p self._libudev = context._libudev self._started = False def __del__(self): self._libudev.udev_monitor_unref(self) @classmethod def from_netlink(cls, context, source='udev'): """ Create a monitor by connecting to the kernel daemon through netlink. ``context`` is the :class:`Context` to use. ``source`` is a string, describing the event source. Two sources are available: ``'udev'`` (the default) Events emitted after udev as registered and configured the device. This is the absolutely recommended source for applications. ``'kernel'`` Events emitted directly after the kernel has seen the device. The device has not yet been configured by udev and might not be usable at all. Never* use this, unless you know what you are doing. Return a new :class:`Monitor` object, which is connected to the given source. Raise :exc:`~exceptions.ValueError`, if an invalid source has been specified. Raise :exc:`~exceptions.EnvironmentError`, if the creation of the monitor failed. """ if source not in ('kernel', 'udev'): raise ValueError('Invalid source: {0!r}. Must be one of "udev" ' 'or "kernel"'.format(source)) monitor = context._libudev.udev_monitor_new_from_netlink( context, ensure_byte_string(source)) if not monitor: raise EnvironmentError('Could not create udev monitor') return cls(context, monitor) @property def started(self): """ ``True``, if this monitor was started, ``False`` otherwise. Readonly. .. seealso:: :meth:`start()` .. versionadded:: 0.16 """ return self._started def fileno(self): # pylint: disable=anomalous-backslash-in-string """ Return the file description associated with this monitor as integer. This is really a real file descriptor ;), which can be watched and :func:`select.select`\ ed. """ return self._libudev.udev_monitor_get_fd(self) def filter_by(self, subsystem, device_type=None): """ Filter incoming events. ``subsystem`` is a byte or unicode string with the name of a subsystem (e.g. ``'input'``). Only events originating from the given subsystem pass the filter and are handed to the caller. If given, ``device_type`` is a byte or unicode string specifying the device type. Only devices with the given device type are propagated to the caller. If ``device_type`` is not given, no additional filter for a specific device type is installed. These filters are executed inside the kernel, and client processes will usually not be woken up for device, that do not match these filters. .. versionchanged:: 0.15 This method can also be after :meth:`start()` now. """ subsystem = ensure_byte_string(subsystem) if device_type is not None: device_type = ensure_byte_string(device_type) self._libudev.udev_monitor_filter_add_match_subsystem_devtype( self, subsystem, device_type) self._libudev.udev_monitor_filter_update(self) def filter_by_tag(self, tag): """ Filter incoming events by the given ``tag``. ``tag`` is a byte or unicode string with the name of a tag. Only events for devices which have this tag attached pass the filter and are handed to the caller. Like with :meth:`filter_by` this filter is also executed inside the kernel, so that client processes are usually not woken up for devices without the given ``tag``. .. udevversion:: 154 .. versionadded:: 0.9 .. versionchanged:: 0.15 This method can also be after :meth:`start()` now. """ self._libudev.udev_monitor_filter_add_match_tag( self, ensure_byte_string(tag)) self._libudev.udev_monitor_filter_update(self) def remove_filter(self): """ Remove any filters installed with :meth:`filter_by()` or :meth:`filter_by_tag()` from this monitor. .. warning:: Up to udev 181 (and possibly even later versions) the underlying ``udev_monitor_filter_remove()`` seems to be broken. If used with affected versions this method always raises :exc:`~exceptions.ValueError`. Raise :exc:`~exceptions.EnvironmentError` if removal of installed filters failed. .. versionadded:: 0.15 """ self._libudev.udev_monitor_filter_remove(self) self._libudev.udev_monitor_filter_update(self) def enable_receiving(self): """ Switch the monitor into listing mode. Connect to the event source and receive incoming events. Only after calling this method, the monitor listens for incoming events. .. note:: This method is implicitly called by :meth:`__iter__`. You don't need to call it explicitly, if you are iterating over the monitor. .. deprecated:: 0.16 Will be removed in 1.0. Use :meth:`start()` instead. """ import warnings warnings.warn('Will be removed in 1.0. Use Monitor.start() instead.', DeprecationWarning) self.start() def start(self): """ Start this monitor. The monitor will not receive events until this method is called. This method does nothing if called on an already started :class:`Monitor`. .. note:: Typically you don't need to call this method. It is implicitly called by :meth:`poll()` and :meth:`__iter__()`. .. seealso:: :attr:`started` .. versionchanged:: 0.16 This method does nothing if the :class:`Monitor` was already started. """ if not self._started: self._libudev.udev_monitor_enable_receiving(self) # Force monitor FD into non-blocking mode pipe.set_fd_status_flag(self, os.O_NONBLOCK) self._started = True def set_receive_buffer_size(self, size): """ Set the receive buffer ``size``. ``size`` is the requested buffer size in bytes, as integer. .. note:: The CAP_NET_ADMIN capability must be contained in the effective capability set of the caller for this method to succeed. Otherwise :exc:`~exceptions.EnvironmentError` will be raised, with ``errno`` set to :data:`~errno.EPERM`. Unprivileged processes typically lack this capability. You can check the capabilities of the current process with the python-prctl_ module: >>> import prctl >>> prctl.cap_effective.net_admin Raise :exc:`~exceptions.EnvironmentError`, if the buffer size could not bet set. .. versionadded:: 0.13 .. _python-prctl: http://packages.python.org/python-prctl """ self._libudev.udev_monitor_set_receive_buffer_size(self, size) def _receive_device(self): """Receive a single device from the monitor. Return the received :class:`Device`, or ``None`` if no device could be received. """ while True: try: device_p = self._libudev.udev_monitor_receive_device(self) return Device(self.context, device_p) if device_p else None except EnvironmentError as error: if error.errno in (errno.EAGAIN, errno.EWOULDBLOCK): # No data available return None elif error.errno == errno.EINTR: # Try again if our system call was interrupted continue else: raise def poll(self, timeout=None): """ Poll for a device event. You can use this method together with :func:`iter()` to synchronously monitor events in the current thread:: for device in iter(monitor.poll, None): print('{0.action} on {0.device_path}'.format(device)) Since this method will never return ``None`` if no ``timeout`` is specified, this is effectively an endless loop. With :func:`functools.partial()` you can also create a loop that only waits for a specified time:: for device in iter(partial(monitor.poll, 3), None): print('{0.action} on {0.device_path}'.format(device)) This loop will only wait three seconds for a new device event. If no device event occurred after three seconds, the loop will exit. ``timeout`` is a floating point number that specifies a time-out in seconds. If omitted or ``None``, this method blocks until a device event is available. If ``0``, this method just polls and will never block. .. note:: This method implicitly calls :meth:`start()`. Return the received :class:`Device`, or ``None`` if a timeout occurred. Raise :exc:`~exceptions.EnvironmentError` if event retrieval failed. .. seealso:: :attr:`Device.action` The action that created this event. :attr:`Device.sequence_number` The sequence number of this event. .. versionadded:: 0.16 """ if timeout is not None and timeout > 0: # .poll() takes timeout in milliseconds timeout = int(timeout * 1000) self.start() if eintr_retry_call(poll.Poll.for_events((self, 'r')).poll, timeout): return self._receive_device() return None def receive_device(self): """ Receive a single device from the monitor. .. warning:: You must call :meth:`start()` before calling this method. The caller must make sure, that there are events available in the event queue. The call blocks, until a device is available. If a device was available, return ``(action, device)``. ``device`` is the :class:`Device` object describing the device. ``action`` is a string describing the action. Usual actions are: ``'add'`` A device has been added (e.g. a USB device was plugged in) ``'remove'`` A device has been removed (e.g. a USB device was unplugged) ``'change'`` Something about the device changed (e.g. a device property) ``'online'`` The device is online now ``'offline'`` The device is offline now Raise :exc:`~exceptions.EnvironmentError`, if no device could be read. .. deprecated:: 0.16 Will be removed in 1.0. Use :meth:`Monitor.poll()` instead. """ import warnings warnings.warn('Will be removed in 1.0. Use Monitor.poll() instead.', DeprecationWarning) device = self.poll() return device.action, device def __iter__(self): """ Wait for incoming events and receive them upon arrival. This methods implicitly calls :meth:`start()`, and starts polling the :meth:`fileno` of this monitor. If a event comes in, it receives the corresponding device and yields it to the caller. The returned iterator is endless, and continues receiving devices without ever stopping. Yields ``(action, device)`` (see :meth:`receive_device` for a description). .. deprecated:: 0.16 Will be removed in 1.0. Use an explicit loop over :meth:`poll()` instead, or monitor asynchronously with :class:`MonitorObserver`. """ import warnings warnings.warn('Will be removed in 1.0. Use an explicit loop over ' '"poll()" instead, or monitor asynchronously with ' '"MonitorObserver".', DeprecationWarning) self.start() while True: device = self.poll() if device is not None: yield device.action, device class MonitorObserver(Thread): """ An asynchronous observer for device events. This class subclasses :class:`~threading.Thread` class to asynchronously observe a :class:`Monitor` in a background thread: >>> from pyudev import Context, Monitor, MonitorObserver >>> context = Context() >>> monitor = Monitor.from_netlink(context) >>> monitor.filter_by(subsystem='input') >>> def print_device_event(device): ... print('background event {0.action}: {0.device_path}'.format(device)) >>> observer = MonitorObserver(monitor, callback=print_device_event, name='monitor-observer') >>> observer.daemon True >>> observer.start() In the above example, input device events will be printed in background, until :meth:`stop()` is called on ``observer``. .. note:: Instances of this class are always created as daemon thread. If you do not want to use daemon threads for monitoring, you need explicitly set :attr:`~threading.Thread.daemon` to ``False`` before invoking :meth:`~threading.Thread.start()`. .. seealso:: :attr:`Device.action` The action that created this event. :attr:`Device.sequence_number` The sequence number of this event. .. versionadded:: 0.14 .. versionchanged:: 0.15 :meth:`Monitor.start()` is implicitly called when the thread is started. """ def __init__(self, monitor, event_handler=None, callback=None, args, kwargs): # pylint: disable=keyword-arg-before-vararg """ Create a new observer for the given ``monitor``. ``monitor`` is the :class:`Monitor` to observe. ``callback`` is the callable to invoke on events, with the signature ``callback(device)`` where ``device`` is the :class:`Device` that caused the event. .. warning:: ``callback`` is invoked in the observer thread, hence the observer is blocked while callback executes. ``args`` and ``kwargs`` are passed unchanged to the constructor of :class:`~threading.Thread`. .. deprecated:: 0.16 The ``event_handler`` argument will be removed in 1.0. Use the ``callback`` argument instead. .. versionchanged:: 0.16 Add ``callback`` argument. """ if callback is None and event_handler is None: raise ValueError('callback missing') elif callback is not None and event_handler is not None: raise ValueError('Use either callback or event handler') Thread.__init__(self, args, *kwargs) self.monitor = monitor # observer threads should not keep the interpreter alive self.daemon = True self._stop_event = None if event_handler is not None: import warnings warnings.warn('"event_handler" argument will be removed in 1.0. ' 'Use Monitor.poll() instead.', DeprecationWarning) callback = lambda d: event_handler(d.action, d) self._callback = callback def start(self): """Start the observer thread.""" if not self.is_alive(): self._stop_event = pipe.Pipe.open() Thread.start(self) def run(self): self.monitor.start() notifier = poll.Poll.for_events((self.monitor, 'r'), (self._stop_event.source, 'r')) while True: for file_descriptor, event in eintr_retry_call(notifier.poll): if file_descriptor == self._stop_event.source.fileno(): # in case of a stop event, close our pipe side, and # return from the thread self._stop_event.source.close() return elif file_descriptor == self.monitor.fileno() and event == 'r': read_device = partial( eintr_retry_call, self.monitor.poll, timeout=0) for device in iter(read_device, None): self._callback(device) else: raise EnvironmentError('Observed monitor hung up') def send_stop(self): """ Send a stop signal to the background thread. The background thread will eventually exit, but it may still be running when this method returns. This method is essentially the asynchronous equivalent to :meth:`stop()`. .. note:: The underlying :attr:`monitor` is not* stopped. """ if self._stop_event is None: return with self._stop_event.sink: # emit a stop event to the thread eintr_retry_call(self._stop_event.sink.write, b'\x01') self._stop_event.sink.flush() def stop(self): """ Synchronously stop the background thread. .. note:: This method can safely be called from the observer thread. In this case it is equivalent to :meth:`send_stop()`. Send a stop signal to the backgroud (see :meth:`send_stop`), and waits for the background thread to exit (see :meth:`~threading.Thread.join`) if the current thread is not the observer thread. After this method returns in a thread that is not the observer thread, the ``callback`` is guaranteed to not be invoked again anymore. .. note:: The underlying :attr:`monitor` is not stopped. .. versionchanged:: 0.16 This method can be called from the observer thread. """ self.send_stop() try: self.join() except RuntimeError: pass	20.6 KB	2020-01-24 21:52
📄 version.py # -- coding: utf-8 -- # Copyright (C) 2015 mulhern <amulhern@redhat.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 """ pyudev.version ============== Version information. .. moduleauthor:: mulhern <amulhern@redhat.com> """ __version_info__ = (0, 22, 0, '') __version__ = "%s%s" % (".".join(str(x) for x in __version_info__[:3]), "".join(str(x) for x in __version_info__[3:]))	1.08 KB	2020-01-24 21:52

Terminal