diff --git a/src/template_finder.py b/src/template_finder.py index e166cae..afc1760 100644 --- a/src/template_finder.py +++ b/src/template_finder.py @@ -1,5 +1,6 @@ import cv2 import threading +import concurrent.futures from screen import convert_screen_to_monitor, grab from dataclasses import dataclass import numpy as np @@ -125,35 +126,55 @@ def _single_template_match(template: Template, inp_img: np.ndarray = None, roi: return template_match +def _match_template_worker(template, inp_img, roi, color_match, use_grayscale): + """Worker for parallel template matching (runs in thread pool).""" + return _single_template_match(template, inp_img, roi, color_match, use_grayscale) + + def search( - ref: str | np.ndarray | list[str], - inp_img: np.ndarray, - threshold: float = 0.68, - roi: list[float] = None, - use_grayscale: bool = False, - color_match: list = False, - best_match: bool = False -) -> TemplateMatch: + ref, + inp_img, + threshold=0.68, + roi=None, + use_grayscale=False, + color_match=False, + best_match=False, + max_workers=4 +): """ - Search for a template in an image + Search for a template in an image. Uses parallel matching for list inputs. + :param ref: Either key of a already loaded template, list of such keys, or a image which is used as template :param inp_img: Image in which the template will be searched :param threshold: Threshold which determines if a template is found or not :param roi: Region of Interest of the inp_img to restrict search area. Format [left, top, width, height] :param use_grayscale: Use grayscale template matching for speed up - :param color_match: Pass a color to be used by misc.color_filter to filter both image of interest and template image (format Config().colors["color"]) + :param color_match: Pass a color to be used by misc.color_filter to filter both image of interest and template image :param best_match: If list input, will search for list of templates by best match. Default behavior is first match. + :param max_workers: Max parallel threads for template matching (default 4) :return: Returns a TemplateMatch object with a valid flag """ templates = _process_template_refs(ref) + + # Single template — no benefit to parallelize + if len(templates) == 1: + match = _single_template_match(templates[0], inp_img, roi, color_match, use_grayscale) + return match if match.score >= threshold else TemplateMatch() + + # Multiple templates — parallel search matches = [] - for template in templates: - match = _single_template_match(template, inp_img, roi, color_match, use_grayscale) - if match.score >= threshold: - if not best_match: - return match - else: + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = [ + executor.submit(_match_template_worker, t, inp_img, roi, color_match, use_grayscale) + for t in templates + ] + for future in concurrent.futures.as_completed(futures): + match = future.result() + if match.score >= threshold: + if not best_match: + return match matches.append(match) + if matches: matches = sorted(matches, key=lambda obj: obj.score, reverse=True) return matches[0] diff --git a/src/utils/custom_mouse.py b/src/utils/custom_mouse.py index 4d140d4..8770eb7 100644 --- a/src/utils/custom_mouse.py +++ b/src/utils/custom_mouse.py @@ -14,6 +14,8 @@ import numpy as np import random import math import time +import threading +from concurrent.futures import Future import screen from config import Config from utils.misc import is_in_roi @@ -341,6 +343,40 @@ class mouse: def wheel(delta): _mouse.wheel(delta) + @staticmethod + def async_move(x, y, absolute=True, randomize=5, delay_factor=[0.4, 0.6]): + """ + Non-blocking mouse move. Returns immediately with a Future-like object. + + :return: A dict with: + - 'done()': callable returning bool + - 'wait(timeout=None)': blocks until move completes or timeout + """ + result = {"_done": False, "_lock": threading.Lock()} + + def _run(): + try: + mouse.move(x, y, absolute=absolute, randomize=randomize, delay_factor=delay_factor) + finally: + with result["_lock"]: + result["_done"] = True + + threading.Thread(target=_run, daemon=True).start() + + def done(): + with result["_lock"]: + return result["_done"] + + def wait(timeout=None): + deadline = None if timeout is None else time.monotonic() + timeout + while not done(): + if deadline is not None and time.monotonic() >= deadline: + return False + time.sleep(0.01) + return True + + return {"done": done, "wait": wait} + if __name__ == "__main__": import os diff --git a/src/utils/npc_auto_label.py b/src/utils/npc_auto_label.py new file mode 100644 index 0000000..6409ee4 --- /dev/null +++ b/src/utils/npc_auto_label.py @@ -0,0 +1,146 @@ +""" +Auto-label NPCs visible on screen. + +Scans for all known NPCs using their template groups (in parallel via +template_finder.search) and returns a dict of {npc_name: {center, score}}. +Can be called from any game loop tick without blocking the main thread. + +Usage: + from utils.npc_auto_label import detect_visible_npcs + npcs_on_screen = detect_visible_npcs(img) + # npcs_on_screen = {"AKARA": {"center": (620, 300), "score": 0.82}, ...} +""" + +import concurrent.futures +import template_finder +from screen import grab + + +# NPCs that exist per act — detected via their template groups in npc_manager +NPC_TEMPLATES = { + # Act 1 + "AKARA": ["AKARA_FRONT", "AKARA_BACK", "AKARA_SIDE", "AKARA_SIDE_2"], + "CHARSI": ["CHARSI_FRONT", "CHARSI_BACK", "CHARSI_SIDE", "CHARSI_SIDE_2", "CHARSI_SIDE_3"], + "KASHYA": ["KASHYA_FRONT", "KASHYA_BACK", "KASHYA_SIDE", "KASHYA_SIDE_2"], + "CAIN": ["CAIN_0", "CAIN_1", "CAIN_2", "CAIN_3"], + # Act 2 + "FARA": ["FARA_LIGHT_1", "FARA_LIGHT_3", "FARA_MEDIUM_1", "FARA_DARK_1"], + "DROGNAN": ["DROGNAN_FRONT", "DROGNAN_LEFT", "DROGNAN_RIGHT_SIDE"], + "LYSANDER": ["LYSANDER_FRONT", "LYSANDER_BACK", "LYSANDER_SIDE", "LYSANDER_SIDE_2"], + # Act 3 + "ORMUS": ["ORMUS_0", "ORMUS_2", "ORMUS_4"], + # Act 4 + "TYRAEL": ["TYRAEL_1", "TYRAEL_2"], + "JAMELLA": ["JAMELLA_FRONT", "JAMELLA_BACK", "JAMELLA_SIDE"], + "HALBU": ["HALBU_FRONT", "HALBU_BACK", "HALBU_SIDE", "HALBU_SIDE_2"], + # Act 5 + "QUAL_KEHK": ["QUAL_0", "QUAL_45", "QUAL_180", "QUAL_270"], + "MALAH": ["MALAH_FRONT", "MALAH_BACK", "MALAH_45", "MALAH_SIDE"], + "LARZUK": ["LARZUK_FRONT", "LARZUK_BACK", "LARZUK_SIDE"], + "ANYA": ["ANYA_FRONT", "ANYA_BACK", "ANYA_SIDE"], +} + +# How many NPCs to search in parallel (limit thread pool) +MAX_WORKERS = 8 + +# Search ROI — skip the bottom skill bar area to reduce false positives +SEARCH_ROI = [0, 0, 1280, 480] + + +def _search_npc(npc_name, template_keys, img): + """Search for one NPC using all its template variants. Returns match or None.""" + best = template_finder.search( + template_keys, + img, + threshold=0.55, + roi=SEARCH_ROI, + best_match=True, + ) + if best.valid: + return { + "center": best.center, + "center_monitor": best.center_monitor, + "score": best.score, + } + return None + + +def detect_visible_npcs(img=None): + """ + Scan the screen for all known NPCs. + + :param img: Screenshot (BGR numpy array). If None, grabs one. + :return: dict {npc_name: {"center": (x,y), "center_monitor": (x,y), "score": float}} + """ + if img is None: + img = grab() + + results = {} + + with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: + futures = { + executor.submit(_search_npc, name, templates, img): name + for name, templates in NPC_TEMPLATES.items() + } + for future in concurrent.futures.as_completed(futures): + npc_name = futures[future] + try: + match = future.result() + if match is not None: + results[npc_name] = match + except Exception: + pass + + return results + + +# ─── Optional: persistent cache that expires after N seconds ─── + +_cache = {} +_cache_time = 0.0 + + +def detect_visible_npcs_cached(img=None, ttl=5.0): + """ + Like detect_visible_npcs() but caches results for ttl seconds. + Use this when scanning every tick but only need fresh data periodically. + """ + import time + now = time.time() + if now - _cache_time < ttl: + return _cache + _cache.clear() + _cache.update(detect_visible_npcs(img)) + global _cache_time + _cache_time = now + return _cache + + +if __name__ == "__main__": + import cv2 + import keyboard + from screen import start_detecting_window + + start_detecting_window() + print("Press F12 to exit. Scanning for NPCs...") + + while True: + keyboard.pause(0.05, True) + if keyboard.is_pressed("f12"): + break + + img = grab() + found = detect_visible_npcs(img) + + display = img.copy() + for name, info in found.items(): + x, y = info["center"] + cv2.putText(display, name, (x - 30, y - 10), + cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2, cv2.LINE_AA) + cv2.circle(display, (x, y), 5, (0, 255, 0), -1) + + if found: + npc_strs = [f"{n}({info['score']:.2f})" for n, info in found.items()] + print(f"NPCs on screen: {', '.join(npc_strs)}") + cv2.imshow("NPC Auto-Label", display) + cv2.waitKey(1) \ No newline at end of file