Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 41 additions & 16 deletions src/input/GameWindowCapturorForMac.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,80 +21,104 @@ def get_window_title(token):
)
# Get all exist windows
for window in window_list:
title = window.get(Quartz.kCGWindowName, '')
logger.debug(f"[get_window_title] Raw window: {window}")
title = window.get(Quartz.kCGWindowOwnerName, '')
if token in title:
logger.info(f"[get_window_title] Found match: {title}")
return title
return None


def get_window_region(window_title):
logger.debug(f"[get_window_region] Looking for: {window_title}")

window_list = Quartz.CGWindowListCopyWindowInfo(
Quartz.kCGWindowListOptionOnScreenOnly | Quartz.kCGWindowListExcludeDesktopElements,
Quartz.kCGNullWindowID
)
# Get all exist windows
logger.debug(f"[get_window_region] Retrieved {len(window_list)} windows")
all_titles = []
for window in window_list:
title = window.get(Quartz.kCGWindowName, '')
logger.debug(f"[get_window_region] Raw window: {window}")
title = window.get(Quartz.kCGWindowOwnerName, '')
owner = window.get(Quartz.kCGWindowOwnerName, '')
if not title:
logger.debug(f"[get_window_region] Skipped unnamed window: {window}")
if title:
all_titles.append(f"{title} (Owner: {owner})")
logger.debug(f"all_titles: {all_titles}")
logger.debug(f"[get_window_region] All titles: {all_titles}")

for window in window_list:
if window.get(Quartz.kCGWindowName, '') == window_title:
current_title = window.get(Quartz.kCGWindowOwnerName, '')
logger.debug(f"[get_window_region] Comparing '{current_title}' to target '{window_title}'")
if current_title == window_title:
bounds = window.get(Quartz.kCGWindowBounds, {})
logger.info(f"[get_window_region] Match found. Bounds: {bounds}")
# Return the region in the format expected by mss
return {
"left": int(bounds.get('X', 0)),
"top": int(bounds.get('Y', 0)),
"width": int(bounds.get('Width', 0)),
"height": int(bounds.get('Height', 0))
}

# No matching window found
logger.warning(f"[get_window_region] No window found with title: {window_title}")
return None


class GameWindowCapturor:
'''
GameWindowCapturor for macOS
'''

def __init__(self, cfg):
self.cfg = cfg
self.frame = None
self.lock = threading.Lock()
self.is_terminated = False

self.window_title = get_window_title(cfg["game_window"]["title"])
if self.window_title is None:
logger.error(
f"[GameWindowCapturor] Unable to find window titles that contain {cfg['game_window']['title']}"
)
return -1
raise RuntimeError(
f"[GameWindowCapturor] Unable to find window titles that contain {cfg['game_window']['title']}")

self.fps = 0
self.fps_limit = cfg["system"]["fps_limit_window_capturor"]
self.t_last_run = 0.0

# 使用 mss 來擷取特定螢幕區域
self.capture = mss.mss()
logger.info("[GameWindowCapturor] mss.mss() called")

# Get game window region
self.update_window_region()

# start game window capture
threading.Thread(target=self.start_capture, daemon=True).start()

# Wait frame init
time.sleep(0.1)
while self.frame is None:
self.limit_fps()
# Wait for initial frame with timeout
start = time.time()
while self.frame is None and time.time() - start < 5:
time.sleep(0.05)

if self.frame is None:
raise RuntimeError("[GameWindowCapturor] Failed to capture initial frame.")

def start_capture(self):
'''
開始螢幕擷取,並不斷更新 frame。
'''
while not self.is_terminated:
# Update self.region
self.update_window_region()

# Update self.frame
self.capture_frame()
try:
self.capture_frame()
except Exception as e:
logger.warning(f"[start_capture] Failed to capture frame: {e}")
# Only update window region if capture fails
self.update_window_region()

# Limit FPS to save systme resources
self.limit_fps()
Expand All @@ -110,9 +134,10 @@ def update_window_region(self):
'''
Update window region
'''
logger.info(f"Searching title: {self.window_title}")
self.region = get_window_region(self.window_title)
if self.region is None:
text = f"Cannot find window: {self.window_title}"
text = f"[update_window_region] Cannot find window: {self.window_title}"
logger.error(text)
raise RuntimeError(text)

Expand Down