feat: Steam game & library detection (M6) — 0.8.0
The first slice of M6 (gaming-environment checks): detect a user's Steam libraries and the games installed in each — also the D12 "pick a game" foundation. - core/steam.py: multi-install/library discovery (libraryfolders.vdf, symlink dedupe, native/Flatpak/Snap), appmanifest_*.acf scan with runtime/Proton/ redist filtering, scan cache + new-game diff. Stdlib only. VDF keys read case-insensitively (e.g. lastupdated vs SizeOnDisk). - Libraries are opt-in (config steam_libraries); the flat TOML writer now emits list/array values. - GUI Games page: library checkboxes with per-library counts, game list, background rescan on every launch, NEW badge + sidebar count for games installed since the last scan (acknowledged when viewed). - CLI: rigdoctor games / games libraries [--enable|--disable|--all|--json] (headless-complete, D17). - Tests for VDF parse, scan, tool filter, cache diff, config list round-trip. - Docs (MODULES/ROADMAP) updated; version 0.7.3 -> 0.8.0. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,3 @@
|
||||
"""RigDoctor — modular hardware monitoring & crash diagnostics for Linux gamers."""
|
||||
|
||||
__version__ = "0.7.3"
|
||||
__version__ = "0.8.0"
|
||||
|
||||
@@ -345,6 +345,73 @@ def cmd_report(args) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_games(args) -> int:
|
||||
from .core import steam
|
||||
|
||||
selected = steam.selected_library_paths()
|
||||
if not selected:
|
||||
print("No Steam libraries selected to scan.")
|
||||
print(" See them with: rigdoctor games libraries")
|
||||
print(" Then enable one: rigdoctor games libraries --enable <path> (or --all)")
|
||||
return 1
|
||||
result = steam.rescan()
|
||||
if args.json:
|
||||
from dataclasses import asdict
|
||||
|
||||
print(json.dumps({
|
||||
"scanned_at": result.scanned_at,
|
||||
"new_appids": result.new_appids,
|
||||
"games": [asdict(g) for g in result.games],
|
||||
}, indent=2, ensure_ascii=False))
|
||||
return 0
|
||||
if not result.games:
|
||||
print("No games found in the selected Steam libraries.")
|
||||
return 0
|
||||
new = set(result.new_appids)
|
||||
print(f"{len(result.games)} game(s) across {len(selected)} librar(y/ies):\n")
|
||||
for g in result.games:
|
||||
flag = " NEW" if g.appid in new else ""
|
||||
print(f" {g.name:<48} {steam.human_size(g.size_bytes):>9}{flag}")
|
||||
if new:
|
||||
print(f"\n{len(new)} newly-installed since the last scan.")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_games_libraries(args) -> int:
|
||||
from .core import steam
|
||||
|
||||
discovered = steam.discover_libraries()
|
||||
selected = {os.path.realpath(p) for p in steam.selected_library_paths()}
|
||||
|
||||
# --all / --enable / --disable adjust the selection, then we list the result.
|
||||
if args.all or args.enable or args.disable:
|
||||
if args.all:
|
||||
selected = {lib.path for lib in discovered}
|
||||
for raw in args.enable or []:
|
||||
selected.add(os.path.realpath(os.path.expanduser(raw)))
|
||||
for raw in args.disable or []:
|
||||
selected.discard(os.path.realpath(os.path.expanduser(raw)))
|
||||
config.update_config(steam_libraries=sorted(selected))
|
||||
|
||||
if not discovered:
|
||||
print("No Steam libraries detected (is Steam installed?).")
|
||||
return 1
|
||||
if args.json:
|
||||
print(json.dumps([
|
||||
{"path": lib.path, "label": lib.label, "selected": lib.path in selected,
|
||||
"games": len(steam.scan_library(lib.path))}
|
||||
for lib in discovered
|
||||
], indent=2, ensure_ascii=False))
|
||||
return 0
|
||||
print("Steam libraries (checked = scanned for games):\n")
|
||||
for lib in discovered:
|
||||
mark = "x" if lib.path in selected else " "
|
||||
count = len(steam.scan_library(lib.path))
|
||||
label = f" [{lib.label}]" if lib.label else ""
|
||||
print(f" [{mark}] {lib.path}{label} ({count} games)")
|
||||
return 0
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
p = argparse.ArgumentParser(
|
||||
prog="rigdoctor",
|
||||
@@ -423,6 +490,17 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
inv.add_argument("--markdown", action="store_true", help="output Markdown (for forum/bug reports)")
|
||||
inv.add_argument("-o", "--output", default=None, help="write to a file instead of stdout")
|
||||
inv.set_defaults(func=cmd_inventory)
|
||||
|
||||
games_p = sub.add_parser("games", help="Steam game & library detection (M6)")
|
||||
games_p.add_argument("--json", action="store_true", help="output JSON")
|
||||
games_p.set_defaults(func=cmd_games)
|
||||
games_sub = games_p.add_subparsers(dest="games_cmd")
|
||||
lib_p = games_sub.add_parser("libraries", help="list/select Steam libraries to scan")
|
||||
lib_p.add_argument("--enable", action="append", metavar="PATH", help="scan this library (repeatable)")
|
||||
lib_p.add_argument("--disable", action="append", metavar="PATH", help="stop scanning this library (repeatable)")
|
||||
lib_p.add_argument("--all", action="store_true", help="scan all detected libraries")
|
||||
lib_p.add_argument("--json", action="store_true", help="output JSON")
|
||||
lib_p.set_defaults(func=cmd_games_libraries)
|
||||
return p
|
||||
|
||||
|
||||
|
||||
@@ -27,6 +27,10 @@ STATUS_FILE = STATE_DIR / "recorder.json"
|
||||
PID_FILE = STATE_DIR / "recorder.pid"
|
||||
SPAWN_LOG = STATE_DIR / "recorder.out"
|
||||
|
||||
# Gaming environment / game detection (M6) — cached Steam game scan (mutable state,
|
||||
# not config: refreshed by the background scan on every launch).
|
||||
GAMES_FILE = STATE_DIR / "games.json"
|
||||
|
||||
# Update access token (M13) — gates updates to Gitea account holders (D18).
|
||||
# Stored in the OS keyring (Secret Service / GNOME Keyring) via `secret-tool` when
|
||||
# available — encrypted at rest, unlocked with the login session — else a 0600 file.
|
||||
@@ -143,6 +147,7 @@ DEFAULTS: dict = {
|
||||
"gpu_temp_alert": 90.0, # °C — alert when GPU reaches this
|
||||
"cpu_temp_alert": 95.0, # °C — alert when CPU reaches this
|
||||
"relay_url": "wss://rigdoctor.jesseyvanofferen.com", # session-sharing relay (M12)
|
||||
"steam_libraries": [], # Steam library paths to scan for games (M6); empty = none picked yet
|
||||
}
|
||||
|
||||
|
||||
@@ -165,6 +170,8 @@ def _toml_value(value) -> str:
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, (int, float)):
|
||||
return repr(value)
|
||||
if isinstance(value, (list, tuple)):
|
||||
return "[" + ", ".join(_toml_value(v) for v in value) + "]"
|
||||
return '"' + str(value).replace("\\", "\\\\").replace('"', '\\"') + '"'
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,340 @@
|
||||
"""Steam library & game detection (M6, the Steam piece of D12 game detection).
|
||||
|
||||
Discovers a user's Steam installs, the library folders they've configured (Steam tracks
|
||||
them all in ``libraryfolders.vdf``, so multiple libraries on multiple drives are covered),
|
||||
and the games installed in each (one ``appmanifest_<appid>.acf`` per app). Stdlib only —
|
||||
no Steam tooling required, every probe degrades gracefully.
|
||||
|
||||
The set of libraries actually scanned is user-chosen (config ``steam_libraries``); nothing
|
||||
is scanned until the user opts a library in. Scan results are cached in ``games.json`` so the
|
||||
GUI can show the list instantly and the launch-time background scan can diff against it to
|
||||
flag newly-installed games.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from ..config import GAMES_FILE, load_config
|
||||
|
||||
# Steam "apps" that aren't games: runtimes, Proton builds, redistributables. Filtered out of
|
||||
# scans by appid (known IDs) or by name prefix (covers future Proton/runtime versions).
|
||||
_TOOL_APPIDS = {
|
||||
"228980", # Steamworks Common Redistributables
|
||||
"1070560", # Steam Linux Runtime 1.0 (scout)
|
||||
"1391110", # Steam Linux Runtime 2.0 (soldier)
|
||||
"1628350", # Steam Linux Runtime 3.0 (sniper)
|
||||
"1493710", # Proton Experimental
|
||||
"2180100", # Proton Hotfix
|
||||
"1826330", # Proton EasyAntiCheat Runtime
|
||||
"1161040", # Proton BattlEye Runtime
|
||||
}
|
||||
_TOOL_NAME_PREFIXES = ("Proton", "Steam Linux Runtime", "Steamworks Common")
|
||||
|
||||
# Where Steam may be installed (native + Flatpak + Snap). Symlinks (~/.steam/steam) are
|
||||
# resolved and de-duplicated by real path.
|
||||
_ROOT_CANDIDATES = (
|
||||
"~/.steam/steam",
|
||||
"~/.steam/root",
|
||||
"~/.local/share/Steam",
|
||||
"~/.var/app/com.valvesoftware.Steam/data/Steam", # Flatpak
|
||||
"~/snap/steam/common/.local/share/Steam", # Snap
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class SteamLibrary:
|
||||
path: str # the library root (contains a steamapps/ dir)
|
||||
label: str = "" # Steam's label for the folder, if any
|
||||
|
||||
|
||||
@dataclass
|
||||
class Game:
|
||||
appid: str
|
||||
name: str
|
||||
library: str # library path the game lives in
|
||||
installdir: str # folder name under <library>/steamapps/common
|
||||
size_bytes: int = 0
|
||||
last_updated: int = 0 # epoch seconds (acf LastUpdated), 0 if unknown
|
||||
|
||||
|
||||
# --- VDF (Valve Data Format) parsing --------------------------------------------------
|
||||
# Minimal text-VDF reader: quoted "key" "value" pairs and "key" { ... } nesting. Enough
|
||||
# for libraryfolders.vdf and appmanifest_*.acf; ignores #base/#include and unquoted tokens.
|
||||
|
||||
def _parse_vdf(text: str) -> dict:
|
||||
pos = 0
|
||||
n = len(text)
|
||||
|
||||
def skip_ws() -> None:
|
||||
nonlocal pos
|
||||
while pos < n:
|
||||
c = text[pos]
|
||||
if c in " \t\r\n":
|
||||
pos += 1
|
||||
elif c == "/" and pos + 1 < n and text[pos + 1] == "/": # // line comment
|
||||
while pos < n and text[pos] != "\n":
|
||||
pos += 1
|
||||
else:
|
||||
break
|
||||
|
||||
def read_string() -> str:
|
||||
nonlocal pos
|
||||
pos += 1 # opening quote
|
||||
out = []
|
||||
while pos < n:
|
||||
c = text[pos]
|
||||
if c == "\\" and pos + 1 < n:
|
||||
nxt = text[pos + 1]
|
||||
out.append({"n": "\n", "t": "\t", "\\": "\\", '"': '"'}.get(nxt, nxt))
|
||||
pos += 2
|
||||
continue
|
||||
if c == '"':
|
||||
pos += 1
|
||||
break
|
||||
out.append(c)
|
||||
pos += 1
|
||||
return "".join(out)
|
||||
|
||||
def parse_obj() -> dict:
|
||||
nonlocal pos
|
||||
obj: dict = {}
|
||||
while True:
|
||||
skip_ws()
|
||||
if pos >= n or text[pos] == "}":
|
||||
pos += 1 # consume closing brace (or run off the end)
|
||||
return obj
|
||||
if text[pos] != '"': # skip unquoted/unsupported tokens defensively
|
||||
pos += 1
|
||||
continue
|
||||
key = read_string()
|
||||
skip_ws()
|
||||
if pos < n and text[pos] == "{":
|
||||
pos += 1
|
||||
obj[key] = parse_obj()
|
||||
elif pos < n and text[pos] == '"':
|
||||
obj[key] = read_string()
|
||||
else: # malformed; bail on this key
|
||||
obj[key] = ""
|
||||
return obj
|
||||
|
||||
skip_ws()
|
||||
if pos < n and text[pos] == '"':
|
||||
root_key = read_string()
|
||||
skip_ws()
|
||||
if pos < n and text[pos] == "{":
|
||||
pos += 1
|
||||
return {root_key: parse_obj()}
|
||||
return {}
|
||||
|
||||
|
||||
def _read_vdf(path: Path) -> dict:
|
||||
try:
|
||||
return _parse_vdf(path.read_text(encoding="utf-8", errors="replace"))
|
||||
except OSError:
|
||||
return {}
|
||||
|
||||
|
||||
# --- discovery ------------------------------------------------------------------------
|
||||
|
||||
def steam_roots() -> list[Path]:
|
||||
"""Existing Steam install roots, de-duplicated by resolved path."""
|
||||
seen: set[Path] = set()
|
||||
roots: list[Path] = []
|
||||
for cand in _ROOT_CANDIDATES:
|
||||
p = Path(os.path.expanduser(cand))
|
||||
if not p.exists():
|
||||
continue
|
||||
real = p.resolve()
|
||||
if real in seen:
|
||||
continue
|
||||
seen.add(real)
|
||||
roots.append(real)
|
||||
return roots
|
||||
|
||||
|
||||
def _libraryfolders_vdf(root: Path) -> Path | None:
|
||||
for rel in ("steamapps/libraryfolders.vdf", "config/libraryfolders.vdf"):
|
||||
p = root / rel
|
||||
if p.exists():
|
||||
return p
|
||||
return None
|
||||
|
||||
|
||||
def discover_libraries() -> list[SteamLibrary]:
|
||||
"""Every Steam library folder configured on this machine, de-duplicated by real path.
|
||||
|
||||
Reads each install's ``libraryfolders.vdf`` (which lists all drives/folders), and
|
||||
always includes the install root itself as a fallback.
|
||||
"""
|
||||
seen: set[Path] = set()
|
||||
libs: list[SteamLibrary] = []
|
||||
|
||||
def add(path: Path, label: str = "") -> None:
|
||||
if not (path / "steamapps").is_dir():
|
||||
return
|
||||
real = path.resolve()
|
||||
if real in seen:
|
||||
return
|
||||
seen.add(real)
|
||||
libs.append(SteamLibrary(path=str(real), label=label))
|
||||
|
||||
for root in steam_roots():
|
||||
vdf = _libraryfolders_vdf(root)
|
||||
folders = _read_vdf(vdf).get("libraryfolders", {}) if vdf else {}
|
||||
if isinstance(folders, dict):
|
||||
for entry in folders.values():
|
||||
if isinstance(entry, dict) and entry.get("path"):
|
||||
add(Path(entry["path"]), entry.get("label", ""))
|
||||
add(root) # the install root is itself a library
|
||||
return libs
|
||||
|
||||
|
||||
# --- game scanning --------------------------------------------------------------------
|
||||
|
||||
def is_tool(appid: str, name: str) -> bool:
|
||||
"""True for non-game Steam apps (runtimes, Proton, redistributables)."""
|
||||
if appid in _TOOL_APPIDS:
|
||||
return True
|
||||
return name.startswith(_TOOL_NAME_PREFIXES)
|
||||
|
||||
|
||||
def scan_library(library: str) -> list[Game]:
|
||||
"""Games installed in one library, parsed from its appmanifest_*.acf files."""
|
||||
steamapps = Path(library) / "steamapps"
|
||||
games: list[Game] = []
|
||||
try:
|
||||
manifests = sorted(steamapps.glob("appmanifest_*.acf"))
|
||||
except OSError:
|
||||
return games
|
||||
for manifest in manifests:
|
||||
state = _read_vdf(manifest).get("AppState", {})
|
||||
if not isinstance(state, dict):
|
||||
continue
|
||||
# Steam treats VDF keys case-insensitively (e.g. "SizeOnDisk" but "lastupdated").
|
||||
state = {k.lower(): v for k, v in state.items()}
|
||||
appid = state.get("appid", "")
|
||||
name = state.get("name", "").strip()
|
||||
if not appid or not name or is_tool(appid, name):
|
||||
continue
|
||||
games.append(Game(
|
||||
appid=appid,
|
||||
name=name,
|
||||
library=str(library),
|
||||
installdir=state.get("installdir", ""),
|
||||
size_bytes=_int(state.get("sizeondisk")),
|
||||
last_updated=_int(state.get("lastupdated")),
|
||||
))
|
||||
return games
|
||||
|
||||
|
||||
def scan_games(libraries: list[str]) -> list[Game]:
|
||||
"""All games across the given libraries, de-duplicated by appid, sorted by name."""
|
||||
by_appid: dict[str, Game] = {}
|
||||
for lib in libraries:
|
||||
for game in scan_library(lib):
|
||||
by_appid.setdefault(game.appid, game) # first library wins on duplicates
|
||||
return sorted(by_appid.values(), key=lambda g: g.name.lower())
|
||||
|
||||
|
||||
def _int(value) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
|
||||
# --- config-driven selection ----------------------------------------------------------
|
||||
|
||||
def selected_library_paths(cfg: dict | None = None) -> list[str]:
|
||||
"""Library paths the user has opted in to scanning (config ``steam_libraries``)."""
|
||||
cfg = cfg or load_config()
|
||||
paths = cfg.get("steam_libraries") or []
|
||||
return [str(p) for p in paths]
|
||||
|
||||
|
||||
# --- scan cache + new-game detection --------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class ScanResult:
|
||||
games: list[Game]
|
||||
new_appids: list[str] # newly-installed since the last scan (badge fuel)
|
||||
scanned_at: float
|
||||
|
||||
|
||||
def load_cache() -> dict | None:
|
||||
try:
|
||||
return json.loads(GAMES_FILE.read_text())
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _save_cache(games: list[Game], known: set[str], new: list[str], when: float) -> None:
|
||||
GAMES_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
data = {
|
||||
"scanned_at": when,
|
||||
"known_appids": sorted(known),
|
||||
"new_appids": new,
|
||||
"games": [asdict(g) for g in games],
|
||||
}
|
||||
GAMES_FILE.write_text(json.dumps(data, indent=2, ensure_ascii=False))
|
||||
|
||||
|
||||
def cached_games() -> list[Game]:
|
||||
"""Games from the last scan (for instant display before a rescan finishes)."""
|
||||
cache = load_cache()
|
||||
if not cache:
|
||||
return []
|
||||
return [Game(**{k: g.get(k) for k in Game.__dataclass_fields__}) for g in cache.get("games", [])]
|
||||
|
||||
|
||||
def rescan(cfg: dict | None = None) -> ScanResult:
|
||||
"""Scan the selected libraries, diff against the cache, and persist the result.
|
||||
|
||||
Newly-installed games (appids never seen before) are reported in ``new_appids``. The
|
||||
very first scan reports nothing as new (so the whole library isn't flagged at once);
|
||||
unacknowledged new games carry forward until they're acknowledged or uninstalled.
|
||||
"""
|
||||
games = scan_games(selected_library_paths(cfg))
|
||||
current = {g.appid for g in games}
|
||||
prev = load_cache()
|
||||
if prev is None:
|
||||
known: set[str] = set(current) # first run: everything is "known", nothing new
|
||||
new = []
|
||||
else:
|
||||
known = set(prev.get("known_appids", []))
|
||||
carried = set(prev.get("new_appids", [])) & current # still-unacknowledged & installed
|
||||
new = sorted((current - known) | carried)
|
||||
known |= current
|
||||
when = time.time()
|
||||
_save_cache(games, known, new, when)
|
||||
return ScanResult(games=games, new_appids=new, scanned_at=when)
|
||||
|
||||
|
||||
def acknowledge_new() -> None:
|
||||
"""Clear the new-game badge (called when the user views the games list)."""
|
||||
cache = load_cache()
|
||||
if not cache or not cache.get("new_appids"):
|
||||
return
|
||||
cache["new_appids"] = []
|
||||
try:
|
||||
GAMES_FILE.write_text(json.dumps(cache, indent=2, ensure_ascii=False))
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
# --- formatting -----------------------------------------------------------------------
|
||||
|
||||
def human_size(num_bytes: int) -> str:
|
||||
if num_bytes <= 0:
|
||||
return "—"
|
||||
size = float(num_bytes)
|
||||
for unit in ("B", "KB", "MB", "GB", "TB"):
|
||||
if size < 1024 or unit == "TB":
|
||||
return f"{size:.0f} {unit}" if unit in ("B", "KB") else f"{size:.1f} {unit}"
|
||||
size /= 1024
|
||||
return f"{size:.1f} TB"
|
||||
@@ -0,0 +1,249 @@
|
||||
"""Games page (M6 in the GUI): pick Steam libraries and browse detected games.
|
||||
|
||||
Libraries are opt-in — the user checks which ones to scan. The list is loaded from the
|
||||
cache instantly, then a background rescan refreshes it and flags games installed since the
|
||||
last scan (a "NEW" badge here + a count on the sidebar nav).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
|
||||
from PySide6.QtCore import Qt, QTimer, Signal
|
||||
from PySide6.QtWidgets import (
|
||||
QCheckBox,
|
||||
QFrame,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QPushButton,
|
||||
QScrollArea,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from ..config import load_config, update_config
|
||||
from .theme import ACCENT, GOOD, MUTED
|
||||
|
||||
|
||||
def _game_row(name: str, sublabel: str, size: str, is_new: bool) -> QFrame:
|
||||
card = QFrame()
|
||||
card.setObjectName("Card")
|
||||
h = QHBoxLayout(card)
|
||||
h.setContentsMargins(16, 10, 16, 10)
|
||||
h.setSpacing(10)
|
||||
|
||||
left = QVBoxLayout()
|
||||
left.setSpacing(2)
|
||||
title = QLabel(name)
|
||||
title.setStyleSheet("font-weight: 600; background: transparent;")
|
||||
title.setWordWrap(True)
|
||||
left.addWidget(title)
|
||||
if sublabel:
|
||||
sub = QLabel(sublabel)
|
||||
sub.setObjectName("Muted")
|
||||
left.addWidget(sub)
|
||||
h.addLayout(left, 1)
|
||||
|
||||
if is_new:
|
||||
badge = QLabel("NEW")
|
||||
badge.setStyleSheet(
|
||||
f"color: {GOOD}; border: 1px solid {GOOD}; border-radius: 6px; "
|
||||
f"padding: 1px 6px; font-weight: 700; background: transparent;"
|
||||
)
|
||||
h.addWidget(badge, 0, Qt.AlignmentFlag.AlignVCenter)
|
||||
|
||||
size_label = QLabel(size)
|
||||
size_label.setObjectName("Muted")
|
||||
size_label.setMinimumWidth(80)
|
||||
size_label.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
|
||||
h.addWidget(size_label, 0)
|
||||
return card
|
||||
|
||||
|
||||
class GamesPage(QWidget):
|
||||
_libraries_ready = Signal(object) # list[dict(path, label, count, selected)]
|
||||
_scanned = Signal(object) # steam.ScanResult
|
||||
new_count_changed = Signal(int) # newly-installed game count (for the nav badge)
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.setObjectName("Page")
|
||||
self._libraries_ready.connect(self._render_libraries)
|
||||
self._scanned.connect(self._render_games)
|
||||
self._busy = False
|
||||
self._new_appids: set[str] = set()
|
||||
|
||||
root = QVBoxLayout(self)
|
||||
root.setContentsMargins(20, 18, 20, 18)
|
||||
root.setSpacing(16)
|
||||
|
||||
header = QHBoxLayout()
|
||||
title = QLabel("Games")
|
||||
title.setObjectName("PageTitle")
|
||||
header.addWidget(title)
|
||||
header.addStretch(1)
|
||||
self._status = QLabel("")
|
||||
self._status.setObjectName("Muted")
|
||||
header.addWidget(self._status)
|
||||
self._rescan_btn = QPushButton("Rescan")
|
||||
self._rescan_btn.setObjectName("PrimaryButton")
|
||||
self._rescan_btn.clicked.connect(self.refresh)
|
||||
header.addWidget(self._rescan_btn)
|
||||
root.addLayout(header)
|
||||
|
||||
# Libraries (opt-in checkboxes)
|
||||
lib_card = QFrame()
|
||||
lib_card.setObjectName("Card")
|
||||
lib_v = QVBoxLayout(lib_card)
|
||||
lib_v.setContentsMargins(16, 12, 16, 12)
|
||||
lib_v.setSpacing(6)
|
||||
lib_head = QLabel("Steam libraries")
|
||||
lib_head.setStyleSheet("font-weight: 700; background: transparent;")
|
||||
lib_v.addWidget(lib_head)
|
||||
self._lib_box = QVBoxLayout()
|
||||
self._lib_box.setSpacing(6)
|
||||
lib_v.addLayout(self._lib_box)
|
||||
self._lib_hint = QLabel("Looking for Steam libraries…")
|
||||
self._lib_hint.setObjectName("Muted")
|
||||
self._lib_hint.setWordWrap(True)
|
||||
lib_v.addWidget(self._lib_hint)
|
||||
root.addWidget(lib_card)
|
||||
|
||||
# Games list
|
||||
scroll = QScrollArea()
|
||||
scroll.setWidgetResizable(True)
|
||||
scroll.setFrameShape(QFrame.Shape.NoFrame)
|
||||
scroll.setStyleSheet("background: transparent;")
|
||||
self._container = QWidget()
|
||||
self._list = QVBoxLayout(self._container)
|
||||
self._list.setContentsMargins(0, 0, 0, 0)
|
||||
self._list.setSpacing(8)
|
||||
self._list.setAlignment(Qt.AlignmentFlag.AlignTop)
|
||||
scroll.setWidget(self._container)
|
||||
root.addWidget(scroll, 1)
|
||||
|
||||
self._load_cached() # instant display from the last scan
|
||||
QTimer.singleShot(400, self.refresh) # then rescan in the background on launch
|
||||
|
||||
# --- loading ----------------------------------------------------------------------
|
||||
|
||||
def _load_cached(self) -> None:
|
||||
from ..core import steam
|
||||
|
||||
cache = steam.load_cache() or {}
|
||||
self._new_appids = set(cache.get("new_appids", []))
|
||||
games = steam.cached_games()
|
||||
if games:
|
||||
self._populate_games(games, self._new_appids)
|
||||
self.new_count_changed.emit(len(self._new_appids))
|
||||
|
||||
def refresh(self) -> None:
|
||||
if self._busy:
|
||||
return
|
||||
self._busy = True
|
||||
self._rescan_btn.setEnabled(False)
|
||||
self._status.setText("Scanning Steam libraries…")
|
||||
threading.Thread(target=self._work, daemon=True).start()
|
||||
|
||||
def _work(self) -> None:
|
||||
from ..core import steam
|
||||
|
||||
try:
|
||||
selected = {os.path.realpath(p) for p in steam.selected_library_paths()}
|
||||
libs = [
|
||||
{"path": lib.path, "label": lib.label, "selected": lib.path in selected,
|
||||
"count": len(steam.scan_library(lib.path))}
|
||||
for lib in steam.discover_libraries()
|
||||
]
|
||||
self._libraries_ready.emit(libs)
|
||||
self._scanned.emit(steam.rescan())
|
||||
except Exception:
|
||||
self._scanned.emit(None)
|
||||
|
||||
# --- rendering --------------------------------------------------------------------
|
||||
|
||||
def _render_libraries(self, libs) -> None:
|
||||
while self._lib_box.count():
|
||||
item = self._lib_box.takeAt(0)
|
||||
w = item.widget()
|
||||
if w is not None:
|
||||
w.deleteLater()
|
||||
if not libs:
|
||||
self._lib_hint.setText("No Steam libraries detected. Is Steam installed?")
|
||||
self._lib_hint.show()
|
||||
return
|
||||
self._lib_hint.hide()
|
||||
for lib in libs:
|
||||
label = lib["path"]
|
||||
if lib["label"]:
|
||||
label += f" [{lib['label']}]"
|
||||
cb = QCheckBox(f"{label} · {lib['count']} games")
|
||||
cb.setChecked(lib["selected"])
|
||||
cb.toggled.connect(lambda checked, p=lib["path"]: self._toggle_library(p, checked))
|
||||
self._lib_box.addWidget(cb)
|
||||
|
||||
def _toggle_library(self, path: str, checked: bool) -> None:
|
||||
selected = {os.path.realpath(p) for p in (load_config().get("steam_libraries") or [])}
|
||||
if checked:
|
||||
selected.add(os.path.realpath(path))
|
||||
else:
|
||||
selected.discard(os.path.realpath(path))
|
||||
update_config(steam_libraries=sorted(selected))
|
||||
self.refresh()
|
||||
|
||||
def _render_games(self, result) -> None:
|
||||
self._busy = False
|
||||
self._rescan_btn.setEnabled(True)
|
||||
if result is None:
|
||||
self._status.setText("scan failed")
|
||||
return
|
||||
self._new_appids = set(result.new_appids)
|
||||
self._populate_games(result.games, self._new_appids)
|
||||
new = len(self._new_appids)
|
||||
suffix = f" · {new} new" if new else ""
|
||||
self._status.setText(
|
||||
f"{len(result.games)} games · {time.strftime('%H:%M:%S')}{suffix}"
|
||||
)
|
||||
self.new_count_changed.emit(new)
|
||||
|
||||
def _populate_games(self, games, new_appids: set[str]) -> None:
|
||||
from ..core import steam
|
||||
|
||||
while self._list.count():
|
||||
item = self._list.takeAt(0)
|
||||
w = item.widget()
|
||||
if w is not None:
|
||||
w.deleteLater()
|
||||
|
||||
if not games:
|
||||
empty = QLabel(
|
||||
"No games to show yet — check a Steam library above to scan it for games."
|
||||
)
|
||||
empty.setObjectName("Muted")
|
||||
empty.setWordWrap(True)
|
||||
self._list.addWidget(empty)
|
||||
self._list.addStretch(1)
|
||||
return
|
||||
|
||||
for g in games:
|
||||
self._list.addWidget(_game_row(
|
||||
g.name,
|
||||
os.path.basename(g.library.rstrip("/")) or g.library,
|
||||
steam.human_size(g.size_bytes),
|
||||
g.appid in new_appids,
|
||||
))
|
||||
self._list.addStretch(1)
|
||||
|
||||
# --- nav badge integration --------------------------------------------------------
|
||||
|
||||
def showEvent(self, event) -> None: # noqa: N802 (Qt override)
|
||||
# Viewing the list acknowledges the new games: clear the sidebar badge. The NEW
|
||||
# tags stay on the rows for this session so the user can still spot them.
|
||||
super().showEvent(event)
|
||||
if self._new_appids:
|
||||
from ..core import steam
|
||||
|
||||
threading.Thread(target=steam.acknowledge_new, daemon=True).start()
|
||||
self.new_count_changed.emit(0)
|
||||
@@ -28,6 +28,7 @@ from .. import __version__
|
||||
from ..config import load_config
|
||||
from ..core import alerts, elevation, updates
|
||||
from .dashboard import Dashboard
|
||||
from .games_page import GamesPage
|
||||
from .health_page import HealthPage
|
||||
from .notifications_page import NotificationsPage
|
||||
from .recorder_page import RecorderPage
|
||||
@@ -36,7 +37,7 @@ from .share_page import SharePage
|
||||
from .theme import ACCENT, GOOD, MUTED
|
||||
from .worker import SamplerWorker
|
||||
|
||||
_NAV_ITEMS = ["Dashboard", "Logs", "Health", "Setup", "Notifications", "Share"]
|
||||
_NAV_ITEMS = ["Dashboard", "Logs", "Health", "Games", "Setup", "Notifications", "Share"]
|
||||
|
||||
|
||||
class MainWindow(QMainWindow):
|
||||
@@ -66,6 +67,8 @@ class MainWindow(QMainWindow):
|
||||
self.dashboard = Dashboard()
|
||||
self.recorder_page = RecorderPage()
|
||||
self.health_page = HealthPage()
|
||||
self.games_page = GamesPage()
|
||||
self.games_page.new_count_changed.connect(self._set_games_badge)
|
||||
self.setup_page = SetupPage()
|
||||
self.notifications_page = NotificationsPage()
|
||||
self.notifications_page.changed.connect(self._apply_alert_settings)
|
||||
@@ -73,9 +76,10 @@ class MainWindow(QMainWindow):
|
||||
self._stack.addWidget(self.dashboard) # 0 Dashboard
|
||||
self._stack.addWidget(self.recorder_page) # 1 Logs
|
||||
self._stack.addWidget(self.health_page) # 2 Health
|
||||
self._stack.addWidget(self.setup_page) # 3 Setup
|
||||
self._stack.addWidget(self.notifications_page) # 4 Notifications
|
||||
self._stack.addWidget(self.share_page) # 5 Share
|
||||
self._stack.addWidget(self.games_page) # 3 Games
|
||||
self._stack.addWidget(self.setup_page) # 4 Setup
|
||||
self._stack.addWidget(self.notifications_page) # 5 Notifications
|
||||
self._stack.addWidget(self.share_page) # 6 Share
|
||||
content_layout.addWidget(self._stack)
|
||||
|
||||
layout.addWidget(self._build_sidebar())
|
||||
@@ -135,6 +139,7 @@ class MainWindow(QMainWindow):
|
||||
|
||||
group = QButtonGroup(self)
|
||||
group.setExclusive(True)
|
||||
self._nav_buttons: dict[str, QPushButton] = {}
|
||||
for i, name in enumerate(_NAV_ITEMS):
|
||||
btn = QPushButton(name)
|
||||
btn.setObjectName("NavButton")
|
||||
@@ -144,6 +149,7 @@ class MainWindow(QMainWindow):
|
||||
btn.clicked.connect(lambda _checked, idx=i: self._stack.setCurrentIndex(idx))
|
||||
group.addButton(btn, i)
|
||||
v.addWidget(btn)
|
||||
self._nav_buttons[name] = btn
|
||||
|
||||
v.addStretch(1)
|
||||
live = QLabel(f'<span style="color:{ACCENT};">●</span> <span style="color:{MUTED};">Live</span>')
|
||||
@@ -229,6 +235,11 @@ class MainWindow(QMainWindow):
|
||||
# collected and used by the relay guest view + the CLI `rigdoctor inventory`.)
|
||||
self.health_page._run()
|
||||
|
||||
def _set_games_badge(self, count: int) -> None:
|
||||
btn = self._nav_buttons.get("Games")
|
||||
if btn is not None:
|
||||
btn.setText(f"Games ● {count}" if count > 0 else "Games")
|
||||
|
||||
def _apply_alert_settings(self) -> None:
|
||||
cfg = load_config()
|
||||
self._alert_monitor.enabled = bool(cfg.get("alerts_enabled", True))
|
||||
|
||||
Reference in New Issue
Block a user