Skip to content

feat: 适配新 anchor #161

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
weinibuliu merged 4 commits into from
Feb 21, 2026
Merged

feat: 适配新 anchor #161

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8250ed4
feat: 支持 new anchor
weinibuliu Feb 19, 2026
7285ba4
vibe
weinibuliu Feb 20, 2026
ab0144c
Update runtime_control.py
weinibuliu Feb 20, 2026
177c5ab
fix
weinibuliu Feb 21, 2026
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
1 change: 1 addition & 0 deletions src/MaaDebugger/maafw/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -448,6 +448,7 @@ def on_node_next_list(
"msg": f"NextList.{msg_suffix}",
"name": detail.name,
"next_list": [attr.name for attr in detail.next_list],
"anchor_flags": [attr.anchor for attr in detail.next_list],
Comment thread
weinibuliu marked this conversation as resolved.
}
if debug_mode:
print(f"[DEBUG EventSink] {msg}")
Expand Down
134 changes: 124 additions & 10 deletions src/MaaDebugger/webpage/index_page/runtime_control.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,14 @@ class ItemData:
name: str
reco_id: int = 0
status: Status = Status.PENDING
# 是否通过 anchor 解析得到
is_anchor: bool = False
# 锚点名称（如 "锚点"），仅当 is_anchor=True 时有效
anchor_name: str = ""
# 锚点指向的目标节点名称
anchor_target: str = ""
# 锚点设置者节点名称（即定义 anchor 的节点）
anchor_setter: str = ""
# 嵌套子项的容器引用（用于 RecognitionNode 添加子识别项）
nested_container: Optional[Any] = field(default=None, repr=False)
# 嵌套子项列表
Expand All @@ -59,6 +67,11 @@ class ListData:
row_len: int
current: str
next_list: List[str]
anchor_flags: List[bool] = field(default_factory=list)
# 每个 next_list 项对应的锚点名称（空字符串表示非锚点）
anchor_names: List[str] = field(default_factory=list)
# 每个 next_list 项对应的锚点目标节点名称（空字符串表示非锚点）
anchor_targets: List[str] = field(default_factory=list)


def main():
Expand Down Expand Up @@ -193,8 +206,25 @@ def on_page_change(self, page: int):
for row_len in row_len_list[start_index:end_index]:
self.create_list(self.other_page_row, self.list_data_map[row_len])

def add_item_data(self, index, name, row_len: int):
data = ItemData(row_len, index, name)
def add_item_data(
self,
index,
name,
row_len: int,
is_anchor: bool = False,
anchor_name: str = "",
anchor_target: str = "",
anchor_setter: str = "",
):
data = ItemData(
row_len,
index,
name,
is_anchor=is_anchor,
anchor_name=anchor_name,
anchor_target=anchor_target,
anchor_setter=anchor_setter,
)
self.data[row_len][index] = data

def create_list(self, row: ui.row, data: ListData):
Expand Down Expand Up @@ -227,7 +257,17 @@ def create_items(self, index: int, name: str, row_len: int):
StatusIndicator(data, "status")

with ui.item_section():
ui.item_label(name)
if data.is_anchor:
with ui.row(wrap=False).classes("items-center gap-1"):
display_name = data.anchor_target or name
anchor_label = (
f"{display_name} (anchor:{data.anchor_name or '未知'}, "
f"from:{data.anchor_setter or '未知'})"
)
ui.item_label(anchor_label)
ui.badge("⚓", color="blue-grey-4").props("rounded")
else:
ui.item_label(name)

with ui.item_section().props("side"):
ui.item_label().bind_text_from(data, "reco_id").bind_visibility_from(
Expand Down Expand Up @@ -296,9 +336,12 @@ async def _handle_message(self, msg: Dict[str, Any]):
if msg_type == "NextList.Starting":
name = msg.get("name", "")
next_names = msg.get("next_list", [])
anchor_flags = msg.get("anchor_flags", [])
if debug_mode:
print(f"[DEBUG] NextList.Starting: name={name}, next_list={next_names}")
self._on_next_list_starting(name, next_names)
print(
f"[DEBUG] NextList.Starting: name={name}, next_list={next_names}, anchor_flags={anchor_flags}"
)
self._on_next_list_starting(name, next_names, anchor_flags)
await maafw.screenshotter.refresh(False)

# RecognitionNode.Starting - 标记进入嵌套识别模式
Expand Down Expand Up @@ -407,7 +450,11 @@ def _on_recognized(self, reco_id: int, name: str, hit: bool):
# 如果通过 reco_id 没找到，检查当前 row（非嵌套项）
if not found:
for item in self.data[self.row_len].values():
if item.status == Status.PENDING and item.name == name:
if item.status != Status.PENDING:
continue
name_matched = item.name == name
anchor_matched = item.is_anchor and item.anchor_target == name
if name_matched or anchor_matched:
item.reco_id = reco_id
item.status = Status.SUCCEEDED if hit else Status.FAILED
matched_item = item
Expand All @@ -421,7 +468,11 @@ def _on_recognized(self, reco_id: int, name: str, hit: bool):
if not found:
for row_data in self.data.values():
for item in row_data.values():
if item.status == Status.PENDING and item.name == name:
if item.status != Status.PENDING:
continue
name_matched = item.name == name
anchor_matched = item.is_anchor and item.anchor_target == name
if name_matched or anchor_matched:
item.reco_id = reco_id
item.status = Status.SUCCEEDED if hit else Status.FAILED
matched_item = item
Expand Down Expand Up @@ -458,11 +509,59 @@ def _on_reco_node_ended(self):
f"[DEBUG] RecognitionNode depth decreased to {self._reco_node_depth}, stack size: {len(self._recognition_stack)}"
)

def _on_next_list_starting(self, current: str, next_list: List[str]):
def _on_next_list_starting(
self, current: str, next_list: List[str], anchor_flags: List[bool] = []
):
"""处理 NextList 开始事件"""
Copy link

Copilot AI Feb 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

anchor_flags uses a mutable default ([]) in the function signature, which can leak state across calls and produce incorrect results if the list is ever mutated. Use None as the default and normalize inside the function (e.g., anchor_flags = anchor_flags or []).

Suggested change
self, current: str, next_list: List[str], anchor_flags: List[bool] = []
):
"""处理 NextList 开始事件"""
self, current: str, next_list: List[str], anchor_flags: List[bool] = None
):
"""处理 NextList 开始事件"""
# Normalize mutable default
anchor_flags = anchor_flags or []

Copilot uses AI. Check for mistakes.
self.row_len += 1

list_data = ListData(self.row_len, current, next_list)
# 从 pipeline 的 next 列表中提取 [Anchor] 标记，得到锚点名和目标节点
anchor_names: List[str] = []
anchor_targets: List[str] = []
if any(anchor_flags):
node_data = maafw.get_node_data(current)
node_next = node_data.get("next", [])
anchor_map = (
node_data.get("anchor", {}) if isinstance(node_data, dict) else {}
)
for i, _name in enumerate(next_list):
is_anchor = i < len(anchor_flags) and anchor_flags[i]
if not is_anchor:
anchor_names.append("")
anchor_targets.append("")
continue

anchor_name = ""
anchor_target = ""

if i < len(node_next) and isinstance(node_next[i], str):
raw_next = node_next[i]
if raw_next.startswith("[Anchor]"):
anchor_name = raw_next.replace("[Anchor]", "", 1)
if isinstance(anchor_map, dict):
anchor_target = anchor_map.get(anchor_name, "")

if (
not anchor_name
and isinstance(anchor_map, dict)
and isinstance(_name, str)
):
if _name in anchor_map:
anchor_name = _name
anchor_target = anchor_map.get(_name, "")

if not anchor_target and isinstance(_name, str):
anchor_target = _name

anchor_names.append(anchor_name)
anchor_targets.append(anchor_target)
else:
anchor_names = [""] * len(next_list)
anchor_targets = [""] * len(next_list)

list_data = ListData(
self.row_len, current, next_list, anchor_flags, anchor_names, anchor_targets
)
self.add_list_data(list_data)

# 299/300 -> page:1 | 300/300 -> page:2
Expand Down Expand Up @@ -507,7 +606,22 @@ def add_list_data(self, data: ListData):
self.list_data_map[data.row_len] = data
for index in range(len(data.next_list)):
name = data.next_list[index]
self.add_item_data(index, name, data.row_len)
is_anchor = index < len(data.anchor_flags) and data.anchor_flags[index]
anchor_name = (
data.anchor_names[index] if index < len(data.anchor_names) else ""
)
anchor_target = (
data.anchor_targets[index] if index < len(data.anchor_targets) else ""
)
self.add_item_data(
index,
name,
data.row_len,
is_anchor=is_anchor,
anchor_name=anchor_name,
anchor_target=anchor_target,
anchor_setter=data.current if is_anchor else "",
)

def on_resource_loading(
self,
Expand Down