Skip to content
Merged
Show file tree
Hide file tree
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
1 change: 1 addition & 0 deletions src/MaaDebugger/maafw/__init__.py
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],
}
if debug_mode:
print(f"[DEBUG EventSink] {msg}")
Expand Down
149 changes: 139 additions & 10 deletions src/MaaDebugger/webpage/index_page/runtime_control.py
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,74 @@ 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: Optional[List[bool]] = None,
):
"""处理 NextList 开始事件"""
self.row_len += 1

list_data = ListData(self.row_len, current, next_list)
normalized_anchor_flags = anchor_flags or []

# 从 pipeline 的 next 列表中提取 [Anchor] 标记,得到锚点名和目标节点
anchor_names: List[str] = []
anchor_targets: List[str] = []
if any(normalized_anchor_flags):
node_data = maafw.get_node_data(current)
if isinstance(node_data, dict):
node_next = node_data.get("next", [])
anchor_map = node_data.get("anchor", {})
else:
node_next = []
anchor_map = {}

for i, _name in enumerate(next_list):
is_anchor = (
i < len(normalized_anchor_flags) and normalized_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,
normalized_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 +621,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