remove all references to degirum in frigate (#23882)

the company ceased operations on 1 Aug 2026

frigate/detectors/plugins/degirum.py

@@ -1,157 +0,0 @@
-import logging
-import queue
-from typing import Literal
-
-import numpy as np
-from pydantic import ConfigDict, Field
-
-from frigate.detectors.detection_api import DetectionApi
-from frigate.detectors.detector_config import BaseDetectorConfig
-
-logger = logging.getLogger(__name__)
-DETECTOR_KEY = "degirum"
-
-
-### DETECTOR CONFIG ###
-class DGDetectorConfig(BaseDetectorConfig):
-    """DeGirum detector for running models via DeGirum cloud or local inference services."""
-
-    model_config = ConfigDict(
-        title="DeGirum",
-    )
-
-    type: Literal[DETECTOR_KEY]
-    location: str = Field(
-        default=None,
-        title="Inference Location",
-        description="Location of the DeGirim inference engine (e.g. '@cloud', '127.0.0.1').",
-    )
-    zoo: str = Field(
-        default=None,
-        title="Model Zoo",
-        description="Path or URL to the DeGirum model zoo.",
-    )
-    token: str = Field(
-        default=None,
-        title="DeGirum Cloud Token",
-        description="Token for DeGirum Cloud access.",
-    )
-
-
-### ACTUAL DETECTOR  ###
-class DGDetector(DetectionApi):
-    type_key = DETECTOR_KEY
-
-    def __init__(self, detector_config: DGDetectorConfig):
-        try:
-            import degirum as dg
-        except ModuleNotFoundError:
-            raise ImportError("Unable to import DeGirum detector.") from None
-
-        self._queue = queue.Queue()
-        self._zoo = dg.connect(
-            detector_config.location, detector_config.zoo, detector_config.token
-        )
-
-        logger.debug(f"Models in zoo: {self._zoo.list_models()}")
-
-        self.dg_model = self._zoo.load_model(
-            detector_config.model.path,
-        )
-
-        # Setting input image format to raw reduces preprocessing time
-        self.dg_model.input_image_format = "RAW"
-
-        # Prioritize the most powerful hardware available
-        self.select_best_device_type()
-        # Frigate handles pre processing as long as these are all set
-        input_shape = self.dg_model.input_shape[0]
-        self.model_height = input_shape[1]
-        self.model_width = input_shape[2]
-
-        # Passing in dummy frame so initial connection latency happens in
-        # init function and not during actual prediction
-        frame = np.zeros(
-            (detector_config.model.width, detector_config.model.height, 3),
-            dtype=np.uint8,
-        )
-        # Pass in frame to overcome first frame latency
-        self.dg_model(frame)
-        self.prediction = self.prediction_generator()
-
-    def select_best_device_type(self):
-        """
-        Helper function that selects fastest hardware available per model runtime
-        """
-        types = self.dg_model.supported_device_types
-
-        device_map = {
-            "OPENVINO": ["GPU", "NPU", "CPU"],
-            "HAILORT": ["HAILO8L", "HAILO8"],
-            "N2X": ["ORCA1", "CPU"],
-            "ONNX": ["VITIS_NPU", "CPU"],
-            "RKNN": ["RK3566", "RK3568", "RK3588"],
-            "TENSORRT": ["DLA", "GPU", "DLA_ONLY"],
-            "TFLITE": ["ARMNN", "EDGETPU", "CPU"],
-        }
-
-        runtime = types[0].split("/")[0]
-        # Just create an array of format {runtime}/{hardware} for every hardware
-        # in the value for appropriate key in device_map
-        self.dg_model.device_type = [
-            f"{runtime}/{hardware}" for hardware in device_map[runtime]
-        ]
-
-    def prediction_generator(self):
-        """
-        Generator for all incoming frames. By using this generator, we don't have to keep
-        reconnecting our websocket on every "predict" call.
-        """
-        logger.debug("Prediction generator was called")
-        with self.dg_model as model:
-            while 1:
-                logger.info(f"q size before calling get: {self._queue.qsize()}")
-                data = self._queue.get(block=True)
-                logger.info(f"q size after calling get: {self._queue.qsize()}")
-                logger.debug(
-                    f"Data we're passing into model predict: {data}, shape of data: {data.shape}"
-                )
-                result = model.predict(data)
-                logger.debug(f"Prediction result: {result}")
-                yield result
-
-    def detect_raw(self, tensor_input):
-        # Reshaping tensor to work with pysdk
-        truncated_input = tensor_input.reshape(tensor_input.shape[1:])
-        logger.debug(f"Detect raw was called for tensor input: {tensor_input}")
-
-        # add tensor_input to input queue
-        self._queue.put(truncated_input)
-        logger.debug(f"Queue size after adding truncated input: {self._queue.qsize()}")
-
-        # define empty detection result
-        detections = np.zeros((20, 6), np.float32)
-        # grab prediction
-        res = next(self.prediction)
-
-        # If we have an empty prediction, return immediately
-        if len(res.results) == 0 or len(res.results[0]) == 0:
-            return detections
-
-        i = 0
-        for result in res.results:
-            if i >= 20:
-                break
-
-            detections[i] = [
-                result["category_id"],
-                float(result["score"]),
-                result["bbox"][1] / self.model_height,
-                result["bbox"][0] / self.model_width,
-                result["bbox"][3] / self.model_height,
-                result["bbox"][2] / self.model_width,
-            ]
-            i += 1
-
-        logger.debug(f"Detections output: {detections}")
-        return detections