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-27 02:38
📄 __init__.py import logging from pyvirtualdisplay.about import __version__ from pyvirtualdisplay.display import Display Display # ignore unused log = logging.getLogger(__name__) log = logging.getLogger(__name__) log.debug("version=%s", __version__)	241 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 about.py __version__ = "3.0"	20 B	2026-01-20 12:58
📄 abstractdisplay.py import fnmatch import logging import os import select import subprocess import tempfile import time from threading import Lock from pyvirtualdisplay import xauth from pyvirtualdisplay.util import get_helptext, platform_is_osx log = logging.getLogger(__name__) # try: # import fcntl # except ImportError: # fcntl = None _mutex = Lock() _mutex_popen = Lock() _MIN_DISPLAY_NR = 1000 _USED_DISPLAY_NR_LIST = [] _X_START_TIMEOUT = 10 _X_START_TIME_STEP = 0.1 _X_START_WAIT = 0.1 class XStartTimeoutError(Exception): pass class XStartError(Exception): pass def _lock_files(): tmpdir = "/tmp" try: ls = os.listdir(tmpdir) except FileNotFoundError: log.warning("missing /tmp") return [] pattern = ".X-lock" names = fnmatch.filter(ls, pattern) ls = [os.path.join(tmpdir, child) for child in names] ls = [p for p in ls if os.path.isfile(p)] return ls def _search_for_display(): # search for free display ls = list(map(lambda x: int(x.split("X")[1].split("-")[0]), _lock_files())) if len(ls): display = max(_MIN_DISPLAY_NR, max(ls) + 3) else: display = _MIN_DISPLAY_NR return display class AbstractDisplay(object): """ Common parent for X servers (Xvfb,Xephyr,Xvnc) """ def __init__(self, program, use_xauth, retries, extra_args, manage_global_env): self._extra_args = extra_args self._retries = retries self._program = program self.stdout = None self.stderr = None self.old_display_var = None self._subproc = None self.display = None self._is_started = False self._manage_global_env = manage_global_env self._reset_global_env = False self._pipe_wfd = None self._retries_current = 0 helptext = get_helptext(program) self._has_displayfd = "-displayfd" in helptext if not self._has_displayfd: log.debug("-displayfd flag is missing.") PYVIRTUALDISPLAY_DISPLAYFD = os.environ.get("PYVIRTUALDISPLAY_DISPLAYFD") if PYVIRTUALDISPLAY_DISPLAYFD: log.debug("PYVIRTUALDISPLAY_DISPLAYFD=%s", PYVIRTUALDISPLAY_DISPLAYFD) # '0'->false, '1'->true self._has_displayfd = bool(int(PYVIRTUALDISPLAY_DISPLAYFD)) else: # TODO: macos: displayfd is available on XQuartz-2.7.11 but it doesn't work, always 0 is returned if platform_is_osx(): self._has_displayfd = False self._check_flags(helptext) if use_xauth and not xauth.is_installed(): raise xauth.NotFoundError() self._use_xauth = use_xauth self._old_xauth = None self._xauth_filename = None def _check_flags(self, helptext): pass def _cmd(self): raise NotImplementedError() def _redirect_display(self, on): """ on: True -> set $DISPLAY to virtual screen * False -> set $DISPLAY to original screen :param on: bool """ d = self.new_display_var if on else self.old_display_var if d is None: log.debug("unset $DISPLAY") try: del os.environ["DISPLAY"] except KeyError: log.warning("$DISPLAY was already unset.") else: log.debug("set $DISPLAY=%s", d) os.environ["DISPLAY"] = d def _env(self): env = os.environ.copy() env["DISPLAY"] = self.new_display_var return env def start(self): """ start display :rtype: self """ if self._is_started: raise XStartError(self, "Display was started twice.") self._is_started = True if self._has_displayfd: self._start1_has_displayfd() else: i = 0 while True: self._retries_current = i + 1 try: self._start1() break except XStartError: log.warning("start failed %s", i + 1) time.sleep(0.05) i += 1 if i >= self._retries: raise XStartError( "No success after %s retries. Last stderr: %s" % (self._retries, self.stderr) ) if self._manage_global_env: self._redirect_display(True) self._reset_global_env = True def _popen(self, use_pass_fds): with _mutex_popen: if use_pass_fds: self._subproc = subprocess.Popen( self._command, pass_fds=[self._pipe_wfd], stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, ) else: self._subproc = subprocess.Popen( self._command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, ) def _start1_has_displayfd(self): # stdout doesn't work on osx -> create own pipe rfd, self._pipe_wfd = os.pipe() self._command = self._cmd() + self._extra_args log.debug("command: %s", self._command) self._popen(use_pass_fds=True) self.display = int(self._wait_for_pipe_text(rfd)) os.close(rfd) os.close(self._pipe_wfd) self.new_display_var = ":%s" % int(self.display) if self._use_xauth: self._setup_xauth() # https://github.com/ponty/PyVirtualDisplay/issues/2 # https://github.com/ponty/PyVirtualDisplay/issues/14 self.old_display_var = os.environ.get("DISPLAY", None) def _start1(self): with _mutex: self.display = _search_for_display() while self.display in _USED_DISPLAY_NR_LIST: self.display += 1 self.new_display_var = ":%s" % int(self.display) _USED_DISPLAY_NR_LIST.append(self.display) self._command = self._cmd() + self._extra_args log.debug("command: %s", self._command) self._popen(use_pass_fds=False) self.new_display_var = ":%s" % int(self.display) if self._use_xauth: self._setup_xauth() # https://github.com/ponty/PyVirtualDisplay/issues/2 # https://github.com/ponty/PyVirtualDisplay/issues/14 self.old_display_var = os.environ.get("DISPLAY", None) # wait until X server is active start_time = time.time() d = self.new_display_var ok = False time.sleep(0.05) # give time for early exit while True: if not self.is_alive(): break try: xdpyinfo = subprocess.Popen( ["xdpyinfo"], env=self._env(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, ) _, _ = xdpyinfo.communicate() exit_code = xdpyinfo.returncode except FileNotFoundError: log.warning( "xdpyinfo was not found, X start can not be checked! Please install xdpyinfo!" ) time.sleep(_X_START_WAIT) # old method ok = True break # try: # xdpyinfo = EasyProcess(["xdpyinfo"], env=self._env()) # xdpyinfo.enable_stdout_log = False # xdpyinfo.enable_stderr_log = False # exit_code = xdpyinfo.call().return_code # except EasyProcessError: # log.warning( # "xdpyinfo was not found, X start can not be checked! Please install xdpyinfo!" # ) # time.sleep(_X_START_WAIT) # old method # ok = True # break if exit_code != 0: pass else: log.info('Successfully started X with display "%s".', d) ok = True break if time.time() - start_time >= _X_START_TIMEOUT: break time.sleep(_X_START_TIME_STEP) if not self.is_alive(): log.warning("process exited early. stderr:%s", self.stderr) msg = "Failed to start process: %s" raise XStartError(msg % self) if not ok: msg = 'Failed to start X on display "%s" (xdpyinfo check failed, stderr:[%s]).' raise XStartTimeoutError(msg % (d, xdpyinfo.stderr)) def _wait_for_pipe_text(self, rfd): s = "" start_time = time.time() while True: (rfd_changed_ls, _, _) = select.select([rfd], [], [], 0.1) if not self.is_alive(): raise XStartError( "%s program closed. command: %s stderr: %s" % (self._program, self._command, self.stderr) ) if rfd in rfd_changed_ls: c = os.read(rfd, 1) if c == b"\n": break s += c.decode("ascii") # this timeout is for "eternal" hang. see #62 if time.time() - start_time >= 600: # = 10 minutes raise XStartTimeoutError( "No reply from program %s. command:%s" % ( self._program, self._command, ) ) return s def stop(self): """ stop display :rtype: self """ if not self._is_started: raise XStartError("stop() is called before start().") if self._reset_global_env: self._redirect_display(False) if self.is_alive(): try: self._subproc.kill() except OSError as oserror: log.debug("exception in terminate:%s", oserror) self._subproc.wait() self._read_stdout_stderr() if self._use_xauth: self._clear_xauth() return self def _read_stdout_stderr(self): if self.stdout is None: (self.stdout, self.stderr) = self._subproc.communicate() log.debug("stdout=%s", self.stdout) log.debug("stderr=%s", self.stderr) def _setup_xauth(self): """ Set up the Xauthority file and the XAUTHORITY environment variable. """ handle, filename = tempfile.mkstemp( prefix="PyVirtualDisplay.", suffix=".Xauthority" ) self._xauth_filename = filename os.close(handle) # Save old environment self._old_xauth = {} self._old_xauth["AUTHFILE"] = os.getenv("AUTHFILE") self._old_xauth["XAUTHORITY"] = os.getenv("XAUTHORITY") os.environ["AUTHFILE"] = os.environ["XAUTHORITY"] = filename cookie = xauth.generate_mcookie() xauth.call("add", self.new_display_var, ".", cookie) def _clear_xauth(self): """ Clear the Xauthority file and restore the environment variables. """ os.remove(self._xauth_filename) for varname in ["AUTHFILE", "XAUTHORITY"]: if self._old_xauth[varname] is None: del os.environ[varname] else: os.environ[varname] = self._old_xauth[varname] self._old_xauth = None def __enter__(self): """used by the :keyword:`with` statement""" self.start() return self def __exit__(self, *exc_info): """used by the :keyword:`with` statement""" self.stop() def is_alive(self): if not self._subproc: return False # return self.return_code is None rc = self._subproc.poll() if rc is not None: # proc exited self._read_stdout_stderr() return rc is None # @property # def return_code(self): # if not self._subproc: # return None # rc = self._subproc.poll() # if rc is not None: # # proc exited # self._read_stdout_stderr() # return rc @property def pid(self): """ PID (:attr:`subprocess.Popen.pid`) :rtype: int """ if self._subproc: return self._subproc.pid	12.29 KB	2026-01-20 12:58
📄 display.py from typing import Dict, List, Optional, Tuple from pyvirtualdisplay.xephyr import XephyrDisplay from pyvirtualdisplay.xvfb import XvfbDisplay from pyvirtualdisplay.xvnc import XvncDisplay _class_map = {"xvfb": XvfbDisplay, "xvnc": XvncDisplay, "xephyr": XephyrDisplay} class Display(object): """ Proxy class :param color_depth: [8, 16, 24, 32] :param size: screen size (width,height) :param bgcolor: background color ['black' or 'white'] :param visible: True -> Xephyr, False -> Xvfb :param backend: 'xvfb', 'xvnc' or 'xephyr', ignores ``visible`` :param xauth: If a Xauthority file should be created. :param manage_global_env: if True then $DISPLAY is set in os.environ which is not thread-safe. Use False to make it thread-safe. """ def __init__( self, backend: Optional[str] = None, visible: bool = False, size: Tuple[int, int] = (1024, 768), color_depth: int = 24, bgcolor: str = "black", use_xauth: bool = False, # check_startup=False, retries: int = 10, extra_args: List[str] = [], manage_global_env: bool = True, kwargs ): self._color_depth = color_depth self._size = size self._bgcolor = bgcolor self._visible = visible self._backend = backend if not self._backend: if self._visible: self._backend = "xephyr" else: self._backend = "xvfb" cls = _class_map.get(self._backend) if not cls: raise ValueError("unknown backend: %s" % self._backend) self._obj = cls( size=size, color_depth=color_depth, bgcolor=bgcolor, retries=retries, use_xauth=use_xauth, # check_startup=check_startup, extra_args=extra_args, manage_global_env=manage_global_env, kwargs ) def start(self) -> "Display": """ start display :rtype: self """ self._obj.start() return self def stop(self) -> "Display": """ stop display :rtype: self """ self._obj.stop() return self def __enter__(self): """used by the :keyword:`with` statement""" self.start() return self def __exit__(self, *exc_info): """used by the :keyword:`with` statement""" self.stop() def is_alive(self) -> bool: return self._obj.is_alive() # @property # def return_code(self): # return self._obj.return_code @property def pid(self) -> int: """ PID (:attr:`subprocess.Popen.pid`) :rtype: int """ return self._obj.pid @property def display(self) -> int: """The new $DISPLAY variable as int. Example 1 if $DISPLAY=':1'""" return self._obj.display @property def new_display_var(self) -> str: """The new $DISPLAY variable like ':1'""" return self._obj.new_display_var def env(self) -> Dict[str, str]: """env() copies global os.environ and adds disp.new_display_var""" return self._obj._env()	3.18 KB	2026-01-20 12:58
📁 examples	-	2026-03-21 15:45
📄 py.typed	0 B	2026-01-20 12:58
📄 smartdisplay.py import logging import time from PIL import Image, ImageChops from PIL.ImageGrab import grab from pyvirtualdisplay import Display log = logging.getLogger(__name__) class DisplayTimeoutError(Exception): pass def autocrop(im, bgcolor): """Crop borders off an image. :param im: Source image. :param bgcolor: Background color, using either a color tuple. :return: An image without borders, or None if there's no actual content in the image. """ if im.mode != "RGB": im = im.convert("RGB") bg = Image.new("RGB", im.size, bgcolor) diff = ImageChops.difference(im, bg) bbox = diff.getbbox() if bbox: return im.crop(bbox) return None # no contents class SmartDisplay(Display): def autocrop(self, im): """Crop borders off an image. :param im: Source image. :return: An image without borders, or None if there's no actual content in the image. """ return autocrop(im, self._bgcolor) def grab(self, autocrop=True): # TODO: use Xvfb fbdir option for screenshot img = grab(xdisplay=self.new_display_var) if autocrop: img = self.autocrop(img) return img def waitgrab(self, timeout=60, autocrop=True, cb_imgcheck=None): """start process and create screenshot. Repeat screenshot until it is not empty and cb_imgcheck callback function returns True for current screenshot. :param autocrop: True -> crop screenshot :param timeout: int :param cb_imgcheck: None or callback for testing img, True = accept img, False = reject img """ t = 0 sleep_time = 0.3 # for fast windows repeat_time = 0.5 while 1: log.debug("sleeping %s secs" % str(sleep_time)) time.sleep(sleep_time) t += sleep_time img = self.grab(autocrop=False) img_crop = self.autocrop(img) if autocrop: img = img_crop if img_crop: if not cb_imgcheck: break if cb_imgcheck(img): break sleep_time = repeat_time repeat_time += 0.5 # progressive if t > timeout: msg = "Timeout! elapsed time:%s timeout:%s " % (t, timeout) raise DisplayTimeoutError(msg) # break log.debug("screenshot is empty, next try..") assert img # if not img: # log.debug('screenshot is empty!') return img	2.6 KB	2026-01-20 12:58
📄 util.py import subprocess import sys def get_helptext(program): cmd = [program, "-help"] # py3.7+ # p = subprocess.run(cmd, capture_output=True) # stderr = p.stderr # py3.6 also p = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, ) _, stderr = p.communicate() helptext = stderr.decode("utf-8", "ignore") return helptext def platform_is_osx(): return sys.platform == "darwin"	486 B	2026-01-20 12:58
📄 xauth.py """Utility functions for xauth.""" import hashlib import os import subprocess class NotFoundError(Exception): """Error when xauth was not found.""" def is_installed(): """ Return whether or not xauth is installed. """ try: xauth = subprocess.Popen( ["xauth", "-V"], # env=self._env(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) _, _ = xauth.communicate() # p = EasyProcess(["xauth", "-V"]) # p.enable_stdout_log = False # p.enable_stderr_log = False # p.call() except FileNotFoundError: return False else: return True def generate_mcookie(): """ Generate a cookie string suitable for xauth. """ data = os.urandom(16) # 16 bytes = 128 bit return hashlib.md5(data).hexdigest() def call(*args): """ Call xauth with the given args. """ xauth = subprocess.Popen( ["xauth"] + list(args), # env=self._env(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) _, _ = xauth.communicate() # EasyProcess(["xauth"] + list(args)).call()	1.14 KB	2026-01-20 12:58
📄 xephyr.py import logging from pyvirtualdisplay.abstractdisplay import AbstractDisplay log = logging.getLogger(__name__) PROGRAM = "Xephyr" class XephyrDisplay(AbstractDisplay): """ Xephyr wrapper Xephyr is an X server outputting to a window on a pre-existing X display """ def __init__( self, size=(1024, 768), color_depth=24, bgcolor="black", use_xauth=False, retries=10, extra_args=[], manage_global_env=True, parent=None, ): """ :param bgcolor: 'black' or 'white' """ self._color_depth = color_depth self._size = size self._bgcolor = bgcolor self._parent = parent AbstractDisplay.__init__( self, PROGRAM, use_xauth=use_xauth, retries=retries, extra_args=extra_args, manage_global_env=manage_global_env, ) def _check_flags(self, helptext): self._has_resizeable = "-resizeable" in helptext def _cmd(self): cmd = ( [ PROGRAM, ] + (["-parent", self._parent] if self._parent else []) + [ dict(black="-br", white="-wr")[self._bgcolor], "-screen", "x".join(map(str, list(self._size) + [self._color_depth])), ] ) if self._has_displayfd: cmd += ["-displayfd", str(self._pipe_wfd)] else: cmd += [self.new_display_var] if self._has_resizeable: cmd += ["-resizeable"] return cmd	1.6 KB	2026-01-20 12:58
📄 xvfb.py import logging from pyvirtualdisplay.abstractdisplay import AbstractDisplay log = logging.getLogger(__name__) PROGRAM = "Xvfb" class XvfbDisplay(AbstractDisplay): """ Xvfb wrapper Xvfb is an X server that can run on machines with no display hardware and no physical input devices. It emulates a dumb framebuffer using virtual memory. """ def __init__( self, size=(1024, 768), color_depth=24, bgcolor="black", use_xauth=False, fbdir=None, dpi=None, retries=10, extra_args=[], manage_global_env=True, ): """ :param bgcolor: 'black' or 'white' :param fbdir: If non-null, the virtual screen is memory-mapped to a file in the given directory ('-fbdir' option) :param dpi: screen resolution in dots per inch if not None """ self._screen = 0 self._size = size self._color_depth = color_depth self._bgcolor = bgcolor self._fbdir = fbdir self._dpi = dpi AbstractDisplay.__init__( self, PROGRAM, use_xauth=use_xauth, retries=retries, extra_args=extra_args, manage_global_env=manage_global_env, ) def _check_flags(self, helptext): pass def _cmd(self): cmd = [ dict(black="-br", white="-wr")[self._bgcolor], "-nolisten", "tcp", "-screen", str(self._screen), "x".join(map(str, list(self._size) + [self._color_depth])), ] if self._fbdir: cmd += ["-fbdir", self._fbdir] if self._dpi is not None: cmd += ["-dpi", str(self._dpi)] if self._has_displayfd: cmd += ["-displayfd", str(self._pipe_wfd)] else: cmd += [self.new_display_var] return [PROGRAM] + cmd	1.9 KB	2026-01-20 12:58
📄 xvnc.py import logging from pyvirtualdisplay.abstractdisplay import AbstractDisplay log = logging.getLogger(__name__) PROGRAM = "Xvnc" class XvncDisplay(AbstractDisplay): """ Xvnc wrapper """ def __init__( self, size=(1024, 768), color_depth=24, bgcolor="black", use_xauth=False, rfbport=5900, rfbauth=None, retries=10, extra_args=[], manage_global_env=True, ): """ :param bgcolor: 'black' or 'white' :param rfbport: Specifies the TCP port on which Xvnc listens for connections from viewers (the protocol used in VNC is called RFB - "remote framebuffer"). The default is 5900 plus the display number. :param rfbauth: Specifies the file containing the password used to authenticate viewers. """ self._size = size self._color_depth = color_depth self._bgcolor = bgcolor self._rfbport = rfbport self._rfbauth = rfbauth AbstractDisplay.__init__( self, PROGRAM, use_xauth=use_xauth, retries=retries, extra_args=extra_args, manage_global_env=manage_global_env, ) def _check_flags(self, helptext): pass def _cmd(self): cmd = [ PROGRAM, "-depth", str(self._color_depth), "-geometry", "%dx%d" % (self._size[0], self._size[1]), "-rfbport", str(self._rfbport), ] if self._rfbauth: cmd += ["-rfbauth", str(self._rfbauth)] # default: # -SecurityTypes = VncAuth else: cmd += ["-SecurityTypes", "None"] if self._has_displayfd: cmd += ["-displayfd", str(self._pipe_wfd)] else: cmd += [self.new_display_var] return cmd	1.87 KB	2026-01-20 12:58

Terminal