import os import shutil import stat import subprocess import tempfile import shlex from bottles.backend.globals import ( Paths, gamemode_available, gamescope_available, mangohud_available, obs_vkc_available, vmtouch_available, ) from bottles.backend.logger import Logger from bottles.backend.managers.runtime import RuntimeManager from bottles.backend.managers.sandbox import SandboxManager from bottles.backend.models.config import BottleConfig from bottles.backend.models.result import Result from bottles.backend.utils.display import DisplayUtils from bottles.backend.utils.generic import detect_encoding from bottles.backend.utils.gpu import GPUUtils from bottles.backend.utils.manager import ManagerUtils from bottles.backend.utils.terminal import TerminalUtils from bottles.backend.utils.steam import SteamUtils logging = Logger() class WineEnv: """ This class is used to store and return a command environment. """ __env: dict = {} __result: dict = {"envs": {}, "overrides": []} def __init__(self, clean: bool = False): self.__env = {} if not clean: self.__env = os.environ.copy() def add(self, key, value, override=False): if key in self.__env: if override: self.__result["overrides"].append(f"{key}={value}") else: return self.__env[key] = value def add_bundle(self, bundle, override=False): for key, value in bundle.items(): self.add(key, value, override) def get(self): result = self.__result result["count_envs"] = len(result["envs"]) result["count_overrides"] = len(result["overrides"]) result["envs"] = self.__env return result def remove(self, key): if key in self.__env: del self.__env[key] def is_empty(self, key): return len(self.__env.get(key, "").strip()) == 0 def concat(self, key, values, sep=":"): if isinstance(values, str): values = [values] values = sep.join(values) if self.has(key): values = self.__env[key] + sep + values self.add(key, values, True) def has(self, key): return key in self.__env class WineCommand: """ This class is used to run a wine command with a custom environment. It also handles the launch in a terminal or not. """ def __init__( self, config: BottleConfig, command: str, terminal: bool = False, arguments: str = "", environment: dict = {}, communicate: bool = False, colors: str = "default", minimal: bool = False, # avoid gamemode/gamescope usage pre_script: str | None = None, post_script: str | None = None, cwd: str | None = None, midi_soundfont: str | None = None, ): _environment = environment.copy() self.config = self._get_config(config) self.minimal = minimal self.arguments = arguments self.cwd = self._get_cwd(cwd) self.runner, self.runner_runtime = self._get_runner_info() self.gamescope_activated = ( environment["GAMESCOPE"] == "1" if "GAMESCOPE" in environment else self.config.Parameters.gamescope ) self.command = self.get_cmd( command, pre_script, post_script, midi_soundfont, environment=_environment ) self.terminal = terminal self.env = self.get_env(_environment) self.communicate = communicate self.colors = colors self.vmtouch_files = None def _get_config(self, config: BottleConfig) -> BottleConfig: if cnf := config.data.get("config"): return cnf if not isinstance(config, BottleConfig): logging.error("Invalid config type: %s" % type(config)) return BottleConfig() return config def _get_cwd(self, cwd) -> str: config = self.config if config.Environment == "Steam": bottle = config.Path else: bottle = ManagerUtils.get_bottle_path(config) if not cwd: """ If no cwd is given, use the WorkingDir from the bottle configuration. """ cwd = config.WorkingDir if cwd == "" or not os.path.exists(cwd): """ If the WorkingDir is empty, use the bottle path as working directory. """ cwd = bottle return cwd def get_env( self, environment: dict | None = None, return_steam_env: bool = False, return_clean_env: bool = False, ) -> dict: env = WineEnv(clean=return_steam_env or return_clean_env) config = self.config arch = config.Arch params = config.Parameters # Bottle Path as environment variable env.add("BOTTLE", config.Path) if None in [arch, params]: return env.get()["envs"] if environment is None: environment = {} bottle = ManagerUtils.get_bottle_path(config) runner_path = ManagerUtils.get_runner_path(config.Runner) if config.Environment == "Steam": bottle = config.Path runner_path = config.RunnerPath if SteamUtils.is_proton(runner_path): runner_path = SteamUtils.get_dist_directory(runner_path) # Clean some env variables which can cause trouble # ref: # env.remove("XDG_DATA_HOME") dll_overrides = [] gpu = GPUUtils().get_gpu() DisplayUtils.check_nvidia_device() ld = [] # Bottle environment variables if config.Environment_Variables: for key, value in config.Environment_Variables.items(): env.add(key, value, override=True) # Environment variables from argument if environment: if environment.get("WINEDLLOVERRIDES"): dll_overrides.append(environment["WINEDLLOVERRIDES"]) del environment["WINEDLLOVERRIDES"] if environment.get("DXVK_CONFIG_FILE", "") == "bottle_root": environment["DXVK_CONFIG_FILE"] = os.path.join(bottle, "dxvk.conf") for e in environment: env.add(e, environment[e], override=True) # Language if config.Language != "sys": env.add("LC_ALL", config.Language) # Bottle DLL_Overrides if config.DLL_Overrides: for k, v in config.DLL_Overrides.items(): dll_overrides.append(f"{k}={v}") # Default DLL overrides if not return_steam_env: dll_overrides.append("winemenubuilder=''") # Get Runtime libraries if ( (params.use_runtime or params.use_eac_runtime or params.use_be_runtime) and not self.terminal and not return_steam_env ): _rb = RuntimeManager.get_runtime_env("bottles") if _rb: _eac = RuntimeManager.get_eac() _be = RuntimeManager.get_be() if params.use_runtime: logging.info("Using Bottles runtime") ld += _rb if ( _eac and not self.minimal ): # NOTE: should check for runner compatibility with "eac" (?) logging.info("Using EasyAntiCheat runtime") env.add("PROTON_EAC_RUNTIME", _eac) dll_overrides.append("easyanticheat_x86,easyanticheat_x64=b,n") if ( _be and not self.minimal ): # NOTE: should check for runner compatibility with "be" (?) logging.info("Using BattlEye runtime") env.add("PROTON_BATTLEYE_RUNTIME", _be) dll_overrides.append("beclient,beclient_x64=b,n") else: logging.warning("Bottles runtime was requested but not found") # Get Runner libraries if arch == "win64": runner_libs = [ "lib", "lib64", "lib/wine/x86_64-unix", "lib32/wine/x86_64-unix", "lib64/wine/x86_64-unix", "lib/wine/i386-unix", "lib32/wine/i386-unix", "lib64/wine/i386-unix", ] gst_libs = [ "lib64/gstreamer-1.0", "lib/gstreamer-1.0", "lib32/gstreamer-1.0", ] else: runner_libs = [ "lib", "lib/wine/i386-unix", "lib32/wine/i386-unix", "lib64/wine/i386-unix", ] gst_libs = ["lib/gstreamer-1.0", "lib32/gstreamer-1.0"] for lib in runner_libs: _path = os.path.join(runner_path, lib) if os.path.exists(_path): ld.append(_path) # Embedded GStreamer environment variables if not env.has("BOTTLES_USE_SYSTEM_GSTREAMER") and not return_steam_env: gst_env_path = [] for lib in gst_libs: if os.path.exists(os.path.join(runner_path, lib)): gst_env_path.append(os.path.join(runner_path, lib)) if len(gst_env_path) > 0: env.add("GST_PLUGIN_SYSTEM_PATH", ":".join(gst_env_path), override=True) # DXVK environment variables if params.dxvk and not return_steam_env: env.add("WINE_LARGE_ADDRESS_AWARE", "1") env.add( "DXVK_STATE_CACHE_PATH", os.path.join(bottle, "cache", "dxvk_state") ) env.add("STAGING_SHARED_MEMORY", "1") env.add("__GL_SHADER_DISK_CACHE", "1") env.add( "__GL_SHADER_DISK_CACHE_SKIP_CLEANUP", "1" ) # should not be needed anymore env.add( "__GL_SHADER_DISK_CACHE_PATH", os.path.join(bottle, "cache", "gl_shader"), ) env.add( "MESA_SHADER_CACHE_DIR", os.path.join(bottle, "cache", "mesa_shader") ) # VKD3D environment variables if params.vkd3d and not return_steam_env: env.add( "VKD3D_SHADER_CACHE_PATH", os.path.join(bottle, "cache", "vkd3d_shader") ) # LatencyFleX environment variables if params.latencyflex and not return_steam_env: _lf_path = ManagerUtils.get_latencyflex_path(config.LatencyFleX) _lf_layer_path = os.path.join( _lf_path, "layer/usr/share/vulkan/implicit_layer.d" ) env.concat("VK_ADD_LAYER_PATH", _lf_layer_path) env.add("LFX", "1") ld.append(os.path.join(_lf_path, "layer/usr/lib/x86_64-linux-gnu")) else: env.add("DISABLE_LFX", "1") # Mangohud environment variables if ( params.mangohud and not self.minimal and not (gamescope_available and self.gamescope_activated) ): env.add("MANGOHUD", "1") env.add("MANGOHUD_DLSYM", "1") if not params.mangohud_display_on_game_start: env.add("MANGOHUD_CONFIG", "no_display") # vkBasalt environment variables if params.vkbasalt and not self.minimal: vkbasalt_conf_path = os.path.join( ManagerUtils.get_bottle_path(config), "vkBasalt.conf" ) if os.path.isfile(vkbasalt_conf_path): env.add("VKBASALT_CONFIG_FILE", vkbasalt_conf_path) env.add("ENABLE_VKBASALT", "1") # OBS Vulkan Capture environment variables if params.obsvkc and not self.minimal: env.add("OBS_VKCAPTURE", "1") if DisplayUtils.display_server_type() == "x11": env.add("OBS_USE_EGL", "1") # DXVK-Nvapi environment variables if params.dxvk_nvapi and not return_steam_env: # NOTE: users reported that DXVK_ENABLE_NVAPI and DXVK_NVAPIHACK must be set to make # DLSS works. I don't have a GPU compatible with this tech, so I'll trust them env.add("DXVK_NVAPIHACK", "0") env.add("DXVK_ENABLE_NVAPI", "1") # Esync environment variable if params.sync == "esync": env.add("WINEESYNC", "1") # Fsync environment variable if params.sync == "fsync": env.add("WINEFSYNC", "1") # Wine debug level if not return_steam_env: debug_level = "fixme-all" if params.fixme_logs: debug_level = "+fixme-all" env.add("WINEDEBUG", debug_level) # Aco compiler # if params["aco_compiler"]: # env.add("ACO_COMPILER", "aco") # FSR if params.fsr: env.add("WINE_FULLSCREEN_FSR", "1") env.add("WINE_FULLSCREEN_FSR_STRENGTH", str(params.fsr_sharpening_strength)) if params.fsr_quality_mode: env.add("WINE_FULLSCREEN_FSR_MODE", str(params.fsr_quality_mode)) # PulseAudio latency if params.pulseaudio_latency: env.add("PULSE_LATENCY_MSEC", "60") # Discrete GPU if not return_steam_env: if params.discrete_gpu: discrete = gpu["prime"]["discrete"] if discrete is not None: gpu_envs = discrete["envs"] for p in gpu_envs: env.add(p, gpu_envs[p]) env.concat("VK_ICD_FILENAMES", discrete["icd"]) # VK_ICD if not env.has("VK_ICD_FILENAMES"): if gpu["prime"]["integrated"] is not None: """ System support PRIME but user disabled the discrete GPU setting (previus check skipped), so using the integrated one. """ env.concat("VK_ICD_FILENAMES", gpu["prime"]["integrated"]["icd"]) else: """ System doesn't support PRIME, so using the first result from the gpu vendors list. """ if "vendors" in gpu and len(gpu["vendors"]) > 0: _first = list(gpu["vendors"].keys())[0] env.concat("VK_ICD_FILENAMES", gpu["vendors"][_first]["icd"]) else: logging.warning( "No GPU vendor found, keep going without setting VK_ICD_FILENAMES…" ) # Add ld to LD_LIBRARY_PATH if ld: env.concat("LD_LIBRARY_PATH", ld) # Vblank # env.add("__GL_SYNC_TO_VBLANK", "0") # env.add("vblank_mode", "0") # DLL Overrides env.concat("WINEDLLOVERRIDES", dll_overrides, sep=";") if env.is_empty("WINEDLLOVERRIDES"): env.remove("WINEDLLOVERRIDES") if not return_steam_env: # Wine prefix env.add("WINEPREFIX", bottle, override=True) # Wine arch env.add("WINEARCH", arch) return env.get()["envs"] def _get_runner_info(self) -> tuple[str, str]: config = self.config runner = ManagerUtils.get_runner_path(config.Runner) arch = config.Arch runner_runtime = "" if config.Environment == "Steam": runner = config.RunnerPath if runner in [None, ""]: return "", "" if SteamUtils.is_proton(runner): """ If the runner is Proton, set the path to /dist or /files based on check if files exists. Additionally, check for its corresponding runtime. """ runner_runtime = SteamUtils.get_associated_runtime(runner) runner = os.path.join(SteamUtils.get_dist_directory(runner), "bin/wine") elif runner.startswith("sys-"): """ If the runner type is system, set the runner binary path to the system command. Else set it to the full path. """ runner = shutil.which("wine") else: runner = f"{runner}/bin/wine" if arch == "win64" and os.path.exists(f"{runner}64"): runner = f"{runner}64" runner = shlex.quote(runner) # type: ignore return runner, runner_runtime def get_cmd( self, command, pre_script: str | None = None, post_script: str | None = None, midi_soundfont: str | None = None, return_steam_cmd: bool = False, return_clean_cmd: bool = False, environment: dict | None = None, ) -> str: config = self.config params = config.Parameters runner = self.runner if environment is None: environment = {} if return_clean_cmd: return_steam_cmd = True if not return_steam_cmd and not return_clean_cmd: command = f"{runner} {command}" if not self.minimal: if gamemode_available and params.gamemode: if not return_steam_cmd: command = f"{gamemode_available} {command}" else: command = f"gamemode {command}" if mangohud_available and params.mangohud and not self.gamescope_activated: if not return_steam_cmd: command = f"{mangohud_available} {command}" else: command = f"mangohud {command}" if gamescope_available and self.gamescope_activated: gamescope_run = tempfile.NamedTemporaryFile(mode="w", suffix=".sh").name # Create temporary sh script in /tmp where Gamescope will execute it file = ["#!/usr/bin/env sh\n"] file.append(f"{command} $@") if mangohud_available and params.mangohud: file.append(" &\nmangoapp") with open(gamescope_run, "w") as f: f.write("".join(file)) # Update command command = ( f"{self._get_gamescope_cmd(return_steam_cmd)} -- {gamescope_run}" ) logging.info(f"Running Gamescope command: '{command}'") logging.info(f"{gamescope_run} contains:") with open(gamescope_run) as f: logging.info(f"\n\n{f.read()}") # Set file as executable st = os.stat(gamescope_run) os.chmod(gamescope_run, st.st_mode | stat.S_IEXEC) if obs_vkc_available and params.obsvkc: command = f"{obs_vkc_available} {command}" if params.use_steam_runtime: _rs = RuntimeManager.get_runtimes("steam") _picked = {} if _rs: if "sniper" in _rs.keys() and "sniper" in self.runner_runtime: """ Sniper is the default runtime used by Proton version >= 8.0 """ _picked = _rs["sniper"] elif "soldier" in _rs.keys() and "soldier" in self.runner_runtime: """ Sniper is the default runtime used by Proton version >= 5.13 and < 8.0 """ _picked = _rs["soldier"] elif "scout" in _rs.keys(): """ For Wine runners, we cannot make assumption about which runtime would suits them the best, as it would depend on their build environment. Sniper/Soldier are not backward-compatible, defaulting to Scout should maximize compatibility. """ _picked = _rs["scout"] else: logging.warning("Steam runtime was requested but not found") if _picked: logging.info(f"Using Steam runtime {_picked['name']}") command = f"{_picked['entry_point']} {command}" else: logging.warning( "Steam runtime was requested and found but there are no valid combinations" ) if self.arguments: prefix, suffix, extracted_env = SteamUtils.handle_launch_options( self.arguments ) if prefix: command = f"{prefix} {command}" if suffix: command = f"{command} {suffix}" if extracted_env: if extracted_env.get("WINEDLLOVERRIDES") and environment.get( "WINEDLLOVERRIDES" ): environment["WINEDLLOVERRIDES"] += ";" + extracted_env.get( "WINEDLLOVERRIDES" ) del extracted_env["WINEDLLOVERRIDES"] environment.update(extracted_env) if post_script not in (None, ""): command = f"{command} ; sh '{post_script}'" if pre_script not in (None, ""): command = f"sh '{pre_script}' ; {command}" return command def _get_gamescope_cmd(self, return_steam_cmd: bool = False) -> str: config = self.config params = config.Parameters gamescope_cmd = [] if gamescope_available and self.gamescope_activated: gamescope_cmd = [gamescope_available] if return_steam_cmd: gamescope_cmd = ["gamescope"] if params.gamescope_fullscreen: gamescope_cmd.append("-f") if params.gamescope_borderless: gamescope_cmd.append("-b") if params.gamescope_scaling: gamescope_cmd.append("-S integer") if params.fsr: gamescope_cmd.append("-F fsr") # Upscaling sharpness is from 0 to 20. There are 5 FSR upscaling levels, # so multiply by 4 to reach 20 gamescope_cmd.append( f"--fsr-sharpness {params.fsr_sharpening_strength * 4}" ) if params.gamescope_fps > 0: gamescope_cmd.append(f"-r {params.gamescope_fps}") if params.gamescope_fps_no_focus > 0: gamescope_cmd.append(f"-o {params.gamescope_fps_no_focus}") if params.gamescope_game_width > 0: gamescope_cmd.append(f"-w {params.gamescope_game_width}") if params.gamescope_game_height > 0: gamescope_cmd.append(f"-h {params.gamescope_game_height}") if params.gamescope_window_width > 0: gamescope_cmd.append(f"-W {params.gamescope_window_width}") if params.gamescope_window_height > 0: gamescope_cmd.append(f"-H {params.gamescope_window_height}") return " ".join(gamescope_cmd) def _vmtouch_preload(self): vmtouch_flags = "-t -v -l -d" vmtouch_file_size = " -m 1024M" if self.command.find("C:\\") > 0: s = ( self.cwd + "/" + (self.command.split(" ")[-1].split("\\")[-1]) ).replace("'", "") else: s = self.command.split(" ")[-1] self.vmtouch_files = shlex.quote(s) # if self.config.Parameters.vmtouch_cache_cwd: # self.vmtouch_files = "'"+self.vmtouch_files+"' '"+self.cwd+"/'" Commented out as fix for #1941 self.command = f"{vmtouch_available} {vmtouch_flags} {vmtouch_file_size} {self.vmtouch_files} && {self.command}" def _vmtouch_free(self): subprocess.Popen( "kill $(pidof vmtouch)", shell=True, env=self.env, cwd=self.cwd, ) if not self.vmtouch_files: return vmtouch_flags = "-e -v" command = f"{vmtouch_available} {vmtouch_flags} {self.vmtouch_files}" subprocess.Popen( command, shell=True, env=self.env, cwd=self.cwd, ) def _get_sandbox_manager(self) -> SandboxManager: return SandboxManager( envs=self.env, chdir=self.cwd, share_paths_rw=[ManagerUtils.get_bottle_path(self.config)], share_paths_ro=[Paths.runners, Paths.temp], share_net=self.config.Sandbox.share_net, share_sound=self.config.Sandbox.share_sound, ) def run(self) -> Result[str | None]: """ Run command with pre-configured parameters :return: `status` is True if command executed successfully, `data` may be available even if `status` is False. """ if None in [self.runner, self.env]: return Result( False, message="runner or env is not ready, Wine command terminated." ) if vmtouch_available and self.config.Parameters.vmtouch and not self.terminal: self._vmtouch_preload() sandbox = ( self._get_sandbox_manager() if self.config.Parameters.sandbox else None ) # run command in external terminal if terminal is True if self.terminal: if sandbox: return Result( status=TerminalUtils().execute( sandbox.get_cmd(self.command), self.env, self.colors, self.cwd ) ) else: return Result( status=TerminalUtils().execute( self.command, self.env, self.colors, self.cwd ) ) # prepare proc if we are going to execute command internally # proc should always be `Popen[bytes]` to make sure # stdout_data's type is `bytes` proc: subprocess.Popen[bytes] if sandbox: proc = sandbox.run(self.command) else: try: proc = subprocess.Popen( self.command, stdout=subprocess.PIPE, shell=True, env=self.env, cwd=self.cwd, ) except FileNotFoundError: return Result(False, message="File not found") stdout_data, _ = proc.communicate() if vmtouch_available and self.config.Parameters.vmtouch: # don't call vmtouch_free while running via external terminal self._vmtouch_free() # Consider changing the locale to C.UTF-8 when # executing commands, to ensure consistent output and # enable callers to make use of the returned value, # also without requiring the encoding detection dance codec = detect_encoding(stdout_data) rv: str try: rv = stdout_data.decode(codec) except (UnicodeDecodeError, LookupError, TypeError): # UnicodeDecodeError: codec mismatch # LookupError: unknown codec name # TypeError: codec is None logging.warning("stdout decoding failed") rv = str(stdout_data)[2:-1] # trim b'' # "ShellExecuteEx" exception may occur while executing command, # previously we rerun the command without `cwd` and `stdout=PIPE` # to fix it, which is removed since it may lead to unexpected behavior if "ShellExecuteEx" in rv: logging.warning("ShellExecuteEx exception seems occurred.") return Result( False, data=rv, message="ShellExecuteEx exception seems occurred." ) return Result(True, data=rv)