Add MTGA Assistant: native Linux deck and game tracker

PySide6 app that follows MTG Arena's Player.log under Steam/Proton and shows
my remaining library with draw odds, my zones, and everything revealed about
the opponent (board, graveyard, exile, known hand/library, seen this match).

- Auto-detects the Steam library, Proton prefix log and MTGA install
- Card names/costs/types from MTGA's local SQLite card database
- Tracks card identity across zone changes, shuffles and Bo3 games
- Scryfall card images with a lazy, size-capped LRU disk cache
- Live log following with restart detection, plus --replay mode
- Flags log gaps where MTGA omits oversized game-state messages

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Aeth3randClaude Opus 5.5 committed 2026-10-16 09:56:12 -04:00
1 parent 734336168c
commit 7dc2ce0da8
21 files changed
+2736

No files matched your search

+173
View File
@@ -0,0 +1,173 @@
"""Card images from Scryfall with a size-capped on-disk LRU cache.
Images are fetched lazily, only for cards that actually show up, and stored as
``<cache>/images/<grpId>.jpg``. A file's mtime is bumped whenever it is used, and
the least recently used files are evicted when the cache exceeds its cap.
"""
from __future__ import annotations
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from PySide6.QtCore import QObject, Signal
from PySide6.QtGui import QPixmap
from .carddb import CardDB
USER_AGENT = "mtga-assistant/0.1 (personal desktop tracker)"
SCRYFALL = "https://api.scryfall.com"
MIN_REQUEST_INTERVAL = 0.11 # Scryfall asks for <= 10 requests/second
class ImageCache(QObject):
image_ready = Signal(int) # grpId
size_changed = Signal(int) # bytes on disk
_downloaded = Signal(int)
def __init__(self, cache_dir: Path, cards: CardDB, max_bytes: int):
super().__init__()
self.dir = cache_dir / "images"
self.dir.mkdir(parents=True, exist_ok=True)
self.cards = cards
self.max_bytes = max_bytes
self._pool = ThreadPoolExecutor(max_workers=3, thread_name_prefix="scryfall")
self._lock = threading.Lock()
self._inflight: set[int] = set()
self._failed: set[int] = set()
self._last_request = 0.0
self._pixmaps: dict[int, QPixmap] = {}
# Emitted from worker threads; Qt queues delivery onto this object's (GUI) thread.
self._downloaded.connect(self._on_downloaded)
self.evict()
def path_for(self, grp_id: int) -> Path:
return self.dir / f"{grp_id}.jpg"
def pixmap(self, grp_id: int) -> QPixmap | None:
"""Return the image if cached; otherwise start a download and return None."""
if pm := self._pixmaps.get(grp_id):
return pm
path = self.path_for(grp_id)
if path.exists():
try:
path.touch()
except OSError:
pass
pm = QPixmap(str(path))
if not pm.isNull():
if len(self._pixmaps) > 200:
self._pixmaps.clear()
self._pixmaps[grp_id] = pm
return pm
self.request(grp_id)
return None
def request(self, grp_id: int) -> None:
with self._lock:
if grp_id in self._inflight or grp_id in self._failed:
return
self._inflight.add(grp_id)
self._pool.submit(self._download, grp_id)
def is_failed(self, grp_id: int) -> bool:
return grp_id in self._failed
# -- worker side ---------------------------------------------------------
def _throttle(self) -> None:
with self._lock:
wait = self._last_request + MIN_REQUEST_INTERVAL - time.monotonic()
self._last_request = max(time.monotonic(), self._last_request + MIN_REQUEST_INTERVAL)
if wait > 0:
time.sleep(wait)
def _candidate_urls(self, grp_id: int) -> list[str]:
q = "format=image&version=normal"
urls = [f"{SCRYFALL}/cards/arena/{grp_id}?{q}"]
info = self.cards.get(grp_id)
if info.set_code and info.collector_number:
set_code = urllib.parse.quote(info.set_code.lower())
cn = urllib.parse.quote(info.collector_number)
urls.append(f"{SCRYFALL}/cards/{set_code}/{cn}?{q}")
if info.name and not info.name.startswith(("Unknown card", "Card #")):
exact = urllib.parse.quote(info.name.split(" // ")[0])
urls.append(f"{SCRYFALL}/cards/named?exact={exact}&{q}")
return urls
def _download(self, grp_id: int) -> None:
ok = False
try:
for url in self._candidate_urls(grp_id):
self._throttle()
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*"})
try:
with urllib.request.urlopen(req, timeout=20) as resp:
data = resp.read()
except urllib.error.HTTPError as e:
if e.code == 404:
continue
break
except (urllib.error.URLError, TimeoutError, OSError):
break
tmp = self.path_for(grp_id).with_suffix(".part")
tmp.write_bytes(data)
tmp.replace(self.path_for(grp_id))
ok = True
break
finally:
with self._lock:
self._inflight.discard(grp_id)
if not ok:
self._failed.add(grp_id)
if ok:
self._downloaded.emit(grp_id)
def _on_downloaded(self, grp_id: int) -> None:
self.evict()
self.image_ready.emit(grp_id)
# -- cache management ------------------------------------------------------
def disk_usage(self) -> int:
return sum(p.stat().st_size for p in self.dir.glob("*.jpg"))
def evict(self) -> None:
files = []
for p in self.dir.glob("*.jpg"):
try:
st = p.stat()
except OSError:
continue
files.append((st.st_mtime, st.st_size, p))
total = sum(f[1] for f in files)
if total > self.max_bytes:
for _, size, p in sorted(files):
try:
p.unlink()
except OSError:
continue
total -= size
if p.stem.isdigit():
self._pixmaps.pop(int(p.stem), None)
if total <= self.max_bytes * 0.9:
break
self.size_changed.emit(total)
def clear(self) -> None:
for p in self.dir.glob("*"):
try:
p.unlink()
except OSError:
pass
self._pixmaps.clear()
self._failed.clear()
self.size_changed.emit(0)
def shutdown(self) -> None:
self._pool.shutdown(wait=False, cancel_futures=True)