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:45
📄 __init__.py from .inventory import * from .device_matcher import * from .device_matcher_udev import *	90 B	2025-08-24 21:48
📁 __pycache__	-	2026-03-21 15:45
📄 device_matcher.py import fnmatch import re __all__ = ["DeviceMatcher"] class DeviceMatcher(object): """ Device name matching against the devices specification in tuning profiles. The devices specification consists of multiple rules separated by spaces. The rules have a syntax of shell-style wildcards and are either positive or negative. The negative rules are prefixed with an exclamation mark. """ def match(self, rules, device_name): """ Match a device against the specification in the profile. If there is no positive rule in the specification, implicit rule which matches all devices is added. The device matches if and only if it matches some positive rule, but no negative rule. """ if isinstance(rules, str): rules = re.split(r"\s\|,\s", rules) positive_rules = [rule for rule in rules if not rule.startswith("!") and not rule.strip() == ''] negative_rules = [rule[1:] for rule in rules if rule not in positive_rules] if len(positive_rules) == 0: positive_rules.append("") matches = False for rule in positive_rules: if fnmatch.fnmatch(device_name, rule): matches = True break for rule in negative_rules: if fnmatch.fnmatch(device_name, rule): matches = False break return matches def match_list(self, rules, device_list): """ Match a device list against the specification in the profile. Returns the list, which is a subset of devices which match. """ matching_devices = [] for device in device_list: if self.match(rules, device): matching_devices.append(device) return matching_devices	1.54 KB	2025-08-24 21:48
📄 device_matcher_udev.py from . import device_matcher import re __all__ = ["DeviceMatcherUdev"] class DeviceMatcherUdev(device_matcher.DeviceMatcher): def match(self, regex, device): """ Match a device against the udev regex in tuning profiles. device is a pyudev.Device object """ properties = '' try: items = device.properties.items() except AttributeError: try: items = device.items() except AttributeError: return False for key, val in sorted(list(items)): properties += key + '=' + val + '\n' return re.search(regex, properties, re.MULTILINE) is not None	581 B	2025-08-24 21:48
📄 inventory.py import pyudev import tuned.logs from tuned import consts __all__ = ["Inventory"] log = tuned.logs.get() class Inventory(object): """ Inventory object can handle information about available hardware devices. It also informs the plugins about related hardware events. """ def __init__(self, udev_context=None, udev_monitor_cls=None, monitor_observer_factory=None, buffer_size=None, set_receive_buffer_size=True): if udev_context is not None: self._udev_context = udev_context else: self._udev_context = pyudev.Context() if udev_monitor_cls is None: udev_monitor_cls = pyudev.Monitor self._udev_monitor = udev_monitor_cls.from_netlink(self._udev_context) if buffer_size is None: buffer_size = consts.CFG_DEF_UDEV_BUFFER_SIZE if (set_receive_buffer_size): try: self._udev_monitor.set_receive_buffer_size(buffer_size) except EnvironmentError: log.warning("cannot set udev monitor receive buffer size, we are probably running inside " + "container or with limited capabilites, TuneD functionality may be limited") if monitor_observer_factory is None: monitor_observer_factory = _MonitorObserverFactory() self._monitor_observer_factory = monitor_observer_factory self._monitor_observer = None self._subscriptions = {} def get_device(self, subsystem, sys_name): """Get a pyudev.Device object for the sys_name (e.g. 'sda').""" try: try: d = pyudev.Devices.from_name(self._udev_context, subsystem, sys_name) except pyudev.DeviceNotFoundByNameError: d = None # workaround for pyudev < 0.18 except AttributeError: try: d = pyudev.Device.from_name(self._udev_context, subsystem, sys_name) except pyudev.DeviceNotFoundByNameError: d = None return d def get_devices(self, subsystem): """Get list of devices on a given subsystem.""" return self._udev_context.list_devices(subsystem=subsystem) def _handle_udev_event(self, event, device): if not device.subsystem in self._subscriptions: return retry = consts.HOTPLUG_WAIT_FOR_DEV_INIT_RETRIES while not device.is_initialized and retry > 0: log.debug("Device '%s' is uninitialized, waiting '%.2f' seconds for its initialization." % (device, consts.HOTPLUG_WAIT_FOR_DEV_INIT_DELAY)) time.sleep(consts.HOTPLUG_WAIT_FOR_DEV_INIT_DELAY) retry =- 1 if not device.is_initialized: log.warn("Unsuccessfully waited for device '%s' initialization, continuing with uninitialized device, problems may occur." % device) for (plugin, callback) in self._subscriptions[device.subsystem]: try: callback(event, device) except Exception as e: log.error("Exception occured in event handler of '%s'." % plugin) log.exception(e) def subscribe(self, plugin, subsystem, callback): """Register handler of device events on a given subsystem.""" log.debug("adding handler: %s (%s)" % (subsystem, plugin)) callback_data = (plugin, callback) if subsystem in self._subscriptions: self._subscriptions[subsystem].append(callback_data) else: self._subscriptions[subsystem] = [callback_data, ] self._udev_monitor.filter_by(subsystem) # After start(), HW events begin to get queued up self._udev_monitor.start() def start_processing_events(self): if self._monitor_observer is None: log.debug("starting monitor observer") self._monitor_observer = self._monitor_observer_factory.create(self._udev_monitor, self._handle_udev_event) self._monitor_observer.start() def stop_processing_events(self): if self._monitor_observer is not None: log.debug("stopping monitor observer") self._monitor_observer.stop() self._monitor_observer = None def _unsubscribe_subsystem(self, plugin, subsystem): for callback_data in self._subscriptions[subsystem]: (_plugin, callback) = callback_data if plugin == _plugin: log.debug("removing handler: %s (%s)" % (subsystem, plugin)) self._subscriptions[subsystem].remove(callback_data) def unsubscribe(self, plugin, subsystem=None): """Unregister handler registered with subscribe method.""" empty_subsystems = [] for _subsystem in self._subscriptions: if subsystem is None or _subsystem == subsystem: self._unsubscribe_subsystem(plugin, _subsystem) if len(self._subscriptions[_subsystem]) == 0: empty_subsystems.append(_subsystem) for _subsystem in empty_subsystems: del self._subscriptions[_subsystem] class _MonitorObserverFactory(object): def create(self, args, kwargs): return pyudev.MonitorObserver(args, **kwargs)	4.41 KB	2025-08-24 21:48

Terminal