Merge aa43b017d1 into e92df8c062

docs/licenses/vdf/LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2015 Rossen Georgiev <rossen@rgp.io>
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+of the Software, and to permit persons to whom the Software is furnished to do
+so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

dzgui/api/acf.py

@@ -1,12 +1,18 @@
+import logging
 import re
+
 from collections.abc import Iterator
+from pathlib import Path
 from typing import Any
 from warnings import deprecated
 
+from dzgui.const.constants import APP_NAME
 
-@deprecated("Use dzgui.api.steam.unsubscribe()")
+logger = logging.getLogger(APP_NAME)
-class WorkshopACF:
+
-    def __init__(self, file: str) -> None:
+
+class ACF:
+    def __init__(self, file: Path) -> None:
         super().__init__()
 
         self.dict: dict[str, Any]
@@ -15,9 +21,10 @@ class WorkshopACF:
     def as_dict(self) -> dict[str, Any]:
         return self.dict
 
-    def load(self, file: str) -> None:
+    def load(self, file: Path) -> None:
         delimiter = r"\t\t"
         lines = []
+        # TODO: illegal file handling
         with open(file, "r", encoding="utf-8") as f:
             for line in f:
                 line = line.strip()
@@ -27,6 +34,14 @@ class WorkshopACF:
                 lines.append(els)
         self.dict = self.parse(iter(lines))
 
+    @classmethod
+    def enquote(cls, s: str) -> str:
+        return f'"{s}"'
+
+    @classmethod
+    def dequote(cls, s: str) -> str:
+        return s.rstrip('"').lstrip('"')
+
     def parse(self, lines: Iterator[list[str]]) -> dict[str, str]:
         acf: dict[str, Any] = {}
         try:
@@ -54,6 +69,24 @@ class WorkshopACF:
         except StopIteration:
             return acf
 
+    def get_allows_downloads(self) -> int | None:
+        try:
+            # TODO: consider coercing strs to ints on initial import
+            flag = int(self.dict["AppState"]["AllowOtherDownloadsWhileRunning"])
+            return flag
+        except Exception as e:
+            logger.critical(e)
+            return None
+
+
+@deprecated(
+    "Serializing workshop files is no longer supported, use dzgui.api.steam.unsubscribe"
+)
+class WorkshopACF(ACF):
+    def __init__(self, file: Path) -> None:
+        super().__init__(file)
+
+    # TODO: break into workshop parser and generic acf parser
     def unpack(self, d: dict, lines: list[Any] = []) -> str:
         t1 = "AppWorkshop"
         t2 = ("WorkshopItemsInstalled", "WorkshopItemDetails")
@@ -78,18 +111,9 @@ class WorkshopACF:
             s += line + "\n"
         return s
 
-    def to_file(self, file: str) -> None:
+    def to_file(self, file: Path) -> None:
         s = self.unpack(self.dict)
-        with open(file, "w") as f:
+        file.write_text(s)
-            f.write(s)
-
-    @classmethod
-    def enquote(cls, s: str) -> str:
-        return f'"{s}"'
-
-    @classmethod
-    def dequote(cls, s: str) -> str:
-        return s.rstrip('"').lstrip('"')
 
     def delete(self, modid: int) -> None:
         del self.dict["AppWorkshop"]["WorkshopItemsInstalled"][modid]

dzgui/api/mods.py

@@ -7,13 +7,12 @@ from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from pathlib import Path
 
-import dzgui.api.pefile as PeFile
+from dzgui.api.steam import get_app_path
 from dzgui.api.servers import get_rules, fqip_to_record
 from dzgui.const.constants import (
     APP_NAME,
     APPID_DAYZ,
     DAYZ_COMMUNITY_ROOT,
-    LIBRARYFOLDERS_PATH,
     WORKSHOP_PATH,
 )
 
@@ -40,7 +39,7 @@ def get_local_mod_ids(steam_path: Path) -> list[int]:
 
 
 def get_local_mod_path(steam_path: Path) -> Path:
-    p = PeFile.get_app_path(steam_path / Path(LIBRARYFOLDERS_PATH), APPID_DAYZ)
+    p = get_app_path(steam_path, APPID_DAYZ)
     workshop_path = p / WORKSHOP_PATH
     return workshop_path
 

dzgui/api/pefile.py

@@ -1,4 +1,3 @@
-import json
 import struct
 
 from dataclasses import dataclass
@@ -12,9 +11,8 @@ from dzgui.const.constants import (
     APPNAME_DAYZ,
     APPNAME_DAYZ_EXP,
     DAYZ_BINARY,
-    LIBRARYFOLDERS_PATH,
 )
-from dzgui.api.steam import vdf2json
+from dzgui.api.steam import get_app_path
 
 # https://learn.microsoft.com/en-us/windows/win32/debug/pe-format
 endian = "<"
@@ -257,25 +255,6 @@ class PeFileError(Exception):
 
     pass
 
-
-class AppNotInstalledError(Exception):
-    """App not present in user's libraryfolders"""
-
-    pass
-
-
-class AppMovedError(Exception):
-    """VDF points to a nonexistent location on disk"""
-
-    pass
-
-
-class VDFLoadError(Exception):
-    """Malformed VDF or JSON conversion"""
-
-    pass
-
-
 def parse_version_number(data: BinaryIO) -> FileVersion:
     # https://learn.microsoft.com/en-us/windows/win32/api/verrsrc/ns-verrsrc-vs_fixedfileinfo
     minor = struct.unpack("<L", data.read(4))[0] >> 16 & 0xFFFF
@@ -419,37 +398,11 @@ def get_pefile_path(steam_path: Path, appid: int) -> Path:
     name = identifier[appid]
     binary = DAYZ_BINARY
 
-    app_path = get_app_path(steam_path / LIBRARYFOLDERS_PATH, appid)
+    app_path = get_app_path(steam_path, appid)
     pe_path = app_path / f"steamapps/common/{name}/{binary}"
     return pe_path
 
 
-def get_app_path(folders_path: Path, appid: int) -> Path:
-    app_path = None
-
-    try:
-        j = json.loads(vdf2json(folders_path))
-    except Exception:
-        raise VDFLoadError("Failed to parse libraryfolders")
-
-    for obj in j["libraryfolders"]:
-        if str(appid) in j["libraryfolders"][obj]["apps"]:
-            app_path = j["libraryfolders"][obj]["path"]
-            if Path(app_path).exists():
-                break
-
-    if app_path is None:
-        raise AppNotInstalledError(
-            f"Failed to find a libraryfolder for the appid {appid}"
-        )
-    if Path(app_path).exists() is False:
-        raise AppMovedError(
-            f"The location '{app_path}' pointed to by '{appid}' no longer exists and may have been changed on the disk."
-        )
-
-    return Path(app_path)
-
-
 def get_pretty_version(steam_path: Path, appid: int) -> str | None:
     try:
         pe_file_path = get_pefile_path(steam_path, appid)

dzgui/api/steam.py

@@ -1,33 +1,59 @@
-import json
 import logging
 import os
+import psutil
 import requests
 import subprocess
-from typing import Union
+import vdf  # type: ignore
+
+from pathlib import Path
+from typing import Any, Union
 from warnings import deprecated
 
-from shlex import shlex
+from dzgui.api.acf import ACF
-from pathlib import Path
-
 from dzgui.init.prereqs import has_steam_client
 from dzgui.const.constants import (
     APPID_DAYZ,
+    APPID_DAYZ_EXP,
     APP_NAME,
     DEBIAN_STEAM_PATH,
     DEFAULT_STEAM_PATH,
     FLATPAK_STEAM_PATH,
+    LIBRARYFOLDERS_PATH,
     UBUNTU_STEAM_PATH,
     REQUEST_TIMEOUT,
     VDF_PATH,
 )
-from dzgui.const.endpoints import SUB_ENDPOINT, STEAM_PUBLISHED_FILES, UNSUB_ENDPOINT
+from dzgui.const.endpoints import (
+    APP_DETAILS,
+    SUB_ENDPOINT,
+    STEAM_PUBLISHED_FILES,
+    UNSUB_ENDPOINT,
+)
 from dzgui.strings import wizard
 from dzgui.util.bash import concat_bash_args
-
+from dzgui.util.strings import unknown
 
 logger = logging.getLogger(APP_NAME)
 
 
+class AppNotInstalledError(Exception):
+    """App not present in user's libraryfolders"""
+
+    pass
+
+
+class AppMovedError(Exception):
+    """VDF points to a nonexistent location on disk"""
+
+    pass
+
+
+class VDFLoadError(Exception):
+    """Malformed VDF or JSON conversion"""
+
+    pass
+
+
 def get_steam_paths() -> list[tuple[Path, str]]:
     paths = []
     if has_steam_client():
@@ -183,57 +209,17 @@ def launch_offline(
 def find_user_id(path: Path) -> str | None:
     resolved_path = path / "config" / "loginusers.vdf"
     try:
-        vdf = vdf2json(resolved_path)
+        with open(resolved_path, "r") as f:
-        j = json.loads(vdf)
+            v = vdf.load(f)
-        for user in j["users"]:
+            for user in v["users"]:
-            if j["users"][user]["MostRecent"] == "1":
+                if v["users"][user]["MostRecent"] == "1":
-                return str(user)
+                    return str(user)
-        return None
+            return None
     except Exception as e:
         logger.warn(e)
         return None
 
 
-def vdf2json(path: Path) -> str:
-    def _istr(indent: int, string: str) -> str:
-        return (indent * "  ") + string
-
-    jbuf = "{\n"
-    indent = 1
-
-    with open(path, "r") as f:
-        st = f.read()
-    lex = shlex(st)
-
-    while True:
-        tok = lex.get_token()
-        if not tok:
-            return jbuf + "}\n"
-        if tok == "}":
-            indent -= 1
-            jbuf += _istr(indent, "}")
-            ntok = lex.get_token()
-            if ntok is not None:
-                lex.push_token(ntok)
-            if ntok and ntok != "}":
-                jbuf += ","
-            jbuf += "\n"
-        else:
-            ntok = lex.get_token()
-            if ntok == "{":
-                jbuf += _istr(indent, tok + ": {\n")
-                indent += 1
-            else:
-                if ntok is not None:
-                    jbuf += _istr(indent, tok + ": " + ntok)
-                    ntok = lex.get_token()
-                    if ntok is not None:
-                        lex.push_token(ntok)
-                    if ntok != "}":
-                        jbuf += ","
-                    jbuf += "\n"
-
-
 def update_workshop(key: str, mod: int, endpoint: str) -> None:
     payload: dict[str, Union[int, str]] = {
         "publishedfileid": mod,
@@ -275,3 +261,152 @@ def gen_shortcut() -> None:
 def enqueue_mod(client: str, mod: str, appid: int) -> None:
     client_args = concat_bash_args(client)
     subprocess.Popen([*client_args, "+workshop_download_item", str(appid), mod])
+
+
+@deprecated("Cf. https://github.com/ValveSoftware/steam-for-linux/issues/9672")
+def get_registry() -> Any | None:
+    home = os.getenv("HOME")
+    try:
+        with open(f"{home}/.steam/registry.vdf") as f:
+            registry = vdf.load(f)
+        return registry
+    except Exception as e:
+        logger.critical(e)
+        return None
+
+
+def _is_dayz_running() -> bool:
+    registry = get_registry()
+    if registry is None:
+        return False
+    apps = registry["Registry"]["HKCU"]["Software"]["Valve"]["Steam"]["apps"].items()
+    for app in apps:
+        k, v = app
+        if k in (APPID_DAYZ, APPID_DAYZ_EXP):
+            try:
+                state = v["Running"]
+                # NOTE: 0 denotes False
+                return bool(int(state))
+            except Exception as e:
+                logger.critical(e)
+                return False
+    return False
+
+
+def _get_running_app() -> int | None:
+    registry = get_registry()
+    if registry is None:
+        return None
+    try:
+        return int(
+            registry["Registry"]["HKCU"]["Software"]["Valve"]["Steam"]["RunningAppID"]
+        )
+    except Exception:
+        return None
+
+
+# TODO: write tests
+# TODO: consider moving to proc module
+def get_running_app() -> int | None:
+    PROC_NAME = "steam"
+    SUBPROC_NAME = "reaper"
+    FLAG = "AppId"
+
+    for proc in psutil.process_iter():
+        if proc.name() == PROC_NAME:
+            subprocs = proc.children()
+            filtered = (proc for proc in subprocs if proc.name() == SUBPROC_NAME)
+            try:
+                proc = next(filtered)
+            except StopIteration:
+                return None
+            args = proc.cmdline()
+            appid = (row for row in args if FLAG in row)
+            try:
+                return int(next(appid).split("=")[1])
+            except StopIteration:
+                return None
+    return None
+
+
+# TODO: write tests
+def get_app_allows_downloads(path: Path, appid: int) -> bool:
+    root_path = get_app_path(path, appid)
+    acf = root_path.joinpath(f"steamapps/appmanifest_{appid}.acf")
+    flag = ACF(acf).get_allows_downloads()
+    match flag:
+        # NOTE: adheres to global client setting
+        case 0:
+            return get_client_allows_downloads(path)
+        # NOTE: always allow
+        case 1:
+            return True
+        # NOTE: never allow
+        case 2:
+            return False
+        # NOTE: no other known values at this time
+        case _:
+            return True
+
+
+def get_client_allows_downloads(path: Path) -> bool:
+    config = path.joinpath("config/config.vdf")
+    try:
+        with open(config) as f:
+            settings = vdf.load(f)
+            # NOTE: "1" denotes "allow"
+            allow = bool(
+                int(
+                    settings["InstallConfigStore"]["Software"]["Valve"]["Steam"][
+                        "AllowDownloadsDuringGameplay"
+                    ]
+                )
+            )
+            return allow
+    except Exception as e:
+        logger.critical(e)
+        return True
+
+
+def is_dayz_running() -> bool:
+    appid = get_running_app()
+    return appid in (APPID_DAYZ, APPID_DAYZ_EXP)
+
+
+def get_app_path(folders_path: Path, appid: int) -> Path:
+    app_path = None
+
+    try:
+        with open(folders_path / LIBRARYFOLDERS_PATH) as f:
+            folders = vdf.load(f)
+    except Exception:
+        raise VDFLoadError("Failed to parse libraryfolders")
+
+    for obj in folders["libraryfolders"]:
+        if str(appid) in folders["libraryfolders"][obj]["apps"]:
+            app_path = folders["libraryfolders"][obj]["path"]
+            if Path(app_path).exists():
+                break
+
+    if app_path is None:
+        raise AppNotInstalledError(
+            f"Failed to find a libraryfolder for the appid {appid}"
+        )
+    if Path(app_path).exists() is False:
+        raise AppMovedError(
+            f"The location '{app_path}' pointed to by '{appid}' no longer exists and may have been changed on the disk."
+        )
+
+    return Path(app_path)
+
+
+def get_app_name(appid: int) -> str:
+    payload = {"appids": [appid]}
+    res = requests.get(APP_DETAILS, params=payload)
+    if res.status_code != 200:
+        return unknown
+    try:
+        return str(res.json()[str(appid)]["data"]["name"])
+    except Exception as e:
+        logger.debug(e)
+        return unknown

dzgui/const/endpoints.py

@@ -5,6 +5,7 @@ STEAM_PUBLISHED_FILES = (
 STEAM_SERVERS = "https://api.steampowered.com/IGameServersService/GetServerList/v1"
 SUB_ENDPOINT = "https://api.steampowered.com/IPublishedFileService/Subscribe/v1"
 UNSUB_ENDPOINT = "https://api.steampowered.com/IPublishedFileService/Unsubscribe/v1"
+APP_DETAILS = "https://store.steampowered.com/api/appdetails?"
 
 BM_SERVERS = "https://api.battlemetrics.com/servers?"
 GITHUB = "https://github.com/aclist"

dzgui/init/dayz.py

@@ -2,11 +2,11 @@ import logging
 
 from pathlib import Path
 
+from dzgui.api.steam import get_app_path
 from dzgui.config.query import lookup
-from dzgui.const.constants import APPID_DAYZ, APP_NAME, LIBRARYFOLDERS_PATH
+from dzgui.const.constants import APPID_DAYZ, APP_NAME
 from dzgui.const.enum import Preferences
 
-import dzgui.api.pefile as PeFile
 
 logger = logging.getLogger(APP_NAME)
 
@@ -14,7 +14,7 @@ logger = logging.getLogger(APP_NAME)
 def is_dayz_installed(config: Path) -> None:
     try:
         path = lookup(config, Preferences.DEFAULT)
-        PeFile.get_app_path(Path(path) / LIBRARYFOLDERS_PATH, APPID_DAYZ)
+        get_app_path(Path(path), APPID_DAYZ)
     except Exception as e:
         logger.critical(e)
         raise e

dzgui/init/proc.py

@@ -3,11 +3,8 @@ import subprocess
 import shutil
 import logging
 
-from warnings import deprecated
-
 from dzgui.const.constants import (
     APP_NAME,
-    DAYZ_BINARY,
     STEAM_CMD,
     FLATPAK_APPID,
     FLATPAK_CMD,
@@ -15,26 +12,9 @@ from dzgui.const.constants import (
     FLATPAK_SANDBOX,
 )
 
-from dzgui.util.format import format_exception
-
 logger = logging.getLogger(APP_NAME)
 
 
-# TODO: move to util.proc
-def is_dayz_running() -> bool:
-    """Subprocesses spawned from Steam will not show up in regular process tree"""
-    procs = []
-    substring = DAYZ_BINARY
-    for proc in psutil.process_iter():
-        try:
-            procs.append(proc.cmdline())
-        except Exception as e:
-            msg = format_exception(e)
-            logger.warning(msg)
-            continue
-    return any(substring in item for sublist in procs for item in sublist)
-
-
 def is_steam_running(cmd: str) -> bool:
     if cmd == STEAM_CMD:
         if has_cmd(STEAM_CMD) is False:
@@ -69,25 +49,3 @@ def has_cmd(cmd: str) -> bool:
     if shutil.which(cmd) is not None:
         return True
     return False
-
-
-@deprecated("dropped in favor of Gtk native methods")
-def foreground(cmd: str, pid: int) -> None:
-    if cmd == "wmctrl":
-        proc = subprocess.run(["wmctrl", "-ilp"], capture_output=True, text=True)
-        lines = proc.stdout.splitlines()
-        for line in lines:
-            els = line.split(" ")
-            if str(pid) in els:
-                wid = els[0]
-                break
-        subprocess.run(["wmctrl", "-ia", wid])
-    elif cmd == "xdotool":
-        args = [cmd, "search", "--pid", str(pid)]
-        proc = subprocess.run([*args], capture_output=True, text=True)
-        lines = proc.stdout.splitlines()
-        ## NOTE: some forked subprocesses may fail, so skip over them
-        for line in lines:
-            subprocess.run(
-                ["xdotool", "windowactivate", line], stderr=subprocess.DEVNULL
-            )

dzgui/managers/connection.py

@@ -32,8 +32,9 @@ from dzgui.const.constants import (
     APPNAME_DAYZ,
     APPNAME_DAYZ_EXP_HUMAN,
 )
+from dzgui.api.steam import is_dayz_running
 from dzgui.const.enum import NotebookPage, Preferences
-from dzgui.init.proc import is_dayz_running, is_steam_running
+from dzgui.init.proc import is_steam_running
 from dzgui.managers.threading import call_on_thread, StoredFunc, ThreadingManager
 from dzgui.strings.dialogs import (
     waiting_for_launch,

dzgui/managers/mods.py

@@ -60,7 +60,7 @@ class ModManager:
 
     @call_on_thread(dialogs.fetching_mods)
     def load_mods(self) -> None:
-        mods = get_delimited_mods(self.path)
+        mods = get_delimited_mods(Path(self.path))
         if len(mods) < 1:
             msg = self.format_mod_statusbar()
             func = StoredFunc(lambda: self.emitter.emit("mods_updated", msg, 0))
@@ -141,7 +141,7 @@ class ModManager:
             app_path_exp = PeFile.get_nested_app_path(steam_path, APPID_DAYZ_EXP)
             symlink = app_path_exp / md5
             symlink.unlink()
-        except PeFile.AppNotInstalledError:
+        except Exception:
             pass
         time.sleep(API_RATE_LIMIT)
 

dzgui/managers/offline.py

@@ -5,10 +5,9 @@ from typing import Callable, TYPE_CHECKING, Union
 
 import dzgui.api.pefile as PeFile
 from dzgui.api.mods import is_mission, get_custom_mods
-from dzgui.api.steam import launch_offline
+from dzgui.api.steam import is_dayz_running, launch_offline
 from dzgui.const.constants import APP_NAME, APPID_DAYZ_EXP
 from dzgui.const.enum import Preferences
-from dzgui.init.proc import is_dayz_running
 from dzgui.managers.threading import call_on_thread, StoredFunc, ThreadingManager
 from dzgui.model.model_factory import ModelFactory
 from dzgui.strings import dialogs

pyproject.toml

@@ -27,6 +27,7 @@ dependencies = [
     "psutil==7.1.3",
     "python-a2s",
     "requests==2.32.5",
+    "vdf==3.4"
 ]
 
 [project.scripts]

tests/test_bare_conf_files.py

@@ -1,6 +1,5 @@
 import pytest
 import tempfile
-import os
 
 from dzgui.app_init import copy_bare_configs
 from dzgui.config.xdg import get_xdg_paths, parse_filepaths

tests/test_pefile.py

@@ -2,7 +2,7 @@ import pytest
 from pathlib import Path
 
 import dzgui.api.pefile as PeFile
-from dzgui.api.pefile import VDFLoadError, AppNotInstalledError, AppMovedError
+from dzgui.api.steam import VDFLoadError, AppNotInstalledError, AppMovedError, get_app_path
 
 from dzgui.config.query import get_config
 from dzgui.config.xdg import get_xdg_paths, parse_filepaths
@@ -56,11 +56,11 @@ def test_not_in_library(fixture, exception):
     fixture = fixture_path(fixture)
     with pytest.raises(exception):
         try:
-            PeFile.get_app_path(fixture, APPID_DAYZ)
+            get_app_path(fixture, APPID_DAYZ)
         except Exception as e:
             raise e
 
 
 def test_on_second_drive(second_drive):
-    path = PeFile.get_app_path(second_drive, APPID_DAYZ)
+    path = get_app_path(second_drive, APPID_DAYZ)
     assert path == Path("/tmp")