2022-11-04 03:23:09 +01:00
|
|
|
import logging
|
|
|
|
import numpy as np
|
|
|
|
|
|
|
|
from frigate.detectors.detection_api import DetectionApi
|
2022-12-15 14:12:52 +01:00
|
|
|
from frigate.detectors.detector_config import BaseDetectorConfig
|
|
|
|
from typing import Literal
|
|
|
|
from pydantic import Extra, Field
|
2023-03-04 00:44:17 +01:00
|
|
|
|
|
|
|
try:
|
|
|
|
from tflite_runtime.interpreter import Interpreter
|
|
|
|
except ModuleNotFoundError:
|
|
|
|
from tensorflow.lite.python.interpreter import Interpreter
|
2022-11-04 03:23:09 +01:00
|
|
|
|
2022-12-15 14:12:52 +01:00
|
|
|
|
2022-11-04 03:23:09 +01:00
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
2022-12-15 14:12:52 +01:00
|
|
|
DETECTOR_KEY = "cpu"
|
|
|
|
|
|
|
|
|
|
|
|
class CpuDetectorConfig(BaseDetectorConfig):
|
|
|
|
type: Literal[DETECTOR_KEY]
|
|
|
|
num_threads: int = Field(default=3, title="Number of detection threads")
|
|
|
|
|
2022-11-04 03:23:09 +01:00
|
|
|
|
|
|
|
class CpuTfl(DetectionApi):
|
2022-12-15 14:12:52 +01:00
|
|
|
type_key = DETECTOR_KEY
|
|
|
|
|
|
|
|
def __init__(self, detector_config: CpuDetectorConfig):
|
2023-03-04 00:44:17 +01:00
|
|
|
self.interpreter = Interpreter(
|
2023-04-24 14:24:28 +02:00
|
|
|
model_path=detector_config.model.path,
|
2022-12-15 14:12:52 +01:00
|
|
|
num_threads=detector_config.num_threads or 3,
|
2022-11-04 03:23:09 +01:00
|
|
|
)
|
|
|
|
|
|
|
|
self.interpreter.allocate_tensors()
|
|
|
|
|
|
|
|
self.tensor_input_details = self.interpreter.get_input_details()
|
|
|
|
self.tensor_output_details = self.interpreter.get_output_details()
|
|
|
|
|
|
|
|
def detect_raw(self, tensor_input):
|
|
|
|
self.interpreter.set_tensor(self.tensor_input_details[0]["index"], tensor_input)
|
|
|
|
self.interpreter.invoke()
|
|
|
|
|
|
|
|
boxes = self.interpreter.tensor(self.tensor_output_details[0]["index"])()[0]
|
|
|
|
class_ids = self.interpreter.tensor(self.tensor_output_details[1]["index"])()[0]
|
|
|
|
scores = self.interpreter.tensor(self.tensor_output_details[2]["index"])()[0]
|
|
|
|
count = int(
|
|
|
|
self.interpreter.tensor(self.tensor_output_details[3]["index"])()[0]
|
|
|
|
)
|
|
|
|
|
|
|
|
detections = np.zeros((20, 6), np.float32)
|
|
|
|
|
|
|
|
for i in range(count):
|
|
|
|
if scores[i] < 0.4 or i == 20:
|
|
|
|
break
|
|
|
|
detections[i] = [
|
|
|
|
class_ids[i],
|
|
|
|
float(scores[i]),
|
|
|
|
boxes[i][0],
|
|
|
|
boxes[i][1],
|
|
|
|
boxes[i][2],
|
|
|
|
boxes[i][3],
|
|
|
|
]
|
|
|
|
|
|
|
|
return detections
|