"""AutoPOE2 - Closed-Loop Motion Feedback Sentinel (SSoT Doc 69 / Doc 72). ============================================================================= Cảm biến phản hồi chuyển động vòng lặp kín (Dual-Tier Motion Feedback): - Khắc phục triệt để hiện tượng "tưởng là đang chạy nhưng thực tế đứng yên" (Ghost Moving). - Kết hợp 2 cơ chế kiểm chứng độc lập: 1. Tier 1 (RAM Velocity Interlock - 0ms): Đo vi sai Delta XYZ trước và sau khi gửi phím WASD. 2. Tier 2 (Optical Frame Differencing): Đo vi sai điểm ảnh / template shift khi RAM mất tín hiệu. - Ngăn chặn triệt để Coordinator thử lại mù quáng 5 lần (30-60s) và ném cảnh báo sớm. - Tuân thủ nghiêm ngặt: - Rule 4: Module hóa tối đa, trần file < 400 dòng. - Rule 5: 100% dữ liệu thực nghiệm, tính toán vi sai toán học chuẩn xác. - Bất biến INV-MOTION-IMMOBILE: Phát hiện chôn chân sau 2-3 nhịp để fail-fast an toàn. """ from __future__ import annotations import math import time from dataclasses import dataclass from typing import Any, Callable, Optional, Tuple, Union import numpy as np class ImmobileCharacterError(RuntimeError): """Ngoại lệ phát sinh khi nhân vật bị chôn chân / liệt di chuyển hoàn toàn.""" pass @dataclass class MotionSample: """Mẫu đo lường trạng thái chuyển động tại thời điểm t.""" timestamp: float pos_xyz: Optional[Tuple[float, float, float]] = None frame_crop: Optional[np.ndarray] = None @dataclass class MotionResult: """Kết quả phân tích vi sai chuyển động sau một nhịp điều khiển WASD.""" dwell_sec: float delta_xyz: float velocity_xyz: float delta_optical_px: float is_moving: bool is_stalled: bool stall_count: int diagnostic_message: str class MotionFeedbackSentinel: """Cảm biến giám sát chuyển động thời gian thực cho Hideout & Combat Navigation.""" def __init__( self, min_move_threshold_u: float = 1.0, min_optical_shift_px: float = 3.0, max_consecutive_stalls: int = 3, expected_min_velocity_u_s: float = 3.0, log_callback: Optional[Callable[[str], None]] = None, ) -> None: self.min_move_threshold_u = min_move_threshold_u self.min_optical_shift_px = min_optical_shift_px self.max_consecutive_stalls = max_consecutive_stalls self.expected_min_velocity_u_s = expected_min_velocity_u_s self.log_callback = log_callback or (lambda msg: None) self._consecutive_stalls: int = 0 self._current_sample: Optional[MotionSample] = None @property def consecutive_stalls(self) -> int: return self._consecutive_stalls def reset(self) -> None: """Đặt lại bộ đếm kẹt chuyển động.""" self._consecutive_stalls = 0 self._current_sample = None def sample_pre_move( self, pos_xyz: Optional[Tuple[float, float, float]] = None, frame_crop: Optional[np.ndarray] = None, ) -> MotionSample: """Lấy mẫu trạng thái ngay TRƯỚC KHI gửi tín hiệu WASD.""" self._current_sample = MotionSample( timestamp=time.perf_counter(), pos_xyz=pos_xyz, frame_crop=frame_crop, ) return self._current_sample def sample_post_move( self, pos_xyz: Optional[Tuple[float, float, float]] = None, frame_crop: Optional[np.ndarray] = None, enforce_invariant: bool = True, ) -> MotionResult: """Lấy mẫu trạng thái ngay SAU KHI nhả phím WASD và đánh giá vi sai chuyển động.""" if self._current_sample is None: raise RuntimeError("sample_post_move gọi khi chưa lấy mẫu sample_pre_move!") t_now = time.perf_counter() dwell_sec = max(0.001, t_now - self._current_sample.timestamp) delta_xyz = 0.0 has_ram_data = False # 1. Đo lường Tier 1: Vi sai bộ nhớ RAM (0ms) if self._current_sample.pos_xyz is not None and pos_xyz is not None: p0 = self._current_sample.pos_xyz p1 = pos_xyz dx = p1[0] - p0[0] dy = p1[1] - p0[1] dz = p1[2] - p0[2] if len(p1) > 2 and len(p0) > 2 else 0.0 delta_xyz = math.sqrt(dx * dx + dy * dy + dz * dz) has_ram_data = True velocity_xyz = delta_xyz / dwell_sec if has_ram_data else 0.0 # 2. Đo lường Tier 2: Vi sai quang học Optical Shift delta_optical_px = 0.0 has_optical_data = False if self._current_sample.frame_crop is not None and frame_crop is not None: delta_optical_px = self._compute_optical_shift( self._current_sample.frame_crop, frame_crop ) has_optical_data = True # 3. Tổng hợp phán quyết di chuyển (Movement Verdict) is_moving = False diagnostic = "" if has_ram_data: # Nếu có RAM: Ưu tiên tuyệt đối chỉ số toán học RAM if delta_xyz >= self.min_move_threshold_u: is_moving = True diagnostic = f"RAM_CONFIRMED (Delta: {delta_xyz:.2f}u, Vel: {velocity_xyz:.1f}u/s)" else: is_moving = False diagnostic = f"RAM_STALLED (Delta: {delta_xyz:.2f}u < {self.min_move_threshold_u}u, Dwell: {dwell_sec:.2f}s)" elif has_optical_data: # Fallback khi mất RAM: Đo lường qua thị giác máy tính if delta_optical_px >= self.min_optical_shift_px: is_moving = True diagnostic = f"OPTICAL_CONFIRMED (Pixel Shift: {delta_optical_px:.1f}px)" else: is_moving = False diagnostic = f"OPTICAL_STALLED (Pixel Shift: {delta_optical_px:.1f}px < {self.min_optical_shift_px}px)" else: # Không có cả 2 cảm biến: Không thể thẩm định diagnostic = "SENSOR_UNAVAILABLE" is_moving = True # Không chặn mù nếu cảm biến chưa sẵn sàng # 4. Quản lý trạng thái kẹt (Stall Counter) if not is_moving and (has_ram_data or has_optical_data): self._consecutive_stalls += 1 is_stalled = True msg = ( f"⚠️ [INV-MOTION-IMMOBILE] Phát hiện nhân vật đứng yên/kẹt! " f"Lần {self._consecutive_stalls}/{self.max_consecutive_stalls} | {diagnostic}" ) self.log_callback(msg) else: self._consecutive_stalls = 0 is_stalled = False res = MotionResult( dwell_sec=dwell_sec, delta_xyz=delta_xyz, velocity_xyz=velocity_xyz, delta_optical_px=delta_optical_px, is_moving=is_moving, is_stalled=is_stalled, stall_count=self._consecutive_stalls, diagnostic_message=diagnostic, ) # 5. Khóa chặn bất biến (Invariant Enforcement) if enforce_invariant and self._consecutive_stalls >= self.max_consecutive_stalls: err_msg = ( f"🚨 [INV-MOTION-IMMOBILE-FATAL] Nhân vật bị chôn chân hoàn toàn sau " f"{self._consecutive_stalls} nhịp bấm WASD! Chi tiết: {diagnostic}. " f"Ngắt khẩn cấp để ngăn chặn báo cáo ảo và lãng phí thời gian OCR!" ) self.log_callback(err_msg) raise ImmobileCharacterError(err_msg) # Chuẩn bị cho chu kỳ tiếp theo self._current_sample = None return res def _compute_optical_shift(self, img_a: np.ndarray, img_b: np.ndarray) -> float: """Tính toán độ trôi điểm ảnh (Pixel Shift) giữa 2 frame.""" if img_a.shape != img_b.shape: return 0.0 # Chuyển sang ảnh xám nếu là ảnh 3 kênh if img_a.ndim == 3: gray_a = np.mean(img_a, axis=2).astype(np.float32) gray_b = np.mean(img_b, axis=2).astype(np.float32) else: gray_a = img_a.astype(np.float32) gray_b = img_b.astype(np.float32) # Phase correlation hoặc Absolute Difference vi sai diff = np.abs(gray_a - gray_b) mean_diff = float(np.mean(diff)) # Chuẩn hóa ước lượng pixel shift dựa trên gradient vi sai grad_y, grad_x = np.gradient(gray_a) avg_grad = float(np.mean(np.abs(grad_x) + np.abs(grad_y))) + 1e-5 estimated_shift = mean_diff / avg_grad return float(min(100.0, max(0.0, estimated_shift)))