Sync from arm/Optimized-Runtimes via hub-sync

9422383 verified 6 days ago

9.92 kB

"""End-to-end public video and Picamera2 example behavior on Linux AArch64.
The real detector, native runtime, rendering code, and OpenCV are exercised.
Only the camera device API is replaced because physical capture is the external
boundary of these deterministic tests.
"""

from __future__ import annotations

import hashlib
import importlib
import io
import json
import sys
import types
import unittest
from contextlib import AbstractContextManager, redirect_stdout
from pathlib import Path
from tempfile import TemporaryDirectory
from types import SimpleNamespace
from unittest.mock import patch

import cv2
import numpy as np
from tests.support import configured_model, configured_runtime, run_module


class _Request:
    def __init__(self, image: np.ndarray) -> None:
        self.image = image
        self.released = False

    def release(self) -> None:
        self.released = True


class _MappedArray(AbstractContextManager["_MappedArray"]):
    def __init__(self, request: _Request, _stream: str) -> None:
        self.array = request.image

    def __enter__(self) -> _MappedArray:
        return self

    def __exit__(self, *_args: object) -> None:
        return None


class _Camera:
    instances: list[_Camera] = []

    def __init__(self, _index: int) -> None:
        self.configuration: dict[str, object] = {}
        self.closed = False
        self.post_callback = None
        self.requests: list[_Request] = []
        self.preview_frames: list[np.ndarray] = []
        self.texture_overlays: list[np.ndarray] = []
        self.instances.append(self)

    def create_video_configuration(self, **configuration: object) -> dict[str, object]:
        return configuration

    def configure(self, configuration: dict[str, object]) -> None:
        self.configuration = configuration

    def _shape(self, channels: int) -> tuple[int, int, int]:
        main = self.configuration["main"]
        assert isinstance(main, dict)
        width, height = main["size"]
        return height, width, channels

    def capture_request(self, *, wait: bool) -> object:
        assert not wait
        return "request"

    def capture_array(self, _stream: str, *, wait: bool) -> object:
        assert not wait
        return "array"

    def wait(self, job: object) -> _Request | np.ndarray:
        if job == "request":
            request = _Request(np.zeros(self._shape(3), dtype=np.uint8))
            self.requests.append(request)
            return request
        request = _Request(np.zeros(self._shape(4), dtype=np.uint8))
        # Picamera2 draws its preview callback on the completed camera request
        # before returning a copied array to the detection loop.
        if self.post_callback is not None:
            self.post_callback(request)
            self.preview_frames.append(request.image.copy())
        return request.image.copy()

    def start(self) -> None:
        pass

    def stop(self) -> None:
        pass

    def close(self) -> None:
        self.closed = True

    def start_preview(self, *_args: object, **_kwargs: object) -> None:
        pass

    def stop_preview(self) -> None:
        pass

    def set_overlay(self, overlay: np.ndarray) -> None:
        self.texture_overlays.append(overlay.copy())


def _camera_module() -> types.ModuleType:
    module = types.ModuleType("picamera2")
    module.MappedArray = _MappedArray
    module.Picamera2 = _Camera
    module.Preview = SimpleNamespace(DRM="drm", QTGL="qtgl")
    return module


def _import_camera_example(module: str) -> types.ModuleType:
    sys.modules.pop(module, None)
    previous = sys.modules.get("picamera2")
    sys.modules["picamera2"] = _camera_module()
    try:
        return importlib.import_module(module)
    finally:
        # Restore only the unavailable device module. Replacing the entire
        # module table would remove real dependencies imported by the example.
        if previous is None:
            sys.modules.pop("picamera2", None)
        else:
            sys.modules["picamera2"] = previous


def _write_video(path: Path) -> None:
    writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"MJPG"), 30.0, (64, 48))
    if not writer.isOpened():
        raise RuntimeError("OpenCV cannot create the test video")
    try:
        writer.write(np.zeros((48, 64, 3), dtype=np.uint8))
        writer.write(np.full((48, 64, 3), 32, dtype=np.uint8))
    finally:
        writer.release()


class VideoDetectionBehaviorTests(unittest.TestCase):
    def test_command_processes_real_detector_results(self) -> None:
        with TemporaryDirectory() as directory:
            root = Path(directory)
            video = root / "input.avi"
            output = root / "detections.json"
            _write_video(video)
            completed = run_module(
                "examples.video.detect",
                "--runtime",
                configured_runtime(),
                "--model",
                configured_model(),
                "--video",
                video,
                "--output",
                output,
                "--frames",
                2,
            )
            self.assertEqual(completed.returncode, 0, completed.stderr)
            self.assertIn("Detection performance:", completed.stdout)
            self.assertIn(" FPS", completed.stdout)
            self.assertIn(" ms/frame", completed.stdout)
            document = json.loads(output.read_text(encoding="utf-8"))
            self.assertEqual(document["version"], 1)
            self.assertEqual(document["coordinate_space"], "inference_frame_pixels_xyxy")
            self.assertEqual(document["runtime_library"], str(configured_runtime()))
            self.assertEqual(document["model_directory"], str(configured_model()))
            self.assertEqual(document["source"]["video"], str(video.resolve()))
            self.assertEqual(
                document["source"]["sha256"], hashlib.sha256(video.read_bytes()).hexdigest()
            )
            self.assertEqual((document["source"]["width"], document["source"]["height"]), (64, 48))
            self.assertEqual(document["summary"]["frames"], 2)
            self.assertEqual([frame["frame_index"] for frame in document["frames"]], [0, 1])
            self.assertGreater(document["summary"]["inference_mean_ms"], 0)
            for frame in document["frames"]:
                self.assertGreaterEqual(frame["timestamp_seconds"], 0)
                self.assertGreater(frame["completed_seconds"], 0)
                self.assertLessEqual(
                    {"preprocess_ms", "inference_ms", "postprocess_ms"},
                    set(frame["timings"]),
                )
                self.assertIsInstance(frame["detections"], list)
                for detection in frame["detections"]:
                    self.assertLessEqual(
                        {"box", "confidence", "class_id", "class_name"},
                        set(detection),
                    )


class PicameraExampleBehaviorTests(unittest.TestCase):
    def setUp(self) -> None:
        _Camera.instances.clear()

    def test_headless_example_processes_a_camera_frame_and_closes(self) -> None:
        app = _import_camera_example("examples.picamera2.detect")
        output = io.StringIO()
        with (
            patch.object(
                sys,
                "argv",
                [
                    "detect.py",
                    "--runtime",
                    str(configured_runtime()),
                    "--model",
                    str(configured_model()),
                    "--frames",
                    "1",
                ],
            ),
            redirect_stdout(output),
        ):
            self.assertEqual(app.main(), 0)
        camera = _Camera.instances[0]
        self.assertTrue(camera.closed)
        self.assertEqual(len(camera.requests), 1)
        self.assertTrue(camera.requests[0].released)
        self.assertIn("frames=1", output.getvalue())

    def test_live_example_renders_a_camera_frame_and_closes(self) -> None:
        app = _import_camera_example("examples.picamera2.live")
        output = io.StringIO()
        with (
            patch.object(
                sys,
                "argv",
                [
                    "live.py",
                    "--runtime",
                    str(configured_runtime()),
                    "--model",
                    str(configured_model()),
                    "--preview",
                    "drm",
                    "--frames",
                    "2",
                ],
            ),
            redirect_stdout(output),
        ):
            self.assertEqual(app.main(), 0)
        camera = _Camera.instances[0]
        self.assertTrue(camera.closed)
        self.assertEqual(len(camera.preview_frames), 2)
        self.assertTrue(np.any(camera.preview_frames[-1]))
        self.assertIn("frames=2", output.getvalue())

    def test_live_example_uploads_a_texture_overlay_and_closes(self) -> None:
        app = _import_camera_example("examples.picamera2.live")
        output = io.StringIO()
        with (
            patch.object(
                sys,
                "argv",
                [
                    "live.py",
                    "--runtime",
                    str(configured_runtime()),
                    "--model",
                    str(configured_model()),
                    "--preview",
                    "qtgl",
                    "--frames",
                    "2",
                ],
            ),
            redirect_stdout(output),
        ):
            self.assertEqual(app.main(), 0)
        camera = _Camera.instances[0]
        self.assertTrue(camera.closed)
        self.assertTrue(camera.texture_overlays)
        self.assertTrue(np.any(camera.texture_overlays[-1]))
        self.assertIn("frames=2", output.getvalue())


if __name__ == "__main__":
    unittest.main()

	"""End-to-end public video and Picamera2 example behavior on Linux AArch64.

	The real detector, native runtime, rendering code, and OpenCV are exercised.
	Only the camera device API is replaced because physical capture is the external
	boundary of these deterministic tests.
	"""

	from __future__ import annotations

	import hashlib
	import importlib
	import io
	import json
	import sys
	import types
	import unittest
	from contextlib import AbstractContextManager, redirect_stdout
	from pathlib import Path
	from tempfile import TemporaryDirectory
	from types import SimpleNamespace
	from unittest.mock import patch

	import cv2
	import numpy as np
	from tests.support import configured_model, configured_runtime, run_module


	class _Request:
	def __init__(self, image: np.ndarray) -> None:
	self.image = image
	self.released = False

	def release(self) -> None:
	self.released = True


	class _MappedArray(AbstractContextManager["_MappedArray"]):
	def __init__(self, request: _Request, _stream: str) -> None:
	self.array = request.image

	def __enter__(self) -> _MappedArray:
	return self

	def __exit__(self, *_args: object) -> None:
	return None


	class _Camera:
	instances: list[_Camera] = []

	def __init__(self, _index: int) -> None:
	self.configuration: dict[str, object] = {}
	self.closed = False
	self.post_callback = None
	self.requests: list[_Request] = []
	self.preview_frames: list[np.ndarray] = []
	self.texture_overlays: list[np.ndarray] = []
	self.instances.append(self)

	def create_video_configuration(self, **configuration: object) -> dict[str, object]:
	return configuration

	def configure(self, configuration: dict[str, object]) -> None:
	self.configuration = configuration

	def _shape(self, channels: int) -> tuple[int, int, int]:
	main = self.configuration["main"]
	assert isinstance(main, dict)
	width, height = main["size"]
	return height, width, channels

	def capture_request(self, *, wait: bool) -> object:
	assert not wait
	return "request"

	def capture_array(self, _stream: str, *, wait: bool) -> object:
	assert not wait
	return "array"

	def wait(self, job: object) -> _Request \| np.ndarray:
	if job == "request":
	request = _Request(np.zeros(self._shape(3), dtype=np.uint8))
	self.requests.append(request)
	return request
	request = _Request(np.zeros(self._shape(4), dtype=np.uint8))
	# Picamera2 draws its preview callback on the completed camera request
	# before returning a copied array to the detection loop.
	if self.post_callback is not None:
	self.post_callback(request)
	self.preview_frames.append(request.image.copy())
	return request.image.copy()

	def start(self) -> None:
	pass

	def stop(self) -> None:
	pass

	def close(self) -> None:
	self.closed = True

	def start_preview(self, _args: object, *_kwargs: object) -> None:
	pass

	def stop_preview(self) -> None:
	pass

	def set_overlay(self, overlay: np.ndarray) -> None:
	self.texture_overlays.append(overlay.copy())


	def _camera_module() -> types.ModuleType:
	module = types.ModuleType("picamera2")
	module.MappedArray = _MappedArray
	module.Picamera2 = _Camera
	module.Preview = SimpleNamespace(DRM="drm", QTGL="qtgl")
	return module


	def _import_camera_example(module: str) -> types.ModuleType:
	sys.modules.pop(module, None)
	previous = sys.modules.get("picamera2")
	sys.modules["picamera2"] = _camera_module()
	try:
	return importlib.import_module(module)
	finally:
	# Restore only the unavailable device module. Replacing the entire
	# module table would remove real dependencies imported by the example.
	if previous is None:
	sys.modules.pop("picamera2", None)
	else:
	sys.modules["picamera2"] = previous


	def _write_video(path: Path) -> None:
	writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"MJPG"), 30.0, (64, 48))
	if not writer.isOpened():
	raise RuntimeError("OpenCV cannot create the test video")
	try:
	writer.write(np.zeros((48, 64, 3), dtype=np.uint8))
	writer.write(np.full((48, 64, 3), 32, dtype=np.uint8))
	finally:
	writer.release()


	class VideoDetectionBehaviorTests(unittest.TestCase):
	def test_command_processes_real_detector_results(self) -> None:
	with TemporaryDirectory() as directory:
	root = Path(directory)
	video = root / "input.avi"
	output = root / "detections.json"
	_write_video(video)
	completed = run_module(
	"examples.video.detect",
	"--runtime",
	configured_runtime(),
	"--model",
	configured_model(),
	"--video",
	video,
	"--output",
	output,
	"--frames",
	2,
	)
	self.assertEqual(completed.returncode, 0, completed.stderr)
	self.assertIn("Detection performance:", completed.stdout)
	self.assertIn(" FPS", completed.stdout)
	self.assertIn(" ms/frame", completed.stdout)
	document = json.loads(output.read_text(encoding="utf-8"))
	self.assertEqual(document["version"], 1)
	self.assertEqual(document["coordinate_space"], "inference_frame_pixels_xyxy")
	self.assertEqual(document["runtime_library"], str(configured_runtime()))
	self.assertEqual(document["model_directory"], str(configured_model()))
	self.assertEqual(document["source"]["video"], str(video.resolve()))
	self.assertEqual(
	document["source"]["sha256"], hashlib.sha256(video.read_bytes()).hexdigest()
	)
	self.assertEqual((document["source"]["width"], document["source"]["height"]), (64, 48))
	self.assertEqual(document["summary"]["frames"], 2)
	self.assertEqual([frame["frame_index"] for frame in document["frames"]], [0, 1])
	self.assertGreater(document["summary"]["inference_mean_ms"], 0)
	for frame in document["frames"]:
	self.assertGreaterEqual(frame["timestamp_seconds"], 0)
	self.assertGreater(frame["completed_seconds"], 0)
	self.assertLessEqual(
	{"preprocess_ms", "inference_ms", "postprocess_ms"},
	set(frame["timings"]),
	)
	self.assertIsInstance(frame["detections"], list)
	for detection in frame["detections"]:
	self.assertLessEqual(
	{"box", "confidence", "class_id", "class_name"},
	set(detection),
	)


	class PicameraExampleBehaviorTests(unittest.TestCase):
	def setUp(self) -> None:
	_Camera.instances.clear()

	def test_headless_example_processes_a_camera_frame_and_closes(self) -> None:
	app = _import_camera_example("examples.picamera2.detect")
	output = io.StringIO()
	with (
	patch.object(
	sys,
	"argv",
	[
	"detect.py",
	"--runtime",
	str(configured_runtime()),
	"--model",
	str(configured_model()),
	"--frames",
	"1",
	],
	),
	redirect_stdout(output),
	):
	self.assertEqual(app.main(), 0)
	camera = _Camera.instances[0]
	self.assertTrue(camera.closed)
	self.assertEqual(len(camera.requests), 1)
	self.assertTrue(camera.requests[0].released)
	self.assertIn("frames=1", output.getvalue())

	def test_live_example_renders_a_camera_frame_and_closes(self) -> None:
	app = _import_camera_example("examples.picamera2.live")
	output = io.StringIO()
	with (
	patch.object(
	sys,
	"argv",
	[
	"live.py",
	"--runtime",
	str(configured_runtime()),
	"--model",
	str(configured_model()),
	"--preview",
	"drm",
	"--frames",
	"2",
	],
	),
	redirect_stdout(output),
	):
	self.assertEqual(app.main(), 0)
	camera = _Camera.instances[0]
	self.assertTrue(camera.closed)
	self.assertEqual(len(camera.preview_frames), 2)
	self.assertTrue(np.any(camera.preview_frames[-1]))
	self.assertIn("frames=2", output.getvalue())

	def test_live_example_uploads_a_texture_overlay_and_closes(self) -> None:
	app = _import_camera_example("examples.picamera2.live")
	output = io.StringIO()
	with (
	patch.object(
	sys,
	"argv",
	[
	"live.py",
	"--runtime",
	str(configured_runtime()),
	"--model",
	str(configured_model()),
	"--preview",
	"qtgl",
	"--frames",
	"2",
	],
	),
	redirect_stdout(output),
	):
	self.assertEqual(app.main(), 0)
	camera = _Camera.instances[0]
	self.assertTrue(camera.closed)
	self.assertTrue(camera.texture_overlays)
	self.assertTrue(np.any(camera.texture_overlays[-1]))
	self.assertIn("frames=2", output.getvalue())


	if __name__ == "__main__":
	unittest.main()