"""Recompute this study from original ETL and CapFrameX files. Python 3, stdlib only. Windows WPT 10+ is required for ETL analysis. No tracing is started, no settings are changed. Only --output is written. Original inputs are never modified. Example from an extracted bundle: python analysis/analyze.py --wpr-root raw/wpr --captures-root raw/capframex --output recalculated """ import argparse import bisect import collections import csv import json import math from pathlib import Path import re import statistics import subprocess MODES = {"WindowsClean": "clean", "BoosterXOptimal": "safe", "BoosterXAdvanced": "advanced", "BoosterXHardcore": "hardcore"} COMMENTS = {"clean": "clean", "safe advanced": "safe", "advanced advanced": "advanced", "hardcore": "hardcore"} ORDER = list(MODES.values()) VARIANCE_BINS = ["<2 ms", "2–<4 ms", "4–<8 ms", "8–<12 ms", ">=12 ms"] VARIANCE_SOURCE = "https://github.com/CXWorld/CapFrameX/blob/383bc029adba29bdbedf96b922713dec48acf76b/source/CapFrameX.Statistics.NetStandard/FrametimeStatisticProvider.cs" def variance_counts(frames): """CapFrameX v1.8.6 GetFrametimeVariancePercentages bins, before rounding. Absolute adjacent differences, separate runs, no outlier removal. This is not statistical variance. Exact thresholds belong to the next bin. """ counts = [0] * 5 for previous, current in zip(frames, frames[1:]): counts[bisect.bisect_right([2, 4, 8, 12], abs(current - previous))] += 1 return counts def read_json(path): return json.loads(path.read_text(encoding="utf-8-sig")) def percentile(values, q): ordered = sorted(values) i = (len(ordered) - 1) * q return ordered[math.floor(i)] + (ordered[math.ceil(i)] - ordered[math.floor(i)]) * (i % 1) def xperf(path, trace, args): result = subprocess.run([str(path), "-i", str(trace), "-a", *args], capture_output=True) if result.returncode: raise RuntimeError(result.stderr.decode("utf-8", "replace")) return result.stdout.decode("utf-8", "replace") def population_at_end(path, trace, end): counts = collections.Counter() for line in xperf(path, trace, ["process", "-thread"]).splitlines(): match = re.match(r"\s*(MIN|\d+),\s*(MAX|\d+),\s*(Process|Thread),.*?,\s*(.*?)\s*\(\s*(\d+)\)", line) if match and match[5] != "0": a = 0 if match[1] == "MIN" else int(match[1]) b = math.inf if match[2] == "MAX" else int(match[2]) counts[match[3]] += a <= end < b return counts def analyze_trace(path, trace, mode, cpus): marks = xperf(path, trace, ["marks", "-summary"]) phase = "IDLE" if trace.stem == "Idle" else "POSTBOOT" endpoints = [] for suffix in ("START", "END"): matches = re.findall(r"Mark,\s*(\d+),\s*BX_" + phase + "_" + suffix + r"\s*$", marks, re.M) if len(matches) != 1: raise ValueError(f"Ambiguous {phase}/{suffix} marker") endpoints.append(int(matches[0])) start, end = endpoints if end <= start: raise ValueError("Invalid marker order") seconds = (end - start) / 1e6 limits = ["-range", str(start), str(end)] header = xperf(path, trace, ["tracestats"]) loss = {} for kind in ("Events", "Buffers"): matches = re.findall(r"Total # Lost " + kind + r"\s*:\s*(\d+)", header) if len(matches) != 1: raise ValueError("Missing/ambiguous trace loss metadata") loss[kind.lower()] = int(matches[0]) # Weight all per-CPU sampled utilization rows by their exact duration. bins = [[] for _ in range(5)] for line in xperf(path, trace, ["profile", "-util", "1000000", *limits]).splitlines(): columns = line.split(",") if len(columns) == cpus + 2 and columns[0].strip().isdigit(): values = list(map(float, columns)) a, b = max(start, values[0]), min(end, values[1]) if b > a: bins[min(4, int((a - start) / 60000000))].append((b - a, statistics.mean(values[2:]))) if not all(bins): raise ValueError("Incomplete CPU sampling coverage") weighted = lambda rows: sum(w * v for w, v in rows) / sum(w for w, v in rows) cpu = weighted([pair for bucket in bins for pair in bucket]) # Stream the full trace: windowed events plus end-of-trace process rundown. # No stack payloads, paths, names or identifiers are copied to the aggregates. proc = subprocess.Popen([str(path), "-i", str(trace), "-a", "dumper"], stdout=subprocess.PIPE, stderr=subprocess.PIPE) counts = collections.Counter() disk_bytes = collections.Counter() memory = {} wanted = {b"CSwitch", b"ReadyThread", b"DPC", b"DPCTmr", b"Interrupt", b"DiskRead", b"DiskWrite", b"HardFault", b"P-Start", b"T-Start", b"T-End", b"T-DCStart", b"ProcessPerfCtrRundown"} for line in proc.stdout: kind = line.split(b",", 1)[0].strip() if kind not in wanted: continue fields = [v.strip() for v in line.split(b",")] if not fields[1].isdigit(): continue timestamp = int(fields[1]) if kind == b"ProcessPerfCtrRundown": pid = int(re.search(rb"\(\s*(\d+)\)", fields[2])[1]) if pid and timestamp >= end: memory[pid] = (timestamp, int(fields[8]), int(fields[10]), int(fields[13])) if not start <= timestamp < end: continue counts[kind] += 1 if kind in {b"DiskRead", b"DiskWrite"}: size = int(fields[7], 0) disk_bytes[kind] += size if kind == b"DiskWrite" and b"wpr_initiated_" in line.lower(): disk_bytes[b"WPR"] += size error = proc.stderr.read() if proc.wait(): raise RuntimeError(error.decode("utf-8", "replace")) if not memory: raise ValueError("Missing end-of-trace process memory rundown") # Lifetimes at the end marker; excludes all Idle threads, includes System. population = population_at_end(path, trace, end) return dict(mode=mode, phase=trace.stem, seconds=seconds, logical_processors=cpus, cpu_sampled_percent=cpu, cpu_peak_1s_percent=max(v for bucket in bins for _, v in bucket), cpu_estimated_core_seconds=cpu / 100 * seconds * cpus, cswitch_count=counts[b"CSwitch"], cswitch_per_second=counts[b"CSwitch"] / seconds, processes_at_end=population["Process"], lost_events=loss["events"], lost_buffers=loss["buffers"], processes_started=counts[b"P-Start"], threads_at_end=population["Thread"], threads_started=counts[b"T-Start"], ready_thread_per_second=counts[b"ReadyThread"] / seconds, dpc_count=counts[b"DPC"] + counts[b"DPCTmr"], dpc_per_second=(counts[b"DPC"] + counts[b"DPCTmr"]) / seconds, isr_count=counts[b"Interrupt"], isr_per_second=counts[b"Interrupt"] / seconds, hard_fault_count=counts[b"HardFault"], disk_read_count=counts[b"DiskRead"], disk_write_count=counts[b"DiskWrite"], disk_read_mib=disk_bytes[b"DiskRead"] / 1048576, disk_write_mib=disk_bytes[b"DiskWrite"] / 1048576, wpr_disk_write_mib=disk_bytes[b"WPR"] / 1048576, process_working_sets_mib=sum(r[1] for r in memory.values()) / 1048576, process_commit_mib=sum(r[2] for r in memory.values()) / 1048576, process_handles=sum(r[3] for r in memory.values()), memory_rundown_min_offset_seconds=(min(r[0] for r in memory.values()) - end) / 1e6, memory_rundown_max_offset_seconds=(max(r[0] for r in memory.values()) - end) / 1e6, cpu_percent_by_minute=[weighted(bucket) for bucket in bins]) def analyze_fps(path): document = read_json(path) mode = COMMENTS[document["Info"]["Comment"].strip().lower()] if len(document["Runs"]) != 1: raise ValueError("Expected one run per capture") run = document["Runs"][0] data = run["CaptureData"] frames, times = data["MsBetweenPresents"], data["TimeInSeconds"] if len(frames) != len(times) or not frames or not all(math.isfinite(v) and v > 0 for v in frames): raise ValueError("Invalid frame samples") if not all(b > a for a, b in zip(times, times[1:])): raise ValueError("Non-monotonic frame timestamps") sensor_summary = {} for sensor in (run.get("SensorData2") or {}).values(): if sensor["Type"] in ("Clock", "Load", "Temperature", "Power"): values = [v for v in sensor["Values"] if isinstance(v, (int, float)) and math.isfinite(v)] if values: sensor_summary[sensor["Name"] + " / " + sensor["Type"]] = dict( mean=statistics.mean(values), min=min(values), max=max(values)) return dict(mode=mode, frames=len(frames), seconds=times[-1] - times[0], mean_frametime_ms=statistics.mean(frames), mean_adjacent_difference_ms=statistics.mean(abs(a-b) for a, b in zip(frames, frames[1:])), variance_pair_count=len(frames)-1, variance_bin_counts=variance_counts(frames), avg_fps=1000 / statistics.mean(frames), p1_fps=1000 / percentile(frames, .99), p01_fps=1000 / percentile(frames, .999), slowest_one_percent_fps=1000 / statistics.mean(sorted(frames, reverse=True)[:math.ceil(len(frames) * .01)]), p99_frametime_ms=percentile(frames, .99), max_frametime_ms=max(frames), frames_over_8ms=sum(v > 8 for v in frames), sensors=sensor_summary) def write_csv(path, rows, excluded=()): columns = [key for key in rows[0] if key not in excluded] with path.open("w", encoding="utf-8-sig", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=columns, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--wpr-root", type=Path, required=True) parser.add_argument("--captures-root", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--xperf", type=Path, default=Path(r"C:\Program Files (x86)\Windows Kits\10\Windows Performance Toolkit\xperf.exe")) args = parser.parse_args() wpr, excluded = [], [] for folder in sorted(args.wpr_root.iterdir()): if not folder.is_dir() or not (folder / "run.json").exists(): continue record = read_json(folder / "run.json") if record["Kind"] != "Measurement": continue mode = MODES[record["Mode"]] traces = [folder / "PostBoot.etl", folder / "Idle.etl"] if not all(f.exists() for f in traces): excluded.append(dict(mode=mode, reason="Incomplete attempt: no complete ETL pair")) continue cpus = int(read_json(folder / "environment.json")["LogicalProcessors"]) for trace in traces: print(f"Analyzing {mode}/{trace.stem}", flush=True) wpr.append(analyze_trace(args.xperf, trace, mode, cpus)) fps = [] # CreationDate, not the unpadded time in filenames, defines run order. captures = sorted(args.captures_root.glob("*.json"), key=lambda p: read_json(p)["Info"]["CreationDate"]) counters = collections.Counter() for capture in captures: result = analyze_fps(capture) counters[result["mode"]] += 1 result["run"] = counters[result["mode"]] fps.append(result) if len(wpr) != 8 or any(sum(r["mode"] == m for r in wpr) != 2 for m in ORDER): raise ValueError("Expected four complete WPR pairs") if dict(counters) != dict.fromkeys(ORDER, 2): raise ValueError("Expected two FPS captures per stage") wpr.sort(key=lambda r: (ORDER.index(r["mode"]), r["phase"] != "PostBoot")) fps.sort(key=lambda r: (ORDER.index(r["mode"]), r["run"])) summary = [] for mode in ORDER: rows = [r for r in fps if r["mode"] == mode] summary.append(dict(mode=mode, runs=len(rows), **{ key: statistics.mean(r[key] for r in rows) for key in ("avg_fps", "p1_fps", "p01_fps", "slowest_one_percent_fps", "mean_frametime_ms", "p99_frametime_ms", "mean_adjacent_difference_ms")}, min_run_fps=min(r["avg_fps"] for r in rows), max_run_fps=max(r["avg_fps"] for r in rows))) for row in summary: row["fps_change_percent"] = (row["avg_fps"] / summary[0]["avg_fps"] - 1) * 100 variance = summarize_variance(fps) args.output.mkdir(parents=True, exist_ok=True) output = dict(protocol="boosterx-windows-cs2/3", measured="2026-10-24", wpr=wpr, fps=fps, fps_summary=summary, frametime_variance=variance, frametime_variance_source=VARIANCE_SOURCE, excluded_attempts=excluded) (args.output / "results.json").write_text(json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") write_csv(args.output / "wpr.csv", wpr, ("cpu_percent_by_minute",)) write_csv(args.output / "fps.csv", fps, ("sensors", "variance_bin_counts")) write_csv(args.output / "fps-summary.csv", summary) write_csv(args.output / "frametime-variance.csv", variance) print("Written results.json, wpr.csv, fps.csv, fps-summary.csv, frametime-variance.csv", flush=True) def summarize_variance(fps): rows = [] for mode in ORDER: runs = [r for r in fps if r["mode"] == mode] pairs = sum(r["variance_pair_count"] for r in runs) for i, label in enumerate(VARIANCE_BINS): count = sum(r["variance_bin_counts"][i] for r in runs) percentage = count / pairs * 100 base = percentage if mode == "clean" else rows[i]["percent"] rows.append(dict(mode=mode, bin=label, count=count, pairs=pairs, percent=percentage, delta_percentage_points=percentage-base, relative_change_percent=(percentage/base-1)*100 if base else None)) return rows if __name__ == "__main__": main()