From a public demo to a coaching clip: the pipeline, end to end

Why this post exists

The Lineups section of CSReplays is a library of pro utility — smokes, mollies and flashes mined from real professional matches, each clustered from where pros *actually* land the nade and paired with a reference clip you can watch.

None of that is hand-authored. It is the output of a five-stage pipeline that starts at a public demo on HLTV and ends at an MP4 sitting in our object storage. This post walks the whole thing, with the real functions that run at each stage. We would rather show you exactly how it works than hand-wave about "AI-powered analysis."

The stages, in dependency order:

• Scrape — pull GOTV demos off HLTV, for free, no scraping service
• Parse — run awpy over every demo to extract every grenade's release + landing position
• Cluster — group thousands of landings into the handful of spots pros repeatedly target
• Render — turn the representative throw of each cluster into a clip via DEMO-SLAP
• Upload — store the clip privately in R2 and serve it through a signed, auth-gated URL

One honest note on ordering: the parse step has to run *before* clustering, because you cannot cluster grenade landings you have not extracted yet. So the real flow is scrape → parse → cluster → render → upload.

Stage 1 — Scraping HLTV for free

HLTV sits behind Cloudflare's "managed challenge." Plain curl, cloudscraper, and even headless Chromium all bounce. The trick that works is curl_cffi, which impersonates a real Chrome TLS + HTTP/2 fingerprint — Cloudflare's bot check waves it through. So we get HLTV's HTML directly, for free, from anywhere:

from curl_cffi import requests as cr

_session = None
def session():
    global _session
    if _session is None:
        _session = cr.Session(impersonate="chrome", timeout=60)
    return _session


def _is_challenge(text: str) -> bool:
    t = text[:4000].lower()
    return any(w in t for w in ("just a moment", "cf-challenge",
                                "attention required", "enable javascript and cookies"))


def get_html(url: str) -> str:
    r = session().get(url)
    if r.status_code != 200 or _is_challenge(r.text):
        raise SystemExit(f"HLTV fetch failed ({url}): status {r.status_code}")
    return r.text

That single impersonate="chrome" flag is the entire bypass. The rest is parsing. From a match page we pull the demo download links and build a folder name that follows our corpus convention so the downstream ingest can recover the event and the two teams:

def parse_match(match_url: str):
    html = get_html(match_url)
    links = sorted({HLTV + rel for rel in re.findall(r'/download/demo/\d+', html)})
    if not links:
        return [], None
    mid = (re.search(r'/matches/(\d+)/', match_url) or [None, "0"])[1]
    teams = re.findall(r'class="teamName"[^>]*>([^<]+)<', html)
    bo = (re.search(r'Best of (\d)', html) or [None, "1"])[1]
    a = _slug(teams[0]) if len(teams) >= 1 else "team-a"
    b = _slug(teams[1]) if len(teams) >= 2 else "team-b"
    return links, f"{a}-vs-{b}-bo{bo}-{mid}"

The demo archives themselves live on an open CDN (r2-demos.hltv.org) that is *not* behind Cloudflare, so the big downloads stream straight down with range/resume support — no quota, no key. Two gotchas we hit and baked into the code:

• curl_cffi's Response is not a context manager, so you stream with an explicit try/finally: r.close() rather than a with block.
• HLTV ships RAR5 archives. p7zip/7z can only *list* RAR5, not decompress it ("Unsupported Method") — you need unrar (RARLAB) or unar. We try the real RAR tools first and fall back to 7z only for plain zips.

def download_demo(demo_url: str, out_dir: str) -> str | None:
    # peek the first bytes for the archive type + a stable filename from the CDN
    head = session().get(demo_url, headers={"Range": "bytes=0-7"})
    ext = next((e for magic, e in ARCHIVE_MAGIC.items()
                if head.content.startswith(magic)), "rar")
    ...
    tmp = dest + ".part"
    r = session().get(demo_url, stream=True)
    try:
        with open(tmp, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)
    finally:
        r.close()           # curl_cffi Response isn't a context manager
    os.replace(tmp, dest)   # atomic — a half-download never looks complete
    return dest

The whole downloader is one file, gate-2/hltv_download.py. Point it at an event's results page, optionally filter to a few teams, and it pulls + extracts every demo into a per-tournament folder. So far that has built a corpus of about 100 pro demos across four events.

Stage 2 — Parsing with awpy

Every demo is parsed with awpy, a Python library built on the Source 2 demoparser2. For the lineup corpus we do *not* filter to one player — we want every grenade thrown by everyone, so we can cluster on the full population of throws.

The subtle part is getting an *accurate* "where did they throw it from" position. The grenade event tables (smokes, infernos) carry a thrower_X/Y — but it is sampled at detonation, by which point the player has often run off (we measured 70–339 units of drift on Dust2). The honest release position is the *first* point of the projectile's trajectory in demo.grenades. So we aggregate each projectile's trajectory down to its first (release) and last (detonation) tick:

def _projectiles(demo: Demo) -> dict[tuple, dict]:
    """Per (entity_id, round_num) projectile: first (release) + last (detonate)
    tick + position, from demo.grenades."""
    g = demo.grenades
    agg = g.group_by(["entity_id", "round_num"]).agg([
        pl.col("grenade_type").first().alias("gtype"),
        pl.col("thrower_steamid").first().alias("tsid"),
        pl.col("tick").min().alias("throw_tick"),
        pl.col("tick").max().alias("det_tick"),
        pl.col("X").sort_by("tick").first().alias("rx"),   # release x
        pl.col("Y").sort_by("tick").first().alias("ry"),
        pl.col("X").sort_by("tick").last().alias("dx"),    # detonate x
        pl.col("Y").sort_by("tick").last().alias("dy"),
    ])
    return {(p["entity_id"], p["round_num"]): p for p in agg.iter_rows(named=True)}

Each grenade type is anchored differently. A smoke's landing is the smoke event's cloud-center X/Y, linked to its projectile by (entity_id, round). A molotov is harder: the resulting inferno entity has a *different* id from its projectile, so we link by (round, thrower) plus nearest-landing distance. A flash has no landing at all — its "detonation" is the air-pop, i.e. the last trajectory point. Here is the smoke path:

for s in smokes:
    p = projectiles.get((s.get("entity_id"), s.get("round_num")))
    thrower = (p["rx"], p["ry"], p["rz"]) if p else (s.get("thrower_X"), s.get("thrower_Y"), ...)
    grenades.append(_row(
        "smoke", map_name,
        side=s.get("thrower_side"), round_num=s.get("round_num"),
        land=(s.get("X"), s.get("Y"), s.get("Z")),       # cloud center
        thrower=thrower,                                  # true release point
        thrower_steamid=s.get("thrower_steamid"),
        detonate_tick=s.get("start_tick"),
        throw_tick=(p["throw_tick"] if p else None),
    ))

Flashes carry no side column anywhere, so we reconstruct it from the smoke/inferno events (which do carry thrower_side) with a per-player history. A player's side flips at most once, at halftime, so we detect that single flip and pick the side from the *same half* as the flash rather than doing a naive nearest-round lookup that could cross the boundary:

def _side_for(side_map, steamid, round_num):
    exact, history = side_map
    if (str(steamid), round_num) in exact:
        return exact[(str(steamid), round_num)]
    hist = sorted(set(history.get(str(steamid), [])), key=lambda rs: rs[0])
    if not hist:
        return None
    flip_round = next((r2 for (_, s1), (r2, s2) in zip(hist, hist[1:]) if s1 != s2), None)
    if flip_round is None:
        return hist[0][1]                       # constant side all match
    pre, post = hist[0][1], next(s for r, s in hist if r >= flip_round)
    return pre if round_num < flip_round else post

The output of this stage is one row per grenade in a corpus_grenades table: type, side, round, the landing (x,y,z), the release (x,y,z), the thrower's Steam ID, and the throw + detonate ticks. That last pair — throw_tick — is what lets us render a clip later.

Stage 3 — Spectre: clustering landings into lineups

We now have thousands of grenade landings per map. A "lineup" is just a spot that pros target *repeatedly*. So we project every landing from world coordinates onto the 1024×1024 radar image, then run a greedy single-pass clustering in pixel space. The projection matches the in-game radar exactly:

// webapp/lib/spectre/cluster.ts
export function worldToRadar(map: string, worldX: number, worldY: number): RadarPoint | null {
  const c = MAP_COORDS[map];
  if (!c) return null;
  return { x: (worldX - c.pos_x) / c.scale, y: (c.pos_y - worldY) / c.scale }; // y flips
}

The clustering is deliberately simple — each point joins the first existing cluster whose running centroid is within a radius, otherwise it seeds a new one. It is order-dependent and approximate, which is exactly right for eyeballing lineup spots; it is not trying to be k-means. Clusters below a minimum size are dropped as one-off "random toss" landings:

export function greedyCluster<T extends RadarPoint>(points: T[], radiusPx: number, minSize: number) {
  const clusters: { cx: number; cy: number; members: T[] }[] = [];
  for (const p of points) {
    let best = null, bestD = Infinity;
    for (const c of clusters) {
      const d = Math.hypot(p.x - c.cx, p.y - c.cy);
      if (d <= radiusPx && d < bestD) { best = c; bestD = d; }
    }
    if (best) {                                    // join + update running centroid
      best.members.push(p);
      const n = best.members.length;
      best.cx += (p.x - best.cx) / n;
      best.cy += (p.y - best.cy) / n;
    } else {
      clusters.push({ cx: p.x, cy: p.y, members: [p] });
    }
  }
  return clusters
    .filter((c) => c.members.length >= minSize)
    .sort((a, b) => b.members.length - a.members.length);
}

The batch builder (gate-2/build_lineup_drafts.py) runs the same algorithm in Python over the whole corpus, per side, then summarizes each cluster back in world space: the average landing, a tolerance radius, the average thrower position, and — critically — the single representative throw closest to the cluster center. That representative carries the demo_id, round_num, thrower_steamid and throw_tick that the render stage needs:

def summarize(members):
    gs = [m["g"] for m in members]
    lx, ly = avg([g["land_x"] for g in gs]), avg([g["land_y"] for g in gs])
    pool = [g for g in gs if g["throw_tick"] is not None] or gs
    rep = min(pool, key=lambda g: math.hypot(g["land_x"] - lx, g["land_y"] - ly))
    return {"target_pos": [r1(lx), r1(ly)], "member_count": len(gs),
            "clip": {"demo_id": rep["demo_id"], "round_num": rep["round_num"],
                     "thrower_steamid": rep["thrower_steamid"],
                     "throw_tick": rep["throw_tick"], "detonate_tick": rep["detonate_tick"]}}

Each draft is auto-named by the nearest map callout plus side ("Window Smoke (T)"), then a human renames the good ones. One detail we are quietly proud of: rebuilding the corpus on a larger demo set must not orphan clips we already rendered. So when a re-derived cluster lands within 150 world-units of a prior draft that already had a video_url, it inherits that draft's id and clip, because the rendered MP4 in R2 is keyed by the old id:

best, bd = -1, 150.0 ** 2  # match the same spot within 150 world units
for j, pl in enumerate(prior):
    pp = pl.get("target_pos") or [1e9, 1e9]
    d = (pp[0] - tx) ** 2 + (pp[1] - ty) ** 2
    if d < bd: bd, best = d, j
if best >= 0:
    _id, vurl = prior[best]["id"], prior[best].get("video_url")  # keep the rendered clip

This stage currently produces 773 candidate lineups across seven maps. Most are markers; the ones worth showing get a clip.

Stage 4 — Rendering the clip with DEMO-SLAP

A cluster knows its representative throw: a demo, a player, and the exact tick they released the nade. DEMO-SLAP is a third-party CS2 render service that turns that into an MP4. The flow is: tell it to analyze the demo by URL, wait, then request a custom clip windowed around the throw tick.

We render a 16-second clip — 8 seconds of lead before the throw, 8 after — from the thrower's point of view:

def render_custom(job, clips):
    return ds("POST", f"/public-api/render-custom/{job}",
              {"customClips": clips, "highResolution": True, "hideWatermark": True})["jobIds"]

def queue_one(job, c):
    for lead in lead_steps:                         # 8s, then step down: 6, 4, 2, 0
        tick = max(int(c["tick"] - lead * TICK_RATE), 0)
        try:
            return render_custom(job, [{"tick": tick, "duration": 16.0,
                                        "povSteamId": c["pov"]}])[0]
        except RuntimeError as e:
            if any(k in str(e).lower() for k in ("freeze", "round", "tick")):
                continue                            # too close to freeze-time — try less lead
            return None

The lead_steps fallback exists because a throw early in a round can sit inside freeze-time; if 8 seconds of lead is rejected, we step the lead down (6 → 4 → 2 → 0) until the window is legal. The renderer charges credits up-front on accept, so we cap concurrency and treat a 40-minute stall as a hard failure rather than waiting forever:

while wl or inflight:
    while wl and len(inflight) < cap:               # concurrency-capped submit
        job, c = wl.pop(0)
        vj = queue_one(job, c)
        if vj: inflight[vj] = {"job": job, "c": c, "t": time.time()}
    ...
    for vj, st in list(inflight.items()):
        s = statuses.get(st["job"], {}).get(vj, {}).get("status")
        if s == "done":      ...fetch + upload...
        elif s == "error":   errored += 1; del inflight[vj]
        elif time.time() - st["t"] > 2400:          # 40-min ceiling → give up on this clip
            print(f"    render stuck {st['c']['id']} (40 min)"); del inflight[vj]

DEMO-SLAP's renderer is the flakiest link in the chain — some demos render every clip cleanly, others degrade and fail most of a batch. That is why the loop is crash-safe and writes progress after every save: a half-finished batch still keeps everything it managed to render.

Stage 5 — Uploading + serving the clip

A finished clip does not stay on DEMO-SLAP. We pull the MP4 and immediately upload it to our own Cloudflare R2, keyed by map and lineup id, then write the video_url back into the catalog. After this point DEMO-SLAP could vanish and every clip still plays:

def fetch(job, vj, c):
    info = render_data(job).get("highlightClips", {}).get(vj)
    if not info or not info.get("clipUrl"):
        return False
    body = urllib.request.urlopen(info["clipUrl"], timeout=300).read()
    s3.put_object(Bucket=R2_BUCKET, Key=c["r2key"],          # lineups/<map>/<id>.mp4
                  Body=body, ContentType="video/mp4")
    write_catalog(c["map"], plan)                            # persist video_url, crash-safe
    return True

The clips live in the same private bucket as user demos, so they are never public. The catalog stores a durable path — /api/lineups/clip/<map>/<id> — and that route auth-gates the request and 302-redirects to a short-lived presigned R2 URL, fresh on every load:

// webapp/app/api/lineups/clip/[map]/[id]/route.ts
export async function GET(_req, ctx) {
  const user = await getCurrentUser();
  if (!user) return NextResponse.json({ error: 'unauthenticated' }, { status: 401 });

  const { map, id } = await ctx.params;
  if (!/^[a-z0-9_]{1,40}$/i.test(map) || !/^[a-z0-9_]{1,80}$/i.test(id)) {
    return NextResponse.json({ error: 'bad params' }, { status: 400 });
  }
  const url = await presignGetUrl(`lineups/${map}/${id}.mp4`, 3600);
  return NextResponse.redirect(url, 302);          // <video src> + "open in new tab" both follow it
}

A logged-in player loads /learn/lineups, picks a map and a grenade, and the <video> element points at that durable path. The signed URL is minted at request time and expires in an hour.

The whole thing, in one breath

A public demo on HLTV is fetched through a Chrome-fingerprint bypass, parsed by awpy into one row per grenade with an accurate release point, clustered in radar-pixel space into the spots pros actually target, rendered into a 16-second point-of-view clip by DEMO-SLAP, and stored privately in R2 behind a signed, auth-gated URL. Five stages, each a small Python or TypeScript file you could read in a sitting.

The interesting engineering is not any single stage — it is that the whole chain is idempotent and crash-safe, so we can keep pointing it at new tournaments and the library just grows. The next post covers the other half of what we mine from these demos: the heuristics and the trained classifier that read *strategy* out of the same data.