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 .base import * from .repository import *	46 B	2025-08-24 21:48
📁 __pycache__	-	2026-03-21 15:45
📄 base.py import tuned.logs log = tuned.logs.get() __all__ = ["Monitor"] class Monitor(object): """ Base class for all monitors. Monitors provide data about the running system to Plugin objects, which use the data to tune system parameters. Following methods require reimplementation: - _init_available_devices(cls) - update(cls) """ # class properties @classmethod def _init_class(cls): cls._class_initialized = False cls._instances = set() cls._available_devices = set() cls._updating_devices = set() cls._load = {} cls._init_available_devices() assert isinstance(cls._available_devices, set) cls._class_initialized = True log.debug("available devices: %s" % ", ".join(cls._available_devices)) @classmethod def _init_available_devices(cls): raise NotImplementedError() @classmethod def _update_available_devices(cls): cls._init_available_devices() log.debug("available devices updated to: %s" % ", ".join(cls._available_devices)) @classmethod def get_available_devices(cls): return cls._available_devices @classmethod def update(cls): raise NotImplementedError() @classmethod def _register_instance(cls, instance): cls._instances.add(instance) @classmethod def _deregister_instance(cls, instance): cls._instances.remove(instance) @classmethod def _refresh_updating_devices(cls): new_updating = set() for instance in cls._instances: new_updating \|= instance.devices cls._updating_devices.clear() cls._updating_devices.update(new_updating) @classmethod def instances(cls): return cls._instances # instance properties def __init__(self, devices = None): if not hasattr(self, "_class_initialized"): self._init_class() assert hasattr(self, "_class_initialized") self._register_instance(self) if devices is not None: self.devices = devices else: self.devices = self.get_available_devices() self.update() def __del__(self): try: self.cleanup() except: pass def cleanup(self): self._deregister_instance(self) self._refresh_updating_devices() @property def devices(self): return self._devices @devices.setter def devices(self, value): new_devices = self._available_devices & set(value) self._devices = new_devices self._refresh_updating_devices() def add_device(self, device): assert (isinstance(device,str) or isinstance(device,unicode)) self._update_available_devices() if device in self._available_devices: self._devices.add(device) self._updating_devices.add(device) def remove_device(self, device): assert (isinstance(device,str) or isinstance(device,unicode)) if device in self._devices: self._devices.remove(device) self._updating_devices.remove(device) def get_load(self): return dict([dev_load for dev_load in list(self._load.items()) if dev_load[0] in self._devices]) def get_device_load(self, device): return self._load.get(device, None)	2.85 KB	2025-08-24 21:48
📄 monitor_disk.py import tuned.monitors import os class DiskMonitor(tuned.monitors.Monitor): _supported_vendors = ["ATA", "SCSI"] @classmethod def _init_available_devices(cls): block_devices = os.listdir("/sys/block") available = set(filter(cls._is_device_supported, block_devices)) cls._available_devices = available for d in available: cls._load[d] = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] @classmethod def _is_device_supported(cls, device): vendor_file = "/sys/block/%s/device/vendor" % device try: vendor = open(vendor_file).read().strip() except IOError: return False return vendor in cls._supported_vendors @classmethod def update(cls): for device in cls._updating_devices: cls._update_disk(device) @classmethod def _update_disk(cls, dev): with open("/sys/block/" + dev + "/stat") as statfile: cls._load[dev] = list(map(int, statfile.read().split()))	886 B	2025-08-24 21:48
📄 monitor_load.py import tuned.monitors class LoadMonitor(tuned.monitors.Monitor): @classmethod def _init_available_devices(cls): cls._available_devices = set(["system"]) @classmethod def update(cls): with open("/proc/loadavg") as statfile: data = statfile.read().split() cls._load["system"] = float(data[0])	306 B	2025-08-24 21:48
📄 monitor_net.py import tuned.monitors import os import re from tuned.utils.nettool import ethcard from tuned.utils.commands import commands cmd = commands() class NetMonitor(tuned.monitors.Monitor): @classmethod def _init_available_devices(cls): available = [] for root, dirs, files in os.walk("/sys/devices"): if root.endswith("/net") and not root.endswith("/virtual/net"): available += dirs cls._available_devices = set(available) for dev in available: #max_speed = cls._calcspeed(ethcard(dev).get_max_speed()) cls._load[dev] = ['0', '0', '0', '0'] @classmethod def _calcspeed(cls, speed): # 0.6 is just a magical constant (empirical value): Typical workload on netcard won't exceed # that and if it does, then the code is smart enough to adapt it. # 1024 * 1024 as for MB -> B # speed / 8 Mb -> MB return (int) (0.6 * 1024 * 1024 * speed / 8) @classmethod def _set_dev_map(cls, map_fce): cls.map_fce = map_fce @classmethod def _dev_map(cls, dev): try: return cls.map_fce(dev) except AttributeError: return dev @classmethod def _updateStat(cls, dev): files = ["rx_bytes", "rx_packets", "tx_bytes", "tx_packets"] for i,f in enumerate(files): cls._load[dev][i] = cmd.read_file("/sys/class/net/" + cls._dev_map(dev) + "/statistics/" + f, err_ret = "0").strip() @classmethod def update(cls): for device in cls._updating_devices: cls._updateStat(device)	1.38 KB	2025-08-24 21:48
📄 repository.py import tuned.logs import tuned.monitors from tuned.utils.class_loader import ClassLoader log = tuned.logs.get() __all__ = ["Repository"] class Repository(ClassLoader): def __init__(self): super(Repository, self).__init__() self._monitors = set() @property def monitors(self): return self._monitors def _set_loader_parameters(self): self._namespace = "tuned.monitors" self._prefix = "monitor_" self._interface = tuned.monitors.Monitor def create(self, plugin_name, devices): log.debug("creating monitor %s" % plugin_name) monitor_cls = self.load_class(plugin_name) monitor_instance = monitor_cls(devices) self._monitors.add(monitor_instance) return monitor_instance def delete(self, monitor): assert isinstance(monitor, self._interface) monitor.cleanup() self._monitors.remove(monitor)	827 B	2025-08-24 21:48

Terminal