"""Follow a whole draft instead of taking one snapshot at the end. Ranked All Pick reveals picks in waves rather than one at a time (official rules: two rounds of 2 picks per team at 25s, then a final round of 1 at 20s, with each round's picks hidden until the round ends). A single grab at strategy time therefore loses the order completely, which is exactly the information you need to reason about what to counter-pick. This polls the screen for as long as GSI says we are still drafting and appends a timeline event whenever the confirmed set of picks changes. A pick only becomes confirmed after the same hero lands in the same slot on `confirm_polls` consecutive frames, because the top bar animates portraits in and a single frame catches half-faded artwork. While the hero grid is still up it also reads the ban list off it (see grid.py), which the top bar never shows. Top-bar portraits use the default icon until everyone has picked; skins land only after the draft is complete. Vision therefore runs through both HERO_SELECTION and early STRATEGY_TIME until all ten slots are filled - the last reveal often lands right as strategy begins, and a player who already locked may be staring at the strategy UI while others are still picking. Skinned portraits are not templated (too many variants). Instead: - during STRATEGY_TIME only empty slots may be filled; confirmed picks are never revised (skin art must not overwrite a settled default face); - the saved best lineup frame prefers HERO_SELECTION when recognition counts tie, so draft_best_* stays on default faces when possible. Once ten heroes are confirmed, vision stops and only GSI is waited on for self. """ import time import mss from capture import grab_frame, raw_dir_for_match, save_frame from grid import bans, hero_table, read_grid from modes import detect_mode, load_mode_templates from recognize import recognize_image from roles import ROLES, detect_roles, load_role_templates HERO_SELECTION = "DOTA_GAMERULES_STATE_HERO_SELECTION" STRATEGY_TIME = "DOTA_GAMERULES_STATE_STRATEGY_TIME" DRAFT_STATES = (HERO_SELECTION, STRATEGY_TIME) def pick_round(per_team_max: int) -> int: """Which of the three All Pick rounds a given pick count belongs to.""" if per_team_max <= 2: return 1 if per_team_max <= 4: return 2 return 3 class DraftSession: def __init__(self, cfg: dict, library, log=print): self.cfg = cfg self.library = library self.log = log g = cfg.get("gsi", {}) self.poll_interval = g.get("poll_interval", 1.0) self.confirm_polls = g.get("confirm_polls", 2) self.timeout = g.get("session_timeout", 300) # how much better a later reading must score before it may overwrite # an already confirmed pick self.revise_gain = g.get("revise_gain", 0.15) self.target = g.get("target_slots", 10) self.keep_frames = g.get("keep_event_frames", True) # >0 saves a frame every N seconds while the hero grid is up self.dump_every = g.get("dump_selection_every", 0) # after selection ends, keep reading strategy frames until 10/10 or # this many polls - catches the last reveal without hanging forever self.strategy_tail = g.get("strategy_tail_polls", 8) self.gsi_wait = g.get("strategy_gsi_wait", 3.0) self.role_templates = load_role_templates() self.mode_templates = load_mode_templates() self.hero_names = {h["key"]: h["name_loc"] for h in hero_table()} self.frame_dir = raw_dir_for_match(None) def run(self, match_id: str, state_fn, gsi_fn=None) -> dict: """Poll until the draft is over. state_fn returns the live GSI state.""" self.frame_dir = raw_dir_for_match(match_id) self.log(f"[draft] frames -> {self.frame_dir}") started = time.monotonic() pending: dict[int, tuple[str, int]] = {} confirmed: dict[int, str] = {} scores: dict[int, float] = {} revisions: list[dict] = [] timeline: list[dict] = [] info = { "self_slot": None, "self_team": None, "roles": {}, "unavailable": None, "mode": None, } polls = 0 last_frame = None best: dict | None = None next_dump = 0.0 saved_milestones: set[int] = set() vision_done = False strategy_polls = 0 freeze_at = 0.0 with mss.MSS() as sct: while True: state = state_fn() if state not in DRAFT_STATES: self.log(f"[draft] session ended (state={state})") break if time.monotonic() - started > self.timeout: self.log(f"[draft] session timed out after {self.timeout}s") break elapsed = time.monotonic() - started frame = grab_frame(sct) last_frame = frame polls += 1 # Keep reading in strategy until the roster is full - the last # pick is often revealed on the same tick selection ends, and # a player who already locked may only see strategy UI. do_vision = not vision_done and ( state == HERO_SELECTION or (state == STRATEGY_TIME and strategy_polls < self.strategy_tail) ) if state == STRATEGY_TIME: strategy_polls += 1 if do_vision: if self.dump_every and state == HERO_SELECTION and elapsed >= next_dump: next_dump = elapsed + self.dump_every self.log(f"[draft] grid frame: {save_frame(frame, self.frame_dir, prefix='select')}") if info["mode"] is None: found = detect_mode(frame, self.cfg, self.mode_templates) if found: info["mode"] = found self.log(f"[draft] mode: {found['label']} ({found['score']:.2f})") result = recognize_image(frame, self.cfg, self.library) self._absorb_roles(frame, info) if state == HERO_SELECTION: self._absorb_grid(frame, info) best = self._remember_best(best, frame, result, state, elapsed) # Strategy frames may show skins; only fill empty slots there. added, revised = self._absorb_picks( result, pending, confirmed, scores, allow_revise=(state == HERO_SELECTION), ) for rev in revised: rev["t"] = round(elapsed, 1) revisions.append(rev) self._log_revision(rev) if added: event = self._event(added, confirmed, state, elapsed) if self.keep_frames: event["frame"] = save_frame(frame, self.frame_dir, prefix="draft") timeline.append(event) self._log_event(event, info) n = len(confirmed) if self.keep_frames and n in (4, 8, 10) and n not in saved_milestones: saved_milestones.add(n) path = save_frame(frame, self.frame_dir, prefix=f"draft_n{n}") self.log(f"[draft] milestone {n}/10: {path}") if n >= self.target or ( state == STRATEGY_TIME and strategy_polls >= self.strategy_tail ): self._finalize_vision(best, confirmed, scores) vision_done = True freeze_at = elapsed self.log("[draft] vision done - waiting on GSI for self hero") if vision_done: gsi_hero = (gsi_fn() or {}).get("hero") if gsi_fn else None if gsi_hero or (elapsed - freeze_at) >= self.gsi_wait: break time.sleep(self.poll_interval) if best and not vision_done: self._finalize_vision(best, confirmed, scores) keep = best["frame"] if best is not None else last_frame return self._summary(match_id, timeline, confirmed, info, polls, started, keep, gsi_fn, revisions, best) def _finalize_vision(self, best: dict | None, confirmed: dict, scores: dict) -> None: if not best: return if best.get("path"): return self.log(f"[draft] best lineup frame: {best['recognized']}/10 " f"at t={best['t']:.1f}s ({best['state']})") filled = self._backfill(confirmed, scores, best) if filled: self.log(f"[draft] backfilled {filled} slots from best frame") if self.keep_frames: best["path"] = save_frame(best["frame"], self.frame_dir, prefix="draft_best") self.log(f"[draft] saved best: {best['path']}") def _backfill(self, confirmed: dict, scores: dict, best: dict) -> int: """Copy threshold-passed heroes from the best frame into empty slots.""" n = 0 for i, hero in enumerate(best.get("heroes") or [], 1): if hero and i not in confirmed: confirmed[i] = hero scores[i] = 0.0 n += 1 return n def _remember_best(self, best: dict | None, frame, result: dict, state: str, t: float) -> dict: """Track the clearest top-bar reading seen so far. Prefer more recognized slots first (so a late last-pick still wins). On a tie, prefer HERO_SELECTION over STRATEGY_TIME so skinned strategy portraits do not replace a cleaner default-face frame. Score sum is the final tie-breaker within the same state preference. """ n = int(result.get("recognized") or 0) if n == 0: return best score_sum = sum(float(r.get("score") or 0) for r in result["slots"] if r.get("hero")) selection = state == HERO_SELECTION cand = { "recognized": n, "score_sum": score_sum, "selection": selection, "t": t, "state": state.replace("DOTA_GAMERULES_STATE_", ""), "frame": frame.copy(), "heroes": [r.get("hero") for r in result["slots"]], } if best is None: return cand if n > best["recognized"]: return cand if n < best["recognized"]: return best # same recognized count: prefer selection-phase default faces if selection and not best.get("selection", False): return cand if selection == best.get("selection", False) and score_sum > best["score_sum"]: return cand return best def _absorb_roles(self, frame, info: dict) -> None: """Roles and your own slot never change, so stop looking once found.""" if info["self_slot"] is not None and info["roles"]: return found = detect_roles(frame, self.cfg, self.role_templates) if found["self_slot"] is not None and info["self_slot"] is None: info["self_slot"] = found["self_slot"] if found["roles"] and not info["roles"]: info["roles"] = found["roles"] info["self_team"] = found["self_team"] or info["self_team"] def _absorb_grid(self, frame, info: dict) -> None: """Read the ban list off the hero grid, once. The set of banned heroes is fixed before the first pick, so the earliest readable frame is also the cleanest: nothing has been taken yet, so everything greyed out is a ban. Frames where a hover tooltip covers the grid fail the layout check and are simply skipped. """ if info["unavailable"] is not None: return res = read_grid(frame, self.cfg) if not res["ok"]: return info["unavailable"] = res["unavailable"] names = ", ".join(self.hero_names.get(k, k) for k in res["unavailable"]) self.log(f"[draft] grid: {len(res['unavailable'])} heroes unavailable " f"(contrast margin {res['margin']}) - {names}") def _absorb_picks(self, result: dict, pending: dict, confirmed: dict, scores: dict, *, allow_revise: bool = True, ) -> tuple[list[dict], list[dict]]: """Promote picks seen on enough consecutive frames. Returns (new picks, revisions). During HERO_SELECTION a slot stays open to revision because the frame that first reveals a portrait is the worst one to judge it on: the art is still fading in and the ranked title bar covers the lower face. Once the portrait settles it scores far higher, and a clearly better reading may overwrite. During STRATEGY_TIME set allow_revise=False: only empty slots may be filled. Skinned portraits must not replace a confirmed default face. """ added, revised = [], [] for r in result["slots"]: slot, hero, score = r["slot"], r["hero"], r["score"] if hero is None: continue if slot in confirmed: if hero == confirmed[slot]: scores[slot] = max(scores.get(slot, 0.0), score) pending.pop(slot, None) continue if not allow_revise: continue if score < scores.get(slot, 0.0) + self.revise_gain: continue prev_hero, streak = pending.get(slot, (None, 0)) streak = streak + 1 if hero == prev_hero else 1 pending[slot] = (hero, streak) if streak >= self.confirm_polls: revised.append({"slot": slot, "team": team_of(slot), "hero": hero, "was": confirmed[slot], "score": score, "was_score": scores.get(slot, 0.0)}) confirmed[slot] = hero scores[slot] = score pending.pop(slot, None) continue prev_hero, streak = pending.get(slot, (None, 0)) streak = streak + 1 if hero == prev_hero else 1 pending[slot] = (hero, streak) if streak >= self.confirm_polls: confirmed[slot] = hero scores[slot] = score pending.pop(slot, None) added.append({"slot": slot, "team": team_of(slot), "hero": hero}) return added, revised def _event(self, added: list[dict], confirmed: dict, state: str, elapsed: float) -> dict: radiant = sorted(s for s in confirmed if s <= 5) dire = sorted(s for s in confirmed if s > 5) return { "t": round(elapsed, 1), "state": state.replace("DOTA_GAMERULES_STATE_", ""), "round": pick_round(max(len(radiant), len(dire))), "added": added, "radiant": [confirmed[s] for s in radiant], "dire": [confirmed[s] for s in dire], "count": len(confirmed), } def _log_event(self, event: dict, info: dict) -> None: for a in event["added"]: mine = " <- you" if a["slot"] == info["self_slot"] else "" role = info["roles"].get(a["slot"]) tag = f" [{role['label']}]" if role else "" self.log( f"[draft] +{event['t']:6.1f}s round{event['round']} " f"{a['team']:7s} slot{a['slot']:<2d} {loc(a['hero'], self.hero_names)}{tag}{mine}" ) def _log_revision(self, rev: dict) -> None: self.log( f"[draft] ~{rev['t']:6.1f}s slot{rev['slot']:<2d} " f"{loc(rev['was'], self.hero_names)} -> {loc(rev['hero'], self.hero_names)} " f"(score {rev['was_score']:.2f} -> {rev['score']:.2f})" ) def _summary(self, match_id, timeline, confirmed, info, polls, started, frame, gsi_fn, revisions=None, best=None) -> dict: gsi = gsi_fn() if gsi_fn else {} self_slot = gsi_slot(gsi) or info["self_slot"] if self_slot and info["self_slot"] and self_slot != info["self_slot"]: self.log(f"[draft] self slot {info['self_slot']} -> {self_slot} (GSI team_slot)") # GSI knows your own hero with certainty once it is locked. Prefer it. gsi_hero = gsi.get("hero") if self_slot and gsi_hero and confirmed.get(self_slot) != gsi_hero: prev = confirmed.get(self_slot) confirmed[self_slot] = gsi_hero self.log( f"[draft] self hero {loc(prev, self.hero_names)} -> " f"{loc(gsi_hero, self.hero_names)} (GSI)" ) role = info["roles"].get(self_slot) if self_slot else None summary = { "match_id": match_id, "captured_at": time.strftime("%Y-%m-%d %H:%M:%S"), "duration_s": round(time.monotonic() - started, 1), "polls": polls, "mode": info.get("mode"), "self": { "slot": self_slot, "team": gsi.get("team") or info["self_team"] or (team_of(self_slot) if self_slot else None), "hero": (confirmed.get(self_slot) if self_slot else None) or gsi_hero, "role": role["role"] if role else None, "role_label": role["label"] if role else None, "position": role["position"] if role else None, "gsi_name": gsi.get("name"), }, "team_roles": { str(s): {"position": r["position"], "label": r["label"], "hero": confirmed.get(s)} for s, r in sorted(info["roles"].items()) }, "final": { "radiant": [confirmed.get(s) for s in range(1, 6)], "dire": [confirmed.get(s) for s in range(6, 11)], }, "recognized": len(confirmed), "revisions": revisions or [], "timeline": timeline, } if info["unavailable"] is not None: banned = bans({"unavailable": info["unavailable"]}, list(confirmed.values())) summary["bans"] = banned summary["bans_loc"] = [self.hero_names.get(k, k) for k in banned] if best: summary["best_lineup"] = { "recognized": best["recognized"], "t": round(best["t"], 1), "state": best["state"], "heroes": best["heroes"], "frame": best.get("path"), } if frame is not None and self.keep_frames: # `frame` is already the best readable lineup when one was found summary["last_frame"] = best.get("path") if best and best.get("path") else \ save_frame(frame, self.frame_dir, prefix="draft") return summary def team_of(slot: int) -> str: return "radiant" if slot <= 5 else "dire" def gsi_slot(gsi: dict) -> int | None: """Top-bar slot from GSI's own team_slot, which beats any pixel heuristic. The bar is ordered by team slot, radiant on the left. GSI leaves the player block out until a match is loaded, hence the None path. """ team_slot = gsi.get("team_slot") team = gsi.get("team") if team_slot is None or team not in ("radiant", "dire"): return None return int(team_slot) + (1 if team == "radiant" else 6) def loc(key: str | None, names: dict[str, str] | None = None) -> str: """English hero key -> in-client Chinese name, for human-facing output.""" if not key: return "?" if names is None: names = {h["key"]: h["name_loc"] for h in hero_table()} return names.get(key, key) def describe(summary: dict) -> list[str]: """Human-readable recap printed when a session ends.""" names = {h["key"]: h["name_loc"] for h in hero_table()} me = summary["self"] lines = [] mode = summary.get("mode") if mode: lines.append(f"mode : {mode.get('label') or mode.get('key')}") for side in ("radiant", "dire"): heroes = [loc(h, names) for h in summary["final"][side]] lines.append(f"{side:7s}: {', '.join(heroes)}") if me["slot"]: pos = f"position {me['position']} ({me['role_label']})" if me["position"] else "position unknown" lines.append(f"you : slot {me['slot']} {me['team']} {loc(me['hero'], names)} - {pos}") if summary["team_roles"]: order = ", ".join( f"{v['position']}:{loc(v['hero'], names)}" for v in sorted(summary["team_roles"].values(), key=lambda v: v["position"]) ) lines.append(f"lanes : {order}") lines.append(f"rounds : {len(summary['timeline'])} reveal events over {summary['duration_s']}s") if summary.get("bans_loc"): lines.append(f"bans : {len(summary['bans_loc'])} - {', '.join(summary['bans_loc'])}") return lines __all__ = ["DraftSession", "DRAFT_STATES", "HERO_SELECTION", "STRATEGY_TIME", "describe", "loc", "ROLES"]