hookvision-rk/tests/test_suppression.py

import time
import math

import pytest

from hookvision.models import HookState, Config, BucketModel
from hookvision.main_window_v2 import MainWindow


def _entry(t, hx, hy, lx, ly):
    dist = math.hypot(hx - lx, hy - ly)
    return {"t": t, "hook_cx": hx, "hook_cy": hy, "lug_cx": lx, "lug_cy": ly, "dist": dist}


class TestSuppressWindow:
    def _feed(self, mw, side, points, step=0.1):
        t0 = time.time()
        mw._pair_traj[side] = [_entry(t0 + i * step, *p) for i, p in enumerate(points)]
        mw._update_suppress_window()

    def test_releasing_suppresses(self, qapp):
        mw = MainWindow(Config())
        pts = [(50, 100 + i * 2, 60, 100 + i * 2) for i in range(6)]
        self._feed(mw, "left", pts)
        assert mw._trace_phase == "放包"
        assert mw._is_voice_suppressed() is True

    def test_separation_diagonal_suppresses(self, qapp):
        mw = MainWindow(Config())
        # 挂钩向左上远离吊耳（任意方向相离），距离由 0 增大到 ~29
        pts = [(60 - i * 5, 100 - i * 3, 60, 100) for i in range(6)]
        self._feed(mw, "left", pts)
        assert mw._trace_phase == "分离"
        assert mw._is_voice_suppressed() is True

    def test_separation_rightward_suppresses(self, qapp):
        mw = MainWindow(Config())
        # 挂钩向右远离吊耳，距离由 0 增大到 25
        pts = [(60 + i * 5, 100, 60, 100) for i in range(6)]
        self._feed(mw, "right", pts)
        assert mw._trace_phase == "分离"
        assert mw._is_voice_suppressed() is True

    def test_genuine_hooking_not_suppressed(self, qapp):
        mw = MainWindow(Config())
        # 真挂包：挂钩下移去咬合，吊耳静止，距离缩小
        pts = [(50, 80 + i * 2, 60, 100) for i in range(6)]
        self._feed(mw, "left", pts)
        assert mw._trace_phase == ""
        assert mw._is_voice_suppressed() is False

    def test_lift_together_not_suppressed(self, qapp):
        mw = MainWindow(Config())
        # 挂包成功后吊起：挂钩+吊耳同步上移，距离保持贴合
        pts = [(50, 120 - i * 2, 60, 120 - i * 2) for i in range(6)]
        self._feed(mw, "left", pts)
        assert mw._trace_phase == ""
        assert mw._is_voice_suppressed() is False

    def test_jitter_not_suppressed(self, qapp):
        mw = MainWindow(Config())
        pts = []
        for i in range(6):
            y = 100 + (i % 3)
            pts.append((50, y, 60, y + 1))
        self._feed(mw, "left", pts)
        assert mw._trace_phase == ""
        assert mw._is_voice_suppressed() is False

    def test_no_pair_not_suppressed(self, qapp):
        mw = MainWindow(Config())
        mw._pair_traj = {"left": [], "right": []}
        mw._update_suppress_window()
        assert mw._trace_phase == ""
        assert mw._is_voice_suppressed() is False

    def test_far_hook_moving_not_suppressed(self, qapp):
        mw = MainWindow(Config())
        # 挂钩远离吊耳（从未贴合），即使距离增大也不应命中分离
        pts = [(200 + i * 5, 100, 60, 100) for i in range(6)]
        self._feed(mw, "left", pts)
        assert mw._trace_phase == ""
        assert mw._is_voice_suppressed() is False


class TestVoiceGate:
    def test_voice_suppressed_when_in_window(self, qapp):
        mw = MainWindow(Config())
        pts = [(50, 100 + i * 2, 60, 100 + i * 2) for i in range(6)]
        t0 = time.time()
        mw._pair_traj["left"] = [_entry(t0 + i * 0.1, *p) for i, p in enumerate(pts)]
        mw._update_suppress_window()
        mw.last_voice_state = None
        mw.update_ui_for_state(HookState.READY)
        assert mw.last_voice_state is None

    def test_voice_plays_when_not_suppressed(self, qapp):
        mw = MainWindow(Config())
        mw._pair_traj = {"left": [], "right": []}
        mw._voice_suppress_until = 0.0
        mw.last_voice_state = None
        mw.update_ui_for_state(HookState.READY)
        assert mw.last_voice_state == HookState.READY


class TestStaleOverlayClear:
    def test_cleared_after_timeout(self, qapp):
        mw = MainWindow(Config())
        mw._last_mqtt_process = time.time() - 10
        mw._overlay_cleared = False
        mw._last_left_draw = [{"x": 0}]
        mw._last_right_draw = [{"x": 0}]
        mw._maybe_clear_stale_overlay()
        assert mw._last_left_draw == []
        assert mw._last_right_draw == []
        assert mw._overlay_cleared is True
        assert mw._trace_phase == ""
        assert mw._voice_suppress_until == 0.0

    def test_not_cleared_within_timeout(self, qapp):
        mw = MainWindow(Config())
        mw._last_mqtt_process = time.time()
        mw._overlay_cleared = False
        mw._last_left_draw = [{"x": 0}]
        mw._maybe_clear_stale_overlay()
        assert mw._last_left_draw == [{"x": 0}]
        assert mw._overlay_cleared is False

    def test_not_cleared_before_first_message(self, qapp):
        mw = MainWindow(Config())
        mw._last_mqtt_process = 0.0
        mw._overlay_cleared = False
        mw._maybe_clear_stale_overlay()
        assert mw._last_mqtt_process == 0.0

    def test_cleared_once_until_new_message(self, qapp):
        mw = MainWindow(Config())
        mw._last_mqtt_process = time.time() - 10
        mw._overlay_cleared = False
        mw._maybe_clear_stale_overlay()
        assert mw._overlay_cleared is True
        mw._maybe_clear_stale_overlay()
        assert mw._last_left_draw == []

    def test_disabled_ai_no_clear(self, qapp):
        mw = MainWindow(Config())
        mw.ai_enabled = False
        mw._last_mqtt_process = time.time() - 10
        mw._overlay_cleared = False
        mw._last_left_draw = [{"x": 0}]
        mw._maybe_clear_stale_overlay()
        assert mw._last_left_draw == [{"x": 0}]


def _obj(label, x, y, w=20, h=20):
    return {"x": x, "y": y, "w": w, "h": h, "label": label}


def _calibrated_bucket(x_off=0.0, y_off=0.0):
    b = BucketModel(id=1, description="test", code_x="X")
    b.hook_lug_x_offset_ratio_left = x_off
    b.hook_lug_y_offset_ratio_left = y_off
    b.hook_lug_x_offset_ratio_right = x_off
    b.hook_lug_y_offset_ratio_right = y_off
    return b


def _matched_pair():
    """标定偏移为 0：hook/lug 中心对齐且框相交 → 可以起吊"""
    return _obj("hook", 40, 40, 20, 20), _obj("lug", 40, 40, 20, 20)


def _misaligned_pair():
    """中心偏开超出容差（xr/yr = -0.5）但框仍相交 → 挂包中"""
    return _obj("hook", 30, 30, 20, 20), _obj("lug", 40, 40, 20, 20)


def _fresh(mw):
    mw.hook_state = HookState.IDLE
    mw.prev_state = HookState.IDLE
    mw.state_counter = 0
    mw._pair_traj = {"left": [], "right": []}
    mw._voice_suppress_until = 0.0
    mw.prev_left_hook = None
    mw.prev_left_lug = None
    mw.prev_right_hook = None
    mw.prev_right_lug = None


class TestSideMatch:
    def test_uncalibrated_side_false(self, qapp):
        mw = MainWindow(Config())
        h, l = _matched_pair()
        assert mw._side_match("left", (h, l)) is False

    def test_calibrated_match_true(self, qapp):
        mw = MainWindow(Config())
        mw.active_bucket = _calibrated_bucket()
        h, l = _matched_pair()
        assert mw._side_match("left", (h, l)) is True

    def test_out_of_tolerance_false(self, qapp):
        mw = MainWindow(Config())
        mw.active_bucket = _calibrated_bucket()
        h, l = _misaligned_pair()
        assert mw._side_match("left", (h, l)) is False

    def test_non_intersecting_false(self, qapp):
        mw = MainWindow(Config())
        mw.active_bucket = _calibrated_bucket()
        h = _obj("hook", 0, 0, 20, 20)
        l = _obj("lug", 100, 100, 20, 20)
        assert mw._side_match("left", (h, l)) is False

    def test_none_pair_false(self, qapp):
        mw = MainWindow(Config())
        mw.active_bucket = _calibrated_bucket()
        assert mw._side_match("left", None) is False

    def test_y_uses_bottom_edge_match_true(self, qapp):
        # 底边对齐（yr=0）但中心偏开（中心 yr 会是 -0.5）：只有底边语义下才判定可以起吊
        mw = MainWindow(Config())
        mw.active_bucket = _calibrated_bucket(x_off=-0.5, y_off=0.0)
        h = _obj("hook", 30, 30, 20, 40)
        l = _obj("lug", 40, 50, 20, 20)
        assert mw._side_match("left", (h, l)) is True

    def test_lug_touches_top_side_match_false(self, qapp):
        # 吊耳贴着画面上沿（y=0）：不再判定可以起吊
        mw = MainWindow(Config())
        mw.active_bucket = _calibrated_bucket()
        h = _obj("hook", 40, 0, 20, 20)
        l = _obj("lug", 40, 0, 20, 20)
        assert mw._side_match("left", (h, l)) is False


class TestCanLiftStateMachine:
    def test_cold_start_stays_idle(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        for _ in range(5):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.IDLE

    def test_uncalibrated_never_ready(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        mw.active_bucket = BucketModel(id=1, description="test", code_x="X")
        mw.left_objects = list(_matched_pair())
        for _ in range(6):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.HOOKING

    def test_calibrated_match_ready(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        mw.active_bucket = _calibrated_bucket()
        mw.left_objects = list(_matched_pair())
        for _ in range(3):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.READY

    def test_misaligned_hooking_not_ready(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        mw.active_bucket = _calibrated_bucket()
        mw.left_objects = list(_misaligned_pair())
        for _ in range(6):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.HOOKING

    def test_no_pair_idle(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        mw.active_bucket = _calibrated_bucket()
        for _ in range(5):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.IDLE

    def test_lug_at_top_clears_display_and_skips_judgment(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        mw.active_bucket = _calibrated_bucket()
        mw.left_objects = list(_matched_pair())
        for _ in range(3):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.READY
        # 吊耳贴着画面上沿 → 不再判定，状态框清空（文字空、边框恢复默认灰），状态重置
        mw.left_objects = [_obj("hook", 40, 0, 20, 20), _obj("lug", 40, 0, 20, 20)]
        mw.evaluate_hook_state()
        assert mw.hook_status_label.text() == ""
        assert "#dcdfe6" in mw.left_label.styleSheet()
        assert mw.hook_state == HookState.IDLE
        # 吊耳回到画面内 → 恢复判定
        mw.left_objects = list(_matched_pair())
        for _ in range(3):
            mw.evaluate_hook_state()
        assert mw.hook_status_label.text() == "挂钩状态: 挂包成功"

    def test_non_intersecting_not_ready(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        mw.active_bucket = _calibrated_bucket()
        mw.left_objects = [_obj("hook", 0, 0, 20, 20), _obj("lug", 100, 100, 20, 20)]
        for _ in range(5):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.IDLE

    def test_releasing_still_detected(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        mw.active_bucket = _calibrated_bucket()
        mw.left_objects = list(_matched_pair())
        for _ in range(3):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.READY
        # 放包：hook/lug 同步下移超过阈值
        for step in range(1, 4):
            mw.left_objects = [
                _obj("hook", 40, 40 + step * 6, 20, 20),
                _obj("lug", 40, 40 + step * 6, 20, 20),
            ]
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.RELEASING


class TestReadyStickyAntiFlicker:
    """READY 进出用 ready_debounce 防抖：短时失配不闪红绿"""

    def _engage(self, mw):
        mw.active_bucket = _calibrated_bucket()
        mw.left_objects = list(_matched_pair())
        for _ in range(3):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.READY

    def test_ready_holds_through_short_flicker(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        self._engage(mw)
        # 短暂 2 帧失配（框仍相交）→ 仍保持 READY
        for _ in range(2):
            mw.left_objects = list(_misaligned_pair())
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.READY

    def test_ready_holds_while_matched(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        self._engage(mw)
        for _ in range(6):
            mw.left_objects = list(_matched_pair())
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.READY

    def test_ready_releases_to_hooking_after_persistent_misalign(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        self._engage(mw)
        for _ in range(3):
            mw.left_objects = list(_misaligned_pair())
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.HOOKING

    def test_ready_releases_to_idle_when_detached(self, qapp):
        mw = MainWindow(Config())
        _fresh(mw)
        self._engage(mw)
        mw.left_objects = []
        for _ in range(3):
            mw.evaluate_hook_state()
        assert mw.hook_state == HookState.IDLE