mirror of
https://github.com/spoopyghosty0/frameport.git
synced 2026-10-06 10:00:36 +02:00
Merge the Screenshots tab (agent list/delete_screenshots, play-session log for attribution, PC cache + download, GUI tab); agent v46
This commit is contained in:
commit
2ecf7f8624
9 files changed
+1149
-11
No files matched your search
@@ -21,7 +21,12 @@ Read `docs/PLAYBOOK.md` (symptom → fix) before debugging a game, and `docs/FRA
|
||||
the "Send files" dialog, game menu → "Add videos & files…" = `go("files", pkg)`; multi-select bar (download/
|
||||
delete); drag-and-drop from the OS via the `flet-dropzone` extension (Apache-2.0, Flutter `desktop_drop`), which
|
||||
only a `flet build` bundle contains: `files.dropzone_available()` keeps it out of source runs and the PyInstaller
|
||||
fallback, which would show an unknown control); settings; welcome; activity panel).
|
||||
fallback, which would show an unknown control); screenshots (Steam screenshots on the Frame, `install/screenshots.py`
|
||||
+ agent v46 `list_screenshots`/`delete_screenshots`: the Frame files every headset shot under SteamVR 250820, so
|
||||
launch.sh logs `start/end <unix>` to `<anchor>/plays.log` (upgrade_launchers adds it; Proton launchers exec → start
|
||||
only) and shots are matched by time; thumbnails cached in `<data>/screenshots-cache/<frame>/`, shown by asset URL;
|
||||
delete leaves screenshots.vdf alone (Steam rewrites it at exit); game menu → "Screenshots" = `go("screenshots",
|
||||
pkg)`); settings; welcome; activity panel).
|
||||
User tags live in library entries (`tags`), filters in library setting `ui.library`.
|
||||
**Performance rules** (the app froze before): never put image bytes in controls — artwork is served by URL from the
|
||||
GUI assets dir (= user data dir; `ft.run(assets_dir=…)`), as thumbnails (`artwork/thumbs.py`, Pillow); the
|
||||
|
||||
@@ -28,6 +28,8 @@ Install games that target the Meta Quest, Android, or general PCVR onto your **V
|
||||
curved and 360° layers); game settings as simple switches.
|
||||
- **Beyond Quest:** Android apps as windows, PC VR via Proton or Revive, Windows (non-VR) games via Proton, a Files
|
||||
tab with drag and drop.
|
||||
- **Screenshots tab:** the screenshots you took in the headset, sorted by game (matched by play time) and day;
|
||||
view them and download them to your computer.
|
||||
- **Self-updating** releases, redacted diagnostics, one-click problem reports and working-config sharing.
|
||||
|
||||
## Quick start
|
||||
|
||||
+257
-7
@@ -6,7 +6,7 @@ The PC app uploads this file to ~/.local/share/frameport/agent/ and calls:
|
||||
|
||||
Commands: info, prepare, finalize, shortcuts, shortcut_status, launch_test, stop, set_settings, uninstall,
|
||||
install_lepton, list_installed, proton_status, install_proton, prepare_pcvr, finalize_pcvr,
|
||||
controller_models.
|
||||
controller_models, list_screenshots, delete_screenshots (and more: see the cmd_* functions).
|
||||
Streaming: python3 frameport_agent.py _keyboard (a virtual keyboard: JSON lines on stdin, see keyboard_session)
|
||||
|
||||
Install layout (one Lepton container per game; same as the manual installs from 2026-09):
|
||||
@@ -35,7 +35,7 @@ import sys
|
||||
import time
|
||||
import zlib
|
||||
|
||||
AGENT_VERSION = 45
|
||||
AGENT_VERSION = 46
|
||||
HOME = os.path.expanduser("~")
|
||||
STEAM = os.path.join(HOME, ".local/share/Steam")
|
||||
ANCHORS = os.path.join(HOME, "Applications/quest-frame")
|
||||
@@ -641,7 +641,7 @@ def ensure_host_fixes():
|
||||
except Exception: # noqa: BLE001
|
||||
upgraded = []
|
||||
if upgraded:
|
||||
changed.append(f"launchers: exit watchdog, dashboard ({len(upgraded)})")
|
||||
changed.append(f"launchers: exit watchdog, dashboard, play log ({len(upgraded)})")
|
||||
return changed
|
||||
|
||||
|
||||
@@ -1565,6 +1565,237 @@ def cmd_list_files(args):
|
||||
return {"roots": roots, "missing": missing[:1000], "truncated": truncated, "kind": dep.get("kind", "quest")}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------------------ screenshots
|
||||
STEAMVR_APPID = "250820" # the Frame files every headset screenshot under SteamVR, whatever game was shown
|
||||
PLAYS_LOG = "plays.log" # <anchor>/plays.log: "start <unix>" / "end <unix>" lines written by launch.sh
|
||||
OPEN_SESSION = 12 * 3600 # a session without an "end" (Proton launchers exec the game; power loss) lasts at most this
|
||||
SESSION_GRACE = 5 # s after "end": the screenshot's time can trail the key press slightly
|
||||
SHOT_NAME = re.compile(r"^(\d{14})_\d+\.(jpe?g|png)$", re.I)
|
||||
KV_TOKEN = re.compile(r'"((?:[^"\\]|\\.)*)"|([{}])|//[^\n]*|([^\s"{}]+)')
|
||||
|
||||
|
||||
def kv_parse(text):
|
||||
"""Steam's text KeyValues (screenshots.vdf, ...) as nested dicts. Tolerant: an unbalanced or cut-off file gives
|
||||
what was read so far."""
|
||||
root, stack, key = {}, [], None
|
||||
cur = root
|
||||
for m in KV_TOKEN.finditer(text):
|
||||
quoted, brace, bare = m.groups()
|
||||
if brace == "{":
|
||||
new = {}
|
||||
if key is not None:
|
||||
cur[key] = new
|
||||
stack.append(cur)
|
||||
cur, key = new, None
|
||||
elif brace == "}":
|
||||
if not stack:
|
||||
break
|
||||
cur, key = stack.pop(), None
|
||||
elif quoted is not None or bare is not None:
|
||||
tok = quoted.replace('\\"', '"').replace("\\\\", "\\") if quoted is not None else bare
|
||||
if key is None:
|
||||
key = tok
|
||||
else:
|
||||
cur[key], key = tok, None
|
||||
return root
|
||||
|
||||
|
||||
def play_sessions():
|
||||
"""[(start, end, package)] from every FramePort game's plays.log, by start time. A start without an end (Proton
|
||||
launchers exec the game, so they log no end; a crash or power loss) lasts until the next game's start (one game
|
||||
runs at a time), at most OPEN_SESSION."""
|
||||
events = []
|
||||
for path in glob.glob(os.path.join(ANCHORS, "*", PLAYS_LOG)):
|
||||
pkg = os.path.basename(os.path.dirname(path))
|
||||
try:
|
||||
lines = open(path, encoding="utf-8", errors="replace").read().splitlines()
|
||||
except OSError:
|
||||
continue
|
||||
for line in lines:
|
||||
parts = line.split()
|
||||
if len(parts) >= 2 and parts[0] in ("start", "end") and parts[1].isdigit():
|
||||
events.append((int(parts[1]), parts[0], pkg))
|
||||
sessions, open_ = [], {} # open_: pkg -> start
|
||||
for t, kind, pkg in sorted(events):
|
||||
if kind == "start":
|
||||
if pkg in open_: # an earlier start never ended: it ended by this one
|
||||
sessions.append([open_[pkg], t, pkg])
|
||||
open_[pkg] = t
|
||||
elif pkg in open_:
|
||||
sessions.append([open_.pop(pkg), t, pkg])
|
||||
sessions += [[s, None, pkg] for pkg, s in open_.items()]
|
||||
sessions.sort(key=lambda s: s[0])
|
||||
for i, s in enumerate(sessions):
|
||||
if s[1] is None:
|
||||
later = [x[0] for x in sessions[i + 1:] if x[0] > s[0]]
|
||||
s[1] = min([s[0] + OPEN_SESSION] + later)
|
||||
return [tuple(s) for s in sessions]
|
||||
|
||||
|
||||
def session_game(t, sessions):
|
||||
"""Package whose play session contains time t (the latest start wins), else None."""
|
||||
hit = None
|
||||
for start, end, pkg in sessions:
|
||||
if start <= t <= end + SESSION_GRACE:
|
||||
hit = pkg
|
||||
return hit
|
||||
|
||||
|
||||
def shot_time_from_name(name):
|
||||
m = SHOT_NAME.match(name)
|
||||
if not m:
|
||||
return None
|
||||
try:
|
||||
return int(time.mktime(time.strptime(m[1], "%Y%m%d%H%M%S"))) # Steam names the files in local time
|
||||
except (ValueError, OverflowError):
|
||||
return None
|
||||
|
||||
|
||||
def _int(v):
|
||||
try:
|
||||
return int(v)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def user_screenshots(user):
|
||||
"""Every screenshot file of one Steam account: [{path, thumb, time, width, height, appid}]. screenshots.vdf gives
|
||||
times and sizes; files it doesn't list (Steam writes it at exit) come from the folders, timed by their name."""
|
||||
remote = os.path.join(STEAM, "userdata", user, "760", "remote")
|
||||
shots, seen = [], set()
|
||||
|
||||
def under(rel):
|
||||
p = os.path.normpath(os.path.join(remote, rel)) if isinstance(rel, str) else None
|
||||
return p if p and p.startswith(remote + os.sep) and os.path.isfile(p) else None
|
||||
try:
|
||||
text = open(os.path.join(STEAM, "userdata", user, "760", "screenshots.vdf"), encoding="utf-8",
|
||||
errors="replace").read()
|
||||
except OSError:
|
||||
text = ""
|
||||
entries = (kv_parse(text).get("screenshots") or {})
|
||||
for game in entries.values() if isinstance(entries, dict) else ():
|
||||
for e in game.values() if isinstance(game, dict) else ():
|
||||
path = under(e.get("filename")) if isinstance(e, dict) else None
|
||||
if not path or path in seen:
|
||||
continue # deleted (Steam lists it until it rewrites the file)
|
||||
seen.add(path)
|
||||
shots.append({"path": path, "thumb": under(e.get("thumbnail")),
|
||||
"time": _int(e.get("creation")) or shot_time_from_name(os.path.basename(path)),
|
||||
"width": _int(e.get("width")), "height": _int(e.get("height")),
|
||||
"appid": os.path.relpath(path, remote).split(os.sep)[0]})
|
||||
for path in sorted(glob.glob(os.path.join(remote, "*", "screenshots", "*"))):
|
||||
name = os.path.basename(path)
|
||||
if path in seen or not SHOT_NAME.match(name) or not os.path.isfile(path):
|
||||
continue
|
||||
thumb = os.path.join(os.path.dirname(path), "thumbnails", name)
|
||||
shots.append({"path": path, "thumb": thumb if os.path.isfile(thumb) else None,
|
||||
"time": shot_time_from_name(name) or int(os.path.getmtime(path)), "width": None,
|
||||
"height": None, "appid": os.path.relpath(path, remote).split(os.sep)[0]})
|
||||
return shots
|
||||
|
||||
|
||||
def shortcut_names(user):
|
||||
"""{shortcut appid (str): name} of an account's non-Steam shortcuts."""
|
||||
try:
|
||||
data = vdf_decode(open(os.path.join(STEAM, "userdata", user, "config", "shortcuts.vdf"), "rb").read())
|
||||
except (OSError, ValueError, IndexError, struct.error, AgentError):
|
||||
return {}
|
||||
out = {}
|
||||
for e in (data.get("shortcuts") or {}).values():
|
||||
if isinstance(e, dict) and isinstance(e.get("appid"), int):
|
||||
out[str(e["appid"] & 0xFFFFFFFF)] = e.get("AppName") or e.get("appname") or ""
|
||||
return out
|
||||
|
||||
|
||||
def cmd_list_screenshots(args):
|
||||
"""Steam screenshots on the Frame (every account), newest first: {shots: [{path, thumb, time, width, height, size,
|
||||
account, appid, package, title}], total, games: [{package, title, count}]}. The Frame files every headset
|
||||
screenshot under SteamVR (250820), so those are matched to the FramePort game whose play session (plays.log)
|
||||
contains their time; the rest stay "SteamVR" (package null). args: offset, limit, package (a package, or "" for
|
||||
the shots of no FramePort game); `total` and `games` count before offset/limit."""
|
||||
titles, by_appid = {}, {}
|
||||
for dep_path in glob.glob(os.path.join(ANCHORS, "*/deployment.json")):
|
||||
try:
|
||||
dep = json.load(open(dep_path))
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
pkg = os.path.basename(os.path.dirname(dep_path))
|
||||
titles[pkg] = dep.get("title") or pkg
|
||||
if dep.get("appid") is not None:
|
||||
by_appid[str(int(dep["appid"]) & 0xFFFFFFFF)] = pkg
|
||||
sessions = play_sessions()
|
||||
shots, names = [], {}
|
||||
for user in steam_users():
|
||||
names.update(shortcut_names(user))
|
||||
for s in user_screenshots(user):
|
||||
pkg = by_appid.get(s["appid"])
|
||||
if pkg is None and s["appid"] == STEAMVR_APPID and s["time"]:
|
||||
pkg = session_game(s["time"], sessions)
|
||||
if pkg:
|
||||
title = titles.get(pkg, pkg)
|
||||
elif s["appid"] == STEAMVR_APPID:
|
||||
title = "SteamVR"
|
||||
else:
|
||||
app = find_app_id(s["appid"]) if s["appid"] not in names else None
|
||||
title = names.get(s["appid"]) or (app or {}).get("name") or f"App {s['appid']}"
|
||||
try:
|
||||
size = os.path.getsize(s["path"])
|
||||
except OSError:
|
||||
continue
|
||||
shots.append({**s, "size": size, "account": user, "package": pkg, "title": title})
|
||||
shots.sort(key=lambda s: (-(s["time"] or 0), s["path"]))
|
||||
games = {}
|
||||
for s in shots:
|
||||
g = games.setdefault(s["package"] or s["title"], {"package": s["package"], "title": s["title"], "count": 0})
|
||||
g["count"] += 1
|
||||
want = args.get("package")
|
||||
if want is not None:
|
||||
shots = [s for s in shots if (s["package"] or "") == want]
|
||||
offset = max(0, int(args.get("offset") or 0))
|
||||
limit = int(args.get("limit") or 0)
|
||||
return {"shots": shots[offset:offset + limit] if limit > 0 else shots[offset:], "total": len(shots),
|
||||
"games": sorted(games.values(), key=lambda g: g["title"].lower())}
|
||||
|
||||
|
||||
def screenshot_file(path):
|
||||
"""`path` if it is a screenshot image in a Steam account's screenshot folder (userdata/<id>/760/remote/<appid>/
|
||||
screenshots/<name>), else AgentError. Links and '..' are refused: nothing outside those folders can be deleted."""
|
||||
if not isinstance(path, str) or "\0" in path or not path.startswith("/"):
|
||||
raise AgentError(f"not a screenshot: {path!r}")
|
||||
norm = os.path.normpath(path)
|
||||
parts = os.path.relpath(norm, os.path.join(STEAM, "userdata")).split(os.sep)
|
||||
if (norm != path or len(parts) != 6 or not parts[0].isdigit() or parts[1:3] != ["760", "remote"]
|
||||
or not parts[3].isdigit() or parts[4] != "screenshots" or not SHOT_NAME.match(parts[5])):
|
||||
raise AgentError(f"not a screenshot: {path}")
|
||||
if os.path.islink(norm) or not os.path.isfile(norm) or not inside_userdata(os.path.dirname(norm)):
|
||||
raise AgentError(f"not a screenshot file: {path}")
|
||||
return norm
|
||||
|
||||
|
||||
def inside_userdata(folder):
|
||||
"""The folder, links resolved, is inside Steam's userdata (a linked screenshots folder could point anywhere)."""
|
||||
return os.path.realpath(folder).startswith(os.path.realpath(os.path.join(STEAM, "userdata")) + os.sep)
|
||||
|
||||
|
||||
def cmd_delete_screenshots(args):
|
||||
"""Delete screenshots (+ their thumbnails); only image files in Steam's screenshot folders (screenshot_file; every
|
||||
path is checked before anything is deleted). screenshots.vdf is left alone: Steam keeps it in memory and rewrites
|
||||
it at exit, so an edit would be lost; Steam drops entries whose file is gone, list_screenshots skips them. Steam's
|
||||
own screenshot list may show the deleted ones until Steam restarts."""
|
||||
paths = [screenshot_file(p) for p in args.get("paths") or []]
|
||||
deleted = []
|
||||
for p in paths:
|
||||
os.unlink(p)
|
||||
deleted.append(p)
|
||||
thumbs = os.path.join(os.path.dirname(p), "thumbnails")
|
||||
if inside_userdata(thumbs):
|
||||
try:
|
||||
os.unlink(os.path.join(thumbs, os.path.basename(p)))
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
return {"deleted": deleted}
|
||||
|
||||
|
||||
def cmd_prepare(args):
|
||||
"""Where to upload, and what the Frame already has (so unchanged data is not re-sent)."""
|
||||
pkg = check_pkg(args["package"])
|
||||
@@ -1633,10 +1864,12 @@ stop() {{
|
||||
kill $permfix 2>/dev/null || true
|
||||
[[ -n "$child" ]] && kill -TERM -- "-$child" 2>/dev/null || true
|
||||
podman kill "lepton-steamlaunch-$SteamAppId" >/dev/null 2>&1 || true
|
||||
{plays_end}
|
||||
}}
|
||||
trap stop EXIT
|
||||
trap 'exit 130' INT
|
||||
trap 'exit 143' TERM
|
||||
{plays_start}
|
||||
setsid {lepton_q} start >"$app_dir/launch.log" 2>&1 &
|
||||
child=$!
|
||||
{dashboard}
|
||||
@@ -1662,9 +1895,17 @@ def dashboard_line():
|
||||
return DASHBOARD_LINE.format(agent_q=shlex.quote(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def plays_lines(anchor):
|
||||
"""launch.sh lines that log play sessions to <anchor>/plays.log (screenshots are matched to games by time: the
|
||||
Frame files every headset screenshot under SteamVR). Never fail the launcher."""
|
||||
q = shlex.quote(os.path.join(anchor, PLAYS_LOG))
|
||||
return (f'echo "start $(date +%s)" >>{q} 2>/dev/null || true',
|
||||
f'echo "end $(date +%s)" >>{q} 2>/dev/null || true')
|
||||
|
||||
|
||||
def upgrade_launchers():
|
||||
"""Give launchers written by older agents the parent watchdog (in place: a new file, so a running launcher keeps
|
||||
reading the old one). Returns the packages changed."""
|
||||
"""Give launchers written by older agents the parent watchdog, the dashboard closer and the play-session log (in
|
||||
place: a new file, so a running launcher keeps reading the old one). Returns the packages changed."""
|
||||
changed = []
|
||||
for path in glob.glob(os.path.join(ANCHORS, "*/launch.sh")):
|
||||
try:
|
||||
@@ -1676,6 +1917,13 @@ def upgrade_launchers():
|
||||
new = new.replace(OLD_WATCHDOG, WATCHDOG, 1)
|
||||
if "_dashboard_worker" not in new and 'child=$!\nwait "$child"' in new:
|
||||
new = new.replace('child=$!\nwait "$child"', 'child=$!\n' + dashboard_line() + '\nwait "$child"', 1)
|
||||
if PLAYS_LOG not in new:
|
||||
start, end = plays_lines(os.path.dirname(path))
|
||||
if "\nsetsid " in new and "\n trap - EXIT INT TERM\n" in new: # Lepton launcher
|
||||
new = new.replace("\nsetsid ", f"\n{start}\nsetsid ", 1)
|
||||
new = new.replace("\n trap - EXIT INT TERM\n", f"\n trap - EXIT INT TERM\n {end}\n", 1)
|
||||
elif "\nexec " in new: # Proton launcher: exec replaces the shell, so only the start is logged
|
||||
new = new.replace("\nexec ", f"\n{start}\nexec ", 1)
|
||||
if new == text:
|
||||
continue
|
||||
tmp = path + ".tmp"
|
||||
@@ -1692,7 +1940,8 @@ def write_launcher(anchor, base, pkg, title, appid, lepton, env):
|
||||
if re.fullmatch(r"[A-Z_][A-Z0-9_]*", k))
|
||||
text = LAUNCH_SH.format(title=title.replace("\n", " "), pkg=pkg, base_q=shlex.quote(base), appid=appid,
|
||||
lepton_q=shlex.quote(lepton), extra_env=extra, watchdog=WATCHDOG,
|
||||
dashboard=dashboard_line())
|
||||
dashboard=dashboard_line(), plays_start=plays_lines(anchor)[0],
|
||||
plays_end=plays_lines(anchor)[1])
|
||||
path = os.path.join(anchor, "launch.sh")
|
||||
with open(path + ".tmp", "w") as f:
|
||||
f.write(text)
|
||||
@@ -2175,6 +2424,7 @@ fi
|
||||
mkdir -p "$STEAM_COMPAT_DATA_PATH" "$STEAM_COMPAT_SHADER_PATH"
|
||||
cd "$base/game"{workdir}
|
||||
echo "FramePort: launching {pkg} with {tool}" >"$base/launch.log"
|
||||
{plays_start}
|
||||
exec {command} >>"$base/launch.log" 2>&1
|
||||
"""
|
||||
|
||||
@@ -2242,7 +2492,7 @@ def write_proton_launcher(anchor, base, pkg, title, appid, tool, exe_rel, revive
|
||||
tool=tool["name"], extra_env=extra, workdir=("/" + shlex.quote(workdir)) if workdir else "",
|
||||
# Proton sets up VR (vrclient, wineopenxr) only when SteamGameId is set: a flat Windows game goes without
|
||||
steam_game_id=f"export SteamGameId={appid}\n" if vr else "",
|
||||
xr_layer=XR_LAYER_ENV if xr_layer else "",
|
||||
xr_layer=XR_LAYER_ENV if xr_layer else "", plays_start=plays_lines(anchor)[0],
|
||||
command=" ".join(shlex.quote(a) for a in argv))
|
||||
path = os.path.join(anchor, "launch.sh")
|
||||
with open(path + ".tmp", "w") as f:
|
||||
|
||||
+42
-2
@@ -44,12 +44,44 @@ class FakeFS:
|
||||
else:
|
||||
with open(os.path.join(base, n), "wb") as f:
|
||||
f.truncate(sizes.get(os.path.splitext(n)[1], 48_000)) # sparse: no real disk use
|
||||
self.shots = self._screenshots()
|
||||
|
||||
def _screenshots(self) -> list[dict]:
|
||||
"""Sample Steam screenshots (colour gradients) for the Screenshots tab, two days, two games + SteamVR."""
|
||||
from PIL import Image
|
||||
|
||||
folder = os.path.join(self.home, "shots")
|
||||
os.makedirs(folder, exist_ok=True)
|
||||
games = [g for g in library.games() if g.get("kind") != "rift"][:2]
|
||||
owners = [(g["package"], g.get("title") or g["package"]) for g in games] + [(None, "SteamVR")]
|
||||
now, shots = time.time(), []
|
||||
for i in range(9):
|
||||
t = now - i * 2400 - (86400 if i >= 5 else 0)
|
||||
name = time.strftime("%Y%m%d%H%M%S", time.localtime(t)) + "_1.jpg"
|
||||
path = os.path.join(folder, name)
|
||||
im = Image.linear_gradient("L").resize((640, 360)).convert("RGB")
|
||||
im = Image.merge("RGB", (im.getchannel(0).point(lambda v, i=i: (v + 40 * i) % 256),
|
||||
im.getchannel(1).point(lambda v: 255 - v), im.getchannel(2)))
|
||||
im.save(path, "JPEG", quality=80)
|
||||
pkg, title = owners[i % len(owners)]
|
||||
shots.append({"path": path, "thumb": None, "time": int(t), "width": 1920, "height": 1080,
|
||||
"size": os.path.getsize(path), "account": "1", "appid": "250820", "package": pkg,
|
||||
"title": title})
|
||||
return shots
|
||||
|
||||
def agent(self, command, **args):
|
||||
if command == "storage_targets":
|
||||
return {"targets": [{"id": n.lower(), "path": os.path.join(self.home, n), "android": a, "shared": True}
|
||||
for n, a in (("Videos", "/sdcard/Movies"), ("Downloads", "/sdcard/Download"),
|
||||
("Documents", "/sdcard/Documents"))]}
|
||||
if command == "list_screenshots":
|
||||
want = args.get("package")
|
||||
shots = [s for s in self.shots if want is None or (s["package"] or "") == want]
|
||||
games = {}
|
||||
for s in self.shots:
|
||||
games.setdefault(s["package"] or "", {"package": s["package"], "title": s["title"], "count": 0})
|
||||
games[s["package"] or ""]["count"] += 1
|
||||
return {"shots": shots, "total": len(shots), "games": list(games.values())}
|
||||
raise RuntimeError(f"fake Frame: {command} not available")
|
||||
|
||||
@property
|
||||
@@ -67,6 +99,11 @@ class FakeFS:
|
||||
|
||||
def stat(self, path):
|
||||
return paramiko.SFTPAttributes.from_stat(os.stat(path))
|
||||
|
||||
def get(self, remote, local, callback=None):
|
||||
import shutil
|
||||
|
||||
shutil.copyfile(remote, local)
|
||||
return SFTP()
|
||||
|
||||
|
||||
@@ -169,7 +206,8 @@ def main() -> int:
|
||||
args.out.mkdir(parents=True, exist_ok=True)
|
||||
game = args.game or (library.games()[0]["package"] if library.games() else None)
|
||||
steps = [("library", lambda a: a.navigate(0)), ("frame", lambda a: a.navigate(1)),
|
||||
("files", lambda a: a.go("files")), ("tools", lambda a: a.go("settings"))]
|
||||
("files", lambda a: a.go("files")), ("screenshots", lambda a: a.go("screenshots")),
|
||||
("tools", lambda a: a.go("settings"))]
|
||||
if game:
|
||||
steps.insert(1, ("game", lambda a: a.open_game(game)))
|
||||
steps.insert(2, ("game-customize", lambda a: a.open_game(game, advanced=True)))
|
||||
@@ -188,7 +226,9 @@ def main() -> int:
|
||||
steps.append(("files", lambda a: a.go("files")))
|
||||
steps.append(("files-select", lambda a: [a.files_view._toggle(e.path, True)
|
||||
for e in a.files_view.entries[1:3]]))
|
||||
steps.append(("type-on-frame", open_type_dialog))
|
||||
steps.append(("screenshots", lambda a: a.go("screenshots")))
|
||||
steps.append(("screenshot-viewer", lambda a: a.screenshots_view.viewer(0)))
|
||||
steps.append(("type-on-frame", lambda a: (a.page.pop_dialog(), open_type_dialog(a))))
|
||||
if args.install_questions:
|
||||
queued: list[str] = []
|
||||
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Steam screenshots taken on the Frame (the GUI's Screenshots tab).
|
||||
|
||||
Steam keeps them per account in userdata/<id>/760/remote/<appid>/screenshots/ (+ thumbnails/). The Frame files every
|
||||
headset screenshot under SteamVR (250820), so the agent (`list_screenshots`) matches them to FramePort games by time:
|
||||
launch.sh logs each play session to <anchor>/plays.log. Thumbnails and opened images are cached on this PC in
|
||||
<user data>/screenshots-cache/<frame>/ and shown by asset URL (never image bytes in controls).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import posixpath
|
||||
import re
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
from ..core.events import Reporter
|
||||
from ..core.paths import user_data_dir
|
||||
from . import installer
|
||||
|
||||
THUMB_WIDTH = 480
|
||||
_locks: dict[str, threading.Lock] = {}
|
||||
_guard = threading.Lock()
|
||||
|
||||
|
||||
def list_shots(frame, package: str | None = None, offset: int = 0, limit: int = 0) -> dict:
|
||||
"""{shots, total, games} from the agent, newest first. package: a game ("" = shots of no FramePort game)."""
|
||||
args = {"offset": offset, "limit": limit}
|
||||
if package is not None:
|
||||
args["package"] = package
|
||||
return frame.agent("list_screenshots", **args)
|
||||
|
||||
|
||||
def frame_id(frame) -> str:
|
||||
target = getattr(frame, "target", None)
|
||||
raw = (getattr(target, "name", "") or getattr(target, "host", "") or "frame") if target else "frame"
|
||||
return re.sub(r"[^A-Za-z0-9._-]+", "_", raw).strip("._") or "frame"
|
||||
|
||||
|
||||
def cache_dir(frame) -> Path:
|
||||
d = user_data_dir() / "screenshots-cache" / frame_id(frame)
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
return d
|
||||
|
||||
|
||||
def _key(shot: dict) -> str:
|
||||
return hashlib.sha1(f"{shot['path']}:{shot.get('size')}".encode()).hexdigest()[:16]
|
||||
|
||||
|
||||
def _lock(key: str) -> threading.Lock:
|
||||
with _guard:
|
||||
return _locks.setdefault(key, threading.Lock())
|
||||
|
||||
|
||||
def _fetch(frame, remote: str, dest: Path) -> Path:
|
||||
with _lock(str(dest)):
|
||||
if not dest.exists():
|
||||
tmp = dest.with_suffix(dest.suffix + ".part")
|
||||
frame.sftp.get(remote, str(tmp))
|
||||
tmp.replace(dest)
|
||||
return dest
|
||||
|
||||
|
||||
def cached(frame, shot: dict, kind: str = "thumb") -> Path | None:
|
||||
"""The cached thumbnail/full image if it was already downloaded (no network: for render paths)."""
|
||||
p = cache_dir(frame) / _cache_name(shot, kind)
|
||||
return p if p.exists() else None
|
||||
|
||||
|
||||
def _cache_name(shot: dict, kind: str) -> str:
|
||||
ext = posixpath.splitext(shot["path"])[1].lower() if kind == "full" else ".jpg"
|
||||
return f"{kind}_{_key(shot)}{ext or '.jpg'}"
|
||||
|
||||
|
||||
def image_path(frame, shot: dict) -> Path:
|
||||
"""The full image, downloaded once into the cache (the viewer shows it from there)."""
|
||||
return _fetch(frame, shot["path"], cache_dir(frame) / _cache_name(shot, "full"))
|
||||
|
||||
|
||||
def thumb_path(frame, shot: dict) -> Path:
|
||||
"""A small image for the grid: Steam's own thumbnail, downloaded once; without one, made from the full image."""
|
||||
dest = cache_dir(frame) / _cache_name(shot, "thumb")
|
||||
if dest.exists():
|
||||
return dest
|
||||
if shot.get("thumb"):
|
||||
try:
|
||||
return _fetch(frame, shot["thumb"], dest)
|
||||
except OSError:
|
||||
pass # thumbnail gone: make one
|
||||
full = image_path(frame, shot)
|
||||
with _lock(str(dest)):
|
||||
if not dest.exists():
|
||||
try:
|
||||
from PIL import Image
|
||||
|
||||
with Image.open(full) as im:
|
||||
im.thumbnail((THUMB_WIDTH, THUMB_WIDTH))
|
||||
tmp = dest.with_suffix(".jpg.part")
|
||||
im.convert("RGB").save(tmp, "JPEG", quality=85)
|
||||
tmp.replace(dest)
|
||||
except Exception: # noqa: BLE001 (unreadable image: show the original)
|
||||
return full
|
||||
return dest
|
||||
|
||||
|
||||
def safe_name(text: str) -> str:
|
||||
return re.sub(r'[<>:"/\\|?*\x00-\x1f]+', "_", text or "").strip(" .") or "Screenshots"
|
||||
|
||||
|
||||
def download(frame, shots: list[dict], dest_dir: Path, reporter: Reporter | None = None) -> dict:
|
||||
"""Copy the full images into dest_dir/<game title>/ (files already there with the same size are skipped).
|
||||
Interruptible (Cancel)."""
|
||||
reporter = reporter or Reporter()
|
||||
dest_dir = Path(dest_dir)
|
||||
items = []
|
||||
for s in shots:
|
||||
dest = dest_dir / safe_name(s.get("title") or "") / posixpath.basename(s["path"])
|
||||
if dest.exists() and dest.stat().st_size == s.get("size"):
|
||||
continue
|
||||
items.append((s, dest))
|
||||
total = sum(s.get("size") or 0 for s, _ in items)
|
||||
skipped = len(shots) - len(items)
|
||||
reporter.log(f"{len(items)} screenshot(s), {total / 2**20:.1f} MiB → {dest_dir}"
|
||||
+ (f" ({skipped} already there)" if skipped else ""))
|
||||
done = 0
|
||||
if items:
|
||||
with installer.transfer_link(frame, reporter, total) as xfer:
|
||||
reporter.stage(f"Download {len(items)} screenshot(s)")
|
||||
for s, dest in items:
|
||||
reporter.check_cancel()
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = dest.with_name(dest.name + ".part")
|
||||
|
||||
def progress(got, _total, base=done, name=dest.name):
|
||||
reporter.check_cancel()
|
||||
reporter.progress((base + got) / max(total, 1), name)
|
||||
xfer.sftp.get(s["path"], str(tmp), callback=progress)
|
||||
tmp.replace(dest)
|
||||
done += s.get("size") or 0
|
||||
return {"files": len(items), "skipped": skipped, "bytes": total, "folder": str(dest_dir)}
|
||||
|
||||
|
||||
def delete(frame, shots: list[dict]) -> int:
|
||||
"""Delete screenshots on the Frame (the agent only deletes images in Steam's screenshot folders) and their cached
|
||||
copies here. Steam's own screenshot list may still show them until Steam restarts."""
|
||||
if not shots:
|
||||
return 0
|
||||
r = frame.agent("delete_screenshots", paths=[s["path"] for s in shots])
|
||||
for s in shots:
|
||||
for kind in ("thumb", "full"):
|
||||
(cache_dir(frame) / _cache_name(s, kind)).unlink(missing_ok=True)
|
||||
return len(r.get("deleted") or [])
|
||||
@@ -92,6 +92,7 @@
|
||||
"Advanced": "",
|
||||
"All": "",
|
||||
"All at their defaults": "",
|
||||
"All games": "",
|
||||
"All of FramePort's data on this PC is gone": "",
|
||||
"Allow swapchains larger than 4096 px": "",
|
||||
"Allows data backup.": "",
|
||||
@@ -204,7 +205,9 @@
|
||||
"Copy path": "",
|
||||
"Couldn't check this PC": "",
|
||||
"Couldn't connect: {exc}": "",
|
||||
"Couldn't list the screenshots: {exc}": "",
|
||||
"Couldn't list {path}: {exc}": "",
|
||||
"couldn't load the full image: {exc}": "",
|
||||
"Couldn't open the game's storage: {exc}": "",
|
||||
"Couldn't read the file list: {exc}": "",
|
||||
"Couldn't read the Frame's folders: {exc}": "",
|
||||
@@ -222,6 +225,14 @@
|
||||
"Declares Meta's permissions some mixed-reality games ask for, so they don't stop.": "",
|
||||
"Delete": "",
|
||||
"Delete {names}?": "",
|
||||
"Delete {n} screenshot?": [
|
||||
"",
|
||||
""
|
||||
],
|
||||
"Deleted {n} screenshot": [
|
||||
"",
|
||||
""
|
||||
],
|
||||
"Deletes the rollback copies kept from each game's previous install and leftover uploads. The games and saves stay.": "",
|
||||
"Details": "",
|
||||
"Developer Mode (on the Frame: Settings → System → Developer) lets FramePort find the Frame on your network and connect to it over SSH. The first-time setup command turns it on for you.": "",
|
||||
@@ -236,14 +247,24 @@
|
||||
"done": "",
|
||||
"Done": "",
|
||||
"Download": "",
|
||||
"Download all": "",
|
||||
"Download selected": "",
|
||||
"Download to this PC": "",
|
||||
"Download to which folder on this PC?": "",
|
||||
"Download {name}": "",
|
||||
"Download {n} screenshot": [
|
||||
"",
|
||||
""
|
||||
],
|
||||
"Downloaded when first needed": "",
|
||||
"Downloaded {n} file to {folder}": [
|
||||
"",
|
||||
""
|
||||
],
|
||||
"Downloaded {n} screenshot to {folder}": [
|
||||
"",
|
||||
""
|
||||
],
|
||||
"Downloading…": "",
|
||||
"Downloads": "",
|
||||
"Draws each eye separately; try it when one eye shows a grey or broken picture.": "",
|
||||
@@ -345,6 +366,7 @@
|
||||
"From the tested recipe for this game": "",
|
||||
"From the {src}": "",
|
||||
"Full launch log": "",
|
||||
"Game": "",
|
||||
"Game arguments (select SteamVR / OpenXR)": "",
|
||||
"Game config files": "",
|
||||
"Game launch capacity": "",
|
||||
@@ -529,12 +551,14 @@
|
||||
"No matching files": "",
|
||||
"No new games in {name}": "",
|
||||
"No release notes.": "",
|
||||
"No screenshots yet": "",
|
||||
"No Unreal crash reporter": "",
|
||||
"None": "",
|
||||
"None found": "",
|
||||
"Normal": "",
|
||||
"Not enough space on the Frame": "",
|
||||
"Not enough space on the Frame ({detail}). Remove games you don't play or use Free up space on the Steam Frame page.": "",
|
||||
"Not from a FramePort game": "",
|
||||
"Not installed": "",
|
||||
"Not now": "",
|
||||
"Not offered by Steam on this Frame yet": "",
|
||||
@@ -615,6 +639,7 @@
|
||||
"Pico/YVR/Quest metadata": "",
|
||||
"Picture": "",
|
||||
"pictures for the Steam library": "",
|
||||
"Pictures you took in the headset (Steam screenshots on the Frame)": "",
|
||||
"Platform compat (ovrMessageType_ToString)": "",
|
||||
"Play on Frame": "",
|
||||
"Play on this PC": "",
|
||||
@@ -729,6 +754,7 @@
|
||||
"Runs xrSyncActions one at a time with xrPollEvent and skips it for 250 ms after focus returns. For games that crash in the runtime's input code right after focus comes back (SIGSEGV in vrclient.so UpdateActionStateInternal / xrSyncActions, e.g. Myst).": "",
|
||||
"Save": "",
|
||||
"Save as known-good recipe": "",
|
||||
"Save the screenshots to which folder on this PC?": "",
|
||||
"Saved as a known-good recipe": "",
|
||||
"Saved as known-good. Check the issue on GitHub and submit it": "",
|
||||
"Saved {name} (in {parent})": "",
|
||||
@@ -740,6 +766,8 @@
|
||||
"Scan a folder…": "",
|
||||
"Scan {name}": "",
|
||||
"Scans the folders you added games from again and adds games that appeared since. Games already in the library keep their recipe, tags and art; unchanged ones aren't analyzed again.": "",
|
||||
"Screenshots": "",
|
||||
"Screenshots on the Frame can be shown once FramePort is connected to it.": "",
|
||||
"SDL apps: start without a clipboard service": "",
|
||||
"Search": "",
|
||||
"Search again": "",
|
||||
@@ -841,6 +869,7 @@
|
||||
"Switch Frame…": "",
|
||||
"Tags": "",
|
||||
"Tags:": "",
|
||||
"Take one in the headset with Steam's screenshot shortcut; it shows up here. Screenshots are matched to the FramePort game that was running.": "",
|
||||
"tall, 600 × 900": "",
|
||||
"Test": "",
|
||||
"Test Proton on the Frame": "",
|
||||
@@ -876,6 +905,7 @@
|
||||
"These games can't run on the Steam Frame": "",
|
||||
"These games check their Oculus license": "",
|
||||
"These installs need about {need} on the Frame, and it has {free} free. Remove games you don't play (or use Free up space on the Steam Frame page), or install fewer at once.": "",
|
||||
"They are deleted on the Frame and can't be restored. Steam's own screenshot list may still show them until Steam restarts.": "",
|
||||
"They use the Oculus Platform SDK, which comes with the Meta Horizon (Oculus) app — it isn't installed on this PC, so they may quit right after starting. FramePort doesn't change how a game checks its license. Untick the ones you'd rather skip.": "",
|
||||
"They use the Oculus Platform SDK, which comes with the Meta Horizon (Oculus) app. It doesn't exist on the Steam Frame, so there they crash right at startup (seen with Robo Recall, Lies Beneath and Lone Echo) — play them on this PC. Tick any you still want to try on the Frame.": "",
|
||||
"They're Oculus games that need Revive to reach VR, and Revive can't run on the Frame. Play them on this PC instead (Install on this PC — SteamVR + Revive). Tick any you still want to put on the Frame to experiment (they'll likely run flat or crash).": "",
|
||||
@@ -894,6 +924,7 @@
|
||||
"This source checkout has uncommitted changes: commit or stash them, then update": "",
|
||||
"Tilts the controllers' pointing ray up (+) or down (−), for games whose pointer doesn't hit what you aim at.": "",
|
||||
"To finish, close FramePort and delete its program folder.": "",
|
||||
"Today": "",
|
||||
"Tools": "",
|
||||
"Tools are up to date": "",
|
||||
"Translator for games built on Meta's oldest VR interface.": "",
|
||||
@@ -937,6 +968,7 @@
|
||||
"Unity: disable MSAA": "",
|
||||
"Unity: keep MSAA off at runtime (OVRManager)": "",
|
||||
"Unity: render each eye separately (no multiview)": "",
|
||||
"Unknown date": "",
|
||||
"Unreal builds of Meta XR Audio abort ('ClassNotFoundException com.oculus.os.AnalyticsEvent') on non-Quest devices. Patches that one JNI FindClass call so the library continues (e.g. NOPE Challenge).": "",
|
||||
"Unreal games start CrashReportClient when they crash, which leaves a crash dialog instead of simply closing. This passes -nocrashreports to the game and, on the Frame, renames the game's copy of CrashReportClient.exe so it can't start (your game files on this PC aren't changed).": "",
|
||||
"Untested": "",
|
||||
@@ -1027,6 +1059,7 @@
|
||||
"Writes liboverport.config.so with runtime options.": "",
|
||||
"Writes the conversion's settings into the game.": "",
|
||||
"XR_FB_passthrough via ALPHA_BLEND (Frame greyscale cameras).": "",
|
||||
"Yesterday": "",
|
||||
"You have the latest version as of the last check ({when}).": "",
|
||||
"You have {version}. {how}": "",
|
||||
"Your copy is used as it is (only the Frame's copy gets launch patches)": "",
|
||||
@@ -1094,6 +1127,10 @@
|
||||
"",
|
||||
""
|
||||
],
|
||||
"{n} screenshot": [
|
||||
"",
|
||||
""
|
||||
],
|
||||
"{n} selected": "",
|
||||
"{n} setting changed from the default": [
|
||||
"",
|
||||
|
||||
+15
-1
@@ -30,7 +30,9 @@ from .jobs import Job
|
||||
|
||||
NAV = [("library", tr("Library"), ft.Icons.GRID_VIEW_ROUNDED),
|
||||
("frame", tr("Steam Frame"), ft.Icons.VIEW_IN_AR_ROUNDED),
|
||||
("files", tr("Files"), ft.Icons.FOLDER_OPEN_ROUNDED), ("settings", tr("Settings"), ft.Icons.TUNE_ROUNDED)]
|
||||
("files", tr("Files"), ft.Icons.FOLDER_OPEN_ROUNDED),
|
||||
("screenshots", tr("Screenshots"), ft.Icons.PHOTO_LIBRARY_OUTLINED),
|
||||
("settings", tr("Settings"), ft.Icons.TUNE_ROUNDED)]
|
||||
POLL_SECONDS = 30
|
||||
|
||||
|
||||
@@ -67,6 +69,7 @@ class FramePortApp:
|
||||
self._pc_cache: tuple[float, dict] | None = None
|
||||
self.library_view = None # created once (views/library.LibraryView), re-mounted on every visit
|
||||
self.files_view = None # likewise (views/files.FilesView): keeps the location/folder between visits
|
||||
self.screenshots_view = None # likewise (views/screenshots.ScreenshotsView): keeps the game filter
|
||||
self.exe_queue: list[str] = [] # games whose executable the user should confirm (after a scan)
|
||||
self._failures: list[Job] = [] # failed installs/tests, shown together when the queue is done
|
||||
self.jobs = jobs_module.shared() # one queue per process, shared by every window session
|
||||
@@ -318,6 +321,12 @@ class FramePortApp:
|
||||
if self.files_view is None:
|
||||
self.files_view = FilesView(self)
|
||||
view = self.files_view.mount(*self.route[1:])
|
||||
elif kind == "screenshots":
|
||||
from .views.screenshots import ScreenshotsView
|
||||
|
||||
if self.screenshots_view is None:
|
||||
self.screenshots_view = ScreenshotsView(self)
|
||||
view = self.screenshots_view.mount(*self.route[1:])
|
||||
elif kind == "settings":
|
||||
view = SettingsView(self).build()
|
||||
else:
|
||||
@@ -344,6 +353,9 @@ class FramePortApp:
|
||||
elif self.route[0] == "files" and (self.files_view is None or self.files_view.root is None
|
||||
or self.frame_state != "connected"):
|
||||
self.render() # connected / disconnected: switch between the browser and "connect first"
|
||||
elif self.route[0] == "screenshots" and (self.screenshots_view is None or self.screenshots_view.root is None
|
||||
or self.frame_state != "connected"):
|
||||
self.render()
|
||||
else:
|
||||
self._refresh_sidebar()
|
||||
|
||||
@@ -672,6 +684,8 @@ class FramePortApp:
|
||||
lambda e: (lv.selected.add(pkg), lv.set_select_mode(True))))
|
||||
if rift:
|
||||
out.append((tr("Change executable…"), ft.Icons.TERMINAL_ROUNDED, lambda e: self.choose_exe(pkg)))
|
||||
if self.frame_state == "connected":
|
||||
out.append((tr("Screenshots"), ft.Icons.PHOTO_LIBRARY_OUTLINED, lambda e: self.go("screenshots", pkg)))
|
||||
out.append((tr("Find artwork…"), ft.Icons.IMAGE_SEARCH_ROUNDED, lambda e: self.find_artwork(pkg)))
|
||||
out.append((tr("Use your own artwork…"), ft.Icons.UPLOAD_FILE_ROUNDED, lambda e: self.custom_artwork(pkg)))
|
||||
if not job and self.frame_state == "connected" and \
|
||||
|
||||
@@ -0,0 +1,362 @@
|
||||
"""Screenshots: the Steam screenshots taken on the Frame, newest first, grouped by day, filterable by game.
|
||||
|
||||
The Frame files every headset screenshot under SteamVR; the agent matches them to FramePort games by play session
|
||||
(install/screenshots.py). Persistent like the Library: the filter survives switching tabs. Thumbnails are downloaded
|
||||
once into the cache and shown by asset URL; cards stream in batches from a background thread. Downloads are jobs.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import flet as ft
|
||||
|
||||
from ...artwork import thumbs
|
||||
from ...errors import explain
|
||||
from ...i18n import tr, tr_n
|
||||
from .. import components as C
|
||||
from .. import theme as T
|
||||
from .files_dialog import human
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..app import FramePortApp
|
||||
|
||||
BATCH = 12
|
||||
ALL = "__all__"
|
||||
OTHER = "__other__" # shots of no FramePort game (SteamVR home, Steam games)
|
||||
|
||||
|
||||
def day_label(t: float | None, now: float | None = None) -> str:
|
||||
if not t:
|
||||
return tr("Unknown date")
|
||||
now = now or time.time()
|
||||
day = time.strftime("%Y-%m-%d", time.localtime(t))
|
||||
if day == time.strftime("%Y-%m-%d", time.localtime(now)):
|
||||
return tr("Today")
|
||||
if day == time.strftime("%Y-%m-%d", time.localtime(now - 86400)):
|
||||
return tr("Yesterday")
|
||||
return day
|
||||
|
||||
|
||||
def group_by_day(shots: list[dict], now: float | None = None) -> list[tuple[str, list[dict]]]:
|
||||
"""[(day label, shots)] in the order given (newest first)."""
|
||||
out: list[tuple[str, list[dict]]] = []
|
||||
for s in shots:
|
||||
label = day_label(s.get("time"), now)
|
||||
if out and out[-1][0] == label:
|
||||
out[-1][1].append(s)
|
||||
else:
|
||||
out.append((label, [s]))
|
||||
return out
|
||||
|
||||
|
||||
def filter_key(package: str | None) -> str:
|
||||
return ALL if package is None else (package or OTHER)
|
||||
|
||||
|
||||
def agent_filter(key: str) -> str | None:
|
||||
"""The agent's `package` argument for a dropdown key."""
|
||||
return None if key == ALL else "" if key == OTHER else key
|
||||
|
||||
|
||||
class ScreenshotsView:
|
||||
def __init__(self, app: FramePortApp):
|
||||
self.app = app
|
||||
self.filter = ALL
|
||||
self.shots: list[dict] = []
|
||||
self.games: list[dict] = []
|
||||
self.selected: set[str] = set() # paths
|
||||
self.checks: dict[str, ft.Checkbox] = {}
|
||||
self._gen = 0
|
||||
self._loaded = False
|
||||
self._lock = threading.Lock()
|
||||
self.dropdown = ft.Dropdown(label=tr("Game"), value=ALL, width=T.px(300), dense=True,
|
||||
options=[ft.DropdownOption(key=ALL, text=tr("All games"))],
|
||||
on_select=lambda e: self.set_filter(e.control.value))
|
||||
self.select_all = ft.Checkbox(value=False, active_color=T.ACCENT, check_color=T.ON_ACCENT,
|
||||
tooltip=tr("Select all"), on_change=self._toggle_all)
|
||||
self.toolbar = ft.Row([
|
||||
self.dropdown, self.select_all, ft.Container(expand=True),
|
||||
C.secondary(tr("Download all"), ft.Icons.DOWNLOAD_ROUNDED, self._download_all),
|
||||
C.icon_btn(ft.Icons.REFRESH_ROUNDED, tr("Refresh"), lambda e: self.load()),
|
||||
], spacing=T.S2, vertical_alignment=ft.CrossAxisAlignment.CENTER)
|
||||
self.sel_label = C.body("", T.TEXT, weight=ft.FontWeight.W_500)
|
||||
self.sel_bar = ft.Container(ft.Row([
|
||||
self.sel_label, ft.Container(expand=True),
|
||||
C.secondary(tr("Download selected"), ft.Icons.DOWNLOAD_ROUNDED, self._download_selected),
|
||||
C.ghost(tr("Delete"), ft.Icons.DELETE_OUTLINE_ROUNDED, lambda e: self.delete(self._chosen())),
|
||||
C.ghost(tr("Clear"), ft.Icons.CLOSE_ROUNDED, lambda e: self._clear_selection()),
|
||||
], spacing=T.S2), padding=ft.Padding(T.S3, T.px(6), T.S2, T.px(6)), border_radius=T.RADIUS_SM,
|
||||
bgcolor=T.ACCENT_SOFT, visible=False)
|
||||
self.grid = ft.Column(spacing=T.S3, scroll=ft.ScrollMode.AUTO, expand=True)
|
||||
self.status = C.meta("")
|
||||
self.root = None
|
||||
|
||||
# ---------------------------------------------------------------- building
|
||||
def mount(self, package: str | None = None) -> ft.Control:
|
||||
app = self.app
|
||||
heading, sub = tr("Screenshots"), tr("Pictures you took in the headset (Steam screenshots on the Frame)")
|
||||
if not (app.target and app.frame_state == "connected"):
|
||||
self._loaded = False # load again once connected
|
||||
return ft.Column([
|
||||
app.top_bar(heading, sub),
|
||||
C.empty_state(ft.Icons.PHOTO_LIBRARY_OUTLINED, tr("Connect your Frame first"),
|
||||
tr("Screenshots on the Frame can be shown once FramePort is connected to it."),
|
||||
C.primary(tr("Connect"), ft.Icons.LINK_ROUNDED, lambda e: app.go("frame")))], expand=True)
|
||||
if self.root is None:
|
||||
self.root = ft.Column([app.top_bar(heading, sub), self.toolbar, self.sel_bar,
|
||||
C.card(self.grid, padding=T.S3, expand=True), self.status],
|
||||
spacing=T.S3, expand=True, horizontal_alignment=ft.CrossAxisAlignment.STRETCH)
|
||||
if package is not None and filter_key(package) != self.filter:
|
||||
self.filter = filter_key(package)
|
||||
self.dropdown.value = self.filter
|
||||
self.selected.clear()
|
||||
self._loaded = False
|
||||
if not self._loaded: # later visits keep what was shown (Refresh reloads)
|
||||
self._loaded = True
|
||||
self.load()
|
||||
return self.root
|
||||
|
||||
def set_filter(self, key: str | None) -> None:
|
||||
self.filter = key or ALL
|
||||
self.selected.clear()
|
||||
self.load()
|
||||
|
||||
# ---------------------------------------------------------------- loading (background thread)
|
||||
def load(self) -> None:
|
||||
self._gen += 1
|
||||
self.status.value = tr("Loading…")
|
||||
C.update(self.status)
|
||||
self.app.run_bg(self._load, self._gen, self.filter)
|
||||
|
||||
def _load(self, gen: int, key: str) -> None:
|
||||
from ...install import screenshots
|
||||
|
||||
frame = self.app.target.frame
|
||||
with self._lock:
|
||||
if gen != self._gen:
|
||||
return
|
||||
try:
|
||||
r = screenshots.list_shots(frame, package=agent_filter(key))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.status.value = tr("Couldn't list the screenshots: {exc}").format(exc=explain(exc))
|
||||
C.update(self.status)
|
||||
return
|
||||
if gen != self._gen:
|
||||
return
|
||||
self.shots, self.games = r.get("shots") or [], r.get("games") or []
|
||||
self.selected &= {s["path"] for s in self.shots}
|
||||
self._fill_dropdown()
|
||||
self.checks = {}
|
||||
self.grid.controls = []
|
||||
if not self.shots:
|
||||
empty = C.empty_state(
|
||||
ft.Icons.PHOTO_CAMERA_OUTLINED, tr("No screenshots yet"),
|
||||
tr("Take one in the headset with Steam's screenshot shortcut; it shows up here. Screenshots "
|
||||
"are matched to the FramePort game that was running."))
|
||||
empty.expand, empty.padding = False, T.px(48) # (expand inside a scrolling column: invalid layout)
|
||||
self.grid.controls = [empty]
|
||||
self._update_selection(render=False)
|
||||
self.status.value = tr_n("{n} screenshot", "{n} screenshots", len(self.shots)) + \
|
||||
((" · " + human(sum(s.get("size") or 0 for s in self.shots))) if self.shots else "")
|
||||
for c in (self.grid, self.status, self.dropdown):
|
||||
C.update(c)
|
||||
sections: dict[str, ft.Row] = {}
|
||||
index = {s["path"]: i for i, s in enumerate(self.shots)}
|
||||
for start in range(0, len(self.shots), BATCH):
|
||||
if gen != self._gen:
|
||||
return # the filter changed or a refresh started: that load takes over
|
||||
batch = self.shots[start:start + BATCH]
|
||||
for s in batch:
|
||||
try:
|
||||
screenshots.thumb_path(frame, s)
|
||||
except Exception: # noqa: BLE001 (the card shows a placeholder)
|
||||
pass
|
||||
for label, shots in group_by_day(batch):
|
||||
row = sections.get(label)
|
||||
if row is None:
|
||||
row = sections[label] = ft.Row(spacing=T.S3, run_spacing=T.S3, wrap=True)
|
||||
self.grid.controls += [C.h2(label), row]
|
||||
row.controls += [self.card(s, index[s["path"]]) for s in shots]
|
||||
C.update(self.grid)
|
||||
|
||||
def _fill_dropdown(self) -> None:
|
||||
opts = [ft.DropdownOption(key=ALL, text=tr("All games"))]
|
||||
for g in self.games:
|
||||
key = g.get("package") or OTHER
|
||||
title = g.get("title") if g.get("package") else tr("Not from a FramePort game")
|
||||
opts.append(ft.DropdownOption(key=key, text=f"{title} ({g.get('count', 0)})"))
|
||||
if self.filter not in {o.key for o in opts}: # a game without screenshots (opened from its page)
|
||||
from ...core import library
|
||||
|
||||
pkg = agent_filter(self.filter) or ""
|
||||
opts.append(ft.DropdownOption(key=self.filter, text=(library.game(pkg) or {}).get("title") or pkg))
|
||||
self.dropdown.options = opts # the agent counts `games` before filtering: the same list for every filter
|
||||
self.dropdown.value = self.filter
|
||||
|
||||
# ---------------------------------------------------------------- cards
|
||||
def _thumb_url(self, s: dict) -> str | None:
|
||||
from ...install import screenshots
|
||||
|
||||
p = screenshots.cached(self.app.target.frame, s, "thumb")
|
||||
return thumbs.asset_url(p) if p else None
|
||||
|
||||
def card(self, s: dict, i: int) -> ft.Control:
|
||||
when = time.strftime("%H:%M", time.localtime(s["time"])) if s.get("time") else ""
|
||||
check = ft.Checkbox(value=s["path"] in self.selected, active_color=T.ACCENT, check_color=T.ON_ACCENT,
|
||||
on_change=lambda e, p=s["path"]: self._toggle(p, e.control.value))
|
||||
self.checks[s["path"]] = check
|
||||
caption = ft.Container(ft.Row([C.meta(s.get("title") or "", T.TEXT, expand=True, max_lines=1,
|
||||
overflow=ft.TextOverflow.ELLIPSIS), C.meta(when, T.TEXT_2)],
|
||||
spacing=T.S2),
|
||||
left=0, right=0, bottom=0, padding=ft.Padding(T.S2, T.px(14), T.S2, T.px(6)),
|
||||
gradient=ft.LinearGradient(begin=ft.Alignment.TOP_CENTER,
|
||||
end=ft.Alignment.BOTTOM_CENTER,
|
||||
colors=[ft.Colors.TRANSPARENT, T.soft("#000000", 0.8)]),
|
||||
border_radius=ft.BorderRadius(0, 0, T.RADIUS_SM, T.RADIUS_SM))
|
||||
return ft.Container(
|
||||
ft.Stack([C.art_fill(self._thumb_url(s), radius=T.RADIUS_SM, placeholder_icon=ft.Icons.IMAGE_OUTLINED,
|
||||
left=0, right=0, top=0, bottom=0),
|
||||
caption, ft.Container(check, left=0, top=0)]),
|
||||
width=T.px(256), height=T.px(144), border_radius=T.RADIUS_SM, ink=True,
|
||||
tooltip=f"{s.get('title') or ''} · {day_label(s.get('time'))} {when}",
|
||||
on_click=lambda e, i=i: self.viewer(i))
|
||||
|
||||
# ---------------------------------------------------------------- selection
|
||||
def _toggle(self, path: str, on: bool) -> None:
|
||||
(self.selected.add if on else self.selected.discard)(path)
|
||||
check = self.checks.get(path)
|
||||
if check is not None and check.value != on:
|
||||
check.value = on
|
||||
C.update(check)
|
||||
self._update_selection()
|
||||
|
||||
def _toggle_all(self, e) -> None:
|
||||
on = bool(e.control.value)
|
||||
self.selected = {s["path"] for s in self.shots} if on else set()
|
||||
for check in self.checks.values():
|
||||
if check.value != on:
|
||||
check.value = on
|
||||
C.update(check)
|
||||
self._update_selection()
|
||||
|
||||
def _clear_selection(self) -> None:
|
||||
self.selected.clear()
|
||||
for check in self.checks.values():
|
||||
if check.value:
|
||||
check.value = False
|
||||
C.update(check)
|
||||
self._update_selection()
|
||||
|
||||
def _update_selection(self, render: bool = True) -> None:
|
||||
n = len(self.selected)
|
||||
size = sum(s.get("size") or 0 for s in self._chosen())
|
||||
self.sel_label.value = tr("{n} selected").format(n=n) + (f" · {human(size)}" if size else "")
|
||||
self.sel_bar.visible = bool(n)
|
||||
self.select_all.value = bool(self.shots) and n == len(self.shots)
|
||||
if render:
|
||||
for c in (self.sel_bar, self.select_all):
|
||||
C.update(c)
|
||||
|
||||
def _chosen(self) -> list[dict]:
|
||||
return [s for s in self.shots if s["path"] in self.selected]
|
||||
|
||||
# ---------------------------------------------------------------- viewer
|
||||
def viewer(self, index: int) -> None:
|
||||
if not self.shots:
|
||||
return
|
||||
page, shots = self.app.page, list(self.shots)
|
||||
state = {"i": index}
|
||||
img = ft.Image(src=self._thumb_url(shots[index]) or "", fit=ft.BoxFit.CONTAIN, width=T.px(1100),
|
||||
height=T.px(620), border_radius=T.RADIUS_SM)
|
||||
caption = C.body("", T.TEXT, weight=ft.FontWeight.W_500)
|
||||
details = C.meta("")
|
||||
|
||||
def show(delta: int = 0) -> None:
|
||||
state["i"] = (state["i"] + delta) % len(shots)
|
||||
s = shots[state["i"]]
|
||||
img.src = self._thumb_url(s) or img.src
|
||||
caption.value = s.get("title") or ""
|
||||
dims = f"{s['width']}×{s['height']} · " if s.get("width") and s.get("height") else ""
|
||||
when = time.strftime("%Y-%m-%d %H:%M", time.localtime(s["time"])) if s.get("time") else ""
|
||||
details.value = f"{state['i'] + 1} / {len(shots)} · {when} · {dims}{human(s.get('size') or 0)}"
|
||||
C.update(img, caption, details)
|
||||
self.app.run_bg(full, state["i"])
|
||||
|
||||
def full(i: int) -> None:
|
||||
from ...install import screenshots
|
||||
|
||||
try:
|
||||
p = screenshots.image_path(self.app.target.frame, shots[i])
|
||||
except Exception as exc: # noqa: BLE001
|
||||
details.value += " · " + tr("couldn't load the full image: {exc}").format(exc=explain(exc))
|
||||
C.update(details)
|
||||
return
|
||||
if state["i"] == i:
|
||||
img.src = thumbs.asset_url(p)
|
||||
C.update(img)
|
||||
|
||||
async def download(e):
|
||||
await self.download([shots[state["i"]]])
|
||||
|
||||
def delete(e):
|
||||
page.pop_dialog()
|
||||
self.delete([shots[state["i"]]])
|
||||
page.show_dialog(ft.AlertDialog(
|
||||
content=ft.Container(ft.Column([img, ft.Row([
|
||||
C.icon_btn(ft.Icons.CHEVRON_LEFT_ROUNDED, tr("Previous"), lambda e: show(-1)),
|
||||
ft.Column([caption, details], spacing=0, expand=True),
|
||||
C.icon_btn(ft.Icons.CHEVRON_RIGHT_ROUNDED, tr("Next"), lambda e: show(1)),
|
||||
C.secondary(tr("Download"), ft.Icons.DOWNLOAD_ROUNDED, download),
|
||||
C.icon_btn(ft.Icons.DELETE_OUTLINE_ROUNDED, tr("Delete"), delete),
|
||||
C.ghost(tr("Close"), on_click=lambda e: page.pop_dialog())], spacing=T.S2,
|
||||
vertical_alignment=ft.CrossAxisAlignment.CENTER)], spacing=T.S2, tight=True), width=T.px(1100)),
|
||||
bgcolor=T.BG, shape=ft.RoundedRectangleBorder(radius=T.RADIUS), content_padding=T.S3))
|
||||
show()
|
||||
|
||||
# ---------------------------------------------------------------- actions
|
||||
async def _download_selected(self, e=None):
|
||||
if self.selected:
|
||||
await self.download(self._chosen())
|
||||
|
||||
async def _download_all(self, e=None):
|
||||
if self.shots:
|
||||
await self.download(list(self.shots))
|
||||
|
||||
async def download(self, shots: list[dict]) -> None:
|
||||
folder = await ft.FilePicker().get_directory_path(
|
||||
dialog_title=tr("Save the screenshots to which folder on this PC?"))
|
||||
if not folder:
|
||||
return
|
||||
app = self.app
|
||||
|
||||
def run(job):
|
||||
from ...install import screenshots
|
||||
|
||||
r = screenshots.download(app.target.frame, shots, Path(folder), job.reporter)
|
||||
return tr_n("Downloaded {n} screenshot to {folder}", "Downloaded {n} screenshots to {folder}",
|
||||
r["files"], folder=r["folder"])
|
||||
app.submit(tr_n("Download {n} screenshot", "Download {n} screenshots", len(shots)), run, None,
|
||||
kind="tool-frame", open_panel=True)
|
||||
|
||||
def delete(self, shots: list[dict]) -> None:
|
||||
if not shots:
|
||||
return
|
||||
|
||||
def go():
|
||||
def work():
|
||||
from ...install import screenshots
|
||||
|
||||
try:
|
||||
n = screenshots.delete(self.app.target.frame, shots)
|
||||
self.app.toast(tr_n("Deleted {n} screenshot", "Deleted {n} screenshots", n))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.app.toast(explain(exc), error=True)
|
||||
self.selected.difference_update(s["path"] for s in shots)
|
||||
self.load()
|
||||
self.app.run_bg(work)
|
||||
C.confirm(self.app.page, tr_n("Delete {n} screenshot?", "Delete {n} screenshots?", len(shots)),
|
||||
tr("They are deleted on the Frame and can't be restored. Steam's own screenshot list may still "
|
||||
"show them until Steam restarts."), tr("Delete"), go, danger=True)
|
||||
@@ -0,0 +1,277 @@
|
||||
"""Screenshots tab: the agent's listing (screenshots.vdf + folders, games matched by play session), its delete path
|
||||
checks, the launchers' play log, and the PC-side cache/download helpers."""
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from test_agent import load_agent
|
||||
|
||||
VDF = """"screenshots"
|
||||
{
|
||||
\t"250820"
|
||||
\t{
|
||||
\t\t"0"
|
||||
\t\t{
|
||||
\t\t\t"type"\t\t"1"
|
||||
\t\t\t"filename"\t\t"250820/screenshots/20260928184431_1.jpg"
|
||||
\t\t\t"thumbnail"\t\t"250820/screenshots/thumbnails/20260928184431_1.jpg"
|
||||
\t\t\t"imported"\t\t"1"
|
||||
\t\t\t"externalfilename"\t\t"/run/user/1000/Screenshot_Mon_Sep_28_18-44-31_2026.png"
|
||||
\t\t\t"width"\t\t"1920"
|
||||
\t\t\t"height"\t\t"1080"
|
||||
\t\t\t"gameid"\t\t"250820"
|
||||
\t\t\t"creation"\t\t"1790635471"
|
||||
\t\t}
|
||||
\t\t"1"
|
||||
\t\t{
|
||||
\t\t\t"filename"\t\t"250820/screenshots/20260927100000_1.jpg"
|
||||
\t\t\t"creation"\t\t"1790500000"
|
||||
\t\t}
|
||||
\t\t"2"
|
||||
\t\t{
|
||||
\t\t\t"filename"\t\t"250820/screenshots/20260101000000_1.jpg"
|
||||
\t\t\t"creation"\t\t"1700000000"
|
||||
\t\t}
|
||||
\t\t"3"
|
||||
\t\t{
|
||||
\t\t\t"filename"\t\t"../../../../config/escape.jpg"
|
||||
\t\t}
|
||||
\t}
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
def make_tree(a, home):
|
||||
user = home / ".local/share/Steam/userdata/123"
|
||||
shots = user / "760/remote/250820/screenshots"
|
||||
(shots / "thumbnails").mkdir(parents=True)
|
||||
(user / "760/screenshots.vdf").write_text(VDF)
|
||||
for name in ("20260928184431_1.jpg", "20260927100000_1.jpg"):
|
||||
(shots / name).write_bytes(b"\xff\xd8" + b"x" * 100)
|
||||
(shots / "thumbnails/20260928184431_1.jpg").write_bytes(b"\xff\xd8thumb")
|
||||
# 20260101000000_1.jpg is listed but deleted; one file isn't in the vdf yet (Steam writes it at exit)
|
||||
(shots / "20260929120000_1.jpg").write_bytes(b"\xff\xd8" + b"y" * 50)
|
||||
(user / "config").mkdir()
|
||||
# a FramePort game installed with its own Steam shortcut + plays.log sessions
|
||||
anchor = home / "Applications/quest-frame/com.x.game"
|
||||
anchor.mkdir(parents=True)
|
||||
(anchor / "deployment.json").write_text(json.dumps({"package": "com.x.game", "title": "X Game",
|
||||
"appid": 3000000001, "base": str(anchor)}))
|
||||
(anchor / "plays.log").write_text("start 1790635000\nend 1790636000\nstart 1790499000\n")
|
||||
other = home / "Applications/quest-frame/com.y.other"
|
||||
other.mkdir(parents=True)
|
||||
(other / "deployment.json").write_text(json.dumps({"package": "com.y.other", "title": "Other", "appid": 5}))
|
||||
(other / "plays.log").write_text("start 1790499500\nend 1790500100\n")
|
||||
# a screenshot filed under a FramePort game's own shortcut appid
|
||||
own = user / "760/remote/3000000001/screenshots"
|
||||
own.mkdir(parents=True)
|
||||
(own / "20260930080000_1.jpg").write_bytes(b"\xff\xd8z")
|
||||
return user
|
||||
|
||||
|
||||
def test_kv_parse(monkeypatch, tmp_path):
|
||||
a = load_agent(monkeypatch, tmp_path)
|
||||
d = a.kv_parse(VDF)
|
||||
assert d["screenshots"]["250820"]["0"]["creation"] == "1790635471"
|
||||
assert a.kv_parse('"a" { "b" "x\\"y" // c\n "c" { "d" "1"')["a"] == {"b": 'x"y', "c": {"d": "1"}} # cut off
|
||||
assert a.kv_parse("") == {}
|
||||
|
||||
|
||||
def test_list_screenshots_attributes_by_session(monkeypatch, tmp_path):
|
||||
a = load_agent(monkeypatch, tmp_path)
|
||||
make_tree(a, tmp_path)
|
||||
r = a.cmd_list_screenshots({})
|
||||
names = [os.path.basename(s["path"]) for s in r["shots"]]
|
||||
assert names == ["20260930080000_1.jpg", "20260929120000_1.jpg", "20260928184431_1.jpg",
|
||||
"20260927100000_1.jpg"] # newest first; the deleted and the escaping entry are left out
|
||||
by = {os.path.basename(s["path"]): s for s in r["shots"]}
|
||||
# inside X Game's closed session
|
||||
assert by["20260928184431_1.jpg"]["package"] == "com.x.game"
|
||||
assert by["20260928184431_1.jpg"]["title"] == "X Game" and by["20260928184431_1.jpg"]["width"] == 1920
|
||||
assert by["20260928184431_1.jpg"]["thumb"].endswith("thumbnails/20260928184431_1.jpg")
|
||||
# X Game's open session (no end) was ended by Other's later start, so this one is Other's
|
||||
assert by["20260927100000_1.jpg"]["package"] == "com.y.other"
|
||||
# not in the vdf: timed by its name, outside every session -> SteamVR
|
||||
assert by["20260929120000_1.jpg"]["package"] is None and by["20260929120000_1.jpg"]["title"] == "SteamVR"
|
||||
assert by["20260929120000_1.jpg"]["time"] == int(time.mktime((2026, 9, 29, 12, 0, 0, 0, 0, -1)))
|
||||
assert by["20260929120000_1.jpg"]["thumb"] is None
|
||||
# filed under the game's own shortcut
|
||||
assert by["20260930080000_1.jpg"]["package"] == "com.x.game" and by["20260930080000_1.jpg"]["account"] == "123"
|
||||
assert r["total"] == 4
|
||||
assert {(g["package"], g["count"]) for g in r["games"]} == {("com.x.game", 2), ("com.y.other", 1), (None, 1)}
|
||||
page = a.cmd_list_screenshots({"package": "com.x.game", "offset": 1, "limit": 5})
|
||||
assert page["total"] == 2 and [s["path"] for s in page["shots"]] == [by["20260928184431_1.jpg"]["path"]]
|
||||
assert [s["title"] for s in a.cmd_list_screenshots({"package": ""})["shots"]] == ["SteamVR"]
|
||||
|
||||
|
||||
def test_list_screenshots_without_steam(monkeypatch, tmp_path):
|
||||
a = load_agent(monkeypatch, tmp_path)
|
||||
assert a.cmd_list_screenshots({}) == {"shots": [], "total": 0, "games": []}
|
||||
|
||||
|
||||
def test_open_session_ends_after_a_while(monkeypatch, tmp_path):
|
||||
a = load_agent(monkeypatch, tmp_path)
|
||||
anchor = tmp_path / "Applications/quest-frame/com.x.game"
|
||||
anchor.mkdir(parents=True)
|
||||
(anchor / "plays.log").write_text("start 1000\ngarbage\nend x\n")
|
||||
sessions = a.play_sessions()
|
||||
assert sessions == [(1000, 1000 + a.OPEN_SESSION, "com.x.game")]
|
||||
assert a.session_game(1500, sessions) == "com.x.game" and a.session_game(999, sessions) is None
|
||||
assert a.session_game(1000 + a.OPEN_SESSION + 60, sessions) is None
|
||||
|
||||
|
||||
def test_delete_screenshots_only_in_screenshot_folders(monkeypatch, tmp_path):
|
||||
a = load_agent(monkeypatch, tmp_path)
|
||||
user = make_tree(a, tmp_path)
|
||||
shots = user / "760/remote/250820/screenshots"
|
||||
target = str(shots / "20260928184431_1.jpg")
|
||||
bad = ["/etc/passwd", str(user / "760/screenshots.vdf"), str(shots / "thumbnails/20260928184431_1.jpg"),
|
||||
str(shots / "../screenshots/20260928184431_1.jpg"), str(shots / "notashot.jpg"), "relative.jpg",
|
||||
str(shots / "20260101000000_1.jpg"), 42]
|
||||
(shots / "notashot.jpg").write_bytes(b"x")
|
||||
for p in bad:
|
||||
with pytest.raises(a.AgentError):
|
||||
a.cmd_delete_screenshots({"paths": [target, p]})
|
||||
assert os.path.exists(target) # nothing deleted when any path is refused
|
||||
outside = tmp_path / "secret"
|
||||
outside.mkdir()
|
||||
(outside / "20260101000000_1.jpg").write_bytes(b"s")
|
||||
link_dir = user / "760/remote/999/screenshots"
|
||||
link_dir.parent.mkdir(parents=True)
|
||||
link_dir.symlink_to(outside)
|
||||
with pytest.raises(a.AgentError):
|
||||
a.cmd_delete_screenshots({"paths": [str(link_dir / "20260101000000_1.jpg")]})
|
||||
assert (outside / "20260101000000_1.jpg").exists()
|
||||
r = a.cmd_delete_screenshots({"paths": [target]})
|
||||
assert r == {"deleted": [target]}
|
||||
assert not os.path.exists(target) and not (shots / "thumbnails/20260928184431_1.jpg").exists()
|
||||
assert "20260928184431_1.jpg" in (user / "760/screenshots.vdf").read_text() # Steam's index is left alone
|
||||
assert target not in [s["path"] for s in a.cmd_list_screenshots({})["shots"]]
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == "win32", reason="bash launcher")
|
||||
def test_launcher_logs_play_sessions(monkeypatch, tmp_path):
|
||||
a = load_agent(monkeypatch, tmp_path)
|
||||
base, anchor = tmp_path / "base", tmp_path / "anchor dir"
|
||||
(base / "lepton-app").mkdir(parents=True)
|
||||
anchor.mkdir()
|
||||
lepton = tmp_path / "lepton"
|
||||
lepton.write_text("#!/bin/bash\nexit 0\n")
|
||||
lepton.chmod(0o755)
|
||||
a.write_launcher(str(anchor), str(base), "com.x.y", "X", 123, str(lepton), {"FRAMEPORT_KEEP_DASHBOARD": "1"})
|
||||
env = {**os.environ, "PATH": f"{tmp_path}:{os.environ['PATH']}"}
|
||||
(tmp_path / "podman").write_text("#!/bin/sh\nexit 0\n")
|
||||
(tmp_path / "podman").chmod(0o755)
|
||||
subprocess.run(["bash", str(anchor / "launch.sh")], env=env, timeout=30, capture_output=True)
|
||||
lines = (anchor / "plays.log").read_text().split("\n")
|
||||
assert lines[0].startswith("start ") and lines[1].startswith("end ") and lines[2] == ""
|
||||
assert int(lines[0].split()[1]) <= int(lines[1].split()[1])
|
||||
|
||||
|
||||
def test_upgrade_launchers_adds_play_log(monkeypatch, tmp_path):
|
||||
a = load_agent(monkeypatch, tmp_path)
|
||||
root = tmp_path / "Applications/quest-frame"
|
||||
quest, pcvr = root / "com.x.y", root / "rift.z"
|
||||
quest.mkdir(parents=True)
|
||||
pcvr.mkdir(parents=True)
|
||||
a.write_launcher(str(quest), "/b", "com.x.y", "X", 1, "/lepton", {})
|
||||
old = "\n".join(line for line in (quest / "launch.sh").read_text().split("\n") if "plays.log" not in line)
|
||||
(quest / "launch.sh").write_text(old)
|
||||
(pcvr / "launch.sh").write_text('#!/bin/bash\necho hi >"$base/launch.log"\nexec wine game.exe\n')
|
||||
assert sorted(a.upgrade_launchers()) == ["com.x.y", "rift.z"]
|
||||
q = (quest / "launch.sh").read_text()
|
||||
assert q.count("plays.log") == 2 and q.index('"start $(date') < q.index("\nsetsid ")
|
||||
assert ' echo "end $(date +%s)"' in q
|
||||
assert subprocess.run(["bash", "-n", str(quest / "launch.sh")]).returncode == 0
|
||||
p = (pcvr / "launch.sh").read_text()
|
||||
assert p.count("plays.log") == 1 and p.index("plays.log") < p.index("\nexec ")
|
||||
assert a.upgrade_launchers() == []
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ PC side
|
||||
class FakeSFTP:
|
||||
def __init__(self):
|
||||
self.gets = []
|
||||
|
||||
def get(self, remote, local, callback=None):
|
||||
self.gets.append(remote)
|
||||
shutil.copyfile(remote, local)
|
||||
if callback:
|
||||
callback(os.path.getsize(remote), os.path.getsize(remote))
|
||||
|
||||
|
||||
class FakeFrame:
|
||||
def __init__(self, agent_mod):
|
||||
self.sftp = FakeSFTP()
|
||||
self.a = agent_mod
|
||||
self.target = type("T", (), {"host": "10.0.0.5", "name": ""})()
|
||||
|
||||
def agent(self, command, **args):
|
||||
return self.a.COMMANDS[command](args)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pc(monkeypatch, tmp_path):
|
||||
from frameport.install import installer
|
||||
|
||||
monkeypatch.setattr(installer, "transfer_link", lambda f, r, n: contextlib.nullcontext(f))
|
||||
frame_home = tmp_path / "frame"
|
||||
frame_home.mkdir()
|
||||
a = load_agent(monkeypatch, frame_home)
|
||||
make_tree(a, frame_home)
|
||||
monkeypatch.setenv("FRAMEPORT_HOME", str(tmp_path / "pc"))
|
||||
return FakeFrame(a)
|
||||
|
||||
|
||||
def test_pc_cache_and_download(pc, tmp_path):
|
||||
from frameport.core.paths import user_data_dir
|
||||
from frameport.install import screenshots
|
||||
|
||||
r = screenshots.list_shots(pc, package="com.x.game")
|
||||
assert r["total"] == 2
|
||||
shot = next(s for s in r["shots"] if s["thumb"])
|
||||
assert screenshots.cached(pc, shot) is None
|
||||
t = screenshots.thumb_path(pc, shot)
|
||||
assert t.read_bytes() == b"\xff\xd8thumb" and t.is_relative_to(user_data_dir() / "screenshots-cache" / "10.0.0.5")
|
||||
assert screenshots.thumb_path(pc, shot) == t and pc.sftp.gets == [shot["thumb"]] # downloaded once
|
||||
assert screenshots.cached(pc, shot) == t
|
||||
full = screenshots.image_path(pc, shot)
|
||||
assert full.read_bytes() == open(shot["path"], "rb").read()
|
||||
# without a Steam thumbnail one is made from the full image (here: not an image -> the original is used)
|
||||
no_thumb = next(s for s in r["shots"] if not s["thumb"])
|
||||
assert screenshots.thumb_path(pc, no_thumb).exists()
|
||||
out = tmp_path / "out"
|
||||
got = screenshots.download(pc, r["shots"], out)
|
||||
assert got["files"] == 2 and (out / "X Game" / os.path.basename(shot["path"])).exists()
|
||||
assert screenshots.download(pc, r["shots"], out)["skipped"] == 2 # same size: not copied again
|
||||
assert screenshots.delete(pc, [shot]) == 1 and not os.path.exists(shot["path"])
|
||||
assert screenshots.cached(pc, shot) is None
|
||||
|
||||
|
||||
def test_pc_download_cancel(pc, tmp_path):
|
||||
from frameport.core.events import Cancelled, Reporter
|
||||
from frameport.install import screenshots
|
||||
|
||||
rep = Reporter()
|
||||
rep.cancelled.set()
|
||||
with pytest.raises(Cancelled):
|
||||
screenshots.download(pc, screenshots.list_shots(pc)["shots"], tmp_path / "out", rep)
|
||||
assert not list((tmp_path / "out").rglob("*.jpg"))
|
||||
|
||||
|
||||
def test_group_by_day():
|
||||
from frameport.ui.views.screenshots import agent_filter, filter_key, group_by_day
|
||||
|
||||
now = time.mktime((2026, 10, 4, 15, 0, 0, 0, 0, -1))
|
||||
shots = [{"time": now - 60}, {"time": now - 3600}, {"time": now - 86400}, {"time": now - 5 * 86400},
|
||||
{"time": None}]
|
||||
groups = group_by_day(shots, now)
|
||||
assert [(label, len(s)) for label, s in groups] == [("Today", 2), ("Yesterday", 1), ("2026-09-29", 1),
|
||||
("Unknown date", 1)]
|
||||
assert agent_filter(filter_key(None)) is None and agent_filter(filter_key("")) == ""
|
||||
assert agent_filter(filter_key("com.x")) == "com.x"
|
||||
Reference in new issue
Block a user