Skip to content
Closed
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
3 changes: 2 additions & 1 deletion aao/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@
from aao.measure.overlay import OverlayWindow # noqa: E402
from aao.measure.worker import MeasurementWorker # noqa: E402
from aao.timeline.editor_window import EditorWindow # noqa: E402
from aao.types import MeasureState # noqa: E402
from aao.ui.about_page import AboutPage # noqa: E402
from aao.ui.background import BackgroundContainer # noqa: E402
from aao.ui.calibration_page import CalibrationPage # noqa: E402
Expand Down Expand Up @@ -493,7 +494,7 @@ def _reset_measure_timer(self, node_name: str) -> None:
logger.debug("计时器重置(pipeline 节点: %s)", node_name)
self.worker.request_reset_timer()

def _on_measure_state(self, state: dict) -> None:
def _on_measure_state(self, state: MeasureState) -> None:
from aao.core.timing.time_source import format_timer

total = state.get("totalElapsedFrames", 0)
Expand Down
27 changes: 21 additions & 6 deletions aao/core/avatar.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import json
import time
from pathlib import Path
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, TypedDict, cast

import numpy as np

Expand All @@ -25,6 +25,12 @@
_DETAIL_WAIT = 1 # 等详情页打开


class Slot(TypedDict):
flag_rect: tuple[int, int, int, int]
avatar_rect: tuple[int, int, int, int]
click_pos: tuple[int, int]


def _avatar_dir() -> Path:
from aao.utils.runtime_paths import project_root

Expand All @@ -39,7 +45,14 @@ def _match_avatar_roi_offset() -> tuple[int, int, int, int]:

path = project_root() / "resource" / "base" / "pipeline" / "reco.json"
raw = json.loads(path.read_text(encoding="utf-8"))
offset = raw["MatchAvatar"].get("roi_offset", [0, 0, 0, 0])
if not isinstance(raw, dict):
return (0, 0, 0, 0)
data = cast(dict[str, object], raw)
match_avatar = data.get("MatchAvatar")
if not isinstance(match_avatar, dict):
return (0, 0, 0, 0)
match_avatar = cast(dict[str, object], match_avatar)
offset = match_avatar.get("roi_offset", [0, 0, 0, 0])
return tuple(int(v) for v in offset) # type: ignore[return-value]


Expand All @@ -56,7 +69,9 @@ def _get_char_id(oper_name: str) -> str:
if not mapping_path.exists():
return ""
mapping = json.loads(mapping_path.read_text(encoding="utf-8"))
return mapping.get(oper_name, "")
if not isinstance(mapping, dict):
return ""
return str(cast(dict[str, object], mapping).get(oper_name, ""))


def _normalize_name(name: str) -> str:
Expand All @@ -71,14 +86,14 @@ def _normalize_name(name: str) -> str:
def detect_slots(
context: Context,
image: np.ndarray,
) -> list[dict]:
) -> list[Slot]:
"""用 pipeline 节点 DetectSlots 检测待部署区所有干员槽位。"""
reco_detail = context.run_recognition("DetectSlots", image)

if not reco_detail or not reco_detail.hit:
return []

slots = []
slots: list[Slot] = []
for result in reco_detail.all_results:
box = getattr(result, "box", None)
if box is None:
Expand Down Expand Up @@ -217,7 +232,7 @@ def _ocr_oper_name(context: Context, detail_img: np.ndarray) -> str | None:

def _save_avatar_from_image(
image: np.ndarray,
slot: dict,
slot: Slot,
oper_name: str,
) -> bool:
"""从截图截取槽位头像并存盘。"""
Expand Down
19 changes: 13 additions & 6 deletions aao/core/geometry/map_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,9 @@

import json
from pathlib import Path
from typing import cast

from aao.types import JsonObject
from aao.utils.logger import logger
from aao.utils.runtime_paths import project_root

Expand All @@ -25,7 +27,8 @@ def _level_codes() -> dict[str, str]:
path = project_root() / "data" / "level_codes.json"
if not path.exists():
return {}
return json.loads(path.read_text(encoding="utf-8"))
data = json.loads(path.read_text(encoding="utf-8"))
return cast(dict[str, str], data) if isinstance(data, dict) else {}


def find_map_file(code: str) -> Path | None:
Expand All @@ -52,17 +55,21 @@ def find_map_file(code: str) -> Path | None:
return None


def load_map(code: str) -> dict | None:
def load_map(code: str) -> JsonObject | None:
"""加载关卡数据。code 如 '1-7'。"""
path = find_map_file(code)
if path is None:
logger.error("未找到关卡 %s 的地图数据", code)
return None
data = json.loads(path.read_text(encoding="utf-8"))
logger.info(
"加载关卡 %s (%s): %dx%d", code, data.get("name", "?"), data["height"], data["width"]
)
return data
if not isinstance(data, dict):
logger.error("关卡 %s 的地图数据格式无效", code)
return None
name = str(cast(JsonObject, data).get("name", "?"))
height = int(cast(JsonObject, data)["height"])
width = int(cast(JsonObject, data)["width"])
logger.info("加载关卡 %s (%s): %dx%d", code, name, height, width)
return cast(JsonObject, data)


def list_codes() -> list[str]:
Expand Down
4 changes: 3 additions & 1 deletion aao/core/geometry/view.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@

import numpy as np

from aao.types import JsonObject

# 投影参数(prts-plus ViewCalculationConfig)
_FROM_RATIO = 9 / 16
_NEAR = 0.3
Expand Down Expand Up @@ -69,7 +71,7 @@ def _build_matrix(view_offset: list[float], side: bool) -> np.ndarray:


def transform_map_to_view(
level_data: dict,
level_data: JsonObject,
side: bool = False,
) -> list[list[tuple[float, float]]]:
"""把关卡数据投影为屏幕坐标(0-1 比例)。
Expand Down
5 changes: 4 additions & 1 deletion aao/core/timing/tick.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from __future__ import annotations

import bisect
from typing import cast

import numpy as np

Expand Down Expand Up @@ -245,9 +246,11 @@ def detect_negative_cost(frame: np.ndarray) -> bool:

region = frame[y : y + h, x : x + w].astype(np.int16) # (h, w, 3) BGR
white = (region > config.WHITE_THRESHOLD).all(axis=2) # (h, w) bool
white_rows = cast(np.ndarray, white)

# 逐行求最长连续纯白 run:对每行做行内累计,遇非白清零,取全局最大。
for row in white: # pyright: ignore[reportGeneralTypeIssues]
for row_index in range(int(white_rows.shape[0])):
row = cast(np.ndarray, white_rows[row_index])
if not row.any():
continue
# 累计连续 True 长度:cumsum 在非白处“断点”归零的经典向量化写法。
Expand Down
5 changes: 3 additions & 2 deletions aao/measure/api_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

import websockets

from aao.types import MeasureState
from aao.utils.logger import logger

DEFAULT_PORT = 2606
Expand All @@ -25,7 +26,7 @@ class ApiServer:

def __init__(
self,
get_state: Callable[[], dict],
get_state: Callable[[], MeasureState],
host: str = "localhost",
port: int = DEFAULT_PORT,
rate_hz: float = 60.0,
Expand All @@ -34,7 +35,7 @@ def __init__(
self.host = host
self.port = port
self.rate_hz = rate_hz
self._clients: set = set()
self._clients: set[Any] = set()
self._thread: threading.Thread | None = None

def start(self) -> None:
Expand Down
3 changes: 2 additions & 1 deletion aao/measure/overlay.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from PySide6.QtWidgets import QApplication, QLabel, QSizeGrip, QVBoxLayout, QWidget

from aao.core.timing.time_source import format_timer
from aao.types import MeasureState
from aao.ui import floating_state
from aao.ui.window_snap import (
create_snap_follow,
Expand Down Expand Up @@ -131,7 +132,7 @@ def reset_layout(self, x: int, y: int) -> None:
self.move(x, y)
self._save_window_state()

def on_state(self, state: dict) -> None:
def on_state(self, state: MeasureState) -> None:
running = state.get("isRunning", False)
cf = state.get("currentFrame")
total = state.get("totalFramesInCycle", 0)
Expand Down
9 changes: 5 additions & 4 deletions aao/measure/worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

from aao.core.timing.calibration import FullCalibrationData
from aao.core.timing.time_source import TimeSource
from aao.types import MeasureState
from aao.utils.logger import logger

if TYPE_CHECKING:
Expand Down Expand Up @@ -47,13 +48,13 @@ def __init__(
self._running = False
self._reset_requested = False
self._consecutive_errors = 0
self._latest: dict = {}
self._latest: MeasureState = {}
self._lock = threading.Lock()

@property
def latest_state(self) -> dict:
def latest_state(self) -> MeasureState:
with self._lock:
return dict(self._latest)
return self._latest.copy()

def run(self) -> None:
self._running = True
Expand Down Expand Up @@ -98,7 +99,7 @@ def run(self) -> None:
time.sleep(backoff)
continue

state = {
state: MeasureState = {
"isRunning": self.time_source.is_running,
"currentFrame": self.time_source.current_frame_in_cycle,
"totalFramesInCycle": self.time_source.total_frames_in_cycle,
Expand Down
25 changes: 15 additions & 10 deletions aao/resources/updater.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,9 +25,11 @@
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import cast

from aao import __version__
from aao.resources import syncer
from aao.types import JsonObject
from aao.utils.logger import logger, setup_logging
from aao.utils.runtime_paths import is_frozen, project_root

Expand Down Expand Up @@ -93,7 +95,8 @@ def check_software(self) -> ReleaseInfo | None:
if token:
req.add_header("Authorization", f"Bearer {token}")
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
raw = json.loads(resp.read())
data = cast(JsonObject, raw) if isinstance(raw, dict) else {}
except Exception as e: # noqa: BLE001
logger.warning("检查更新失败: %s", e)
return None
Expand Down Expand Up @@ -140,14 +143,12 @@ def _fetch_changelog_since(self, current: str) -> str:
try:
req = _make_request(_RELEASES_LIST_API, _settings_github_token())
with urllib.request.urlopen(req, timeout=10) as resp:
releases = json.loads(resp.read())
raw = json.loads(resp.read())
releases = cast(list[JsonObject], raw) if isinstance(raw, list) else []
except Exception as e: # noqa: BLE001
logger.warning("拉取 release 列表失败,降级为单版 changelog: %s", e)
return ""

if not isinstance(releases, list):
return ""

# 筛 version > current 的,按版本降序排
entries: list[tuple[str, str]] = [] # (version, body)
for r in releases:
Expand Down Expand Up @@ -209,13 +210,13 @@ def update_resources(
progress_cb(r.message)
return results

def update_all(self, progress_cb: Callable[[str], None] | None = None) -> dict:
def update_all(self, progress_cb: Callable[[str], None] | None = None) -> JsonObject:
"""检查软件更新 + 更新资源。

Returns:
{"software": ReleaseInfo | None, "resources": [SyncResult]}
"""
result: dict = {}
result: JsonObject = {}

if progress_cb:
progress_cb("检查软件更新...")
Expand Down Expand Up @@ -417,7 +418,7 @@ def apply_update(self, zip_path: Path) -> None:
_DOWNLOAD_BACKOFF_SEC = 1.0


def _pick_win_asset(assets: list) -> AssetInfo | None:
def _pick_win_asset(assets: list[JsonObject]) -> AssetInfo | None:
"""从 release assets 里选 win-x64 zip。"""
for a in assets:
name = str(a.get("name", "")).lower()
Expand All @@ -437,7 +438,9 @@ def _settings_proxy() -> str | None:
path = project_root() / "config" / "settings.json"
if not path.exists():
return None
proxy = json.loads(path.read_text(encoding="utf-8")).get("proxy", "")
raw = json.loads(path.read_text(encoding="utf-8"))
data = cast(JsonObject, raw) if isinstance(raw, dict) else {}
proxy = data.get("proxy", "")
return str(proxy).strip() or None
except Exception: # noqa: BLE001
return None
Expand All @@ -450,7 +453,9 @@ def _settings_github_token() -> str | None:
path = project_root() / "config" / "settings.json"
if not path.exists():
return None
enc = json.loads(path.read_text(encoding="utf-8")).get("github_token_enc", "")
raw = json.loads(path.read_text(encoding="utf-8"))
data = cast(JsonObject, raw) if isinstance(raw, dict) else {}
enc = data.get("github_token_enc", "")
return decrypt_text(str(enc)) if enc else None
except Exception: # noqa: BLE001
return None
Expand Down
2 changes: 2 additions & 0 deletions aao/timeline/editor_window.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from __future__ import annotations

import json
from typing import cast

from PySide6.QtCore import QEvent, QObject, QRect, Qt, Signal
from PySide6.QtGui import QColor, QFont, QPainter, QPalette
Expand Down Expand Up @@ -316,6 +317,7 @@ def _restore_side_panel_state(self) -> None:
s = load_settings()
states = s.get("collapsible_sections", {})
if isinstance(states, dict) and "timeline_side_panel" in states:
states = cast(dict[str, object], states)
self._set_side_panel_visible(bool(states["timeline_side_panel"]))

def _style_frame_label(self) -> None:
Expand Down
Loading