diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py index b8cd91636e..a56cb55d93 100644 --- a/frigate/detectors/plugins/axelera.py +++ b/frigate/detectors/plugins/axelera.py @@ -92,9 +92,7 @@ def is_url(value: str) -> bool: URL too: rejecting it would raise FileNotFoundError on a valid remote model. ``_url_with_scheme`` prepends ``http://`` before any fetch. """ - return value.startswith(("http://", "https://")) or bool( - _HOST_URL_RE.match(value) - ) + return value.startswith(("http://", "https://")) or bool(_HOST_URL_RE.match(value)) def _url_with_scheme(value: str) -> str: @@ -212,9 +210,7 @@ def _read_model_manifest(model_dir: str) -> dict: """ candidates = [ os.path.join(model_dir, "manifest.json"), - os.path.join( - os.path.dirname(model_dir), "quantized", "manifest.json" - ), + os.path.join(os.path.dirname(model_dir), "quantized", "manifest.json"), ] for candidate in candidates: if not os.path.isfile(candidate): @@ -317,8 +313,7 @@ class _AxeleraRuntimeInference: in_orig = [int(x) for x in manifest["input_shapes_original"][0]] if len(in_padded) != 4 or in_padded[3] != 4 or in_orig[1] != 3: raise ValueError( - f"unsupported compiled input: padded {in_padded}, " - f"original {in_orig}" + f"unsupported compiled input: padded {in_padded}, original {in_orig}" ) self._padded_h, self._padded_w = in_padded[1], in_padded[2] self._model_h, self._model_w = in_orig[2], in_orig[3] @@ -327,11 +322,7 @@ class _AxeleraRuntimeInference: # buffer: n_padded_ch_inputs = [[N(lo,hi), H(lo,hi), W(lo,hi), # C(lo,hi)]] (one entry per input) npad_data = manifest["n_padded_ch_inputs"] - if ( - isinstance(npad_data, list) - and npad_data - and isinstance(npad_data[0], list) - ): + if isinstance(npad_data, list) and npad_data and isinstance(npad_data[0], list): npad_data = npad_data[0] npad = [int(x) for x in npad_data] if len(npad) == 8: @@ -340,12 +331,9 @@ class _AxeleraRuntimeInference: self._content_top = (self._padded_h - self._model_h) // 2 self._content_left = (self._padded_w - self._model_w) // 2 - out_padded = [ - tuple(int(x) for x in o) for o in manifest["output_shapes"] - ] + out_padded = [tuple(int(x) for x in o) for o in manifest["output_shapes"]] out_real = [ - tuple(int(x) for x in o) - for o in manifest["output_shapes_original"] + tuple(int(x) for x in o) for o in manifest["output_shapes_original"] ] dequant = manifest["dequantize_params"] if not (len(out_padded) == len(out_real) == len(dequant)): @@ -356,9 +344,7 @@ class _AxeleraRuntimeInference: ) # per head: real channel count (crop target) + dequant (scale, zp) self._head_real = [o[1] for o in out_real] - self._head_dequant = [ - (np.float32(d[0]), np.float32(d[1])) for d in dequant - ] + self._head_dequant = [(np.float32(d[0]), np.float32(d[1])) for d in dequant] self._out_shapes = out_padded pp_path = os.path.join(model_dir, "postprocess_graph.onnx") @@ -417,9 +403,7 @@ class _AxeleraRuntimeInference: self._decode_q: queue.Queue = queue.Queue(maxsize=BUFFER_SETS) # letterbox geometry from the frame coords (width x height) into the # model's own workspace (640x640 for the YOLO presets) - self._scale = min( - self._model_w / self._width, self._model_h / self._height - ) + self._scale = min(self._model_w / self._width, self._model_h / self._height) new_w = round(self._width * self._scale) new_h = round(self._height * self._scale) self._new_w, self._new_h = max(1, new_w), max(1, new_h) @@ -498,41 +482,29 @@ class _AxeleraRuntimeInference: try: in_buf, out_bufs = self._pool.get(timeout=1.0) except queue.Empty: - logger.error( - "axelera: no free buffers; dropping %s", connection_id - ) - self._push_result( - connection_id, np.zeros((MAX_ROWS, 6), np.float32) - ) + logger.error("axelera: no free buffers; dropping %s", connection_id) + self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32)) continue try: self._preprocess(frame, in_buf) self._instance.run([in_buf], out_bufs) self._decode_q.put_nowait((connection_id, in_buf, out_bufs)) except Exception: - logger.exception( - "axelera: inference failed for %s", connection_id - ) + logger.exception("axelera: inference failed for %s", connection_id) self._pool.put((in_buf, out_bufs)) - self._push_result( - connection_id, np.zeros((MAX_ROWS, 6), np.float32) - ) + self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32)) def _consumer(self) -> None: """Decode the raw int8 heads (dequant + onnx + NMS) to (20,6) rows.""" while not self._stop.is_set(): try: - connection_id, in_buf, out_bufs = self._decode_q.get( - timeout=0.25 - ) + connection_id, in_buf, out_bufs = self._decode_q.get(timeout=0.25) except queue.Empty: continue try: rows = self._decode(out_bufs) except Exception: - logger.exception( - "axelera: decode failed for %s", connection_id - ) + logger.exception("axelera: decode failed for %s", connection_id) rows = np.zeros((MAX_ROWS, 6), np.float32) self._pool.put((in_buf, out_bufs)) self._push_result(connection_id, rows) @@ -584,9 +556,7 @@ class _AxeleraRuntimeInference: # the model input space. raw = pp_out[0][0] cx, cy, w, h = raw[:, 0], raw[:, 1], raw[:, 2], raw[:, 3] - boxes = np.stack( - [cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2], axis=1 - ) + boxes = np.stack([cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2], axis=1) total = raw.shape[-1] cls = raw[:, 4:] if total == 4 + 1 + len(self._labels):