mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-09-28 10:56:52 +03:00
Nick RogersandGitHub
c959df32c9
* Run ONNX models on a Mac's Neural Engine through lighter's plugin provider lighter's lighter.sh/ane device places an ONNX Runtime plugin execution provider in the container. When it is present, the ONNX session setup registers it once and opens sessions on its Neural Engine device, the same place CUDA, ROCm and OpenVINO are chosen, so the onnx detector (and any model that is not pinned to the CPU) runs there with no configuration. The hardware probe reports it as an onnx unit. * docs: hardware decode on an Apple Silicon Mac under lighter A community section on the video decoding page: lighter's lighter.sh/video device, hwaccel_args -c:v h264_v4l2m2m, and why the Raspberry Pi presets decode a single-stream camera in software. The detector docs link to it. * docs: set the lighter decoder per camera when codecs are mixed * docs: the ONNX detector on a Mac's Neural Engine under lighter * Format the Neural Engine provider setup * Fall back to the default providers when the Neural Engine cannot load a model * Apple Silicon ffmpeg presets for lighter's media engine, recommended when it is present
254 lines
9.6 KiB
Python
254 lines
9.6 KiB
Python
"""Tests for the hardware decoding recommendation."""
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from frigate.detectors.hardware import DetectionHardware
|
|
from frigate.util import hwaccel
|
|
|
|
|
|
def found(key: str) -> DetectionHardware:
|
|
"""A probe result carrying only the fields the recommendation reads."""
|
|
return DetectionHardware(
|
|
key=key,
|
|
detector=key.partition(":")[0],
|
|
name=key,
|
|
units=[],
|
|
count=0,
|
|
unlimited=True,
|
|
)
|
|
|
|
|
|
class HwaccelRecommendationTestCase(unittest.TestCase):
|
|
"""Points every read at an empty fixture tree, so nothing is found by default."""
|
|
|
|
def setUp(self):
|
|
self.root = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.root.cleanup)
|
|
|
|
self.proc_root = os.path.join(self.root.name, "proc")
|
|
os.makedirs(self.proc_root)
|
|
patcher = patch.object(hwaccel, "PROC_ROOT", self.proc_root)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
self.sys_root = os.path.join(self.root.name, "sys")
|
|
os.makedirs(self.sys_root)
|
|
patcher = patch.object(hwaccel, "SYS_ROOT", self.sys_root)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
drm = patch.object(hwaccel, "enumerate_drm_devices", return_value={})
|
|
self.drm = drm.start()
|
|
self.addCleanup(drm.stop)
|
|
|
|
def options(self, keys=(), detector_key=None, codecs=None):
|
|
"""Run the recommendation against a fixed set of hardware keys."""
|
|
with patch.object(
|
|
hwaccel.hardware_prober,
|
|
"probe",
|
|
return_value=[found(key) for key in keys],
|
|
):
|
|
return hwaccel.hwaccel_options(detector_key, codecs)
|
|
|
|
def recommend(self, keys=(), detector_key=None, codecs=None) -> str:
|
|
"""The recommended family key."""
|
|
return self.options(keys, detector_key, codecs)[0]
|
|
|
|
def available(self, keys=(), detector_key=None, codecs=None) -> list[str]:
|
|
"""The keys of the usable families, best first."""
|
|
return [family.key for family in self.options(keys, detector_key, codecs)[1]]
|
|
|
|
def presets(self, keys=(), detector_key=None, codecs=None) -> dict:
|
|
"""The presets each usable family provides."""
|
|
return {
|
|
family.key: family.presets
|
|
for family in self.options(keys, detector_key, codecs)[1]
|
|
}
|
|
|
|
def write_cpuinfo(self, model_name: str) -> None:
|
|
with open(os.path.join(self.proc_root, "cpuinfo"), "w") as f:
|
|
f.write(f"processor\t: 0\nmodel name\t: {model_name}\n")
|
|
|
|
def write_video_device(self, vendor: str) -> None:
|
|
device = os.path.join(self.sys_root, "class", "video4linux", "video0", "device")
|
|
os.makedirs(device)
|
|
with open(os.path.join(device, "vendor"), "w") as f:
|
|
f.write(f"{vendor}\n")
|
|
|
|
def write_device_tree(self) -> None:
|
|
os.makedirs(os.path.join(self.proc_root, "device-tree"), exist_ok=True)
|
|
with open(os.path.join(self.proc_root, "device-tree", "compatible"), "w") as f:
|
|
f.write("raspberrypi,5-model-b\x00brcm,bcm2712\x00")
|
|
|
|
|
|
class TestPriority(HwaccelRecommendationTestCase):
|
|
def test_nothing_found_recommends_nothing(self):
|
|
self.assertEqual(self.recommend(), "")
|
|
|
|
def test_nvidia_wins_over_intel(self):
|
|
self.assertEqual(self.recommend(["onnx:nvidia", "openvino:GPU"]), "nvidia")
|
|
|
|
def test_a_jetson_uses_its_own_family(self):
|
|
self.assertEqual(self.recommend(["tensorrt"]), "jetson")
|
|
|
|
def test_a_rockchip_uses_rkmpp(self):
|
|
self.assertEqual(self.recommend(["rknn"]), "rkmpp")
|
|
|
|
def test_an_amd_gpu_uses_vaapi(self):
|
|
self.assertEqual(self.recommend(["onnx:amd"]), "vaapi")
|
|
|
|
|
|
class TestDetectorBias(HwaccelRecommendationTestCase):
|
|
def test_a_chosen_intel_gpu_beats_a_present_nvidia(self):
|
|
self.assertEqual(
|
|
self.recommend(["onnx:nvidia", "openvino:GPU"], "openvino:GPU"), "vaapi"
|
|
)
|
|
|
|
def test_a_chosen_npu_decodes_through_the_igpu(self):
|
|
self.assertEqual(
|
|
self.recommend(["openvino:NPU", "openvino:GPU"], "openvino:NPU"), "vaapi"
|
|
)
|
|
|
|
def test_an_npu_without_an_igpu_falls_through(self):
|
|
self.assertEqual(self.recommend(["openvino:NPU"], "openvino:NPU"), "")
|
|
|
|
def test_a_cpu_choice_still_recommends_the_present_gpu(self):
|
|
self.assertEqual(self.recommend(["cpu", "openvino:GPU"], "cpu"), "vaapi")
|
|
|
|
|
|
class TestIntelGeneration(HwaccelRecommendationTestCase):
|
|
def test_the_xe_driver_prefers_qsv(self):
|
|
self.drm.return_value = {"0000:00:02.0": "xe"}
|
|
self.assertEqual(self.recommend(["openvino:GPU"], codecs={"h264"}), "intel-qsv")
|
|
|
|
def test_gen13_prefers_qsv(self):
|
|
self.write_cpuinfo("13th Gen Intel(R) Core(TM) i5-13500")
|
|
self.assertEqual(self.recommend(["openvino:GPU"], codecs={"h264"}), "intel-qsv")
|
|
|
|
def test_a_core_ultra_prefers_qsv(self):
|
|
self.write_cpuinfo("Intel(R) Core(TM) Ultra 7 155H")
|
|
self.assertEqual(self.recommend(["openvino:GPU"], codecs={"h264"}), "intel-qsv")
|
|
|
|
def test_gen13_prefers_qsv_for_mixed_codecs(self):
|
|
# each camera resolves the family to its own codec
|
|
self.write_cpuinfo("13th Gen Intel(R) Core(TM) i5-13500")
|
|
self.assertEqual(
|
|
self.recommend(["openvino:GPU"], codecs={"h264", "h265"}), "intel-qsv"
|
|
)
|
|
|
|
def test_gen12_prefers_vaapi(self):
|
|
self.write_cpuinfo("12th Gen Intel(R) Core(TM) i5-12400")
|
|
self.assertEqual(self.recommend(["openvino:GPU"], codecs={"h264"}), "vaapi")
|
|
|
|
def test_gen12_still_offers_qsv(self):
|
|
self.write_cpuinfo("12th Gen Intel(R) Core(TM) i5-12400")
|
|
self.assertEqual(self.available(["openvino:GPU"]), ["vaapi", "intel-qsv"])
|
|
|
|
def test_an_older_model_string_prefers_vaapi(self):
|
|
self.write_cpuinfo("Intel(R) Core(TM) i7-8700K CPU @ 3.70GHz")
|
|
self.assertEqual(self.recommend(["openvino:GPU"], codecs={"h264"}), "vaapi")
|
|
|
|
def test_missing_cpuinfo_prefers_vaapi(self):
|
|
self.assertEqual(self.recommend(["openvino:GPU"], codecs={"h264"}), "vaapi")
|
|
|
|
def test_qsv_is_not_offered_before_gen8(self):
|
|
self.write_cpuinfo("7th Gen Intel(R) Core(TM) i5-7500")
|
|
self.assertEqual(self.available(["openvino:GPU"]), ["vaapi"])
|
|
|
|
|
|
class TestUnknownCodecs(HwaccelRecommendationTestCase):
|
|
def test_a_codec_agnostic_family_wins_when_no_codec_is_known(self):
|
|
# a qsv preset would have to guess a codec for cameras added later
|
|
self.drm.return_value = {"0000:00:02.0": "xe"}
|
|
self.assertEqual(self.recommend(["openvino:GPU"]), "vaapi")
|
|
|
|
def test_hardware_with_no_agnostic_family_still_recommends(self):
|
|
self.assertEqual(self.recommend(["tensorrt"]), "jetson")
|
|
|
|
|
|
class TestAvailableFamilies(HwaccelRecommendationTestCase):
|
|
def test_nothing_found_offers_nothing(self):
|
|
self.assertEqual(self.available(), [])
|
|
|
|
def test_only_families_the_hardware_can_use_are_offered(self):
|
|
self.assertEqual(self.available(["onnx:nvidia"]), ["nvidia"])
|
|
|
|
def test_a_pi_does_not_offer_desktop_gpu_families(self):
|
|
self.write_device_tree()
|
|
self.assertEqual(self.available(), ["rpi"])
|
|
|
|
def test_an_intel_system_does_not_offer_the_pi_family(self):
|
|
offered = self.available(["openvino:GPU"])
|
|
|
|
self.assertIn("vaapi", offered)
|
|
self.assertNotIn("rpi", offered)
|
|
self.assertNotIn("nvidia", offered)
|
|
|
|
def test_every_gpu_present_is_offered(self):
|
|
offered = self.available(["onnx:nvidia", "openvino:GPU"])
|
|
|
|
self.assertEqual(offered[0], "nvidia")
|
|
self.assertIn("vaapi", offered)
|
|
|
|
def test_a_gpu_wins_over_the_pi_fallback(self):
|
|
self.write_device_tree()
|
|
self.assertEqual(self.recommend(["onnx:nvidia"]), "nvidia")
|
|
|
|
def test_the_recommendation_is_always_offered(self):
|
|
recommended, families = self.options(["openvino:GPU"], codecs={"h264"})
|
|
|
|
self.assertIn(recommended, [family.key for family in families])
|
|
|
|
|
|
class TestLighter(HwaccelRecommendationTestCase):
|
|
def test_lighters_media_engine_is_recommended(self):
|
|
self.write_video_device(hwaccel.LIGHTER_VIRTIO_VENDOR)
|
|
|
|
self.assertEqual(self.recommend(codecs={"h264"}), "apple-silicon")
|
|
self.assertEqual(
|
|
self.presets()["apple-silicon"],
|
|
{"h264": "preset-apple-silicon-h264", "h265": "preset-apple-silicon-h265"},
|
|
)
|
|
|
|
def test_another_virtio_media_device_is_not_lighters(self):
|
|
self.write_video_device("0x554d4551")
|
|
|
|
self.assertEqual(self.available(), [])
|
|
|
|
|
|
class TestCodecCoverage(HwaccelRecommendationTestCase):
|
|
def test_a_family_carries_a_preset_per_codec(self):
|
|
self.assertEqual(
|
|
self.presets(["tensorrt"])["jetson"],
|
|
{"h264": "preset-jetson-h264", "h265": "preset-jetson-h265"},
|
|
)
|
|
|
|
def test_a_codec_agnostic_family_carries_one_preset(self):
|
|
self.assertEqual(
|
|
self.presets(["onnx:nvidia"])["nvidia"], {"any": "preset-nvidia"}
|
|
)
|
|
|
|
def test_hevc_is_treated_as_h265(self):
|
|
self.assertEqual(self.available(["tensorrt"], codecs={"hevc"}), ["jetson"])
|
|
|
|
def test_a_family_that_cannot_decode_a_codec_is_dropped(self):
|
|
# a jetson decodes h264 and h265 only, so an mjpeg camera rules it out
|
|
self.assertEqual(self.available(["tensorrt"], codecs={"mjpeg"}), [])
|
|
|
|
def test_codec_agnostic_families_survive_any_codec(self):
|
|
self.assertEqual(
|
|
self.available(["onnx:nvidia"], codecs={"mjpeg", "h265"}), ["nvidia"]
|
|
)
|
|
|
|
def test_a_dropped_family_hands_off_to_the_next_hardware(self):
|
|
self.assertEqual(
|
|
self.recommend(["tensorrt", "onnx:nvidia"], codecs={"mjpeg"}), "nvidia"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|