mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-09-28 10:56:52 +03:00
Add onboarding wizard for new installations (#24102)
* add onboarding wizard for new users * resolve hwaccel per camera and clarify recording retention The hwaccel step listed every preset Frigate ships, so an Intel box was offered Raspberry Pi and Rockchip decoding, and the codec specific presets (`preset-intel-qsv-h264` vs `-h265`) were offered as global values that break as soon as two cameras use different codecs. `/hardware/hwaccel` now returns the decoding families the probed hardware can actually use, each carrying a preset per codec, and the wizard resolves the family against the detect stream codec the camera wizard already probed: one global `ffmpeg.hwaccel_args` when every camera agrees, per-camera `cameras.<name>.ffmpeg.hwaccel_args` when they don't. The global stays on `auto` in that case so cameras added later still resolve at startup. A gen13+ Intel machine keeps its QuickSync recommendation with mixed h264 and h265 cameras instead of dropping to vaapi. The recording step's "Days to retain recordings" only wrote alert and detection retention, and the storage estimate under it assumed continuous recording. It now asks what to record in plain language, writes `record.continuous.days` to match, shows the estimate only for continuous, and drops the spinner arrows on the number input. * clean up * add light/dark mode icon switcher * use yml as default config file extension when not found * i18n tweaks * gate the setup wizard on cameras instead of a config key * render setup wizard steps by key * share the setup wizard e2e helpers and mock users * add an account step to the setup wizard * add setup wizard account step e2e coverage * cover the account step's restart behavior * button consistency * fix test * docs * fixes
This commit is contained in:
Josh Hawkins
committed by
Nicolas Mowen
Nicolas Mowen
parent
96013a0487
commit
fb8ab56c41
@@ -7,6 +7,7 @@ from fastapi import APIRouter, Depends
|
||||
from frigate.api.auth import require_role
|
||||
from frigate.api.defs.tags import Tags
|
||||
from frigate.detectors.hardware import DetectionHardware, hardware_prober
|
||||
from frigate.util.hwaccel import HwaccelRecommendation, hwaccel_options
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -28,3 +29,29 @@ def probe_hardware(refresh: bool = False) -> list[DetectionHardware]:
|
||||
Every kind of detection hardware that was found
|
||||
"""
|
||||
return hardware_prober.probe(refresh=refresh)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/hardware/hwaccel",
|
||||
response_model=HwaccelRecommendation,
|
||||
dependencies=[Depends(require_role(["admin"]))],
|
||||
)
|
||||
def hwaccel_recommendation(
|
||||
detector: str | None = None, codecs: str | None = None
|
||||
) -> HwaccelRecommendation:
|
||||
"""Get the hardware decoding this system can do.
|
||||
|
||||
Args:
|
||||
detector: Hardware key of the detection hardware in use, which biases
|
||||
the recommendation toward that hardware's GPU
|
||||
codecs: Comma separated codecs of the streams that will be decoded,
|
||||
used to drop families that cannot decode one of them
|
||||
|
||||
Returns:
|
||||
The recommended family (empty when none fits) and every usable family
|
||||
"""
|
||||
wanted = {
|
||||
codec.strip().lower() for codec in (codecs or "").split(",") if codec.strip()
|
||||
}
|
||||
recommended, available = hwaccel_options(detector, wanted)
|
||||
return HwaccelRecommendation(recommended=recommended, available=available)
|
||||
|
||||
@@ -0,0 +1,225 @@
|
||||
"""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)
|
||||
|
||||
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_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 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()
|
||||
+11
-1
@@ -79,10 +79,20 @@ def redact_credential(obj: dict[str, Any], key: str) -> None:
|
||||
|
||||
|
||||
def find_config_file() -> str:
|
||||
"""Return the path of the config file to use.
|
||||
|
||||
Both .yml and .yaml are supported, so fall back to the other extension when
|
||||
the configured path does not exist. If neither exists the configured path is
|
||||
returned so a new config is created with the default .yml extension.
|
||||
"""
|
||||
config_path = os.environ.get("CONFIG_FILE", DEFAULT_CONFIG_FILE)
|
||||
|
||||
if not os.path.isfile(config_path):
|
||||
config_path = config_path.replace("yml", "yaml")
|
||||
base, ext = os.path.splitext(config_path)
|
||||
alternate = f"{base}.yaml" if ext == ".yml" else f"{base}.yml"
|
||||
|
||||
if os.path.isfile(alternate):
|
||||
return alternate
|
||||
|
||||
return config_path
|
||||
|
||||
|
||||
@@ -0,0 +1,273 @@
|
||||
"""Recommendation of ffmpeg hwaccel presets from the hardware on the system.
|
||||
|
||||
Every check is a filesystem read, like the detection hardware probes, so this
|
||||
is cheap enough to serve from the API process.
|
||||
|
||||
Presets are grouped into families because some of them only decode the codec
|
||||
they name. A family hides that: callers pick the family their hardware needs
|
||||
and resolve it per camera against that camera's detect stream.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from frigate.const import (
|
||||
FFMPEG_HWACCEL_NVIDIA,
|
||||
FFMPEG_HWACCEL_RKMPP,
|
||||
FFMPEG_HWACCEL_VAAPI,
|
||||
)
|
||||
from frigate.detectors.hardware import hardware_prober
|
||||
from frigate.util.services import enumerate_drm_devices
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# root the /proc reads use, so tests can point them at a fixture tree
|
||||
PROC_ROOT = "/proc"
|
||||
|
||||
ANY_CODEC = "any"
|
||||
|
||||
# a Raspberry Pi has no detection hardware of its own, so it gets a key here
|
||||
RASPBERRY_PI = "raspberrypi"
|
||||
|
||||
# ffprobe names h265 streams hevc
|
||||
CODEC_ALIASES = {"hevc": "h265"}
|
||||
|
||||
# e.g. "13th Gen Intel(R) Core(TM) i5-13500"
|
||||
INTEL_GEN_PATTERN = re.compile(r"(\d+)th Gen")
|
||||
# Core Ultra dropped the generation prefix and is newer than all of them
|
||||
INTEL_ULTRA_PATTERN = re.compile(r"Core\(TM\) Ultra")
|
||||
INTEL_GEN_LATEST = 99
|
||||
|
||||
# per the hwaccel docs, gen13+ and Arc prefer qsv while older is safest on
|
||||
# vaapi, and qsv is not supported at all before gen8
|
||||
INTEL_QSV_MIN_GEN = 13
|
||||
INTEL_QSV_SUPPORTED_GEN = 8
|
||||
|
||||
# decode capable detection hardware, in recommendation priority order
|
||||
DECODE_HARDWARE = (
|
||||
"onnx:nvidia",
|
||||
"tensorrt",
|
||||
"rknn",
|
||||
"openvino:GPU",
|
||||
"onnx:amd",
|
||||
RASPBERRY_PI,
|
||||
)
|
||||
|
||||
|
||||
class HwaccelFamily(BaseModel):
|
||||
"""A kind of hardware decoding, and the presets that drive it."""
|
||||
|
||||
key: str = Field(
|
||||
title="Family key",
|
||||
description="Stable identifier for this kind of hardware decoding.",
|
||||
)
|
||||
presets: dict[str, str] = Field(
|
||||
title="Presets",
|
||||
description="The ffmpeg preset for each codec this family decodes, or a single 'any' preset when it decodes every codec.",
|
||||
)
|
||||
|
||||
|
||||
class HwaccelRecommendation(BaseModel):
|
||||
"""The hardware decoding this system can do."""
|
||||
|
||||
recommended: str = Field(
|
||||
title="Recommended family",
|
||||
description="Key of the family that fits this system best, or an empty string when none does.",
|
||||
)
|
||||
available: list[HwaccelFamily] = Field(
|
||||
default_factory=list,
|
||||
title="Available families",
|
||||
description="Every family this system's hardware can use, best first.",
|
||||
)
|
||||
|
||||
|
||||
FAMILY_NVIDIA = HwaccelFamily(key="nvidia", presets={ANY_CODEC: FFMPEG_HWACCEL_NVIDIA})
|
||||
FAMILY_VAAPI = HwaccelFamily(key="vaapi", presets={ANY_CODEC: FFMPEG_HWACCEL_VAAPI})
|
||||
FAMILY_RKMPP = HwaccelFamily(key="rkmpp", presets={ANY_CODEC: FFMPEG_HWACCEL_RKMPP})
|
||||
FAMILY_QSV = HwaccelFamily(
|
||||
key="intel-qsv",
|
||||
presets={"h264": "preset-intel-qsv-h264", "h265": "preset-intel-qsv-h265"},
|
||||
)
|
||||
FAMILY_JETSON = HwaccelFamily(
|
||||
key="jetson",
|
||||
presets={"h264": "preset-jetson-h264", "h265": "preset-jetson-h265"},
|
||||
)
|
||||
FAMILY_RPI = HwaccelFamily(
|
||||
key="rpi",
|
||||
presets={"h264": "preset-rpi-64-h264", "h265": "preset-rpi-64-h265"},
|
||||
)
|
||||
|
||||
|
||||
def _read(path: str) -> str | None:
|
||||
"""Read a small file, returning None if it cannot be read."""
|
||||
try:
|
||||
with open(path) as f:
|
||||
return f.read().strip()
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def _intel_generation() -> int | None:
|
||||
"""The Intel platform generation, or None when it cannot be determined."""
|
||||
# the xe driver only binds to the newest platforms (Arc and later iGPUs)
|
||||
if "xe" in enumerate_drm_devices().values():
|
||||
return INTEL_GEN_LATEST
|
||||
|
||||
cpuinfo = _read(f"{PROC_ROOT}/cpuinfo") or ""
|
||||
|
||||
for line in cpuinfo.splitlines():
|
||||
if not line.startswith("model name"):
|
||||
continue
|
||||
|
||||
match = INTEL_GEN_PATTERN.search(line)
|
||||
|
||||
if match:
|
||||
return int(match.group(1))
|
||||
|
||||
if INTEL_ULTRA_PATTERN.search(line):
|
||||
return INTEL_GEN_LATEST
|
||||
|
||||
break
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _is_raspberry_pi() -> bool:
|
||||
compatible = _read(f"{PROC_ROOT}/device-tree/compatible") or ""
|
||||
return "raspberrypi" in compatible
|
||||
|
||||
|
||||
def _intel_families(generation: int | None) -> list[HwaccelFamily]:
|
||||
"""vaapi drives every Intel GPU, qsv only those from gen8 on."""
|
||||
if generation is not None and generation < INTEL_QSV_SUPPORTED_GEN:
|
||||
return [FAMILY_VAAPI]
|
||||
|
||||
if generation is not None and generation >= INTEL_QSV_MIN_GEN:
|
||||
return [FAMILY_QSV, FAMILY_VAAPI]
|
||||
|
||||
return [FAMILY_VAAPI, FAMILY_QSV]
|
||||
|
||||
|
||||
def _families(key: str, generation: int | None) -> list[HwaccelFamily]:
|
||||
"""Every family that can decode on this hardware, best first."""
|
||||
if key == "onnx:nvidia":
|
||||
return [FAMILY_NVIDIA]
|
||||
|
||||
if key == "tensorrt":
|
||||
return [FAMILY_JETSON]
|
||||
|
||||
if key == "rknn":
|
||||
return [FAMILY_RKMPP]
|
||||
|
||||
if key == "onnx:amd":
|
||||
return [FAMILY_VAAPI]
|
||||
|
||||
if key == RASPBERRY_PI:
|
||||
return [FAMILY_RPI]
|
||||
|
||||
if key == "openvino:GPU":
|
||||
return _intel_families(generation)
|
||||
|
||||
return []
|
||||
|
||||
|
||||
def _decodes(family: HwaccelFamily, codecs: set[str]) -> bool:
|
||||
"""Whether a family can decode every codec that is in use."""
|
||||
if ANY_CODEC in family.presets:
|
||||
return True
|
||||
|
||||
return all(codec in family.presets for codec in codecs)
|
||||
|
||||
|
||||
def _decode_hardware(detector_key: str | None) -> list[str]:
|
||||
"""Decode capable hardware on this system, best first.
|
||||
|
||||
Args:
|
||||
detector_key: Hardware key of the detection hardware in use, whose GPU
|
||||
is preferred over any other
|
||||
|
||||
Returns:
|
||||
The hardware keys that can decode video, in recommendation order
|
||||
"""
|
||||
present = {found.key for found in hardware_prober.probe()}
|
||||
|
||||
if _is_raspberry_pi():
|
||||
present.add(RASPBERRY_PI)
|
||||
|
||||
# an Intel NPU decodes through the iGPU next to it
|
||||
if detector_key == "openvino:NPU":
|
||||
detector_key = "openvino:GPU"
|
||||
|
||||
ordered = [key for key in DECODE_HARDWARE if key in present]
|
||||
|
||||
if detector_key in ordered:
|
||||
ordered.remove(detector_key)
|
||||
ordered.insert(0, detector_key)
|
||||
|
||||
return ordered
|
||||
|
||||
|
||||
def hwaccel_options(
|
||||
detector_key: str | None = None, codecs: set[str] | None = None
|
||||
) -> tuple[str, list[HwaccelFamily]]:
|
||||
"""Get the hardware decoding this system can do.
|
||||
|
||||
Args:
|
||||
detector_key: Hardware key of the detection hardware in use, which
|
||||
biases the recommendation toward that hardware's GPU
|
||||
codecs: Codecs of the streams that will be decoded, used to drop
|
||||
families that cannot decode one of them
|
||||
|
||||
Returns:
|
||||
The recommended family key (empty when none fits) and every usable
|
||||
family, best first
|
||||
"""
|
||||
wanted = {CODEC_ALIASES.get(codec, codec) for codec in codecs or set()}
|
||||
hardware = _decode_hardware(detector_key)
|
||||
generation = _intel_generation() if "openvino:GPU" in hardware else None
|
||||
|
||||
available: list[HwaccelFamily] = []
|
||||
recommended = ""
|
||||
|
||||
for key in hardware:
|
||||
usable = [
|
||||
family for family in _families(key, generation) if _decodes(family, wanted)
|
||||
]
|
||||
|
||||
if usable and not recommended:
|
||||
recommended = _recommend(usable, bool(wanted))
|
||||
|
||||
for family in usable:
|
||||
if family.key not in {entry.key for entry in available}:
|
||||
available.append(family)
|
||||
|
||||
return recommended, available
|
||||
|
||||
|
||||
def _recommend(families: list[HwaccelFamily], codecs_known: bool) -> str:
|
||||
"""Pick the family to default to out of the ones this hardware can use."""
|
||||
if not codecs_known:
|
||||
# a codec specific family would have to guess a codec for cameras
|
||||
# that do not exist yet
|
||||
for family in families:
|
||||
if ANY_CODEC in family.presets:
|
||||
return family.key
|
||||
|
||||
return families[0].key
|
||||
|
||||
|
||||
def recommend_hwaccel(
|
||||
detector_key: str | None = None, codecs: set[str] | None = None
|
||||
) -> str:
|
||||
"""Recommend a hardware decoding family for this system.
|
||||
|
||||
Args:
|
||||
detector_key: Hardware key of the detection hardware in use
|
||||
codecs: Codecs of the streams that will be decoded
|
||||
|
||||
Returns:
|
||||
The key of the family that fits, or an empty string when none does
|
||||
"""
|
||||
return hwaccel_options(detector_key, codecs)[0]
|
||||
Reference in New Issue
Block a user