| from pathlib import Path |
| import math |
| import cv2 |
| import numpy as np |
| import onnxruntime as ort |
| from numpy import ndarray |
| from pydantic import BaseModel |
|
|
| class BoundingBox(BaseModel): |
| x1: int |
| y1: int |
| x2: int |
| y2: int |
| cls_id: int |
| conf: float |
|
|
| class TVFrameResult(BaseModel): |
| frame_id: int |
| boxes: list[BoundingBox] |
| keypoints: list[tuple[int, int]] |
|
|
| class Miner: |
| class_names = ['fire', 'smoke', 'fire extinguisher'] |
| _model_class_order = ["fire", "fire extinguisher", "smoke"] |
| iou_thres = 0.55 |
| cross_iou_thresh = 0.8 |
| max_det = 30 |
| _conf_thres_array = np.array([0.22, 0.30, 0.30], dtype=np.float32) |
| _bonus_array = np.array([0.05, 0.05, 0.05], dtype=np.float32) |
| min_box_area = 0 |
| min_side = 8 |
| max_aspect_ratio = 8.0 |
| smoke_merge_overlap = 0.8 |
| fire_merge_overlap = 0.9 |
| fire_suppress_overlap = 0.88 |
| smoke_raw_refine_overlap = 0.9 |
| smoke_ext_shrink = 0.95 |
| fire_expand = 1.05 |
| fire_color_filter_max_conf = 0.45 |
| fire_ext_color_filter_max_conf = 0.0 |
| color_filter_min_saturation = 0.06 |
| use_edge_filter = False |
| edge_filter_max_conf = 0.0 |
| edge_tol = 2.0 |
| use_tta_view_filter = False |
| tta_view_filter_max_conf = 0.0 |
| tta_view_iou_thresh = 0.5 |
|
|
| def __init__(self, path_hf_repo: Path) -> None: |
| model_path = path_hf_repo / 'weights.onnx' |
| print('ORT version:', ort.__version__) |
| try: |
| ort.preload_dlls() |
| print('✅ onnxruntime.preload_dlls() success') |
| except Exception as e: |
| print(f'⚠️ preload_dlls failed: {e}') |
| print('ORT available providers BEFORE session:', ort.get_available_providers()) |
| sess_options = ort.SessionOptions() |
| sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL |
| sess_options.intra_op_num_threads = 2 |
| sess_options.inter_op_num_threads = 1 |
| sess_options.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL |
| try: |
| self.session = ort.InferenceSession(str(model_path), sess_options=sess_options, providers=['CUDAExecutionProvider', 'CPUExecutionProvider']) |
| print('✅ Created ORT session with preferred CUDA provider list') |
| except Exception as e: |
| print(f'⚠️ CUDA session creation failed, falling back to CPU: {e}') |
| self.session = ort.InferenceSession(str(model_path), sess_options=sess_options, providers=['CPUExecutionProvider']) |
| print('ORT session providers:', self.session.get_providers()) |
| model_class_order = self._read_model_class_order() |
| if model_class_order is None: |
| model_class_order = list(self._model_class_order) |
| print(f'cls order: no usable ONNX metadata, FALLBACK {model_class_order}') |
| else: |
| print(f'cls order: from ONNX metadata {model_class_order}') |
| self.cls_remap = np.array([self.class_names.index(n) for n in model_class_order], dtype=np.int32) |
| for inp in self.session.get_inputs(): |
| print('INPUT:', inp.name, inp.shape, inp.type) |
| for out in self.session.get_outputs(): |
| print('OUTPUT:', out.name, out.shape, out.type) |
| self.input_name = self.session.get_inputs()[0].name |
| self.output_names = [output.name for output in self.session.get_outputs()] |
| self.input_shape = self.session.get_inputs()[0].shape |
| self.input_height = self._safe_dim(self.input_shape[2], default=1280) |
| self.input_width = self._safe_dim(self.input_shape[3], default=1280) |
| self.use_tta = False |
| print(f'✅ ONNX model loaded from: {model_path}') |
| print(f'✅ ONNX providers: {self.session.get_providers()}') |
| print(f'✅ ONNX input: name={self.input_name}, shape={self.input_shape}') |
| print('per-class conf: ' + ', '.join((f'{n}={t:.3f}' for n, t in zip(self.class_names, self._conf_thres_array.tolist())))) |
| self._warmup() |
|
|
| def _warmup(self, iters: int=3) -> None: |
| try: |
| dummy = np.zeros((720, 1280, 3), dtype=np.uint8) |
| for _ in range(max(1, iters)): |
| self.predict_batch(batch_images=[dummy], offset=0, n_keypoints=0) |
| print(f'✅ warmup: {iters} dummy predict_batch call(s) done') |
| except Exception as e: |
| print(f'⚠️ warmup skipped: {e}') |
|
|
| def __repr__(self) -> str: |
| return f'ONNXRuntime(session={type(self.session).__name__}, providers={self.session.get_providers()})' |
|
|
| @staticmethod |
| def _safe_dim(value, default: int) -> int: |
| return value if isinstance(value, int) and value > 0 else default |
|
|
| def _read_model_class_order(self) -> list[str] | None: |
| try: |
| import ast |
| meta = self.session.get_modelmeta().custom_metadata_map |
| names = ast.literal_eval(meta['names']) |
| if isinstance(names, dict): |
| order = [str(names[i]) for i in sorted(names)] |
| else: |
| order = [str(n) for n in names] |
| except Exception as e: |
| print(f'cls order: could not read ONNX names metadata ({e})') |
| return None |
| if sorted(order) != sorted(self.class_names): |
| print(f'cls order: ONNX names {order} do not match expected classes {self.class_names}; ignoring metadata') |
| return None |
| return order |
|
|
| def _letterbox(self, image: ndarray, new_shape: tuple[int, int], color=(114, 114, 114)) -> tuple[ndarray, float, tuple[float, float]]: |
| h, w = image.shape[:2] |
| new_w, new_h = new_shape |
| ratio = min(new_w / w, new_h / h) |
| resized_w = int(round(w * ratio)) |
| resized_h = int(round(h * ratio)) |
| if (resized_w, resized_h) != (w, h): |
| interp = cv2.INTER_CUBIC if ratio > 1.0 else cv2.INTER_LINEAR |
| image = cv2.resize(image, (resized_w, resized_h), interpolation=interp) |
| dw = (new_w - resized_w) / 2.0 |
| dh = (new_h - resized_h) / 2.0 |
| left = int(round(dw - 0.1)) |
| right = int(round(dw + 0.1)) |
| top = int(round(dh - 0.1)) |
| bottom = int(round(dh + 0.1)) |
| padded = cv2.copyMakeBorder(image, top, bottom, left, right, borderType=cv2.BORDER_CONSTANT, value=color) |
| return (padded, ratio, (dw, dh)) |
|
|
| def _preprocess(self, image: ndarray) -> tuple[np.ndarray, float, tuple[float, float], tuple[int, int]]: |
| orig_h, orig_w = image.shape[:2] |
| img, ratio, pad = self._letterbox(image, (self.input_width, self.input_height)) |
| blob = cv2.dnn.blobFromImage(img, scalefactor=1.0 / 255.0, swapRB=True) |
| return (blob, ratio, pad, (orig_w, orig_h)) |
|
|
| @staticmethod |
| def _clip_boxes(boxes: np.ndarray, image_size: tuple[int, int]) -> np.ndarray: |
| w, h = image_size |
| boxes[:, 0] = np.clip(boxes[:, 0], 0, w - 1) |
| boxes[:, 1] = np.clip(boxes[:, 1], 0, h - 1) |
| boxes[:, 2] = np.clip(boxes[:, 2], 0, w - 1) |
| boxes[:, 3] = np.clip(boxes[:, 3], 0, h - 1) |
| return boxes |
|
|
| @staticmethod |
| def _xywh_to_xyxy(boxes: np.ndarray) -> np.ndarray: |
| out = np.empty_like(boxes) |
| out[:, 0] = boxes[:, 0] - boxes[:, 2] / 2.0 |
| out[:, 1] = boxes[:, 1] - boxes[:, 3] / 2.0 |
| out[:, 2] = boxes[:, 0] + boxes[:, 2] / 2.0 |
| out[:, 3] = boxes[:, 1] + boxes[:, 3] / 2.0 |
| return out |
|
|
| @staticmethod |
| def _hard_nms(boxes: np.ndarray, scores: np.ndarray, iou_thresh: float) -> np.ndarray: |
| n = len(boxes) |
| if n == 0: |
| return np.array([], dtype=np.intp) |
| order = np.argsort(-scores) |
| keep: list[int] = [] |
| while len(order) > 0: |
| i = int(order[0]) |
| keep.append(i) |
| if len(order) == 1: |
| break |
| rest = order[1:] |
| xx1 = np.maximum(boxes[i, 0], boxes[rest, 0]) |
| yy1 = np.maximum(boxes[i, 1], boxes[rest, 1]) |
| xx2 = np.minimum(boxes[i, 2], boxes[rest, 2]) |
| yy2 = np.minimum(boxes[i, 3], boxes[rest, 3]) |
| inter = np.maximum(0.0, xx2 - xx1) * np.maximum(0.0, yy2 - yy1) |
| a_i = max(0.0, boxes[i, 2] - boxes[i, 0]) * max(0.0, boxes[i, 3] - boxes[i, 1]) |
| a_r = np.maximum(0.0, boxes[rest, 2] - boxes[rest, 0]) * np.maximum(0.0, boxes[rest, 3] - boxes[rest, 1]) |
| iou = inter / (a_i + a_r - inter + 1e-07) |
| order = rest[iou <= iou_thresh] |
| return np.array(keep, dtype=np.intp) |
|
|
| def _per_class_hard_nms(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray, iou_thresh: float) -> np.ndarray: |
| if len(boxes) == 0: |
| return np.array([], dtype=np.intp) |
| all_keep: list[int] = [] |
| for c in np.unique(cls_ids): |
| mask = cls_ids == c |
| indices = np.where(mask)[0] |
| keep = self._hard_nms(boxes[mask], scores[mask], iou_thresh) |
| all_keep.extend(indices[keep].tolist()) |
| all_keep.sort() |
| return np.array(all_keep, dtype=np.intp) |
|
|
| def _cross_class_dedup_op(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray, iou_thresh: float) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| n = len(boxes) |
| if n <= 1: |
| return (boxes, scores, cls_ids) |
| boxes = np.asarray(boxes, dtype=np.float32) |
| scores = np.asarray(scores, dtype=np.float32) |
| cls_ids = np.asarray(cls_ids, dtype=np.int32) |
| areas = np.maximum(0.0, boxes[:, 2] - boxes[:, 0]) * np.maximum(0.0, boxes[:, 3] - boxes[:, 1]) |
| margins = scores - self._conf_thres_array[cls_ids] |
| order = np.lexsort((-areas, -margins)) |
| suppressed = np.zeros(n, dtype=bool) |
| keep: list[int] = [] |
| for i in order: |
| if suppressed[i]: |
| continue |
| keep.append(int(i)) |
| bi = boxes[i] |
| xx1 = np.maximum(bi[0], boxes[:, 0]) |
| yy1 = np.maximum(bi[1], boxes[:, 1]) |
| xx2 = np.minimum(bi[2], boxes[:, 2]) |
| yy2 = np.minimum(bi[3], boxes[:, 3]) |
| inter = np.maximum(0.0, xx2 - xx1) * np.maximum(0.0, yy2 - yy1) |
| a_i = max(1e-07, float((bi[2] - bi[0]) * (bi[3] - bi[1]))) |
| iou = inter / (a_i + areas - inter + 1e-07) |
| dup = iou > iou_thresh |
| dup[i] = False |
| suppressed |= dup |
| keep_idx = np.array(keep, dtype=np.intp) |
| return (boxes[keep_idx], scores[keep_idx], cls_ids[keep_idx]) |
|
|
| def _merge_class_boxes(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray, target_cls: int, overlap: float) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| if overlap > 1.0: |
| return (boxes, scores, cls_ids) |
| idx = np.where(cls_ids == target_cls)[0] |
| if len(idx) <= 1: |
| return (boxes, scores, cls_ids) |
| sb = boxes[idx].astype(np.float32).tolist() |
| ss = scores[idx].astype(np.float32).tolist() |
| merged_any = True |
| while merged_any and len(sb) > 1: |
| merged_any = False |
| for i in range(len(sb)): |
| for j in range(i + 1, len(sb)): |
| a, b = (sb[i], sb[j]) |
| ix1 = max(a[0], b[0]) |
| iy1 = max(a[1], b[1]) |
| ix2 = min(a[2], b[2]) |
| iy2 = min(a[3], b[3]) |
| inter = max(0.0, ix2 - ix1) * max(0.0, iy2 - iy1) |
| area_a = max(0.0, a[2] - a[0]) * max(0.0, a[3] - a[1]) |
| area_b = max(0.0, b[2] - b[0]) * max(0.0, b[3] - b[1]) |
| smaller = min(area_a, area_b) |
| if inter / (smaller + 1e-07) >= overlap: |
| sb[i] = [min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3])] |
| ss[i] = max(ss[i], ss[j]) |
| del sb[j] |
| del ss[j] |
| merged_any = True |
| break |
| if merged_any: |
| break |
| other = cls_ids != target_cls |
| new_boxes = np.concatenate([boxes[other].astype(np.float32), np.array(sb, dtype=np.float32).reshape(-1, 4)]) |
| new_scores = np.concatenate([scores[other].astype(np.float32), np.array(ss, dtype=np.float32)]) |
| new_cls = np.concatenate([cls_ids[other].astype(np.int32), np.full(len(sb), target_cls, dtype=np.int32)]) |
| return (new_boxes, new_scores, new_cls) |
|
|
| def _suppress_contained_lower_conf(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray, target_cls: int, overlap: float) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| if overlap > 1.0: |
| return (boxes, scores, cls_ids) |
| idx = np.where(cls_ids == target_cls)[0] |
| if len(idx) <= 1: |
| return (boxes, scores, cls_ids) |
| order = idx[np.argsort(-scores[idx])] |
| remove: set[int] = set() |
| for a in range(len(order)): |
| i = int(order[a]) |
| if i in remove: |
| continue |
| bi = boxes[i] |
| area_i = max(1e-07, float((bi[2] - bi[0]) * (bi[3] - bi[1]))) |
| for b in range(a + 1, len(order)): |
| j = int(order[b]) |
| if j in remove: |
| continue |
| bj = boxes[j] |
| ix1 = max(bi[0], bj[0]) |
| iy1 = max(bi[1], bj[1]) |
| ix2 = min(bi[2], bj[2]) |
| iy2 = min(bi[3], bj[3]) |
| inter = max(0.0, ix2 - ix1) * max(0.0, iy2 - iy1) |
| if inter <= 0.0: |
| continue |
| area_j = max(1e-07, float((bj[2] - bj[0]) * (bj[3] - bj[1]))) |
| if inter / (min(area_i, area_j) + 1e-07) >= overlap: |
| remove.add(j) |
| if not remove: |
| return (boxes, scores, cls_ids) |
| keep = np.array([k not in remove for k in range(len(boxes))], dtype=bool) |
| return (boxes[keep], scores[keep], cls_ids[keep]) |
|
|
| def _merge_same_class_boxes(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| boxes, scores, cls_ids = self._merge_class_boxes(boxes, scores, cls_ids, self.class_names.index('smoke'), self.smoke_merge_overlap) |
| boxes, scores, cls_ids = self._merge_class_boxes(boxes, scores, cls_ids, self.class_names.index('fire'), self.fire_merge_overlap) |
| boxes, scores, cls_ids = self._suppress_contained_lower_conf(boxes, scores, cls_ids, self.class_names.index('fire'), self.fire_suppress_overlap) |
| return (boxes, scores, cls_ids) |
|
|
| def _merge_smoke_boxes(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| return self._merge_same_class_boxes(boxes, scores, cls_ids) |
|
|
| @staticmethod |
| def _max_score_per_cluster(post_boxes: np.ndarray, post_cls: np.ndarray, full_boxes: np.ndarray, full_scores: np.ndarray, full_cls: np.ndarray, iou_thresh: float) -> np.ndarray: |
| n = len(post_boxes) |
| if n == 0: |
| return np.empty(0, dtype=np.float32) |
| full_areas = np.maximum(0.0, full_boxes[:, 2] - full_boxes[:, 0]) * np.maximum(0.0, full_boxes[:, 3] - full_boxes[:, 1]) |
| out = np.empty(n, dtype=np.float32) |
| for i in range(n): |
| bi = post_boxes[i] |
| xx1 = np.maximum(bi[0], full_boxes[:, 0]) |
| yy1 = np.maximum(bi[1], full_boxes[:, 1]) |
| xx2 = np.minimum(bi[2], full_boxes[:, 2]) |
| yy2 = np.minimum(bi[3], full_boxes[:, 3]) |
| inter = np.maximum(0.0, xx2 - xx1) * np.maximum(0.0, yy2 - yy1) |
| a_i = max(0.0, float((bi[2] - bi[0]) * (bi[3] - bi[1]))) |
| iou = inter / (a_i + full_areas - inter + 1e-07) |
| cluster = (iou >= iou_thresh) & (full_cls == post_cls[i]) |
| out[i] = float(np.max(full_scores[cluster])) if np.any(cluster) else 0.0 |
| return out |
|
|
| def _conf_filter_mask(self, scores: np.ndarray, cls_ids: np.ndarray) -> np.ndarray: |
| if len(scores) == 0: |
| return np.zeros(0, dtype=bool) |
| thr = self._conf_thres_array[cls_ids] |
| keep = scores >= thr |
| for c in np.unique(cls_ids): |
| b = float(self._bonus_array[c]) |
| if b <= 0.0: |
| continue |
| cm = cls_ids == c |
| if keep[cm].any(): |
| continue |
| idx = np.where(cm)[0] |
| top = int(idx[int(np.argmax(scores[idx]))]) |
| if scores[top] >= self._conf_thres_array[c] - b: |
| keep[top] = True |
| return keep |
|
|
| def _filter_sane_boxes(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray, orig_size: tuple[int, int]) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| if len(boxes) == 0: |
| return (boxes, scores, cls_ids) |
| orig_w, orig_h = orig_size |
| image_area = float(orig_w * orig_h) |
| keep = [] |
| for i, box in enumerate(boxes): |
| x1, y1, x2, y2 = box.tolist() |
| bw = x2 - x1 |
| bh = y2 - y1 |
| if bw <= 0 or bh <= 0: |
| continue |
| if bw < self.min_side or bh < self.min_side: |
| continue |
| area = bw * bh |
| if area < self.min_box_area: |
| continue |
| if area > 0.95 * image_area: |
| continue |
| ar = max(bw / max(bh, 1e-06), bh / max(bw, 1e-06)) |
| if ar > self.max_aspect_ratio: |
| continue |
| keep.append(i) |
| if not keep: |
| return (np.empty((0, 4), dtype=np.float32), np.empty((0,), dtype=np.float32), np.empty((0,), dtype=np.int32)) |
| k = np.array(keep, dtype=np.intp) |
| return (boxes[k], scores[k], cls_ids[k]) |
|
|
| def _per_view_pipeline(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| if len(boxes) > 1: |
| keep = self._per_class_hard_nms(boxes, scores, cls_ids, self.iou_thres) |
| boxes, scores, cls_ids = (boxes[keep], scores[keep], cls_ids[keep]) |
| if len(scores) > self.max_det: |
| top = np.argsort(-scores)[:self.max_det] |
| boxes, scores, cls_ids = (boxes[top], scores[top], cls_ids[top]) |
| if len(boxes) > 1: |
| boxes, scores, cls_ids = self._cross_class_dedup_op(boxes, scores, cls_ids, self.cross_iou_thresh) |
| if len(boxes) > 1: |
| boxes, scores, cls_ids = self._merge_same_class_boxes(boxes, scores, cls_ids) |
| return (boxes, scores, cls_ids) |
|
|
| @staticmethod |
| def _roi_for_box(image: np.ndarray, box: BoundingBox) -> np.ndarray | None: |
| h, w = image.shape[:2] |
| x1 = max(0, int(math.floor(box.x1))) |
| y1 = max(0, int(math.floor(box.y1))) |
| x2 = min(w, int(math.ceil(box.x2))) |
| y2 = min(h, int(math.ceil(box.y2))) |
| if x2 <= x1 or y2 <= y1: |
| return None |
| roi = image[y1:y2, x1:x2] |
| return roi if roi.size else None |
|
|
| def _roi_is_near_grayscale(self, roi: np.ndarray) -> bool: |
| mx = roi.max(axis=2).astype(np.float32) |
| mn = roi.min(axis=2).astype(np.float32) |
| sat = (mx - mn) / (mx + 1e-06) |
| return float(sat.mean()) < self.color_filter_min_saturation |
|
|
| @staticmethod |
| def _passes_fire_color(roi: np.ndarray) -> bool: |
| blue = roi[:, :, 0].astype(np.float32) |
| green = roi[:, :, 1].astype(np.float32) |
| red = roi[:, :, 2].astype(np.float32) |
| mean_r = float(np.mean(red)) |
| max_rgb = float(max(np.max(red), np.max(green), np.max(blue))) |
| bright_frac = float(np.mean(np.max(roi, axis=2) >= 150)) |
| if max_rgb >= 200.0 and bright_frac >= 0.01: |
| return True |
| warm = (red > green + 10.0) & (red > blue + 10.0) |
| warm_frac = float(np.mean(warm)) |
| r_minus_g = mean_r - float(np.mean(green)) |
| if warm_frac >= 0.05 and (max_rgb >= 120.0 or mean_r >= 120.0 or warm_frac >= 0.15): |
| return True |
| if bright_frac >= 0.12 and r_minus_g >= 2.0: |
| return True |
| return False |
|
|
| @staticmethod |
| def _passes_fire_ext_red_color(roi: np.ndarray) -> bool: |
| blue = roi[:, :, 0].astype(np.float32) |
| green = roi[:, :, 1].astype(np.float32) |
| red = roi[:, :, 2].astype(np.float32) |
| red_dom = float(np.mean((red > green + 10.0) & (red > blue + 10.0))) |
| if red_dom >= 0.03: |
| return True |
| if float(np.mean(red)) - float(np.mean(green)) >= 0.0 and float(np.mean(red)) >= 50.0: |
| return True |
| return False |
|
|
| def _remove_edge_low_conf(self, results: list[BoundingBox], orig_size: tuple[int, int]) -> list[BoundingBox]: |
| if not self.use_edge_filter or self.edge_filter_max_conf <= 0.0 or (not results): |
| return results |
| w, h = orig_size |
| tol = self.edge_tol |
| out: list[BoundingBox] = [] |
| for b in results: |
| on_edge = b.x1 <= tol or b.y1 <= tol or b.x2 >= w - 1 - tol or (b.y2 >= h - 1 - tol) |
| if on_edge and b.conf <= self.edge_filter_max_conf: |
| continue |
| out.append(b) |
| return out |
|
|
| def _views_corroborated(self, post_boxes: np.ndarray, post_cls: np.ndarray, full_boxes: np.ndarray, full_cls: np.ndarray, full_views: np.ndarray, iou_thresh: float) -> np.ndarray: |
| n = len(post_boxes) |
| if n == 0: |
| return np.zeros(0, dtype=bool) |
| full_areas = np.maximum(0.0, full_boxes[:, 2] - full_boxes[:, 0]) * np.maximum(0.0, full_boxes[:, 3] - full_boxes[:, 1]) |
| out = np.zeros(n, dtype=bool) |
| for i in range(n): |
| bi = post_boxes[i] |
| xx1 = np.maximum(bi[0], full_boxes[:, 0]) |
| yy1 = np.maximum(bi[1], full_boxes[:, 1]) |
| xx2 = np.minimum(bi[2], full_boxes[:, 2]) |
| yy2 = np.minimum(bi[3], full_boxes[:, 3]) |
| inter = np.maximum(0.0, xx2 - xx1) * np.maximum(0.0, yy2 - yy1) |
| a_i = max(0.0, float((bi[2] - bi[0]) * (bi[3] - bi[1]))) |
| iou = inter / (a_i + full_areas - inter + 1e-07) |
| mask = (iou >= iou_thresh) & (full_cls == post_cls[i]) |
| if np.any(mask): |
| out[i] = len(np.unique(full_views[mask])) >= 2 |
| return out |
|
|
| def _filter_low_conf_by_color(self, image: np.ndarray, results: list[BoundingBox]) -> list[BoundingBox]: |
| if not results: |
| return results |
| cls_fire = self.class_names.index('fire') |
| cls_ext = self.class_names.index('fire extinguisher') |
| out: list[BoundingBox] = [] |
| for box in results: |
| check_fire = box.cls_id == cls_fire and box.conf <= self.fire_color_filter_max_conf |
| check_ext = box.cls_id == cls_ext and box.conf <= self.fire_ext_color_filter_max_conf |
| if not check_fire and (not check_ext): |
| out.append(box) |
| continue |
| roi = self._roi_for_box(image, box) |
| if roi is None or self._roi_is_near_grayscale(roi): |
| out.append(box) |
| continue |
| if check_fire and (not self._passes_fire_color(roi)): |
| continue |
| if check_ext and (not self._passes_fire_ext_red_color(roi)): |
| continue |
| out.append(box) |
| return out |
|
|
| @staticmethod |
| def _build_results(boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray) -> list[BoundingBox]: |
| results: list[BoundingBox] = [] |
| for box, conf, cls_id in zip(boxes, scores, cls_ids): |
| x1, y1, x2, y2 = box.tolist() |
| if x2 <= x1 or y2 <= y1: |
| continue |
| results.append(BoundingBox(x1=int(math.floor(x1)), y1=int(math.floor(y1)), x2=int(math.ceil(x2)), y2=int(math.ceil(y2)), cls_id=int(cls_id), conf=float(conf))) |
| return results |
|
|
| @staticmethod |
| def _empty_raw() -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| return (np.empty((0, 4), dtype=np.float32), np.empty((0,), dtype=np.float32), np.empty((0,), dtype=np.int32)) |
|
|
| @staticmethod |
| def _iomin(a: np.ndarray, b: np.ndarray) -> float: |
| ix1 = max(float(a[0]), float(b[0])) |
| iy1 = max(float(a[1]), float(b[1])) |
| ix2 = min(float(a[2]), float(b[2])) |
| iy2 = min(float(a[3]), float(b[3])) |
| inter = max(0.0, ix2 - ix1) * max(0.0, iy2 - iy1) |
| area_a = max(0.0, float(a[2] - a[0]) * float(a[3] - a[1])) |
| area_b = max(0.0, float(b[2] - b[0]) * float(b[3] - b[1])) |
| smaller = min(area_a, area_b) |
| return inter / (smaller + 1e-07) |
|
|
| def _refine_smoke_from_raw(self, finals: list[BoundingBox], raw_boxes: np.ndarray, raw_scores: np.ndarray, raw_cls: np.ndarray) -> list[BoundingBox]: |
| del raw_scores |
| if not finals or len(raw_boxes) == 0: |
| return finals |
| smoke_id = self.class_names.index('smoke') |
| raw_smoke = raw_cls == smoke_id |
| if not np.any(raw_smoke): |
| return finals |
| cand_boxes = raw_boxes[raw_smoke] |
| out: list[BoundingBox] = [] |
| thr = float(self.smoke_raw_refine_overlap) |
| for b in finals: |
| if b.cls_id != smoke_id: |
| out.append(b) |
| continue |
| final_xyxy = np.array([b.x1, b.y1, b.x2, b.y2], dtype=np.float32) |
| best_idx = -1 |
| best_area = None |
| for i, rb in enumerate(cand_boxes): |
| if self._iomin(final_xyxy, rb) < thr: |
| continue |
| area = max(0.0, float(rb[2] - rb[0]) * float(rb[3] - rb[1])) |
| if best_area is None or area < best_area: |
| best_area = area |
| best_idx = i |
| if best_idx < 0: |
| out.append(b) |
| continue |
| rb = cand_boxes[best_idx] |
| out.append(BoundingBox(x1=int(math.floor(rb[0])), y1=int(math.floor(rb[1])), x2=int(math.ceil(rb[2])), y2=int(math.ceil(rb[3])), cls_id=b.cls_id, conf=b.conf)) |
| return out |
|
|
| def _rescale_class_boxes(self, finals: list[BoundingBox], orig_size: tuple[int, int]) -> list[BoundingBox]: |
| if not finals: |
| return finals |
| img_w, img_h = orig_size |
| fire_id = self.class_names.index('fire') |
| smoke_id = self.class_names.index('smoke') |
| ext_id = self.class_names.index('fire extinguisher') |
| out: list[BoundingBox] = [] |
| for b in finals: |
| x1, y1, x2, y2 = (float(b.x1), float(b.y1), float(b.x2), float(b.y2)) |
| w = max(0.0, x2 - x1) |
| h = max(0.0, y2 - y1) |
| if w <= 0.0 or h <= 0.0: |
| continue |
| if b.cls_id in (smoke_id, ext_id): |
| scale = float(self.smoke_ext_shrink) |
| nw, nh = (w * scale, h * scale) |
| cx = 0.5 * (x1 + x2) |
| nx1 = cx - 0.5 * nw |
| nx2 = cx + 0.5 * nw |
| ny2 = y2 |
| ny1 = ny2 - nh |
| elif b.cls_id == fire_id: |
| scale = float(self.fire_expand) |
| nw, nh = (w * scale, h * scale) |
| cx = 0.5 * (x1 + x2) |
| cy = 0.5 * (y1 + y2) |
| nx1 = cx - 0.5 * nw |
| nx2 = cx + 0.5 * nw |
| ny1 = cy - 0.5 * nh |
| ny2 = cy + 0.5 * nh |
| else: |
| out.append(b) |
| continue |
| nx1 = max(0.0, min(float(img_w), nx1)) |
| nx2 = max(0.0, min(float(img_w), nx2)) |
| ny1 = max(0.0, min(float(img_h), ny1)) |
| ny2 = max(0.0, min(float(img_h), ny2)) |
| if nx2 <= nx1 or ny2 <= ny1: |
| continue |
| out.append(BoundingBox(x1=int(math.floor(nx1)), y1=int(math.floor(ny1)), x2=int(math.ceil(nx2)), y2=int(math.ceil(ny2)), cls_id=b.cls_id, conf=b.conf)) |
| return out |
|
|
| def _apply_extra_post(self, finals: list[BoundingBox], raw_boxes: np.ndarray, raw_scores: np.ndarray, raw_cls: np.ndarray, orig_size: tuple[int, int]) -> list[BoundingBox]: |
| finals = self._refine_smoke_from_raw(finals, raw_boxes, raw_scores, raw_cls) |
| return self._rescale_class_boxes(finals, orig_size) |
|
|
| def _candidates_final_dets(self, preds: np.ndarray, ratio: float, pad: tuple[float, float], orig_size: tuple[int, int]) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| if preds.ndim == 3 and preds.shape[0] == 1: |
| preds = preds[0] |
| if preds.ndim != 2 or preds.shape[1] < 6: |
| raise ValueError(f'Unexpected ONNX final-det output shape: {preds.shape}') |
| boxes = preds[:, :4].astype(np.float32) |
| scores = preds[:, 4].astype(np.float32) |
| cls_ids = preds[:, 5].astype(np.int32) |
| cls_ids = self.cls_remap[cls_ids] |
| keep = self._conf_filter_mask(scores, cls_ids) |
| boxes = boxes[keep] |
| scores = scores[keep] |
| cls_ids = cls_ids[keep] |
| if len(boxes) == 0: |
| return self._empty_raw() |
| pad_w, pad_h = pad |
| boxes[:, [0, 2]] -= pad_w |
| boxes[:, [1, 3]] -= pad_h |
| boxes /= ratio |
| boxes = self._clip_boxes(boxes, orig_size) |
| return (boxes, scores, cls_ids) |
|
|
| def _candidates_raw_yolo(self, preds: np.ndarray, ratio: float, pad: tuple[float, float], orig_size: tuple[int, int]) -> tuple[np.ndarray, np.ndarray, np.ndarray]: |
| if preds.ndim != 3 or preds.shape[0] != 1: |
| raise ValueError(f'Unexpected raw ONNX output shape: {preds.shape}') |
| preds = preds[0] |
| if preds.shape[0] <= 16 and preds.shape[1] > preds.shape[0]: |
| preds = preds.T |
| if preds.ndim != 2 or preds.shape[1] < 5: |
| raise ValueError(f'Unexpected raw output shape: {preds.shape}') |
| boxes_xywh = preds[:, :4].astype(np.float32) |
| cls_part = preds[:, 4:].astype(np.float32) |
| if cls_part.shape[1] == 1: |
| scores = cls_part[:, 0] |
| cls_ids = np.zeros(len(scores), dtype=np.int32) |
| else: |
| cls_ids = np.argmax(cls_part, axis=1).astype(np.int32) |
| scores = cls_part[np.arange(len(cls_part)), cls_ids] |
| cls_ids = self.cls_remap[cls_ids] |
| keep = self._conf_filter_mask(scores, cls_ids) |
| boxes_xywh = boxes_xywh[keep] |
| scores = scores[keep] |
| cls_ids = cls_ids[keep] |
| if len(boxes_xywh) == 0: |
| return self._empty_raw() |
| boxes = self._xywh_to_xyxy(boxes_xywh) |
| pad_w, pad_h = pad |
| boxes[:, [0, 2]] -= pad_w |
| boxes[:, [1, 3]] -= pad_h |
| boxes /= ratio |
| boxes = self._clip_boxes(boxes, orig_size) |
| return (boxes, scores, cls_ids) |
|
|
| def _pipeline_from_candidates(self, boxes: np.ndarray, scores: np.ndarray, cls_ids: np.ndarray, orig_size: tuple[int, int]) -> list[BoundingBox]: |
| if len(boxes) == 0: |
| return [] |
| boxes, scores, cls_ids = self._filter_sane_boxes(boxes, scores, cls_ids, orig_size) |
| if len(boxes) == 0: |
| return [] |
| boxes, scores, cls_ids = self._per_view_pipeline(boxes, scores, cls_ids) |
| return self._build_results(boxes, scores, cls_ids) |
|
|
| def _decode_final_dets(self, preds: np.ndarray, ratio: float, pad: tuple[float, float], orig_size: tuple[int, int]) -> tuple[list[BoundingBox], tuple[np.ndarray, np.ndarray, np.ndarray]]: |
| raw = self._candidates_final_dets(preds, ratio, pad, orig_size) |
| return (self._pipeline_from_candidates(*raw, orig_size), raw) |
|
|
| def _decode_raw_yolo(self, preds: np.ndarray, ratio: float, pad: tuple[float, float], orig_size: tuple[int, int]) -> tuple[list[BoundingBox], tuple[np.ndarray, np.ndarray, np.ndarray]]: |
| raw = self._candidates_raw_yolo(preds, ratio, pad, orig_size) |
| return (self._pipeline_from_candidates(*raw, orig_size), raw) |
|
|
| def _postprocess(self, output: np.ndarray, ratio: float, pad: tuple[float, float], orig_size: tuple[int, int]) -> tuple[list[BoundingBox], tuple[np.ndarray, np.ndarray, np.ndarray]]: |
| if output.ndim == 2 and output.shape[1] >= 6: |
| return self._decode_final_dets(output, ratio, pad, orig_size) |
| if output.ndim == 3 and output.shape[0] == 1 and (output.shape[2] == 6): |
| return self._decode_final_dets(output, ratio, pad, orig_size) |
| return self._decode_raw_yolo(output, ratio, pad, orig_size) |
|
|
| def _predict_single(self, image: np.ndarray) -> tuple[list[BoundingBox], tuple[np.ndarray, np.ndarray, np.ndarray]]: |
| if image is None: |
| raise ValueError('Input image is None') |
| if not isinstance(image, np.ndarray): |
| raise TypeError(f'Input is not numpy array: {type(image)}') |
| if image.ndim != 3: |
| raise ValueError(f'Expected HWC image, got shape={image.shape}') |
| if image.shape[0] <= 0 or image.shape[1] <= 0: |
| raise ValueError(f'Invalid image shape={image.shape}') |
| if image.shape[2] != 3: |
| raise ValueError(f'Expected 3 channels, got shape={image.shape}') |
| if image.dtype != np.uint8: |
| image = image.astype(np.uint8) |
| input_tensor, ratio, pad, orig_size = self._preprocess(image) |
| expected = (1, 3, self.input_height, self.input_width) |
| if input_tensor.shape != expected: |
| raise ValueError(f'Bad input tensor shape={input_tensor.shape}, expected={expected}') |
| outputs = self.session.run(self.output_names, {self.input_name: input_tensor}) |
| return self._postprocess(outputs[0], ratio, pad, orig_size) |
|
|
| def _predict_tta(self, image: np.ndarray) -> tuple[list[BoundingBox], tuple[np.ndarray, np.ndarray, np.ndarray]]: |
| boxes_orig, raw_orig = self._predict_single(image) |
| flipped = cv2.flip(image, 1) |
| boxes_flip, raw_flip = self._predict_single(flipped) |
| w = image.shape[1] |
| boxes_flip = [BoundingBox(x1=w - b.x2, y1=b.y1, x2=w - b.x1, y2=b.y2, cls_id=b.cls_id, conf=b.conf) for b in boxes_flip] |
| raw_boxes_o, raw_scores_o, raw_cls_o = raw_orig |
| raw_boxes_f, raw_scores_f, raw_cls_f = raw_flip |
| if len(raw_boxes_f) > 0: |
| mapped_f = raw_boxes_f.copy() |
| mapped_f[:, 0] = w - raw_boxes_f[:, 2] |
| mapped_f[:, 2] = w - raw_boxes_f[:, 0] |
| mapped_f[:, 1] = raw_boxes_f[:, 1] |
| mapped_f[:, 3] = raw_boxes_f[:, 3] |
| else: |
| mapped_f = raw_boxes_f |
| if len(raw_boxes_o) == 0 and len(mapped_f) == 0: |
| raw_all = self._empty_raw() |
| elif len(raw_boxes_o) == 0: |
| raw_all = (mapped_f, raw_scores_f, raw_cls_f) |
| elif len(mapped_f) == 0: |
| raw_all = (raw_boxes_o, raw_scores_o, raw_cls_o) |
| else: |
| raw_all = (np.concatenate([raw_boxes_o, mapped_f], axis=0), np.concatenate([raw_scores_o, raw_scores_f], axis=0), np.concatenate([raw_cls_o, raw_cls_f], axis=0)) |
| all_boxes = boxes_orig + boxes_flip |
| if not all_boxes: |
| return ([], raw_all) |
| coords = np.array([[b.x1, b.y1, b.x2, b.y2] for b in all_boxes], dtype=np.float32) |
| scores = np.array([b.conf for b in all_boxes], dtype=np.float32) |
| cls_ids = np.array([b.cls_id for b in all_boxes], dtype=np.int32) |
| view_ids = np.array([0] * len(boxes_orig) + [1] * len(boxes_flip), dtype=np.int32) |
| hard_keep = self._per_class_hard_nms(coords, scores, cls_ids, self.iou_thres) |
| if len(hard_keep) == 0: |
| return ([], raw_all) |
| if len(hard_keep) > self.max_det: |
| top = np.argsort(-scores[hard_keep])[:self.max_det] |
| hard_keep = hard_keep[top] |
| boosted = self._max_score_per_cluster(coords[hard_keep], cls_ids[hard_keep], coords, scores, cls_ids, self.iou_thres) |
| kept_coords = coords[hard_keep] |
| kept_cls = cls_ids[hard_keep] |
| if self.use_tta_view_filter and self.tta_view_filter_max_conf > 0.0 and (len(kept_coords) > 0): |
| corrob = self._views_corroborated(kept_coords, kept_cls, coords, cls_ids, view_ids, self.tta_view_iou_thresh) |
| keep = ~((boosted <= self.tta_view_filter_max_conf) & ~corrob) |
| kept_coords = kept_coords[keep] |
| boosted = boosted[keep] |
| kept_cls = kept_cls[keep] |
| if len(kept_coords) > 1: |
| kept_coords, boosted, kept_cls = self._cross_class_dedup_op(kept_coords, boosted, kept_cls, self.cross_iou_thresh) |
| if len(kept_coords) > 1: |
| kept_coords, boosted, kept_cls = self._merge_same_class_boxes(kept_coords, boosted, kept_cls) |
| finals = [BoundingBox(x1=int(math.floor(kept_coords[j, 0])), y1=int(math.floor(kept_coords[j, 1])), x2=int(math.ceil(kept_coords[j, 2])), y2=int(math.ceil(kept_coords[j, 3])), cls_id=int(kept_cls[j]), conf=float(boosted[j])) for j in range(len(kept_coords))] |
| return (finals, raw_all) |
|
|
| def predict_batch(self, batch_images: list[ndarray], offset: int, n_keypoints: int) -> list[TVFrameResult]: |
| results: list[TVFrameResult] = [] |
| for frame_number_in_batch, image in enumerate(batch_images): |
| try: |
| if self.use_tta: |
| boxes, raw = self._predict_tta(image) |
| else: |
| boxes, raw = self._predict_single(image) |
| if isinstance(image, np.ndarray) and image.ndim == 3: |
| boxes = self._filter_low_conf_by_color(image, boxes) |
| boxes = self._remove_edge_low_conf(boxes, (image.shape[1], image.shape[0])) |
| boxes = self._apply_extra_post(boxes, raw[0], raw[1], raw[2], (image.shape[1], image.shape[0])) |
| else: |
| boxes = self._apply_extra_post(boxes, raw[0], raw[1], raw[2], (0, 0)) |
| except Exception as e: |
| print(f'⚠️ Inference failed for frame {offset + frame_number_in_batch}: {e}') |
| boxes = [] |
| results.append(TVFrameResult(frame_id=offset + frame_number_in_batch, boxes=boxes, keypoints=[(0, 0) for _ in range(max(0, int(n_keypoints)))])) |
| return results |
|
|
