nik/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-18 07:29:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

style(axelera): ruff format under the CI-pinned 0.15.20

Browse Source 
Pre-existing drift in HEAD: the file failed 'ruff format --check' as-is,
so CI's format gate was already red. Line-joins only, no behavior change.
This commit is contained in:
patrykorwat 2026-08-24 10:43:12 +00:00
parent 4e33dda6da
commit 0d0581494f
1 changed files with 15 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
frigate/detectors/plugins/axelera.py
View File

@ -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):