diff --git a/umu/umu_consts.py b/umu/umu_consts.py index 6d23da001..ee316529b 100644 --- a/umu/umu_consts.py +++ b/umu/umu_consts.py @@ -19,9 +19,9 @@ class Color(Enum): CONFIG = "umu_version.json" -umu_LOCAL: Path = Path.home().joinpath(".local", "share", "umu") +UMU_LOCAL: Path = Path.home().joinpath(".local", "share", "umu") -umu_CACHE: Path = Path.home().joinpath(".cache", "umu") +UMU_CACHE: Path = Path.home().joinpath(".cache", "umu") STEAM_COMPAT: Path = Path.home().joinpath( ".local", "share", "Steam", "compatibilitytools.d" diff --git a/umu/umu_dl_util.py b/umu/umu_dl_util.py index 717d62be5..8493227bd 100644 --- a/umu/umu_dl_util.py +++ b/umu/umu_dl_util.py @@ -11,7 +11,7 @@ from umu_plugins import enable_zenity from socket import gaierror from umu_log import log -from umu_consts import STEAM_COMPAT, umu_CACHE +from umu_consts import STEAM_COMPAT, UMU_CACHE try: from tarfile import tar_filter @@ -30,30 +30,30 @@ def get_umu_proton(env: Dict[str, str]) -> Union[Dict[str, str]]: """ files: List[Tuple[str, str]] = [] - try: - files = _fetch_releases() - except gaierror: - pass # User is offline - - umu_CACHE.mkdir(exist_ok=True, parents=True) + UMU_CACHE.mkdir(exist_ok=True, parents=True) STEAM_COMPAT.mkdir(exist_ok=True, parents=True) # Prioritize the Steam compat - if _get_from_steamcompat(env, STEAM_COMPAT, umu_CACHE, files): + if _get_from_steamcompat(env, STEAM_COMPAT, UMU_CACHE): return env + try: + files = _fetch_releases() + except gaierror: + pass # User is offline + # Use the latest Proton in the cache if it exists - if _get_from_cache(env, STEAM_COMPAT, umu_CACHE, files, True): + if _get_from_cache(env, STEAM_COMPAT, UMU_CACHE, files, True): return env # Download the latest if Proton is not in Steam compat # If the digests mismatched, refer to the cache in the next block - if _get_latest(env, STEAM_COMPAT, umu_CACHE, files): + if _get_latest(env, STEAM_COMPAT, UMU_CACHE, files): return env # Refer to an old version previously downloaded # Reached on digest mismatch, user interrupt or download failure/no internet - if _get_from_cache(env, STEAM_COMPAT, umu_CACHE, files, False): + if _get_from_cache(env, STEAM_COMPAT, UMU_CACHE, files, False): return env # No internet and cache/compat tool is empty, just return and raise an @@ -121,7 +121,7 @@ def _fetch_releases() -> List[Tuple[str, str]]: def _fetch_proton( env: Dict[str, str], steam_compat: Path, cache: Path, files: List[Tuple[str, str]] ) -> Dict[str, str]: - """Download the latest umu-Proton and set it as PROTONPATH.""" + """Download the latest umu-proton and set it as PROTONPATH.""" hash, hash_url = files[0] proton, proton_url = files[1] proton_dir: str = proton[: proton.find(".tar.gz")] # Proton dir @@ -228,29 +228,21 @@ def _cleanup(tarball: str, proton: str, cache: Path, steam_compat: Path) -> None def _get_from_steamcompat( - env: Dict[str, str], steam_compat: Path, cache: Path, files: List[Tuple[str, str]] + env: Dict[str, str], steam_compat: Path, cache: Path ) -> Union[Dict[str, str], None]: """Refer to Steam compat folder for any existing Proton directories.""" - proton_dir: str = "" # Latest Proton - - if len(files) == 2: - proton_dir: str = files[1][0][: files[1][0].find(".tar.gz")] - - for proton in steam_compat.glob("umu-Proton*"): + for proton in sorted( + [ + proton + for proton in steam_compat.glob("*") + if proton.name.startswith("umu-proton") + or proton.name.startswith("ULWGL-Proton") + ] + ): log.console(f"{proton.name} found in: {steam_compat}") log.console(f"Using {proton.name}") - environ["PROTONPATH"] = proton.as_posix() env["PROTONPATH"] = environ["PROTONPATH"] - - # Notify the user that they're not using the latest - if proton_dir and proton.name != proton_dir: - link: str = files[1][1] - log.console( - "umu-Proton is outdated.\n" - f"For latest release, please download {link}" - ) - return env return None @@ -269,40 +261,40 @@ def _get_from_cache( Older Proton versions are only referred to when: digests mismatch, user interrupt, or download failure/no internet """ - path: Path = None - name: str = "" - - for tarball in cache.glob("umu-Proton*.tar.gz"): + resource: Tuple[Path, str] = None # Path to the archive and its file name + + for tarball in [ + tarball + for tarball in cache.glob("*.tar.gz") + if tarball.name.startswith("umu-proton") + or tarball.name.startswith("ULWGL-Proton") + ]: # Online if files and tarball == cache.joinpath(files[1][0]) and use_latest: - path = tarball - name = tarball.name + resource = (tarball, tarball.name) break # Offline, download interrupt, digest mismatch if not files or not use_latest: - path = tarball - name = tarball.name + resource = (tarball, tarball.name) break - if path: - proton_dir: str = name[: name.find(".tar.gz")] # Proton dir + if not resource: + return None + path, name = resource + proton: str = name[: name.find(".tar.gz")] # Proton dir + try: log.console(f"{name} found in: {path}") - try: - _extract_dir(path, steam_compat) - - log.console(f"Using {proton_dir}") - environ["PROTONPATH"] = steam_compat.joinpath(proton_dir).as_posix() - env["PROTONPATH"] = environ["PROTONPATH"] - - return env - except KeyboardInterrupt: - if steam_compat.joinpath(proton_dir).is_dir(): - log.console(f"Purging {proton_dir} in {steam_compat} ...") - rmtree(steam_compat.joinpath(proton_dir).as_posix()) - raise - - return None + _extract_dir(path, steam_compat) + log.console(f"Using {proton}") + environ["PROTONPATH"] = steam_compat.joinpath(proton).as_posix() + env["PROTONPATH"] = environ["PROTONPATH"] + return env + except KeyboardInterrupt: + if steam_compat.joinpath(proton).is_dir(): + log.console(f"Purging {proton} in {steam_compat} ...") + rmtree(steam_compat.joinpath(proton).as_posix()) + raise def _get_latest( @@ -312,37 +304,35 @@ def _get_latest( When the digests mismatched or when interrupted, refer to cache for an old version """ - if files: - log.console("Fetching latest release ...") + if not files: + return None - try: - tarball: str = files[1][0] - proton_dir: str = tarball[: tarball.find(".tar.gz")] # Proton dir - - _fetch_proton(env, steam_compat, cache, files) - - log.console(f"Using {proton_dir}") - env["PROTONPATH"] = environ["PROTONPATH"] - except ValueError: - log.exception("Exception") - tarball: str = files[1][0] - - # Digest mismatched - # Refer to the cache for old version next - # Since we do not want the user to use a suspect file, delete it - cache.joinpath(tarball).unlink(missing_ok=True) - return None - except KeyboardInterrupt: - tarball: str = files[1][0] - proton_dir: str = tarball[: tarball.find(".tar.gz")] # Proton dir - - # Exit cleanly - # Clean up extracted data and cache to prevent corruption/errors - # Refer to the cache for old version next - _cleanup(tarball, proton_dir, cache, steam_compat) - return None - except HTTPException: - # Download failed - return None + try: + log.console("Fetching latest release ...") + tarball: str = files[1][0] + proton: str = tarball[: tarball.find(".tar.gz")] + _fetch_proton(env, steam_compat, cache, files) + log.console(f"Using {proton}") + env["PROTONPATH"] = environ["PROTONPATH"] + except ValueError: + log.exception("Exception") + tarball: str = files[1][0] + + # Digest mismatched + # Refer to the cache for old version next + # Since we do not want the user to use a suspect file, delete it + cache.joinpath(tarball).unlink(missing_ok=True) + return None + except KeyboardInterrupt: + tarball: str = files[1][0] + proton_dir: str = tarball[: tarball.find(".tar.gz")] # Proton dir + + # Exit cleanly + # Clean up extracted data and cache to prevent corruption/errors + # Refer to the cache for old version next + _cleanup(tarball, proton_dir, cache, steam_compat) + return None + except HTTPException: # Download failed + return None return env diff --git a/umu/umu_run.py b/umu/umu_run.py index 4d91f9532..835b0cd42 100755 --- a/umu/umu_run.py +++ b/umu/umu_run.py @@ -9,7 +9,7 @@ from re import match from subprocess import run from umu_dl_util import get_umu_proton -from umu_consts import PROTON_VERBS, DEBUG_FORMAT, STEAM_COMPAT, umu_LOCAL +from umu_consts import PROTON_VERBS, DEBUG_FORMAT, STEAM_COMPAT, UMU_LOCAL from umu_util import setup_umu from umu_log import log, console_handler, CustomFormatter from umu_util import UnixUser @@ -27,7 +27,10 @@ def parse_args() -> Union[Namespace, Tuple[str, List[str]]]: # noqa: D103 parser.add_argument("--config", help="path to TOML file (requires Python 3.11+)") if not sys.argv[1:]: - err: str = "Please see project README.md for more info and examples.\nhttps://github.com/Open-Wine-Components/umu-launcher" + err: str = ( + "Please see project README.md for more info and examples.\n" + "https://github.com/Open-Wine-Components/umu-launcher" + ) parser.print_help(sys.stderr) raise SystemExit(err) @@ -46,21 +49,21 @@ def set_log() -> None: """Adjust the log level for the logger.""" levels: Set[str] = {"1", "warn", "debug"} - if os.environ["umu_LOG"] not in levels: + if os.environ["UMU_LOG"] not in levels: return - if os.environ["umu_LOG"] == "1": + if os.environ["UMU_LOG"] == "1": # Show the envvars and command at this level log.setLevel(level=INFO) - elif os.environ["umu_LOG"] == "warn": + elif os.environ["UMU_LOG"] == "warn": log.setLevel(level=WARNING) - elif os.environ["umu_LOG"] == "debug": + elif os.environ["UMU_LOG"] == "debug": # Show all logs console_handler.setFormatter(CustomFormatter(DEBUG_FORMAT)) log.addHandler(console_handler) log.setLevel(level=DEBUG) - os.environ.pop("umu_LOG") + os.environ.pop("UMU_LOG") def setup_pfx(path: str) -> None: @@ -192,6 +195,7 @@ def set_env( # UMU_ID env["UMU_ID"] = env["GAMEID"] + env["ULWGL_ID"] = env["UMU_ID"] # Set ULWGL_ID for compatibility env["STEAM_COMPAT_APP_ID"] = "0" if match(r"^umu-[\d\w]+$", env["UMU_ID"]): @@ -204,7 +208,7 @@ def set_env( env["PROTONPATH"] = Path(env["PROTONPATH"]).expanduser().as_posix() env["STEAM_COMPAT_DATA_PATH"] = env["WINEPREFIX"] env["STEAM_COMPAT_SHADER_PATH"] = env["STEAM_COMPAT_DATA_PATH"] + "/shadercache" - env["STEAM_COMPAT_TOOL_PATHS"] = env["PROTONPATH"] + ":" + umu_LOCAL.as_posix() + env["STEAM_COMPAT_TOOL_PATHS"] = env["PROTONPATH"] + ":" + UMU_LOCAL.as_posix() env["STEAM_COMPAT_MOUNTS"] = env["STEAM_COMPAT_TOOL_PATHS"] return env @@ -270,6 +274,7 @@ def main() -> int: # noqa: D103 "STORE": "", "PROTON_VERB": "", "UMU_ID": "", + "ULWGL_ID": "", } command: List[str] = [] opts: List[str] = None @@ -284,7 +289,7 @@ def main() -> int: # noqa: D103 err: str = "This script is not designed to run on musl-based systems" raise SystemExit(err) - if "umu_LOG" in os.environ: + if "UMU_LOG" in os.environ: set_log() log.debug("Arguments: %s", args) @@ -292,9 +297,9 @@ def main() -> int: # noqa: D103 # Setup the launcher and runtime files # An internet connection is required for new setups try: - setup_umu(root, umu_LOCAL) + setup_umu(root, UMU_LOCAL) except TimeoutError: # Request to a server timed out - if not umu_LOCAL.exists() or not any(umu_LOCAL.iterdir()): + if not UMU_LOCAL.exists() or not any(UMU_LOCAL.iterdir()): err: str = ( "umu has not been setup for the user\n" "An internet connection is required to setup umu" @@ -304,8 +309,8 @@ def main() -> int: # noqa: D103 except OSError as e: # No internet if ( e.errno == ENETUNREACH - and not umu_LOCAL.exists() - or not any(umu_LOCAL.iterdir()) + and not UMU_LOCAL.exists() + or not any(UMU_LOCAL.iterdir()) ): err: str = ( "umu has not been setup for the user\n" @@ -339,7 +344,7 @@ def main() -> int: # noqa: D103 os.environ[key] = val # Run - build_command(env, umu_LOCAL, command, opts) + build_command(env, UMU_LOCAL, command, opts) log.debug(command) return run(command).returncode @@ -358,6 +363,6 @@ def main() -> int: # noqa: D103 log.exception("Exception") sys.exit(1) finally: - umu_LOCAL.joinpath(".ref").unlink( + UMU_LOCAL.joinpath(".ref").unlink( missing_ok=True ) # Cleanup .ref file on every exit diff --git a/umu/umu_test.py b/umu/umu_test.py index 72d27048c..cddb9116f 100644 --- a/umu/umu_test.py +++ b/umu/umu_test.py @@ -54,9 +54,9 @@ def setUp(self): self.test_cache = Path("./tmp.5HYdpddgvs") # Steam compat dir self.test_compat = Path("./tmp.ZssGZoiNod") - # umu-Proton dir - self.test_proton_dir = Path("umu-Proton-5HYdpddgvs") - # umu-Proton release + # umu-proton dir + self.test_proton_dir = Path("umu-proton-5HYdpddgvs") + # umu-proton release self.test_archive = Path(self.test_cache).joinpath( f"{self.test_proton_dir}.tar.gz" ) @@ -966,7 +966,7 @@ def test_latest_offline(self): self.env, self.test_compat, self.test_cache, files ) self.assertFalse(self.env["PROTONPATH"], "Expected PROTONPATH to be empty") - self.assertTrue(result is self.env, "Expected the same reference") + self.assertFalse(result, "Expected None to be returned from _get_latest") def test_cache_interrupt(self): """Test _get_from_cache on keyboard interrupt when extracting. @@ -1117,10 +1117,9 @@ def test_steamcompat_nodir(self): continue with downloading the latest Proton """ result = None - files = [("", ""), (self.test_archive.name, "")] result = umu_dl_util._get_from_steamcompat( - self.env, self.test_compat, self.test_cache, files + self.env, self.test_compat, self.test_cache ) self.assertFalse(result, "Expected None after calling _get_from_steamcompat") @@ -1133,12 +1132,11 @@ def test_steamcompat(self): when PROTONPATH is unset """ result = None - files = [("", ""), (self.test_archive.name, "")] umu_dl_util._extract_dir(self.test_archive, self.test_compat) result = umu_dl_util._get_from_steamcompat( - self.env, self.test_compat, self.test_cache, files + self.env, self.test_compat, self.test_cache ) self.assertTrue(result is self.env, "Expected the same reference") diff --git a/umu/umu_util.py b/umu/umu_util.py index b5fddb0fd..0c98c5f75 100644 --- a/umu/umu_util.py +++ b/umu/umu_util.py @@ -1,6 +1,6 @@ from tarfile import open as tar_open, TarInfo from os import getuid -from umu_consts import CONFIG, STEAM_COMPAT, umu_LOCAL +from umu_consts import CONFIG, STEAM_COMPAT, UMU_LOCAL from typing import Any, Dict, List, Callable from json import load, dump from umu_log import log @@ -131,7 +131,7 @@ def setup_runtime(root: Path, json: Dict[str, Any]) -> None: # noqa: D103 log.warning("Archive will be extracted insecurely") # Ensure the target directory exists - umu_LOCAL.mkdir(parents=True, exist_ok=True) + UMU_LOCAL.mkdir(parents=True, exist_ok=True) # Extract the 'depot' folder to the target directory log.debug("Extracting archive files -> %s", tmp) @@ -143,12 +143,12 @@ def setup_runtime(root: Path, json: Dict[str, Any]) -> None: # noqa: D103 source_dir = tmp.joinpath("steam-container-runtime", "depot") log.debug("Source: %s", source_dir) - log.debug("Destination: %s", umu_LOCAL) + log.debug("Destination: %s", UMU_LOCAL) # Move each file to the destination directory, overwriting if it exists for file in source_dir.glob("*"): src_file: Path = source_dir.joinpath(file.name) - dest_file: Path = umu_LOCAL.joinpath(file.name) + dest_file: Path = UMU_LOCAL.joinpath(file.name) if dest_file.is_file() or dest_file.is_symlink(): log.debug("Removing file: %s", dest_file) @@ -172,7 +172,7 @@ def setup_runtime(root: Path, json: Dict[str, Any]) -> None: # noqa: D103 log.debug("Renaming: _v2-entry-point -> umu") # Rename _v2-entry-point - umu_LOCAL.joinpath("_v2-entry-point").rename(umu_LOCAL.joinpath("umu")) + UMU_LOCAL.joinpath("_v2-entry-point").rename(UMU_LOCAL.joinpath("umu")) def setup_umu(root: Path, local: Path) -> None: