Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixes after name change #67

Merged
merged 8 commits into from
Mar 21, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions umu/umu_consts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
160 changes: 75 additions & 85 deletions umu/umu_dl_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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(
Expand All @@ -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
35 changes: 20 additions & 15 deletions umu/umu_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)

Expand All @@ -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:
Expand Down Expand Up @@ -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"]):
Expand All @@ -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
Expand Down Expand Up @@ -270,6 +274,7 @@ def main() -> int: # noqa: D103
"STORE": "",
"PROTON_VERB": "",
"UMU_ID": "",
"ULWGL_ID": "",
}
command: List[str] = []
opts: List[str] = None
Expand All @@ -284,17 +289,17 @@ 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)

# 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"
Expand All @@ -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"
Expand Down Expand Up @@ -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
Expand All @@ -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
Loading