feat: add page map recognition and grouped task planning
This commit is contained in:
1 parent
386198e8e0
commit
f7628ab41e
76 files changed
+12437
-49
No files matched your search
@@ -0,0 +1,70 @@
|
||||
"""Read-only page recognition and route planning from the game page map."""
|
||||
import json
|
||||
from collections import deque
|
||||
from pathlib import Path
|
||||
|
||||
from .engine import Matcher
|
||||
|
||||
|
||||
class PageMap:
|
||||
def __init__(self, root):
|
||||
self.root = Path(root)
|
||||
self.data = json.loads((self.root / 'workflows/page_map.json').read_text(encoding='utf-8'))
|
||||
self.pages = {page['id']: page for page in self.data['pages']}
|
||||
self.matcher = Matcher(self.root / 'assets')
|
||||
|
||||
def identify(self, frame, previous_page=None):
|
||||
candidates = []
|
||||
for page in self.pages.values():
|
||||
for variant in page.get('recognition', []):
|
||||
scores = [self.matcher.score(frame, spec) for spec in variant]
|
||||
if scores and all(score >= spec.get('threshold', .86)
|
||||
for score, spec in zip(scores, variant)):
|
||||
candidates.append({'id': page['id'], 'name': page['name'],
|
||||
'kind': page['kind'], 'score': round(min(scores), 4)})
|
||||
break
|
||||
# A specific matching tab supersedes its matching container. Preserve
|
||||
# ambiguity between unrelated pages, even if their templates overlap.
|
||||
def ancestors(page_id):
|
||||
seen = set()
|
||||
current = self.pages[page_id].get('parent')
|
||||
while current in self.pages and current not in seen:
|
||||
seen.add(current)
|
||||
current = self.pages[current].get('parent')
|
||||
return seen
|
||||
|
||||
matched_parents = set().union(*(ancestors(item['id']) for item in candidates)) if candidates else set()
|
||||
specific = [item for item in candidates if item['id'] not in matched_parents]
|
||||
# Tabs inside an overlay retain the overlay's priority. The parent
|
||||
# of a generic reward still comes from history, never from its art.
|
||||
overlays = [item for item in specific if item['kind'] == 'overlay' or any(
|
||||
self.pages[parent]['kind'] == 'overlay' for parent in ancestors(item['id']))]
|
||||
active = overlays or specific
|
||||
# Shared popups may have identical pixels. Only a known navigation
|
||||
# origin can choose between them; absence of origin stays ambiguous.
|
||||
groups = {self.pages[item['id']].get('shared_visual_group') for item in active}
|
||||
if len(active) > 1 and len(groups) == 1 and None not in groups and previous_page in self.pages:
|
||||
contextual = [item for item in active if previous_page in ancestors(item['id'])]
|
||||
if len(contextual) == 1:
|
||||
active = contextual
|
||||
return {'status': 'identified' if len(active) == 1 else 'ambiguous' if active else 'unknown',
|
||||
'page': active[0] if len(active) == 1 else None, 'candidates': candidates}
|
||||
|
||||
def route(self, start, target):
|
||||
"""Plan verified navigation only; this method sends no input events."""
|
||||
if start not in self.pages or target not in self.pages:
|
||||
raise ValueError('未知页面 ID。')
|
||||
queue = deque([(start, [])])
|
||||
seen = {start}
|
||||
while queue:
|
||||
current, steps = queue.popleft()
|
||||
if current == target:
|
||||
return steps
|
||||
for edge in self.data['edges']:
|
||||
if edge['from'] != current or edge['status'] not in ('live_verified', 'workflow_verified'):
|
||||
continue
|
||||
if edge['to'] in seen:
|
||||
continue
|
||||
seen.add(edge['to'])
|
||||
queue.append((edge['to'], steps + [edge]))
|
||||
raise ValueError('没有已验证的导航路径;需要先补充页面与返回关系。')
|
||||
Reference in new issue
Block a user