Skip draft vision when Dota 2 is not the foreground window.

Prevents Alt-Tab desktop frames from advancing confirm streaks or burning strategy_tail polls.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
voson
2026-07-26 12:20:54 +08:00
co-authored by Cursor
parent 7dc3e515e9
commit e567a5cdfc
6 changed files with 101 additions and 11 deletions
+2
View File
@@ -8,6 +8,8 @@
- GSI 全量落盘:默认把每包 POST 追加到 `samples/raw/<matchid>/gsi.jsonl` - GSI 全量落盘:默认把每包 POST 追加到 `samples/raw/<matchid>/gsi.jsonl`
`gsi.dump_payloads` / `--dump-gsi` / `--no-dump-gsi`)。 `gsi.dump_payloads` / `--dump-gsi` / `--no-dump-gsi`)。
- 前台窗口检测:`gsi.require_foreground`(默认 true)。选将跟踪时若前台不是
`dota2.exe` 则跳过本帧截屏/识别,且不消耗 `strategy_tail_polls`;切回后继续。
### Changed ### Changed
+1
View File
@@ -180,6 +180,7 @@ climperor/
| `gsi.revise_gain` | 选人阶段改判所需分数增益 | 0.15 | | `gsi.revise_gain` | 选人阶段改判所需分数增益 | 0.15 |
| `gsi.strategy_tail_polls` | 决策阶段继续视觉轮询 | 8 | | `gsi.strategy_tail_polls` | 决策阶段继续视觉轮询 | 8 |
| `gsi.strategy_gsi_wait` | 视觉结束后等待本人 GSI 英雄 | 3.0 | | `gsi.strategy_gsi_wait` | 视觉结束后等待本人 GSI 英雄 | 3.0 |
| `gsi.require_foreground` | 仅当前台为 `dota2.exe` 时截屏识别 | true |
| `gsi.dump_payloads` | 全量 GSI JSONL 落盘 | true | | `gsi.dump_payloads` | 全量 GSI JSONL 落盘 | true |
--- ---
+47
View File
@@ -21,6 +21,53 @@ import numpy as np
RAW_DIR = Path(__file__).parent / "samples" / "raw" RAW_DIR = Path(__file__).parent / "samples" / "raw"
GSI_JSONL = "gsi.jsonl" GSI_JSONL = "gsi.jsonl"
DOTA_EXE = "dota2.exe"
def is_dota_foreground() -> bool:
"""True when the foreground process is dota2.exe.
Uses Win32 GetForegroundWindow + QueryFullProcessImageNameW. On non-Windows
or if the probe fails, returns True so recognition is not blocked.
"""
if sys.platform != "win32":
return True
try:
import ctypes
from ctypes import wintypes
user32 = ctypes.WinDLL("user32", use_last_error=True)
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
hwnd = user32.GetForegroundWindow()
if not hwnd:
return False
pid = wintypes.DWORD()
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
if not pid.value:
return False
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid.value)
if not handle:
return False
try:
buf = ctypes.create_unicode_buffer(32768)
size = wintypes.DWORD(len(buf))
# QueryFullProcessImageNameW(hProcess, dwFlags, lpExeName, lpdwSize)
QueryFullProcessImageNameW = kernel32.QueryFullProcessImageNameW
QueryFullProcessImageNameW.argtypes = [
wintypes.HANDLE, wintypes.DWORD, wintypes.LPWSTR, ctypes.POINTER(wintypes.DWORD)
]
QueryFullProcessImageNameW.restype = wintypes.BOOL
if not QueryFullProcessImageNameW(handle, 0, buf, ctypes.byref(size)):
return False
return Path(buf.value).name.lower() == DOTA_EXE
finally:
kernel32.CloseHandle(handle)
except Exception:
return True
def raw_dir_for_match(match_id: str | None = None) -> Path: def raw_dir_for_match(match_id: str | None = None) -> Path:
+1
View File
@@ -118,6 +118,7 @@
"dump_selection_every": 0, "dump_selection_every": 0,
"strategy_tail_polls": 8, "strategy_tail_polls": 8,
"strategy_gsi_wait": 3.0, "strategy_gsi_wait": 3.0,
"require_foreground": true,
"capture_interval": 1.0, "capture_interval": 1.0,
"target_slots": 10, "target_slots": 10,
"dump_payloads": true "dump_payloads": true
+25 -7
View File
@@ -34,7 +34,7 @@ import time
import mss import mss
from capture import grab_frame, raw_dir_for_match, save_frame from capture import grab_frame, is_dota_foreground, raw_dir_for_match, save_frame
from grid import bans, hero_table, read_grid from grid import bans, hero_table, read_grid
from modes import detect_mode, load_mode_templates from modes import detect_mode, load_mode_templates
from recognize import recognize_image from recognize import recognize_image
@@ -74,6 +74,8 @@ class DraftSession:
# this many polls - catches the last reveal without hanging forever # this many polls - catches the last reveal without hanging forever
self.strategy_tail = g.get("strategy_tail_polls", 8) self.strategy_tail = g.get("strategy_tail_polls", 8)
self.gsi_wait = g.get("strategy_gsi_wait", 3.0) self.gsi_wait = g.get("strategy_gsi_wait", 3.0)
# skip grab/recognize when Dota is not the foreground window
self.require_foreground = bool(g.get("require_foreground", True))
self.role_templates = load_role_templates() self.role_templates = load_role_templates()
self.mode_templates = load_mode_templates() self.mode_templates = load_mode_templates()
self.hero_names = {h["key"]: h["name_loc"] for h in hero_table()} self.hero_names = {h["key"]: h["name_loc"] for h in hero_table()}
@@ -104,6 +106,7 @@ class DraftSession:
vision_done = False vision_done = False
strategy_polls = 0 strategy_polls = 0
freeze_at = 0.0 freeze_at = 0.0
skipped_fg = False
with mss.MSS() as sct: with mss.MSS() as sct:
while True: while True:
@@ -116,6 +119,26 @@ class DraftSession:
break break
elapsed = time.monotonic() - started elapsed = time.monotonic() - started
# Vision finished: only wait on GSI / short timeout (no grab).
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)
continue
# Desktop / other apps must not burn strategy_tail or confirm streaks.
if self.require_foreground and not is_dota_foreground():
if not skipped_fg:
self.log("[draft] Dota 2 not foreground - skipping frames")
skipped_fg = True
time.sleep(self.poll_interval)
continue
if skipped_fg:
self.log("[draft] Dota 2 foreground again - resuming vision")
skipped_fg = False
frame = grab_frame(sct) frame = grab_frame(sct)
last_frame = frame last_frame = frame
polls += 1 polls += 1
@@ -123,7 +146,7 @@ class DraftSession:
# Keep reading in strategy until the roster is full - the last # Keep reading in strategy until the roster is full - the last
# pick is often revealed on the same tick selection ends, and # pick is often revealed on the same tick selection ends, and
# a player who already locked may only see strategy UI. # a player who already locked may only see strategy UI.
do_vision = not vision_done and ( do_vision = (
state == HERO_SELECTION state == HERO_SELECTION
or (state == STRATEGY_TIME and strategy_polls < self.strategy_tail) or (state == STRATEGY_TIME and strategy_polls < self.strategy_tail)
) )
@@ -177,11 +200,6 @@ class DraftSession:
freeze_at = elapsed freeze_at = elapsed
self.log("[draft] vision done - waiting on GSI for self hero") 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) time.sleep(self.poll_interval)
if best and not vision_done: if best and not vision_done:
+25 -4
View File
@@ -30,7 +30,7 @@ from pathlib import Path
import mss import mss
from capture import append_gsi_payload, grab_frame, raw_dir_for_match, save_frame from capture import append_gsi_payload, grab_frame, is_dota_foreground, raw_dir_for_match, save_frame
from common import ROOT, load_config, load_template_library from common import ROOT, load_config, load_template_library
from draft_session import HERO_SELECTION, DraftSession, describe, gsi_slot, loc from draft_session import HERO_SELECTION, DraftSession, describe, gsi_slot, loc
from recognize import recognize_image from recognize import recognize_image
@@ -155,13 +155,32 @@ class Watcher:
def _capture_only(self, match_id: str, state: str) -> None: def _capture_only(self, match_id: str, state: str) -> None:
"""No calibration yet: just bank a few frames to calibrate from later.""" """No calibration yet: just bank a few frames to calibrate from later."""
out = raw_dir_for_match(match_id) out = raw_dir_for_match(match_id)
require_fg = bool(self.cfg.get("gsi", {}).get("require_foreground", True))
print(f"[run] capture-only -> {out} (state={state})", flush=True) print(f"[run] capture-only -> {out} (state={state})", flush=True)
saved = 0
skipped_fg = False
with mss.MSS() as sct: with mss.MSS() as sct:
for attempt in range(1, 4): # Retry until 3 usable frames or a short deadline so Alt-Tab does not
# bank desktop screenshots for calibration.
deadline = time.monotonic() + max(self.interval * 12, 15.0)
while saved < 3 and time.monotonic() < deadline:
if require_fg and not is_dota_foreground():
if not skipped_fg:
print("[run] Dota 2 not foreground - waiting to capture", flush=True)
skipped_fg = True
time.sleep(self.interval)
continue
if skipped_fg:
print("[run] Dota 2 foreground again - capturing", flush=True)
skipped_fg = False
saved += 1
path = save_frame(grab_frame(sct), out, prefix="draft") path = save_frame(grab_frame(sct), out, prefix="draft")
print(f"[run] capture-only {attempt}/3: {path}", flush=True) print(f"[run] capture-only {saved}/3: {path}", flush=True)
time.sleep(self.interval) time.sleep(self.interval)
print("[run] no calibrated slots - run calibrate.py on one of these frames", flush=True) if saved == 0:
print("[run] capture-only got no frames (Dota never foreground)", flush=True)
else:
print("[run] no calibrated slots - run calibrate.py on one of these frames", flush=True)
return None return None
def report_session(self, summary: dict) -> None: def report_session(self, summary: dict) -> None:
@@ -175,6 +194,8 @@ class Watcher:
def run_once(self) -> dict | None: def run_once(self) -> dict | None:
"""One snapshot of whatever is on screen right now, for manual checks.""" """One snapshot of whatever is on screen right now, for manual checks."""
if not is_dota_foreground():
print("[run] warning: Dota 2 is not the foreground window", flush=True)
frame = grab_frame() frame = grab_frame()
out = raw_dir_for_match(self.last_match_id) out = raw_dir_for_match(self.last_match_id)
if not self.calibrated: if not self.calibrated: