71 lines
3.6 KiB
Python
71 lines
3.6 KiB
Python
"""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('没有已验证的导航路径；需要先补充页面与返回关系。')
|