diff --git a/README.md b/README.md index ed51700..da63917 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,8 @@ SOPS 文件、运行时 env、SSH key 或 mihomo 配置。 - `Enter`:仅在目标在线时进入其 controller 详情。 - `r`:刷新连通性状态。 - `t`:对当前 Selector 组的候选节点做 mihomo delay 探测;默认使用 - `https://www.gstatic.com/generate_204`,最多并发 4 个。 + `https://www.gstatic.com/generate_204`,最多并发 4 个。每个 candidate 完成后会立即 + 更新 delay 和当前推荐,慢节点或超时节点不会阻塞先完成的结果显示。 - `s`:对当前选择的候选节点发起运行态切换。TUI 会显示机器、组、旧节点和新节点, 必须按 `y` 二次确认;成功后立即读取 controller 验证当前节点。 diff --git a/proxy_monitor_tui.py b/proxy_monitor_tui.py index 22363f2..b6c5320 100644 --- a/proxy_monitor_tui.py +++ b/proxy_monitor_tui.py @@ -8,7 +8,7 @@ import json import os from pathlib import Path import sys -from typing import Any +from typing import Any, Iterator import urllib.error import urllib.parse import urllib.request @@ -287,7 +287,7 @@ def recommend_fastest(delay_results: dict[str, tuple[int | None, str]]) -> str | return min(successful)[1] if successful else None -def measure_candidates( +def iter_measure_candidates( client: ProxyMonitorClient, target_name: str, candidates: list[str], @@ -295,9 +295,9 @@ def measure_candidates( test_url: str, timeout_ms: int, concurrency: int, -) -> dict[str, tuple[int | None, str]]: +) -> Iterator[tuple[str, tuple[int | None, str]]]: + """Yield each candidate result as soon as its controller request finishes.""" unique_candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate)) - results: dict[str, tuple[int | None, str]] = {} def measure(candidate: str) -> tuple[int | None, str]: try: @@ -316,9 +316,32 @@ def measure_candidates( for future in as_completed(futures): candidate = futures[future] try: - results[candidate] = future.result() + yield candidate, future.result() except Exception as exc: # noqa: BLE001 - results[candidate] = (None, str(exc)) + yield candidate, (None, str(exc)) + + +def measure_candidates( + client: ProxyMonitorClient, + target_name: str, + candidates: list[str], + *, + test_url: str, + timeout_ms: int, + concurrency: int, +) -> dict[str, tuple[int | None, str]]: + """Collect all candidate results for non-interactive callers.""" + unique_candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate)) + results = dict( + iter_measure_candidates( + client, + target_name, + unique_candidates, + test_url=test_url, + timeout_ms=timeout_ms, + concurrency=concurrency, + ) + ) return {candidate: results.get(candidate, (None, "not run")) for candidate in unique_candidates} @@ -431,6 +454,7 @@ def run_textual( self.target = target self.groups_by_name: dict[str, dict[str, Any]] = {} self.delay_results: dict[str, dict[str, tuple[int | None, str]]] = {} + self.delay_test_runs: dict[str, int] = {} self.selected_group = "" self.selected_node = "" self.pending_switch: tuple[str, str] | None = None @@ -537,27 +561,69 @@ def run_textual( if not group: self._set_operation("choose a Selector group first") return - self._set_operation(f"testing {len(as_list(group.get('candidates')))} candidates...") - self.test_selected_group(str(group["name"]), list(as_list(group.get("candidates")))) + group_name = str(group["name"]) + candidates = list(dict.fromkeys(str(candidate) for candidate in as_list(group.get("candidates")) if candidate)) + run_id = self.delay_test_runs.get(group_name, 0) + 1 + self.delay_test_runs[group_name] = run_id + self.delay_results[group_name] = {} + self._show_group(group_name) + self._set_operation( + f"testing {len(candidates)} candidates; each completed delay appears immediately..." + ) + self.test_selected_group(group_name, candidates, run_id) @work(thread=True, exclusive=True) - def test_selected_group(self, group_name: str, candidates: list[str]) -> None: - results = measure_candidates( + def test_selected_group(self, group_name: str, candidates: list[str], run_id: int) -> None: + total = len(candidates) + for candidate, result in iter_measure_candidates( client, str(self.target["name"]), candidates, test_url=test_url, timeout_ms=test_timeout_ms, concurrency=delay_concurrency, + ): + self.app.call_from_thread( + self._record_delay_result, + group_name, + run_id, + candidate, + result, + total, + ) + self.app.call_from_thread(self._finish_delay_test, group_name, run_id) + + def _record_delay_result( + self, + group_name: str, + run_id: int, + candidate: str, + result: tuple[int | None, str], + total: int, + ) -> None: + if self.delay_test_runs.get(group_name) != run_id: + return + results = self.delay_results.setdefault(group_name, {}) + results[candidate] = result + completed = len(results) + recommended = recommend_fastest(results) + if group_name == self.selected_group: + self._show_group(group_name) + delay, detail = result + latest = f"{delay} ms" if delay is not None else detail + self._set_operation( + f"testing {completed}/{total}; {candidate}: {latest}; " + f"best so far: {recommended or 'none'}" ) - self.app.call_from_thread(self._finish_delay_test, group_name, results) def _finish_delay_test( self, group_name: str, - results: dict[str, tuple[int | None, str]], + run_id: int, ) -> None: - self.delay_results[group_name] = results + if self.delay_test_runs.get(group_name) != run_id: + return + results = self.delay_results.get(group_name, {}) recommended = recommend_fastest(results) if group_name == self.selected_group: self._show_group(group_name) diff --git a/pyproject.toml b/pyproject.toml index f2e01f2..09e822b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "proxy-port-monitor-tui" -version = "0.1.1" +version = "0.1.2" description = "Terminal UI for proxy-port-monitor and mihomo selector groups" readme = "README.md" requires-python = ">=3.11" diff --git a/tests/test_proxy_monitor_tui.py b/tests/test_proxy_monitor_tui.py index ece29e9..1afd5a7 100644 --- a/tests/test_proxy_monitor_tui.py +++ b/tests/test_proxy_monitor_tui.py @@ -163,6 +163,37 @@ class ProxyMonitorTUITests(unittest.TestCase): ) self.assertEqual(recommended, "fast") + def test_delay_results_are_yielded_before_slower_candidates_finish(self) -> None: + class ControlledClient: + def __init__(self) -> None: + self.slow_started = threading.Event() + self.release_slow = threading.Event() + + def measure_delay(self, _target: str, candidate: str, **_kwargs: object) -> int: + if candidate == "slow": + self.slow_started.set() + self.release_slow.wait(timeout=2) + return 300 + self.slow_started.wait(timeout=2) + return 42 + + client = ControlledClient() + stream = self.mod.iter_measure_candidates( + client, + "pc", + ["slow", "fast"], + test_url="https://www.gstatic.com/generate_204", + timeout_ms=5000, + concurrency=2, + ) + try: + candidate, result = next(stream) + self.assertEqual(candidate, "fast") + self.assertEqual(result, (42, "42 ms")) + finally: + client.release_slow.set() + self.assertEqual(list(stream), [("slow", (300, "300 ms"))]) + def test_machine_rows_preserve_monitor_order_and_status(self) -> None: rows = self.mod.normalize_machine_rows( { diff --git a/uv.lock b/uv.lock index 20f24e0..42f6ba3 100644 --- a/uv.lock +++ b/uv.lock @@ -63,7 +63,7 @@ wheels = [ [[package]] name = "proxy-port-monitor-tui" -version = "0.1.1" +version = "0.1.2" source = { editable = "." } dependencies = [ { name = "textual" },