steam-frame/saphid--frame-control
1
0 Fork 0
mirror of https://github.com/saphid/frame-control.git synced 2026-10-06 07:00:37 +02:00
Code Issues Packages Projects Releases 6 Wiki Activity
Files
1203ec0a9e1fcf87e7cb2977f0f561b17d21c9f0
saphid--frame-control/ui/frame_titles.py
T
saphidandClaude Opus 5.5 ca2a991f03 Sideloaded titles: use ids Steam's create-shortcut accepts 
On the Frame, Steam refused to register titles whose id had a hyphen
(fc-smoke-exe) with "missing/invalid arguments", and registered the same
program as FCSmokeProbe (headset smoke test, 2026-09-27, BUILD_ID
20260922.6101926). Valve's client only allows ^[A-Za-z_][A-Za-z0-9_.]+$.

title_id now makes ids of letters, digits and _, not starting with a
digit, 2 to 64 long; new installs are checked against that, while titles
already on the Frame are still listed, launched and removed. Steam's error
text is trimmed before it's quoted, and the "install it again with Steam
running" hint only follows a Steam-not-running error.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-27 17:40:09 +10:00

829 lines
38 KiB
Python
Raw Blame History

"""Linux and Windows builds on the Frame as Steam "Devkit Games".
A .zip, a folder or a single executable becomes a title in the Steam library,
through the same path as Valve's SteamOS Devkit Client: its devkit-utils
(vendored in frame/devkit-utils, synced to ~/devkit-utils on the Frame) make
~/devkit-game/<id>/, the files are copied there, and steam-client-create-shortcut
asks the running Steam client to register it with a runtime:
Windows .exe -> Proton Experimental (steam_play=1; x86-64 runs through FEX)
aarch64 ELF -> Steam Linux Runtime 4.0 ARM64 (steam_play=0)
x86-64 ELF -> Steam Linux Runtime 4.0 (steam_play=0; runs through FEX)
Everything device-side is inferred from Valve's steamos-devkit source until
checked on a headset; see docs/sideloading.md.
Python stdlib only. CLI:
python3 ui/frame_titles.py inspect PATH
python3 ui/frame_titles.py install PATH [--name N] [--exe REL] [--runtime R]
python3 ui/frame_titles.py list | launch ID | remove ID
"""
import hashlib, json, os, posixpath, re, shlex, shutil, stat, struct, subprocess, sys, tempfile, threading, time, zipfile
import frame_android
import frame_host
from frame_android import FrameError
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
UTILS_LOCAL = os.path.join(ROOT, 'frame', 'devkit-utils')
UTILS = 'devkit-utils' # on the Frame, relative to $HOME (where Valve's client puts it)
GAMES = 'devkit-game' # ditto; steamos-prepare-upload makes <id>/ in here
STAMP = '.frame-control-stamp'
PY = 'python3 ~/' + UTILS + '/'
# Valve's reserved sideload names: uploading one of these replaces the Steam client itself.
# devkit-steam is the trampoline file that switches SteamOS to a sideloaded client
# (select_steam.sh); a folder there breaks Valve's devkit tools.
RESERVED_IDS = ('steam', 'steamdeckard', 'steamvr', 'steamvrdeckard', 'devkit-steam')
# Any title folder on the Frame we'll list, launch or remove: safe in a shell and a path.
ID_RE = re.compile(r'^[A-Za-z0-9_][A-Za-z0-9_-]{0,63}$')
# What a new install may use. Steam's create-shortcut refused "fc-smoke-exe" with
# "missing/invalid arguments" and took the same program as "FCSmokeProbe" (headset
# smoke test, 2026-09-27, build 20260922.6101926); Valve's client only allows
# GAMEID_ALLOWED_PATTERN, ^[A-Za-z_][A-Za-z0-9_.]+$ (devkit_client/gui2). No dots here.
NEW_ID_RE = re.compile(r'^[A-Za-z_][A-Za-z0-9_]{1,63}$')
DIR_RE = re.compile(r'^/[A-Za-z0-9_./-]+$')
# Zip limits: well above any real game, well below a zip bomb.
MAX_UNPACKED = 64 * 1024**3
MAX_ENTRIES = 200000
TMP_PREFIX = 'frame-title-' # then the server's PID, so server.sweep_tmp can clear a killed run's
MAX_RATIO = 200 # uncompressed / compressed, once past 1 GB
# The Steam compat tool aliases Valve's client uses (devkit_client RUNTIME_ALIASES).
RUNTIMES = {
'proton-experimental': {'label': 'Proton Experimental', 'steam_play': True},
'proton-stable': {'label': 'Proton (stable)', 'steam_play': True},
'SteamLinuxRuntime_4-arm64': {'label': 'Steam Linux Runtime 4.0 (ARM64)', 'steam_play': False},
'SteamLinuxRuntime_4': {'label': 'Steam Linux Runtime 4.0 (x86-64, through FEX)', 'steam_play': False},
}
# Experimental rather than stable: the Frame's ARM64 Proton + FEX stack is new,
# and Proton fixes reach Experimental first. --runtime proton-stable switches.
DEFAULT_PROTON = 'proton-experimental'
ELF_MACHINES = {0xB7: 'arm64', 0x3E: 'x86_64', 0x03: 'x86', 0x28: 'arm'}
PE_MACHINES = {0x8664: 'x86_64', 0xAA64: 'arm64', 0x14C: 'x86', 0x1C4: 'arm'}
# Executables that are never the game: crash reporters, installers, redistributables.
SKIP_RE = re.compile(r'crash|unins|setup|install|redist|dxsetup|dxwebsetup|dotnet|prereq|'
r'easyanticheat|eac_|updater|uploader|report|sandbox|helper', re.I)
SKIP_DIRS = re.compile(r'^(_*commonredist|redist|redistributables?|directx|vcredist|__installer|'
r'installers?|prereqs?|support|engine|__macosx)$', re.I)
# Trailing words of an archive name that describe the build, not the game.
BUILD_WORDS = re.compile(r'([ ._-]+(win(dows)?(32|64)?|linux(32|64)?|x64|x86(_64)?|amd64|arm64|aarch64|'
r'build|release|portable|steamos|v\d+([._]\d+)*|\d+([._]\d+)+))+$', re.I)
def classify(path):
"""{'format': 'elf'|'pe'|'script', 'arch', 'exe'} for an executable file, else None."""
try:
with open(path, 'rb') as f:
head = f.read(4096)
if head[:4] == b'\x7fELF':
return _elf(f, head)
if head[:2] == b'MZ':
return _pe(f, head)
except (OSError, struct.error, ValueError, OverflowError):
return None # unreadable, or a header that lies about its sizes
if head[:2] == b'#!' or path.lower().endswith('.sh'):
return {'format': 'script', 'arch': None, 'exe': True}
return None
def _elf(f, head):
if len(head) < 64 or head[4] not in (1, 2) or head[5] not in (1, 2):
return None
wide, end = head[4] == 2, '<' if head[5] == 1 else '>'
e_type, machine = struct.unpack_from(end + 'HH', head, 16)
arch = ELF_MACHINES.get(machine, f'elf-0x{machine:x}')
if e_type == 2: # ET_EXEC
return {'format': 'elf', 'arch': arch, 'exe': True}
if e_type != 3: # not ET_DYN either: object file, core dump
return {'format': 'elf', 'arch': arch, 'exe': False}
# ET_DYN is a PIE executable or a shared library; only executables ask for an interpreter.
if wide:
phoff, = struct.unpack_from(end + 'Q', head, 32)
phentsize, phnum = struct.unpack_from(end + 'HH', head, 54)
else:
phoff, = struct.unpack_from(end + 'I', head, 28)
phentsize, phnum = struct.unpack_from(end + 'HH', head, 42)
if phentsize < (56 if wide else 32) or phnum > 256:
return {'format': 'elf', 'arch': arch, 'exe': False}
f.seek(phoff)
table = f.read(phentsize * phnum)
interp = any(struct.unpack_from(end + 'I', table, i * phentsize)[0] == 3 # PT_INTERP
for i in range(len(table) // phentsize))
return {'format': 'elf', 'arch': arch, 'exe': interp}
def _pe(f, head):
if len(head) < 0x40:
return None
lfanew, = struct.unpack_from('<I', head, 0x3C)
f.seek(lfanew)
coff = f.read(24)
if len(coff) < 24 or coff[:4] != b'PE\0\0':
return None # a DOS program, or not an executable at all
machine, = struct.unpack_from('<H', coff, 4)
characteristics, = struct.unpack_from('<H', coff, 22)
return {'format': 'pe', 'arch': PE_MACHINES.get(machine, f'pe-0x{machine:x}'),
'exe': not characteristics & 0x2000} # IMAGE_FILE_DLL
def title_id(name):
"""The Devkit Game id Steam shows as the title's name, in the form Steam accepts
(NEW_ID_RE): letters, digits and _, not starting with a digit, 2 to 64 long."""
s = re.sub(r'[^A-Za-z0-9]+', '_', str(name or '').strip()).strip('_')[:64].strip('_')
if not s:
raise FrameError('the title needs a name with some letters or digits')
if s[0].isdigit():
s = '_' + s[:63]
if len(s) < 2 or s.lower() in RESERVED_IDS:
s += '_game'
return s
def display_name(filename):
"""A title name from a zip, folder or exe name: no extension or build/platform words."""
base = os.path.basename(str(filename).rstrip('/\\'))
stem, ext = os.path.splitext(base)
if ext.lower() in ('.zip', '.exe', '.sh', '.x86_64', '.arm64', '.aarch64', '.bin'):
base = stem
return BUILD_WORDS.sub('', base) or base
def _norm(s):
return re.sub(r'[^a-z0-9]', '', s.lower())
# ---- payload: zip, folder or single file -> a local tree to upload ----------
def extract_zip(zpath, dest):
"""Unpack a zip into dest, refusing paths that escape it and absurd sizes."""
try:
z = zipfile.ZipFile(zpath)
except (zipfile.BadZipFile, OSError) as e:
raise FrameError(f'{os.path.basename(zpath)} is not a readable zip: {e}')
with z:
infos = z.infolist()
if len(infos) > MAX_ENTRIES:
raise FrameError(f'the zip has {len(infos)} entries; the limit is {MAX_ENTRIES}')
total = sum(i.file_size for i in infos)
packed = max(1, os.path.getsize(zpath))
if total > MAX_UNPACKED or (total > 1024**3 and total > packed * MAX_RATIO):
raise FrameError(f'the zip would unpack to {total / 1024**3:.1f} GB, which looks wrong')
free = shutil.disk_usage(dest).free
if total + 256 * 1024**2 > free:
raise FrameError(f'not enough space on this computer to unpack the zip '
f'({total / 1024**3:.1f} GB needed, {free / 1024**3:.1f} GB free)')
try:
_extract_members(z, infos, os.path.realpath(dest))
except FrameError:
raise # a RuntimeError too, but already worded
except (zipfile.BadZipFile, RuntimeError, NotImplementedError, EOFError, OSError) as e:
# Encrypted or corrupt members, unsupported compression, clashing names, a full disk.
raise FrameError(f'could not unpack {os.path.basename(zpath)}: {e}')
def _extract_members(z, infos, root):
# No symlink is ever created here, so no write can be redirected through one
# (chained links, Windows without the privilege). Links inside the zip are
# resolved on paper and materialised as copies once the real files are out.
links = {}
for info in infos:
rel = _safe_member(info.filename)
if rel is None:
continue
target = _inside(root, rel, info.filename)
mode = info.external_attr >> 16
if info.is_dir():
os.makedirs(target, exist_ok=True)
continue
os.makedirs(os.path.dirname(target), exist_ok=True)
if stat.S_ISLNK(mode):
if info.file_size > 4096: # a link's content is a path, never this big
raise FrameError(f'the zip has an oversized link: {info.filename}')
link = z.read(info).decode('utf-8', 'replace').replace('\\', '/')
dest = posixpath.normpath(posixpath.join(posixpath.dirname(rel), link))
if link.startswith('/') or ':' in link or dest == '..' or dest.startswith('../'):
raise FrameError(f'the zip has a link that points outside it: {info.filename}')
links[rel] = link # as written: resolved later, one component at a time
continue
with z.open(info) as src, open(target, 'wb') as out:
shutil.copyfileobj(src, out, 1 << 20)
if mode & 0o111:
os.chmod(target, 0o755)
_materialise_links(root, links)
def _inside(root, rel, name):
"""rel's path under root, refusing anything that resolves outside it."""
target = os.path.join(root, *rel.split('/'))
if not (os.path.realpath(target) + os.sep).startswith(root + os.sep):
raise FrameError(f'the zip has a path that climbs out of it: {name}')
return target
def _resolve_link(path, links):
"""path with every link in it followed, on paper; None if it loops or leaves the zip.
Like the kernel: each component in turn, so 'dirlink/..' is the parent of
where dirlink points, not the folder dirlink sits in.
"""
todo, done, hops = path.split('/'), [], 0
while todo:
part = todo.pop(0)
if part in ('', '.'):
continue
if part == '..':
if not done:
return None
done.pop()
continue
done.append(part)
text = links.get('/'.join(done))
if text is not None:
hops += 1
if hops > 40:
return None
done.pop() # link text is relative to the link's folder
todo = text.split('/') + todo
return '/'.join(done)
def _materialise_links(root, links):
"""Copy the file each link names into its place (lib.so.1 -> lib.so.1.2.3 and the like).
Only links to files: a folder link could hold itself, and game builds link
libraries, not folders. Folder, looping and dangling links are dropped.
"""
copies = []
for rel in sorted(links):
dest = _resolve_link(rel, links)
if not dest:
continue # loops, escapes, or the zip's own top folder
src, target = _inside(root, dest, rel), _inside(root, rel, rel)
if os.path.isfile(src) and not os.path.lexists(target): # a real entry may have the name
copies.append((src, target))
# Many links to one big file could fill the disk: the same limits as the zip itself.
need = sum(os.path.getsize(src) for src, _ in copies)
if need + _tree_size(root) > MAX_UNPACKED:
raise FrameError('the zip\'s links would copy more than it holds, which looks wrong')
if need + 256 * 1024**2 > shutil.disk_usage(root).free:
raise FrameError(f'not enough space on this computer for the zip\'s linked files ({need / 1024**3:.1f} GB)')
for src, target in copies:
os.makedirs(os.path.dirname(target), exist_ok=True)
shutil.copy2(src, target)
def _safe_member(name):
"""The member's relative path with / separators, None to skip it; raises if it escapes."""
rel = name.replace('\\', '/')
if rel.startswith('/') or re.match(r'^[A-Za-z]:', rel):
raise FrameError(f'the zip has an absolute path: {name}')
parts = [p for p in rel.split('/') if p not in ('', '.')]
if any(p == '..' for p in parts):
raise FrameError(f'the zip has a path that climbs out of it: {name}')
if any(':' in p for p in parts):
# Drive-qualified parts ('C:..') climb out on Windows; ':' is an NTFS stream elsewhere.
raise FrameError(f'the zip has a path with a drive or stream name: {name}')
if not parts or parts[0] == '__MACOSX' or parts[-1] in ('.DS_Store', 'Thumbs.db'):
return None
return '/'.join(parts)
def _redirected(path, expected):
"""True if path is a symlink, or resolves somewhere else (a Windows junction isn't islink)."""
return os.path.islink(path) or os.path.normcase(os.path.realpath(path)) != os.path.normcase(expected)
def _has_links(root):
real = os.path.realpath(root)
for dirpath, dirnames, filenames in os.walk(real):
for n in dirnames + filenames:
if _redirected(os.path.join(dirpath, n), os.path.join(dirpath, n)):
return True
return False
def _stage_folder(src, dest):
"""Copy src to dest; links to files inside src become copies, all other links are left out."""
real = os.path.realpath(src)
inside = lambda p: os.path.normcase(p).startswith(os.path.normcase(real) + os.sep) # noqa: E731
folders, copies = [], [] # decide everything first, so the space check sees the same files
for dirpath, dirnames, filenames in os.walk(real):
out = os.path.join(dest, os.path.relpath(dirpath, real))
folders.append(out)
# Only descend into real folders: not symlinked ones, not junctions (os.walk follows those).
linked = [d for d in dirnames if _redirected(os.path.join(dirpath, d), os.path.join(dirpath, d))]
dirnames[:] = [d for d in dirnames if d not in linked]
for fn in filenames + linked:
target = os.path.realpath(os.path.join(dirpath, fn))
if inside(target) and os.path.isfile(target):
copies.append((target, os.path.join(out, fn)))
if shutil.disk_usage(os.path.dirname(dest)).free < sum(os.path.getsize(t) for t, _ in copies) + 256 * 1024**2:
raise FrameError('not enough space on this computer to stage the folder')
for out in folders:
os.makedirs(out, exist_ok=True)
for target, out in copies:
shutil.copy2(target, out)
return dest
def _below(top, root):
"""The folders _unwrap stepped through from top to root, as 'a/b', or ''. Taken
before staging moves root elsewhere (possibly another drive on Windows)."""
rel = os.path.relpath(root, os.path.realpath(top)).replace(os.sep, '/')
return '' if rel == '.' else rel
def _unwrap(root):
"""Step into a single top-level folder, the usual shape of a zipped build.
Never through a link or junction: the folder you chose (or unpacked) stays the boundary.
"""
root = os.path.realpath(root)
for _ in range(4):
entries = [e for e in os.listdir(root) if e not in ('__MACOSX', '.DS_Store', 'Thumbs.db')]
if len(entries) != 1:
break
only = os.path.join(root, entries[0])
if not os.path.isdir(only) or _redirected(only, only):
break
root = only
return root
def candidates(root, title=''):
"""Executables under root, best launch target first."""
found = []
want = _norm(title)
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = sorted(d for d in dirnames if not os.path.islink(os.path.join(dirpath, d)))
rel_dir = os.path.relpath(dirpath, root)
parts = [] if rel_dir == '.' else rel_dir.split(os.sep)
for fn in sorted(filenames):
full = os.path.join(dirpath, fn)
if os.path.islink(full) or not os.path.isfile(full):
continue
c = classify(full)
if not c or not c['exe']:
continue
stem = _norm(os.path.splitext(fn)[0])
skip = bool(SKIP_RE.search(fn)) or any(SKIP_DIRS.match(p) for p in parts)
match = 2 if want and stem == want else 1 if want and stem and (want in stem or stem in want) else 0
found.append({'path': '/'.join(parts + [fn]), 'format': c['format'], 'arch': c['arch'],
'size': os.path.getsize(full), 'depth': len(parts), 'skip': skip, 'match': match})
found.sort(key=_rank)
_prefer_launcher_script(found)
return found
def _platform_rank(c):
# Native ARM64 first, then Proton, then x86-64 Linux through FEX; scripts are placed separately.
order = {('elf', 'arm64'): 0, ('pe', 'x86_64'): 1, ('elf', 'x86_64'): 2, ('pe', 'x86'): 3, ('pe', 'arm64'): 3}
return order.get((c['format'], c['arch']), 5 if c['format'] == 'script' else 6)
def _rank(c):
return (c['skip'], _platform_rank(c), -c['match'], c['depth'], -c['size'], c['path'])
def _prefer_launcher_script(found):
"""A top-level shell script beats a Linux binary one folder down (run.sh + bin/game)."""
if not found or found[0]['format'] != 'elf' or found[0]['depth'] == 0:
return
for i, c in enumerate(found):
if c['format'] == 'script' and c['depth'] == 0 and not c['skip']:
found.insert(0, found.pop(i))
return
def runtime_for(target, found=()):
"""(compat tool alias, note) for a launch target, or raise FrameError if it can't run."""
fmt, arch = target['format'], target['arch']
if fmt == 'script':
# A script runs in the runtime of the binaries next to it; alone, natively.
elf = next((c for c in found if c['format'] == 'elf' and not c['skip']), None)
if elf:
return runtime_for(elf)[0], 'A shell script; runtime chosen from the Linux binary next to it.'
return 'SteamLinuxRuntime_4-arm64', 'A shell script with no Linux binary beside it; run natively.'
if fmt == 'pe':
if arch not in ('x86_64', 'x86', 'arm64'):
raise FrameError(f"{target['path']} is a Windows program for {arch}, which Proton can't run")
note = 'Windows x86-64 build: Proton runs it through FEX.' if arch == 'x86_64' else \
f'Windows {arch} build under Proton.'
return DEFAULT_PROTON, note
if fmt == 'elf' and arch == 'arm64':
return 'SteamLinuxRuntime_4-arm64', 'Native ARM64 Linux build.'
if fmt == 'elf' and arch == 'x86_64':
return 'SteamLinuxRuntime_4', ("x86-64 Linux build: probably won't start. It needs the x86-64 Steam "
"Linux Runtime 4.0, which the Frame didn't install for a sideloaded title "
"(2026-09-26). Use an ARM64 or Windows build if there is one.")
raise FrameError(f"{target['path']} is a {arch} Linux program; the Frame runs ARM64 and x86-64 (through FEX) only")
def allowed_runtimes(target, found=()):
alias, _ = runtime_for(target, found)
return ['proton-experimental', 'proton-stable'] if RUNTIMES[alias]['steam_play'] else [alias]
def inspect(path, name=None):
"""Read a .zip, folder or executable into an install plan (a JSON-safe dict).
A zip is unpacked into a temporary folder, plan['work']; pass the plan to
discard() when done with it. Raises FrameError if nothing in it can run.
"""
path = os.path.abspath(path)
if not os.path.exists(path):
raise FrameError(f'{path} does not exist')
work = None
try:
if os.path.isdir(path):
root = _unwrap(path)
unwrapped = _below(path, root)
if _has_links(root):
# scp -r follows links, so a link out of the folder could upload
# anything; copy the folder with its links made safe first.
work = tempfile.mkdtemp(prefix=f'{TMP_PREFIX}{os.getpid()}-')
root = _stage_folder(root, os.path.join(work, os.path.basename(root)))
elif path.lower().endswith('.zip'):
work = tempfile.mkdtemp(prefix=f'{TMP_PREFIX}{os.getpid()}-')
extract_zip(path, work)
root = _unwrap(work)
unwrapped = _below(work, root)
elif classify(path):
# A single executable is uploaded on its own; don't copy a whole Downloads folder.
work = tempfile.mkdtemp(prefix=f'{TMP_PREFIX}{os.getpid()}-')
shutil.copy2(path, os.path.join(work, os.path.basename(path)))
root, unwrapped = work, ''
else:
raise FrameError(f'{os.path.basename(path)} is not a .zip, a folder or a program')
title = name or display_name(os.path.basename(path.rstrip('/\\')))
found = candidates(root, title)
if not found:
raise FrameError(f'no Linux or Windows program found in {os.path.basename(path)}')
plan = {'source': os.path.basename(path.rstrip('/\\')), 'name': title, 'id': title_id(title),
'root': root, 'work': work, 'candidates': found,
'unwrapped': unwrapped,
'size': _tree_size(root), 'warnings': []}
_choose(plan, found[0]['path'])
return plan
except BaseException:
if work:
shutil.rmtree(work, ignore_errors=True)
raise
def _tree_size(root):
total = 0
for dirpath, _, filenames in os.walk(root):
for fn in filenames:
full = os.path.join(dirpath, fn)
if os.path.isfile(full) and not os.path.islink(full):
total += os.path.getsize(full)
return total
def _choose(plan, rel, runtime=None):
"""Set plan's launch target (a path relative to root) and its runtime."""
rel = rel.replace('\\', '/')
# A manifest may name the program as it is in the archive, above the folder
# _unwrap stepped into. A path that works as it is always wins.
prefix = plan.get('unwrapped') and plan['unwrapped'] + '/'
if (prefix and rel.startswith(prefix) and not any(c['path'] == rel for c in plan['candidates'])
and not os.path.isfile(os.path.join(plan['root'], *rel.split('/')))):
rel = rel[len(prefix):]
target = next((c for c in plan['candidates'] if c['path'] == rel), None)
if target is None:
full = os.path.realpath(os.path.join(plan['root'], *rel.replace('\\', '/').split('/')))
root = os.path.realpath(plan['root'])
if not (full + os.sep).startswith(root + os.sep) or not os.path.isfile(full):
raise FrameError(f'{rel} is not a file in the title')
c = classify(full)
if not c or not c['exe']:
raise FrameError(f"{rel} isn't a program the Frame can start")
target = {'path': os.path.relpath(full, root).replace(os.sep, '/'), 'format': c['format'],
'arch': c['arch'], 'size': os.path.getsize(full), 'depth': rel.count('/'),
'skip': False, 'match': 0}
alias, note = runtime_for(target, plan['candidates'])
allowed = allowed_runtimes(target, plan['candidates'])
if runtime:
if runtime not in allowed:
raise FrameError(f"{target['path']} can't use {runtime}; choose one of {', '.join(allowed)}")
alias = runtime
plan.update(target=target['path'], format=target['format'], arch=target['arch'], runtime=alias,
runtime_label=RUNTIMES[alias]['label'], runtimes=allowed, note=note)
plan['warnings'] = (['This looks like an installer or helper, not the game itself.'] if target['skip'] else [])
return plan
def discard(plan):
if plan and plan.get('work'):
shutil.rmtree(plan['work'], ignore_errors=True)
def argv_for(rel):
# Valve's client sends the start command as one string (it may carry arguments),
# so a path with spaces is quoted. How Steam splits it is inferred.
return ['"' + rel + '"' if re.search(r'\s', rel) else rel]
def shortcut_parms(gameid, directory, rel, runtime):
"""The JSON steam-client-create-shortcut takes, as devkit_client.new_or_ensure_game builds it."""
settings = {'steam_play': '1' if RUNTIMES[runtime]['steam_play'] else '0'}
if RUNTIMES[runtime]['steam_play']:
# gui2._update_game sends these with every Proton title; debugging stays off.
settings.update(steam_play_debug='0', steam_play_debug_version='2019')
settings['compat_tool'] = runtime
return {'gameid': gameid, 'directory': directory, 'argv': argv_for(rel), 'env': {},
'settings': settings, 'clear_settings': True, 'force_appid': '', 'lepton_args': ''}
# ---- the Frame side ----------------------------------------------------------
def ssh(cmd, input=None, timeout=120):
return frame_android.ssh(cmd, input=input, timeout=timeout)
def _json_out(out, what):
"""The JSON object a devkit-utils script prints last (its logging goes to stderr)."""
for line in reversed(out.strip().splitlines()):
line = line.strip()
if line.startswith('{') or line.startswith('['):
try:
return json.loads(line)
except ValueError:
break
raise FrameError(f'{what} gave no usable answer: {out.strip()[-300:]!r}')
def utils_stamp():
"""Hash of the vendored devkit-utils, compared with the copy on the Frame."""
h = hashlib.sha256()
for dirpath, dirnames, filenames in os.walk(UTILS_LOCAL):
dirnames[:] = sorted(d for d in dirnames if d != '__pycache__')
for fn in sorted(filenames):
if fn.endswith('.pyc'):
continue
full = os.path.join(dirpath, fn)
h.update(os.path.relpath(full, UTILS_LOCAL).replace(os.sep, '/').encode() + b'\0')
with open(full, 'rb') as f:
h.update(f.read())
return h.hexdigest()[:20]
def _json_files(gid):
# Exact names: a glob like Game-*.json would also match another title called Game-Deluxe.
return ' '.join(f'{GAMES}/{gid}-{k}.json' for k in ('argv', 'env', 'settings', 'framecontrol'))
_utils_lock = threading.Lock()
def ensure_utils():
"""Copy frame/devkit-utils to ~/devkit-utils on the Frame unless it's already this version."""
with _utils_lock: # one sync at a time: they share a staging folder
return _ensure_utils()
def _ensure_utils():
stamp = utils_stamp()
have = ssh(f'cat {UTILS}/{STAMP} 2>/dev/null || true', timeout=30).strip()
if have == stamp:
return False
# Merge rather than replace: Valve's own client may have put newer files there.
tmp = f'.{UTILS}.frame-control'
ssh(f'rm -rf {tmp}', timeout=30)
_copy_tree(UTILS_LOCAL, tmp, timeout=300)
ssh(f'mkdir -p {UTILS} && cp -R {tmp}/. {UTILS}/ && rm -rf {tmp} {UTILS}/__pycache__ '
f'&& echo {stamp} > {UTILS}/{STAMP}', timeout=60)
return True
def _copy_tree(src, dest, timeout=3 * 3600, delete=False):
"""Copy a local folder's contents to dest on the Frame (dest ends up a copy of src).
rsync where installed (not on Windows; see server.push_file), else scp -r
into a fresh dest, which is what Windows has. dest must be a plain path.
"""
name = os.path.basename(src.rstrip('/\\'))
opts = frame_android.SSH_OPTS # read now: the server swaps in its multiplexed options
if _rsync():
cmd = ['rsync', '-a', *(['--delete'] if delete else []), '-e', shlex.join(['ssh', *opts]),
src.rstrip('/') + '/', f'{frame_android.FRAME}:{dest.rstrip("/")}/']
else:
# scp -r copies src *into* dest when dest exists, so dest must not.
ssh(f'rm -rf {shlex.quote(dest)}', timeout=60)
cmd = ['scp', *opts, '-r', src, f'{frame_android.FRAME}:{dest}']
try:
subprocess.run(cmd, check=True, capture_output=True, stdin=subprocess.DEVNULL, text=True,
errors='replace', timeout=timeout)
except subprocess.TimeoutExpired:
raise FrameError(f'copying {name} to the Frame timed out')
except subprocess.CalledProcessError as e:
raise FrameError(f'copying {name} to the Frame failed: {(e.stderr or "").strip()[-300:]}')
def _rsync():
# Not on Windows: a Windows rsync (cwRsync, MSYS2) wouldn't take the POSIX -e quoting.
return not frame_host.WINDOWS and bool(shutil.which('rsync'))
_install_lock = threading.Lock()
def install(path, name=None, exe=None, runtime=None, progress=None):
"""Sideload a .zip, folder or executable as a Devkit Game; returns the title dict.
name: the Steam name (sanitised to the title id), default from the file name.
exe: launch target relative to the title's root, default the best candidate.
runtime: a compat tool alias from RUNTIMES that suits the target (e.g.
'proton-stable' instead of the default Proton Experimental).
progress: optional callable(stage_text, fraction 0..1).
"""
plan = inspect(path, name)
try:
return install_plan(plan, name=name, exe=exe, runtime=runtime, progress=progress)
finally:
discard(plan)
def install_plan(plan, name=None, exe=None, runtime=None, progress=None):
"""Install an inspect() plan, optionally with another name, target or runtime."""
if name:
plan['name'], plan['id'] = name, title_id(name)
if exe or runtime:
_choose(plan, exe or plan['target'], runtime)
step = progress or (lambda *a: None)
with _install_lock:
return _install(plan, step)
def _install(plan, step):
gid = plan['id']
if not NEW_ID_RE.match(gid) or gid.lower() in RESERVED_IDS:
raise FrameError(f'bad title id {gid!r}')
step("Syncing Valve's devkit tools to the Frame", 0.02)
ensure_utils()
existed = ssh(f'test -d {GAMES}/{gid} && echo yes || true', timeout=30).strip() == 'yes'
step('Preparing the title folder', 0.05)
ready = _json_out(ssh(f'{PY}steamos-prepare-upload --gameid {gid}', timeout=60), 'steamos-prepare-upload')
directory = str(ready.get('directory') or '')
if not DIR_RE.match(directory) or not directory.endswith(f'/{GAMES}/{gid}'):
raise FrameError(f'steamos-prepare-upload returned an unexpected folder {directory!r}')
registered = False
try:
step(f"Copying {plan['size'] / 1e6:.0f} MB to the Frame", 0.1)
if _rsync():
_copy_tree(plan['root'], directory, delete=True)
else:
part = f"{directory.rsplit('/', 1)[0]}/.{gid}.upload"
_copy_tree(plan['root'], part)
ssh(f'rm -rf {directory} && mv {part} {directory}', timeout=120)
# Same modes Valve's client gives an upload (rsync --chmod=Du=rwx,Dgo=rx,Fu=rwx,Fog=rx).
ssh(f'chmod -R 755 {directory}', timeout=300)
step('Registering with Steam', 0.9)
parms = shortcut_parms(gid, directory, plan['target'], plan['runtime'])
reply = _json_out(ssh(f'{PY}steam-client-create-shortcut --parms {shlex.quote(json.dumps(parms))}',
timeout=90), 'steam-client-create-shortcut')
meta = {'id': gid, 'name': plan['name'], 'target': plan['target'], 'runtime': plan['runtime'],
'source': plan['source'], 'size': plan['size'], 'installed': time.strftime('%Y-%m-%dT%H:%M:%S')}
ssh(f'cat > {GAMES}/{gid}-framecontrol.json', input=json.dumps(meta, indent=1), timeout=30)
registered = True # the files stay: Steam registers them once it's running
if 'error' in reply:
err = str(reply['error']).strip().rstrip('.')
hint = ' With Steam running on the Frame, install it again.' if 'not running' in err else ''
raise FrameError(f"Uploaded, but Steam didn't register it: {err}.{hint}")
step('Done', 1.0)
meta.update(runtime_label=RUNTIMES[plan['runtime']]['label'], steam=str(reply.get('success', '')).strip())
return meta
finally:
if not registered and not existed:
# A first install that failed part-way: don't leave an orphan folder behind.
try:
ssh(f'rm -rf {GAMES}/{gid} {GAMES}/.{gid}.upload {_json_files(gid)}', timeout=60)
except FrameError:
pass
LIST_SCRIPT = r'''
import json, os
root = os.path.expanduser('~/devkit-game')
reserved = %r
out = []
for d in sorted(os.listdir(root)) if os.path.isdir(root) else []:
if d.startswith('.') or d.lower() in reserved or not os.path.isdir(os.path.join(root, d)):
continue
t = {'id': d}
for key, suffix in (('settings', '-settings.json'), ('argv', '-argv.json'), ('meta', '-framecontrol.json')):
try:
with open(os.path.join(root, d + suffix)) as f:
t[key] = json.load(f)
except (OSError, ValueError):
t[key] = None
out.append(t)
print(json.dumps(out))
''' % (RESERVED_IDS,)
def list_titles():
"""The Devkit Games on the Frame (any uploaded by Valve's client too)."""
raw = _json_out(ssh('python3 -', input=LIST_SCRIPT, timeout=30), 'the title list')
titles = []
for t in raw if isinstance(raw, list) else []:
if not ID_RE.match(str(t.get('id', ''))):
continue
settings, meta = t.get('settings') or {}, t.get('meta') or {}
argv = t.get('argv') if isinstance(t.get('argv'), list) else []
alias = str(settings.get('compat_tool') or '')
titles.append({'id': t['id'], 'name': str(meta.get('name') or t['id']),
'target': str(meta.get('target') or (argv[0] if argv else '')),
'runtime': alias, 'runtime_label': RUNTIMES.get(alias, {}).get('label', alias or 'not set'),
'source': str(meta.get('source') or ''), 'size': meta.get('size'),
'installed': meta.get('installed'), 'registered': t.get('settings') is not None,
'frame_control': bool(meta)})
return titles
def _check_id(gid):
gid = str(gid or '')
if not ID_RE.match(gid) or gid.lower() in RESERVED_IDS:
raise FrameError(f'bad title id {gid!r}')
if ssh(f'test -d {GAMES}/{gid} && echo yes || true', timeout=30).strip() != 'yes':
raise FrameError(f'{gid} is not installed')
return gid
def launch(gid):
gid = _check_id(gid)
ensure_utils()
# steam-devkit-rpc logs 'success' when Steam answers, but exits 0 after a
# 5 s timeout too, so read its log (stderr) rather than trust the exit code.
out = ssh(f'{PY}steam-devkit-rpc run-game gameid={gid} 2>&1', timeout=60)
if 'success' not in [line.strip() for line in out.splitlines()]:
raise FrameError(f"Steam didn't confirm the launch: {out.strip()[-300:] or 'no answer'}")
return {'id': gid}
def remove(gid):
# Not while an install runs: it could be this title, half copied or about to register.
if not _install_lock.acquire(blocking=False):
raise FrameError('an install is running; remove the title when it has finished')
try:
gid = _check_id(gid)
ensure_utils()
# steamos-delete removes the folder and syncs Steam's shortcuts; its json files stay, so clear them too.
ssh(f'{PY}steamos-delete --delete-title {gid}', timeout=120)
ssh(f'rm -f {_json_files(gid)}', timeout=30)
return {'id': gid}
finally:
_install_lock.release()
def public(plan):
"""A plan without its local paths, for the UI, with the runtimes each candidate may use."""
out = {k: v for k, v in plan.items() if k not in ('root', 'work', 'candidates')}
out['candidates'] = []
for c in plan['candidates']:
c = dict(c)
try:
c['runtimes'] = allowed_runtimes(c, plan['candidates'])
except FrameError as e:
c['runtimes'], c['blocked'] = [], str(e)
out['candidates'].append(c)
out['runtime_labels'] = {k: v['label'] for k, v in RUNTIMES.items()}
return out
def main():
cmd, *args = sys.argv[1:] or ['help']
def opt(flag):
return args[args.index(flag) + 1] if flag in args and args.index(flag) + 1 < len(args) else None
try:
if cmd == 'inspect' and args:
plan = inspect(args[0], opt('--name'))
try:
if opt('--exe') or opt('--runtime'):
_choose(plan, opt('--exe') or plan['target'], opt('--runtime'))
r = public(plan)
finally:
discard(plan)
elif cmd == 'install' and args:
r = install(args[0], name=opt('--name'), exe=opt('--exe'), runtime=opt('--runtime'),
progress=lambda text, _: print(text + '…', file=sys.stderr))
elif cmd == 'list':
r = list_titles()
elif cmd in ('launch', 'remove') and args:
r = globals()[cmd](args[0])
else:
sys.exit(__doc__)
except FrameError as e:
sys.exit(f'error: {e}')
print(json.dumps(r, indent=1))
if __name__ == '__main__':
main()
Reference in new issue View Git Blame Copy Permalink