"""The API on a fake 100 ms camera: parallel requests are not serialized, the contract is unchanged."""

import asyncio
import statistics
import time

import cv2
import numpy as np
import pytest

pytest.importorskip("pyrealsense2")

from httpx import ASGITransport, AsyncClient  # noqa: E402
from starlette.testclient import TestClient  # noqa: E402

from src.backend.api import routes, websocket  # noqa: E402
from src.backend.main import app  # noqa: E402
from src.backend.services.camera import RealSenseD435  # noqa: E402
from tests.fake_realsense import FakePipeline, open_fake_camera  # noqa: E402

# Fields of /api/capture before optimize-streams R01 (the multi-camera service parses them).
CAPTURE_FIELDS = {
    "frame_id", "timestamp", "rgb_shape", "depth_shape", "depth_min_mm", "depth_max_mm",
    "depth_valid_pixels", "rgb_jpeg_b64", "depth_png_b64", "depth_color_b64",
}


@pytest.fixture
def fake_camera():
    cam = RealSenseD435()
    pipe = FakePipeline()
    open_fake_camera(cam, pipe)
    routes.set_camera(cam)       # the lifespan (not run in tests) injects the camera into both
    websocket.set_camera(cam)
    yield cam, pipe
    cam.close()
    routes.set_camera(None)
    websocket.set_camera(None)


async def _wall_time(client, n: int) -> tuple[float, list]:
    t0 = time.perf_counter()
    responses = await asyncio.gather(*(client.get("/api/capture") for _ in range(n)))
    return time.perf_counter() - t0, [r.json() for r in responses]


def test_three_parallel_captures_take_less_than_twice_one(fake_camera):
    """The criterion declared for R01: t(3 parallel) < 2 x t(1 alone), medians of 5."""
    async def run():
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as client:
            await _wall_time(client, 1)
            one = [(await _wall_time(client, 1))[0] for _ in range(5)]
            three = [(await _wall_time(client, 3))[0] for _ in range(5)]
            return statistics.median(one), statistics.median(three)

    one, three = asyncio.run(run())
    assert three < 2 * one, f"1 alone {one*1000:.0f} ms, 3 parallel {three*1000:.0f} ms"


def test_capture_fields_unchanged_and_one_frame_shared(fake_camera):
    _, pipe = fake_camera

    async def run():
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as client:
            return (await _wall_time(client, 3))[1]

    bodies = asyncio.run(run())
    assert all(b["success"] for b in bodies)
    data = [b["data"] for b in bodies]
    assert all(set(d) == CAPTURE_FIELDS for d in data)
    assert len({d["frame_id"] for d in data}) == 3            # frame_id still counts responses
    assert len({d["timestamp"] for d in data}) == 1           # ... all three served by one frame
    assert len({d["rgb_jpeg_b64"] for d in data}) == 1        # ... with the same bytes
    assert pipe.max_concurrent == 1                           # only one thread calls the pipeline


def test_rgb_and_depth_endpoints_decode(fake_camera):
    client = TestClient(app)
    rgb = cv2.imdecode(np.frombuffer(client.get("/api/rgb").content, np.uint8), cv2.IMREAD_COLOR)
    depth = cv2.imdecode(np.frombuffer(client.get("/api/depth").content, np.uint8), cv2.IMREAD_UNCHANGED)
    color = cv2.imdecode(np.frombuffer(client.get("/api/depth/color").content, np.uint8), cv2.IMREAD_COLOR)
    assert rgb.shape == (480, 640, 3)
    assert depth.shape == (480, 640) and depth.dtype == np.uint16
    assert color.shape == (480, 640, 3)


def test_websocket_streams_frames(fake_camera):
    with TestClient(app).websocket_connect("/ws") as ws:
        ws.send_text("start")
        first, second = ws.receive_json(), ws.receive_json()
        ws.send_text("close")
    assert set(first) == CAPTURE_FIELDS
    assert second["timestamp"] > first["timestamp"]


def test_close_stops_the_acquisition_thread():
    cam = RealSenseD435()
    pipe = FakePipeline()
    open_fake_camera(cam, pipe)
    assert cam.capture() is not None
    acquisition = cam._acquisition
    cam.close()
    assert not cam.is_open()
    assert pipe.stopped
    assert not acquisition._thread.is_alive()
    assert cam.capture() is None