tests/test_automation.py

from __future__ import annotations

import json

import pytest
from pydantic import ValidationError

from tiny_trigger.automation import RuleEngine, document_labels, evaluate_video_detections, load_automation_text, rule_labels
from tiny_trigger.models import Detection, FrameSample


def detection(
    label: str,
    box: tuple[float, float, float, float],
    frame: int = 0,
    time: float = 0.0,
    track_id: int | None = None,
) -> Detection:
    return Detection(
        frame_index=frame,
        timestamp_sec=time,
        label=label,
        confidence=0.9,
        bbox_xyxy=(0.0, 0.0, 10.0, 10.0),
        bbox_xyxy_norm=box,
        track_id=track_id,
    )


def test_presence_near_and_cooldown_fire_once_per_window() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "feed-cat",
                        "when": {
                            "all": [
                                {"present": {"label": "cat", "min_count": 1}},
                                {"near": {"a": "cat", "b": "feeder robot", "max_distance": 0.35}},
                                {"cooldown": {"key": "feed-cat", "seconds": 60}},
                            ]
                        },
                        "trigger": {"on": "while"},
                        "then": [{"type": "simulate", "name": "feed cat"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)
    frame = [
        detection("cat", (0.10, 0.10, 0.20, 0.20), time=0.0),
        detection("feeder robot", (0.25, 0.12, 0.35, 0.22), time=0.0),
    ]

    assert len(engine.evaluate_frame(frame, frame_index=0, timestamp_sec=0.0)) == 1
    assert engine.evaluate_frame(frame, frame_index=1, timestamp_sec=30.0) == []
    assert len(engine.evaluate_frame(frame, frame_index=2, timestamp_sec=61.0)) == 1


def test_count_condition() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "two-packages",
                        "when": {"all": [{"count": {"label": "package", "min": 2}}]},
                        "then": [{"type": "simulate", "name": "notify"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)
    one = [detection("package", (0.1, 0.1, 0.2, 0.2))]
    two = [*one, detection("package", (0.4, 0.4, 0.5, 0.5))]

    assert engine.evaluate_frame(one, frame_index=0, timestamp_sec=0.0) == []
    assert len(engine.evaluate_frame(two, frame_index=1, timestamp_sec=1.0)) == 1


def test_default_trigger_is_enter_not_every_frame() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "guitar-amplifier-on",
                        "when": {"all": [{"present": {"label": "guitar"}}]},
                        "then": [{"type": "simulate", "name": "turn on amplifier"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)
    frame = [detection("guitar", (0.1, 0.1, 0.2, 0.2))]

    assert [event.action for event in engine.evaluate_frame(frame, frame_index=0, timestamp_sec=0.0)] == [
        "turn on amplifier"
    ]
    assert engine.evaluate_frame(frame, frame_index=1, timestamp_sec=1.0) == []


def test_rule_labels_include_condition_labels() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "person-near-monitor",
                        "when": {
                            "all": [
                                {"present": {"label": "person"}},
                                {"near": {"a": "person", "b": "monitor"}},
                                {"far": {"a": "cat", "b": "door"}},
                            ]
                        },
                        "then": [{"type": "simulate", "name": "notify"}],
                    }
                ]
            }
        )
    )

    assert rule_labels(document.rules[0]) == ["person", "monitor", "cat", "door"]
    assert document_labels(document) == ["person", "monitor", "cat", "door"]


def test_moving_condition_uses_tracked_centroid_history() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "car-moving",
                        "when": {
                            "all": [
                                {"moving": {"label": "car", "min_displacement_ratio": 0.15, "window_frames": 3}}
                            ]
                        },
                        "then": [{"type": "simulate", "name": "notify"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)

    first = engine.evaluate_frame(
        [detection("car", (0.10, 0.10, 0.20, 0.20), frame=0, track_id=1)],
        frame_index=0,
        timestamp_sec=0.0,
    )
    second = engine.evaluate_frame(
        [detection("car", (0.11, 0.10, 0.21, 0.20), frame=1, time=1.0, track_id=1)],
        frame_index=1,
        timestamp_sec=1.0,
    )
    third = engine.evaluate_frame(
        [detection("car", (0.20, 0.10, 0.30, 0.20), frame=2, time=2.0, track_id=1)],
        frame_index=2,
        timestamp_sec=2.0,
    )

    assert first == []
    assert second == []
    assert [event.action for event in third] == ["notify"]


def test_moving_condition_ignores_box_jitter_on_stationary_object() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "person-moving",
                        "when": {
                            "all": [
                                {"moving": {"label": "person", "min_displacement_ratio": 0.15, "window_frames": 3}}
                            ]
                        },
                        "then": [{"type": "simulate", "name": "notify"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)

    # Stationary person; only the box's top edge flickers (arms/occlusion/shadow),
    # which swings the geometric center but leaves the foot point essentially put.
    # The old center-displacement check fired on this; the foot-point + size-ratio
    # check must not.
    frames = [
        (0.45, 0.50, 0.55, 0.80),
        (0.46, 0.42, 0.56, 0.80),
        (0.44, 0.34, 0.54, 0.80),
    ]
    events = []
    for index, box in enumerate(frames):
        events += engine.evaluate_frame(
            [detection("person", box, frame=index, time=float(index), track_id=1)],
            frame_index=index,
            timestamp_sec=float(index),
        )

    assert events == []


def test_moving_condition_migrates_short_window_to_three() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "legacy-moving",
                        "when": {"all": [{"moving": {"label": "car", "window_frames": 2}}]},
                        "then": [{"type": "simulate", "name": "notify"}],
                    }
                ]
            }
        )
    )

    moving = document.rules[0].when.all_conditions[0].moving
    assert moving is not None
    assert moving.window_frames == 3


def test_moving_condition_does_not_reenter_after_one_missed_frame() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "car-moving",
                        "when": {
                            "all": [
                                {
                                    "moving": {
                                        "label": "car",
                                        "min_displacement_ratio": 0.15,
                                        "window_frames": 3,
                                        "max_missing_frames": 1,
                                    }
                                }
                            ]
                        },
                        "then": [{"type": "simulate", "name": "notify"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)

    assert engine.evaluate_frame(
        [detection("car", (0.10, 0.10, 0.20, 0.20), frame=0, track_id=1)],
        frame_index=0,
        timestamp_sec=0.0,
    ) == []
    assert engine.evaluate_frame(
        [detection("car", (0.14, 0.10, 0.24, 0.20), frame=1, time=1.0, track_id=1)],
        frame_index=1,
        timestamp_sec=1.0,
    ) == []
    first_alert = engine.evaluate_frame(
        [detection("car", (0.18, 0.10, 0.28, 0.20), frame=2, time=2.0, track_id=1)],
        frame_index=2,
        timestamp_sec=2.0,
    )
    missed_frame = engine.evaluate_frame([], frame_index=3, timestamp_sec=3.0)
    same_track_returns = engine.evaluate_frame(
        [detection("car", (0.22, 0.10, 0.32, 0.20), frame=4, time=4.0, track_id=1)],
        frame_index=4,
        timestamp_sec=4.0,
    )

    assert [event.action for event in first_alert] == ["notify"]
    assert missed_frame == []
    assert same_track_returns == []


def test_gate_cooldown_fire_once_per_window() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "turn-on-pc",
                        "when": {
                            "all": [
                                {"present": {"label": "person"}},
                                {"near": {"a": "person", "b": "steering wheel", "max_distance": 0.45}},
                            ]
                        },
                        "trigger": {"on": "while"},
                        "gate": {"enabled": True, "cooldown": {"key": "turn-on-pc", "seconds": 60}},
                        "then": [{"type": "simulate", "name": "turn on pc"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)
    frame = [
        detection("person", (0.10, 0.10, 0.20, 0.20), time=0.0),
        detection("steering wheel", (0.25, 0.12, 0.35, 0.22), time=0.0),
    ]

    assert len(engine.evaluate_frame(frame, frame_index=0, timestamp_sec=0.0)) == 1
    assert engine.evaluate_frame(frame, frame_index=1, timestamp_sec=30.0) == []
    assert len(engine.evaluate_frame(frame, frame_index=2, timestamp_sec=61.0)) == 1


def test_disabled_gate_blocks_rule() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "disabled",
                        "when": {"all": [{"present": {"label": "person"}}]},
                        "gate": {"enabled": False},
                        "then": [{"type": "simulate", "name": "turn on pc"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)

    assert engine.evaluate_frame([detection("person", (0.1, 0.1, 0.2, 0.2))], frame_index=0, timestamp_sec=0.0) == []


def test_evaluate_video_detections_returns_updated_last_fired() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "turn-on-pc",
                        "when": {"all": [{"present": {"label": "person"}}]},
                        "trigger": {"on": "while"},
                        "gate": {"cooldown": {"key": "turn-on-pc", "seconds": 60}},
                        "then": [{"type": "simulate", "name": "turn on pc"}],
                    }
                ]
            }
        )
    )
    events, last_fired = evaluate_video_detections(
        document.rules,
        [detection("person", (0.1, 0.1, 0.2, 0.2), frame=2, time=12.0)],
    )
    blocked_events, unchanged_last_fired = evaluate_video_detections(
        document.rules,
        [detection("person", (0.1, 0.1, 0.2, 0.2), frame=3, time=30.0)],
        last_fired=last_fired,
    )

    assert len(events) == 1
    assert last_fired == {"turn-on-pc": 12.0}
    assert blocked_events == []
    assert unchanged_last_fired == last_fired


def test_video_cooldowns_are_fresh_without_persisted_last_fired() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "turn-on-pc",
                        "when": {"all": [{"present": {"label": "person"}}]},
                        "trigger": {"on": "while"},
                        "gate": {"cooldown": {"key": "turn-on-pc", "seconds": 60}},
                        "then": [{"type": "simulate", "name": "turn on pc"}],
                    }
                ]
            }
        )
    )
    detections = [detection("person", (0.1, 0.1, 0.2, 0.2), frame=2, time=12.0)]

    first_events, first_last_fired = evaluate_video_detections(document.rules, detections)
    second_events, second_last_fired = evaluate_video_detections(document.rules, detections)

    assert len(first_events) == 1
    assert len(second_events) == 1
    assert first_last_fired == second_last_fired == {"turn-on-pc": 12.0}


def test_change_trigger_fires_enter_and_exit() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "monitor-lights",
                        "when": {
                            "all": [
                                {"near": {"a": "person", "b": "monitor", "max_gap_percent": 8}},
                            ]
                        },
                        "trigger": {"on": "change"},
                        "then": {
                            "enter": [{"type": "simulate", "name": "turn on lights"}],
                            "exit": [{"type": "simulate", "name": "turn off lights"}],
                        },
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)
    matching = [
        detection("person", (0.10, 0.10, 0.20, 0.20)),
        detection("monitor", (0.23, 0.10, 0.33, 0.20)),
    ]

    enter_events = engine.evaluate_frame(matching, frame_index=0, timestamp_sec=0.0)
    while_events = engine.evaluate_frame(matching, frame_index=1, timestamp_sec=1.0)
    exit_events = engine.evaluate_frame([], frame_index=2, timestamp_sec=2.0)

    assert [event.action for event in enter_events] == ["turn on lights"]
    assert while_events == []
    assert [event.action for event in exit_events] == ["turn off lights"]


def test_exit_trigger_accepts_plain_then_actions() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "person-leaves-turn-off-lights",
                        "when": {"all": [{"present": {"label": "person", "min_count": 1}}]},
                        "trigger": {"on": "exit"},
                        "gate": {"enabled": True},
                        "then": [{"type": "simulate", "name": "turn off lights"}],
                    }
                ]
            }
        )
    )
    engine = RuleEngine(document.rules)

    first_events = engine.evaluate_frame(
        [detection("person", (0.1, 0.1, 0.2, 0.2))],
        frame_index=0,
        timestamp_sec=0.0,
    )
    exit_events = engine.evaluate_frame([], frame_index=1, timestamp_sec=1.0)

    assert first_events == []
    assert [event.action for event in exit_events] == ["turn off lights"]


def test_video_evaluation_uses_empty_frames_for_exit_triggers() -> None:
    document = load_automation_text(
        json.dumps(
            {
                "rules": [
                    {
                        "name": "person-presence",
                        "when": {"all": [{"present": {"label": "person"}}]},
                        "trigger": {"on": "change"},
                        "then": {
                            "enter": [{"type": "simulate", "name": "arrived"}],
                            "exit": [{"type": "simulate", "name": "left"}],
                        },
                    }
                ]
            }
        )
    )

    events, _last_fired = evaluate_video_detections(
        document.rules,
        [detection("person", (0.1, 0.1, 0.2, 0.2), frame=0, time=0.0)],
        frames=[
            FrameSample(frame_index=0, timestamp_sec=0.0),
            FrameSample(frame_index=1, timestamp_sec=1.0),
        ],
    )

    assert [event.action for event in events] == ["arrived", "left"]


def test_invalid_rule_rejected() -> None:
    with pytest.raises(ValidationError):
        load_automation_text(
            json.dumps(
                {
                    "rules": [
                        {
                            "name": "bad",
                            "when": {"all": [{"present": {"label": "cat"}, "near": {"a": "cat", "b": "bowl"}}]},
                            "then": [{"type": "simulate", "name": "bad"}],
                        }
                    ]
                }
            )
        )