Where Did Each Team Ward, and What Could the Wards See?

Question. Where did each team put its observer and sentry wards, how long did they last, and how much of what an observer's circle covers could the team actually see?

The facts

• match.wards: every ward with its exact position, the tick it went up, and the tick it expired or was killed, with the killer.
• gem.region_of(x, y): the map region, here folded into the warding team's own half, the enemy's half, or the river.
• gem.analysis.vision.OBSERVER_VISION_RADIUS: the vision model's observer radius, 1,600 world units. It is a flat circle: cliffs and trees that block the game's real vision are not in it.
• match.hero_visibility_events (read as gem.hero_visibility_at does): the replay's own record of whether each team could see each hero. This is the game's answer, terrain and trees included.
• player.position_log: each hero's position, sampled about once a second.
• gem.build_smoke_analysis(match): who was under Smoke of Deceit, and when.

A ward is up from its placement tick until it is killed or expires (gone on that tick). Times are in-game seconds, pauses excluded.

The recipe

"""Where did each team ward, how long did the wards last, and what could they see?

Every ward (``match.wards``) with its exact position, when it went up and how it
ended (killed, and by whom, or expired), and its map region (``gem.region_of``).
Then two summaries:

- how long observers lasted, by team and region, and how many were killed;
- how much of each team's observer coverage the replay confirms. gem draws an
  observer's vision as a flat 1,600-unit circle (``OBSERVER_VISION_RADIUS``);
  the game's real vision is cut by cliffs and trees. For every second an enemy
  hero stood inside a live observer circle, the replay's own visibility
  (``gem.hero_visibility_at``) says whether the warding team could see them.

    python examples/cookbook/wards.py replay.dem [more.dem ...]
"""

from __future__ import annotations

import bisect
import math
import sys
from collections import defaultdict

import pandas as pd

import gem
from gem.analysis.vision import OBSERVER_VISION_RADIUS

TEAMS = {2: "radiant", 3: "dire"}
WARD_COLUMNS = [
    "match_id", "team", "type", "placer", "placed", "placed_s", "ended_s",
    "lasted_s", "how", "killer", "region", "side", "x", "y", "placed_tick", "end_tick",
]  # fmt: skip
SAMPLE_COLUMNS = ["match_id", "team", "hero", "tick", "distance", "state"]
STATES = ["visible", "hidden", "unknown", "smoked"]


def ward_table(match: gem.ParsedMatch) -> pd.DataFrame:
    """Return one row per placed ward: who, where, when, and how it ended.

    A ward is up from its placement tick until it was killed or expired (half
    open: it is gone on that tick). ``how`` is ``killed``, ``expired``, or
    ``up at the end`` when the recording ended first. Times are in-game seconds,
    pauses excluded.

    Args:
        match: A parsed match.

    Returns:
        A DataFrame with one row per ward that has a position.
    """
    clock = match.game_clock
    heroes = {player.player_id: player.hero_name for player in match.players}
    end_of_recording = match.post_game_tick or match.game_end_tick
    rows = []
    for ward in match.wards:
        if ward.x is None or ward.y is None or ward.team not in TEAMS:
            continue
        if ward.killed_tick is not None:
            end_tick, how = ward.killed_tick, "killed"
        elif ward.expires_tick is not None:
            end_tick, how = ward.expires_tick, "expired"
        else:
            end_tick, how = end_of_recording, "up at the end"
        region = gem.region_of(ward.x, ward.y)
        placed_s = clock.game_time_at(ward.tick) if clock else None
        ended_s = clock.game_time_at(end_tick) if clock and end_tick is not None else None
        rows.append(
            {
                "match_id": match.match_id,
                "team": TEAMS[ward.team],
                "type": ward.ward_type,
                # The player's hero, joined on the player slot.
                "placer": heroes.get(ward.player_id, ward.placer),
                "placed": clock.format_tick(ward.tick) if clock else str(ward.tick),
                "placed_s": placed_s,
                "ended_s": ended_s,
                "lasted_s": None if placed_s is None or ended_s is None else ended_s - placed_s,
                "how": how,
                "killer": ward.killer if how == "killed" else "",
                "region": region,
                "side": _side(TEAMS[ward.team], region),
                "x": ward.x,
                "y": ward.y,
                "placed_tick": ward.tick,
                "end_tick": end_tick,
            }
        )
    return pd.DataFrame(rows, columns=WARD_COLUMNS)


def _side(team: str, region: str) -> str:
    """The region as the warding team sees it: its own half, the enemy's, or the river."""
    if region.endswith("_half"):
        return "own half" if region == f"{team}_half" else "enemy half"
    return "river"  # the river and its lotus pools


def observer_lifetimes(wards: pd.DataFrame) -> pd.DataFrame:
    """Return how long observers lasted and how many were killed, by team and side of the map."""
    observers = wards[(wards["type"] == "observer") & wards["lasted_s"].notna()]
    return (
        observers.groupby(["team", "side"])
        .agg(
            placed=("lasted_s", "size"),
            median_lasted_s=("lasted_s", "median"),
            killed=("how", lambda how: int((how == "killed").sum())),
        )
        .reset_index()
    )


def _states(match: gem.ParsedMatch) -> dict[int, tuple[list[int], list[object]]]:
    """Each player's visibility events, sorted by tick, for bisecting.

    Equivalent to ``gem.hero_visibility_at``: the last event at or before a
    tick wins, and on one tick the later event in the list.
    """
    events: dict[int, list] = defaultdict(list)
    for event in match.hero_visibility_events:
        events[event.player_id].append(event)
    lookup = {}
    for player_id, rows in events.items():
        rows.sort(key=lambda event: event.tick)  # stable: same-tick order kept
        lookup[player_id] = ([event.tick for event in rows], rows)
    return lookup


def visibility_at(
    lookup: dict[int, tuple[list[int], list]], player_id: int, team: int, tick: int
) -> str:
    """Return ``visible``, ``hidden`` or ``unknown``: the team's view of the hero at the tick."""
    ticks, rows = lookup.get(player_id, ([], []))
    i = bisect.bisect_right(ticks, tick)
    if i == 0:
        return "unknown"
    event = rows[i - 1]
    state = event.radiant_state if team == 2 else event.dire_state
    return gem.VisibilityState(state).value


def circle_samples(match: gem.ParsedMatch, wards: pd.DataFrame | None = None) -> pd.DataFrame:
    """Return every second an enemy hero stood inside one of a team's live observer circles.

    Each row is one of the enemy hero's position samples (``position_log``,
    about one a second) while it was alive and within ``OBSERVER_VISION_RADIUS``
    of at least one of the warding team's live observers, with the distance to
    the nearest one. ``state`` is the replay's visibility of the hero for the
    warding team then: ``visible``, ``hidden`` or ``unknown``; or ``smoked``
    under Smoke of Deceit, which hides a hero from wards whatever the terrain.

    Args:
        match: A parsed match.
        wards: ``ward_table(match)``, if already built.

    Returns:
        One row per sample: ``match_id``, ``team`` (the warding team), ``hero``,
        ``tick``, ``distance`` (world units) and ``state``.
    """
    wards = ward_table(match) if wards is None else wards
    observers = wards[wards["type"] == "observer"]
    lookup = _states(match)
    smoked: dict[int, list[tuple[int, int]]] = defaultdict(list)
    end = match.post_game_tick or match.game_end_tick or 0
    for smoke in gem.build_smoke_analysis(match):
        for member in smoke.members:
            if member.player_id is not None:
                smoked[member.player_id].append((member.applied_tick, member.removed_tick or end))
    rows = []
    for team_id, team in TEAMS.items():
        circles = observers[observers["team"] == team][["x", "y", "placed_tick", "end_tick"]]
        live = [tuple(row) for row in circles.itertuples(index=False)]
        for player in match.players:
            if player.team == team_id or player.team not in TEAMS:
                continue
            hp = dict(zip(player.times, player.hp_t, strict=False))
            for tick, x, y in player.position_log:
                if hp.get(tick, 1) <= 0:
                    continue  # dead: hidden whatever the wards
                distance = min(
                    (
                        math.dist((x, y), (wx, wy))
                        for wx, wy, start, stop in live
                        if start <= tick < stop
                    ),
                    default=math.inf,
                )
                if distance > OBSERVER_VISION_RADIUS:
                    continue
                if any(start <= tick < stop for start, stop in smoked[player.player_id]):
                    state = "smoked"
                else:
                    state = visibility_at(lookup, player.player_id, team_id, tick)
                rows.append(
                    {
                        "match_id": match.match_id,
                        "team": team,
                        "hero": player.hero_name,
                        "tick": tick,
                        "distance": distance,
                        "state": state,
                    }
                )
    return pd.DataFrame(rows, columns=SAMPLE_COLUMNS)


def circle_summary(samples: pd.DataFrame) -> pd.DataFrame:
    """Count each warding team's samples by state: the seconds enemies spent in its circles."""
    counts = samples.pivot_table(
        index=["match_id", "team"], columns="state", aggfunc="size", fill_value=0
    )
    counts = counts.reindex(columns=STATES, fill_value=0).reset_index()
    counts.columns.name = None
    counts.insert(2, "seconds", counts[["visible", "hidden", "unknown"]].sum(axis=1))
    return counts


def by_distance(samples: pd.DataFrame, step: int = 400) -> pd.DataFrame:
    """Return the share of unsmoked seconds the warding team saw the hero, by distance from the ward."""
    seen = samples[samples["state"].isin(["visible", "hidden"])]
    band = (seen["distance"] // step * step).astype(int)
    table = seen.groupby(band)["state"].agg(
        seconds="size", visible=lambda s: int((s == "visible").sum())
    )
    table.index = [f"{lo}-{lo + step}" for lo in table.index]
    table["visible_pct"] = (100 * table["visible"] / table["seconds"]).round(1)
    return table.rename_axis("distance").reset_index()


def main(paths: list[str]) -> None:
    """Print the wards, observer lifetimes by region, and the circle check."""
    if len(paths) == 1:
        matches = [gem.parse(paths[0])]
    else:
        matches = [result.match for result in gem.parse_many(paths) if result.match is not None]
    wards = [ward_table(match) for match in matches]
    table = pd.concat(wards, ignore_index=True)
    if table.empty:
        print("No wards in these replays.")
        return
    shown = ["match_id", "team", "type", "placer", "placed", "lasted_s", "how", "killer", "region"]
    print(table[shown].round({"lasted_s": 0}).to_string(index=False))

    print("\nObservers by team and side of the map:")
    print(observer_lifetimes(table).round({"median_lasted_s": 0}).to_string(index=False))

    samples = pd.concat(
        [circle_samples(match, w) for match, w in zip(matches, wards, strict=True)],
        ignore_index=True,
    )
    print("\nSeconds enemy heroes spent inside live observer circles:")
    print(circle_summary(samples).to_string(index=False))
    print("\nHow often the warding team could see them, by distance from the ward:")
    print(by_distance(samples).to_string(index=False))


if __name__ == "__main__":  # parse_many's worker processes re-import this file
    main(sys.argv[1:])

On one match

python examples/cookbook/wards.py 8856501050.dem lists all 221 wards. The first rows, then the summaries:

  match_id    team     type                           placer placed  lasted_s     how                           killer       region
8856501050    dire observer             npc_dota_hero_treant -01:00     339.0  killed             npc_dota_hero_sniper    dire_half
8856501050 radiant observer           npc_dota_hero_shredder -00:42     360.0 expired                                  radiant_half
8856501050    dire observer             npc_dota_hero_treant -00:26     360.0 expired                                  radiant_half
8856501050 radiant observer               npc_dota_hero_lion  00:17     360.0 expired                                  radiant_half
8856501050 radiant   sentry               npc_dota_hero_lion  00:32      23.0  killed          npc_dota_hero_nevermore    dire_half
...

Observers by team and side of the map:
   team       side  placed  median_lasted_s  killed
   dire enemy half      30            146.0      18
   dire   own half       9            360.0       4
   dire      river       2            360.0       0
radiant enemy half       7            360.0       1
radiant   own half      33            360.0      13
radiant      river       1            360.0       0

Seconds enemy heroes spent inside live observer circles:
  match_id    team  seconds  visible  hidden  unknown  smoked
8856501050    dire     1937     1519     418        0     104
8856501050 radiant     1775     1192     583        0     168

How often the warding team could see them, by distance from the ward:
 distance  seconds  visible  visible_pct
    0-400      337      330         97.9
  400-800      916      807         88.1
 800-1200     1049      773         73.7
1200-1600     1410      801         56.8

Dire put 30 of its 41 observers in Radiant's half, and 18 of those were killed; Radiant kept 33 of its 41 at home. Inside Dire's circles, Dire could see the Radiant hero 1,519 of 1,937 seconds; inside Radiant's, Radiant saw the Dire hero 1,192 of 1,775.

Across the fixtures

Over the 1,140 wards in the 9 fixture replays (429 observers, 711 sentries):

• 165 of the 429 observers were killed; the other 264 expired;
• observers in the enemy's half were killed more often: 98 of 216, against 65 of 200 in the team's own half and 2 of 13 in the river;
• a killed observer lasted a median of 1:43;
• 244 of the 711 sentries were killed.

Enemy heroes spent 22,376 seconds inside live observer circles, leaving out 1,367 seconds under smoke. The warding team could see them for 15,522 of those seconds, and they were hidden for 6,854. How often they were seen falls with the distance from the ward:

What this does not say

• What the ward itself saw. "Seen" is the team's visibility, from any source: a hero, a creep or another ward may have seen the hero instead. "Hidden" is the clearer fact: the circle covered the hero and the team could not see it.
• Why a hero was hidden. Cliffs and trees block a ward's vision, and so does invisibility: Shadow Blade, Glimmer Cape, Treant's Nature's Guise and other invisible heroes count as hidden here. Only smoke is counted apart.
• The true shape of a ward's vision. That needs the map's terrain and trees, which the replay does not carry.
• Positions between samples. Hero positions are sampled about once a second, so a second is the unit of the counts.

Variations

• Change the step of by_distance for finer distance bands.
• Filter ward_table to one placer (table["placer"] == "npc_dota_hero_lion") to follow a support's wards.
• Join circle_samples with match.fights to ask how often the enemy was seen coming into a fight.