#!/usr/bin/env python3 """Mine LlamaCasty's own logs for per-session performance evidence. Why this exists (2026-11-09): a recording and a livestream share the entire pipeline — capture, compositing, one ffmpeg at the selected tier. Only the sink differs. That makes recordings the cleanest evidence for the machine-side minimum configuration, because the network cannot pollute the numbers, and livestreams the evidence for everything else. Usage python3 scripts/session-stats.py # every log under %APPDATA%\\ytLlive python3 scripts/session-stats.py path/to/startup.log ... python3 scripts/session-stats.py --json What a "session" is: from `FramePump started` (which carries the tier and scene) to the next start or end of file, with the sink inferred from the markers inside it: Test session dock / Test chat simulation -> TEST (private broadcast, real socket) Broadcast created / Broadcast bound -> LIVE Recording ended / Recording discarded -> RECORD (both -> LIVE+RECORD) no marker -> UNKNOWN (crashed launch, or a paused session) TEST is the one that looks like LIVE and behaves like RECORD: the full pipeline and the socket, but no audience and simulated chat. It is good machine evidence and useless congestion evidence. `holds` is the published-minimum rule, deliberately strict: zero pump drops, zero stalls, and worst render under HALF the frame budget — headroom, not survivability. """ from __future__ import annotations import argparse import glob import json import os import re import sys from dataclasses import dataclass, field, asdict from statistics import mean TIER_RE = re.compile( r"FramePump started \((\d+)×(\d+) @ (\d+) fps\).*scene='([^']*)'") STATS_RE = re.compile( r"FramePump stats: (\d+)/(\d+) frames per (\d+)s from (\d+) renders, " r"avg render ([\d.]+)ms.*?avg submit ([\d.]+)ms, avg wait ([\d.]+)ms, " r"worst render ([\d.]+)ms, worst submit ([\d.]+)ms, dropped (\d+), stalls (\d+)") CAPTURE_RE = re.compile( r"ScreenCapture telemetry \[([^\]]+)\]: \d+ frames in [\d.]+s \(([\d.]+)/s\), " r"conv avg (\d+)ms max (\d+)ms") STAMP_RE = re.compile(r"^(\d{4}-\d\d-\d\d \d\d:\d\d:\d\d\.\d{3})") BUILD_RE = re.compile(r"Build ([0-9a-f]{6,})") TEST_MARKERS = ("Test session dock open", "Test chat simulation") LIVE_MARKERS = ("Broadcast created:", "Broadcast bound:") RECORD_MARKERS = ("Recording ended", "Recording discarded") @dataclass class Session: log: str started: str = "" build: str = "" scene: str = "" width: int = 0 height: int = 0 fps: int = 0 test: bool = False live: bool = False record: bool = False samples: int = 0 frames: int = 0 dropped: int = 0 stalls: int = 0 avg_render: float = 0.0 worst_render: float = 0.0 avg_submit: float = 0.0 avg_wait: float = 0.0 capture_rates: list[float] = field(default_factory=list) capture_conv: list[int] = field(default_factory=list) stall_events: int = 0 @property def sink(self) -> str: if self.test: return "TEST" if self.live and self.record: return "LIVE+REC" if self.live: return "LIVE" if self.record: return "RECORD" return "UNKNOWN" @property def tier(self) -> str: return f"{self.width}x{self.height}@{self.fps}" @property def seconds(self) -> float: return self.samples * 5.0 @property def measured_fps(self) -> float: return self.frames / self.seconds if self.seconds else 0.0 @property def verdict(self) -> str: # Deliberately judged on the AVERAGE: `worst render` includes one-off startup hitches # (100 ms while the first frame bakes) that say nothing about sustained capability. budget = 1000.0 / self.fps if self.fps else 0.0 if not self.samples: return "no-data" if self.dropped or self.stalls or self.avg_render > budget: return "over-budget" if self.avg_render > budget / 2: return "strained" return "holds" def default_root() -> str: r"""%APPDATA%\ytLlive, including when the script is run from WSL (no APPDATA set).""" if os.environ.get("APPDATA"): return os.path.join(os.environ["APPDATA"], "ytLlive") # WSL / Linux: the Windows profile is mounted under /mnt/c/Users/. candidates = sorted(glob.glob("/mnt/c/Users/*/AppData/Roaming/ytLlive")) if candidates: return max(candidates, key=os.path.getmtime) return os.path.expanduser("~/AppData/Roaming/ytLlive") def default_logs() -> list[str]: root = default_root() paths = sorted(glob.glob(os.path.join(root, "startup*.log"))) paths += sorted(glob.glob(os.path.join(root, "instances", "*", "startup*.log"))) return paths def parse(paths: list[str]) -> list[Session]: sessions: list[Session] = [] for path in paths: try: lines = open(path, encoding="utf-8", errors="ignore").read().splitlines() except OSError as exc: print(f"skip {path}: {exc}", file=sys.stderr) continue current: Session | None = None build = "" for line in lines: build_match = BUILD_RE.search(line) if build_match: build = build_match.group(1) stamp = STAMP_RE.match(line) tier = TIER_RE.search(line) if tier: current = Session( log=os.path.basename(path), started=stamp.group(1) if stamp else "", build=build, width=int(tier.group(1)), height=int(tier.group(2)), fps=int(tier.group(3)), scene=tier.group(4), ) sessions.append(current) continue if current is None: continue if "FramePump stall:" in line: current.stall_events += 1 continue stats = STATS_RE.search(line) if stats: current.samples += 1 current.frames += int(stats.group(1)) current.avg_render += float(stats.group(5)) current.avg_submit += float(stats.group(6)) current.avg_wait += float(stats.group(7)) current.worst_render = max(current.worst_render, float(stats.group(8))) current.dropped += int(stats.group(10)) current.stalls += int(stats.group(11)) continue capture = CAPTURE_RE.search(line) if capture: current.capture_rates.append(float(capture.group(2))) current.capture_conv.append(int(capture.group(3))) continue if any(m in line for m in TEST_MARKERS): current.test = True if any(m in line for m in LIVE_MARKERS): current.live = True if any(m in line for m in RECORD_MARKERS): current.record = True for s in sessions: if s.samples: s.avg_render /= s.samples s.avg_submit /= s.samples s.avg_wait /= s.samples return [s for s in sessions if s.samples] def table(sessions: list[Session]) -> str: head = (f"{'started':<24}{'build':<10}{'sink':<8}{'tier':<15}{'scene':<10}{'dur':>6}" f"{'fps':>6}{'drop':>6}{'stalls':>7}{'avgRen':>8}{'worstRen':>10}" f"{'cap/s':>7}{'conv':>6} verdict") out = [head, "-" * len(head)] for s in sessions: cap = mean(s.capture_rates) if s.capture_rates else 0.0 conv = mean(s.capture_conv) if s.capture_conv else 0 out.append( f"{s.started:<24}{s.build[:8]:<10}{s.sink:<8}{s.tier:<15}{s.scene[:9]:<10}{s.seconds:>5.0f}s" f"{s.measured_fps:>6.1f}{s.dropped:>6}{s.stalls:>7}{s.avg_render:>8.1f}" f"{s.worst_render:>10.1f}{cap:>7.1f}{conv:>6.0f} {s.verdict}") return "\n".join(out) def summary(sessions: list[Session]) -> str: groups: dict[tuple[str, str], list[Session]] = {} for s in sessions: groups.setdefault((s.tier, s.sink), []).append(s) out = ["", f"{'tier':<16}{'sink':<9}{'sessions':>9}{'holds':>7}{'strained':>10}{'over':>6}"] for (tier, sink), rows in sorted(groups.items()): holds = sum(1 for r in rows if r.verdict == "holds") strained = sum(1 for r in rows if r.verdict == "strained") over = sum(1 for r in rows if r.verdict == "over-budget") out.append(f"{tier:<16}{sink:<9}{len(rows):>9}{holds:>7}{strained:>10}{over:>6}") return "\n".join(out) def main() -> int: ap = argparse.ArgumentParser(description="Per-session performance evidence from LlamaCasty logs.") ap.add_argument("logs", nargs="*", help="log files (default: every startup*.log under %%APPDATA%%\\ytLlive)") ap.add_argument("--json", action="store_true", help="machine-readable output") args = ap.parse_args() paths = args.logs or default_logs() if not paths: print("no logs found", file=sys.stderr) return 1 sessions = parse(paths) if args.json: print(json.dumps([asdict(s) | {"sink": s.sink, "verdict": s.verdict, "measured_fps": round(s.measured_fps, 2)} for s in sessions], indent=2)) return 0 print(table(sessions)) print(summary(sessions)) print("\nholds = 0 drops, 0 stalls, worst render under half the frame budget. " "RECORD and TEST isolate the machine; LIVE adds the network and the audience.") return 0 if __name__ == "__main__": raise SystemExit(main())