|
|
|
|
|
import json |
|
|
import os |
|
|
import sys |
|
|
from typing import Any, Dict, List, Tuple |
|
|
|
|
|
import cv2 |
|
|
import numpy as np |
|
|
import onnxruntime as ort |
|
|
from PIL import Image |
|
|
|
|
|
|
|
|
def read_json(path: str) -> Dict[str, Any]: |
|
|
with open(path, "r", encoding="utf-8") as f: |
|
|
return json.load(f) |
|
|
|
|
|
|
|
|
def read_text_lines(path: str) -> List[str]: |
|
|
with open(path, "r", encoding="utf-8") as f: |
|
|
return [line.strip() for line in f.readlines() if line.strip()] |
|
|
|
|
|
|
|
|
def load_environment(data_dir: str) -> Dict[str, Any]: |
|
|
env_path = os.path.join(data_dir, "environment.json") |
|
|
if not os.path.exists(env_path): |
|
|
return {} |
|
|
env = read_json(env_path) |
|
|
preproc = env.get("PREPROCESSING") |
|
|
if isinstance(preproc, str): |
|
|
try: |
|
|
env["PREPROCESSING"] = json.loads(preproc) |
|
|
except json.JSONDecodeError: |
|
|
env["PREPROCESSING"] = {} |
|
|
return env |
|
|
|
|
|
|
|
|
def load_class_names(data_dir: str, environment: Dict[str, Any]) -> List[str]: |
|
|
class_path = os.path.join(data_dir, "class_names.txt") |
|
|
if os.path.exists(class_path): |
|
|
return read_text_lines(class_path) |
|
|
class_map = environment.get("CLASS_MAP") |
|
|
if isinstance(class_map, dict): |
|
|
class_names = [] |
|
|
for i in range(len(class_map.keys())): |
|
|
class_names.append(class_map[str(i)]) |
|
|
return class_names |
|
|
return [] |
|
|
|
|
|
|
|
|
def load_keypoints_metadata(data_dir: str) -> List[Dict[str, Any]]: |
|
|
meta_path = os.path.join(data_dir, "keypoints_metadata.json") |
|
|
if not os.path.exists(meta_path): |
|
|
return [] |
|
|
return read_json(meta_path) |
|
|
|
|
|
|
|
|
def load_image(value: Any) -> Tuple[np.ndarray, bool]: |
|
|
if isinstance(value, np.ndarray): |
|
|
return value, True |
|
|
if isinstance(value, Image.Image): |
|
|
return np.asarray(value.convert("RGB")), False |
|
|
if isinstance(value, (bytes, bytearray)): |
|
|
image = cv2.imdecode(np.frombuffer(value, np.uint8), cv2.IMREAD_COLOR) |
|
|
return image, True |
|
|
if isinstance(value, str): |
|
|
image = cv2.imread(value, cv2.IMREAD_COLOR) |
|
|
if image is None: |
|
|
raise ValueError(f"Could not read image: {value}") |
|
|
return image, True |
|
|
raise ValueError(f"Unsupported image input type: {type(value)}") |
|
|
|
|
|
|
|
|
def static_crop_should_be_applied(preprocessing_config: dict) -> bool: |
|
|
cfg = preprocessing_config.get("static-crop") |
|
|
return bool(cfg and cfg.get("enabled")) |
|
|
|
|
|
|
|
|
def take_static_crop(image: np.ndarray, crop_parameters: Dict[str, int]) -> np.ndarray: |
|
|
height, width = image.shape[:2] |
|
|
x_min = int(crop_parameters["x_min"] / 100 * width) |
|
|
y_min = int(crop_parameters["y_min"] / 100 * height) |
|
|
x_max = int(crop_parameters["x_max"] / 100 * width) |
|
|
y_max = int(crop_parameters["y_max"] / 100 * height) |
|
|
return image[y_min:y_max, x_min:x_max, :] |
|
|
|
|
|
|
|
|
def apply_grayscale_conversion(image: np.ndarray) -> np.ndarray: |
|
|
image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) |
|
|
return cv2.cvtColor(image, cv2.COLOR_GRAY2BGR) |
|
|
|
|
|
|
|
|
def apply_contrast_stretching(image: np.ndarray) -> np.ndarray: |
|
|
p2, p98 = np.percentile(image, (2, 98)) |
|
|
image = np.clip(image, p2, p98) |
|
|
if p98 - p2 > 0: |
|
|
image = (image - p2) * (255.0 / (p98 - p2)) |
|
|
return image.astype(np.uint8) |
|
|
|
|
|
|
|
|
def apply_histogram_equalisation(image: np.ndarray) -> np.ndarray: |
|
|
image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) |
|
|
image = cv2.equalizeHist(image) |
|
|
return cv2.cvtColor(image, cv2.COLOR_GRAY2BGR) |
|
|
|
|
|
|
|
|
def apply_adaptive_equalisation(image: np.ndarray) -> np.ndarray: |
|
|
image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) |
|
|
clahe = cv2.createCLAHE(clipLimit=0.03, tileGridSize=(8, 8)) |
|
|
image = clahe.apply(image) |
|
|
return cv2.cvtColor(image, cv2.COLOR_GRAY2BGR) |
|
|
|
|
|
|
|
|
def apply_preproc(image: np.ndarray, preproc: Dict[str, Any]) -> Tuple[np.ndarray, Tuple[int, int]]: |
|
|
h, w = image.shape[:2] |
|
|
img_dims = (h, w) |
|
|
if static_crop_should_be_applied(preproc): |
|
|
image = take_static_crop(image, preproc["static-crop"]) |
|
|
if preproc.get("contrast", {}).get("enabled"): |
|
|
ctype = preproc.get("contrast", {}).get("type") |
|
|
if ctype == "Contrast Stretching": |
|
|
image = apply_contrast_stretching(image) |
|
|
elif ctype == "Histogram Equalization": |
|
|
image = apply_histogram_equalisation(image) |
|
|
elif ctype == "Adaptive Equalization": |
|
|
image = apply_adaptive_equalisation(image) |
|
|
if preproc.get("grayscale", {}).get("enabled"): |
|
|
image = apply_grayscale_conversion(image) |
|
|
return image, img_dims |
|
|
|
|
|
|
|
|
def resize_image_keeping_aspect_ratio(image: np.ndarray, desired_size: Tuple[int, int]) -> np.ndarray: |
|
|
height, width = image.shape[:2] |
|
|
ratio = min(desired_size[1] / height, desired_size[0] / width) |
|
|
new_width = int(width * ratio) |
|
|
new_height = int(height * ratio) |
|
|
return cv2.resize(image, (new_width, new_height)) |
|
|
|
|
|
|
|
|
def letterbox_image(image: np.ndarray, desired_size: Tuple[int, int], color: Tuple[int, int, int]) -> np.ndarray: |
|
|
resized = resize_image_keeping_aspect_ratio(image, desired_size) |
|
|
new_height, new_width = resized.shape[:2] |
|
|
top = (desired_size[1] - new_height) // 2 |
|
|
bottom = desired_size[1] - new_height - top |
|
|
left = (desired_size[0] - new_width) // 2 |
|
|
right = desired_size[0] - new_width - left |
|
|
return cv2.copyMakeBorder(resized, top, bottom, left, right, cv2.BORDER_CONSTANT, value=color) |
|
|
|
|
|
|
|
|
def get_resize_method(preproc: Dict[str, Any]) -> str: |
|
|
resize = preproc.get("resize") |
|
|
if not resize: |
|
|
return "Stretch to" |
|
|
method = resize.get("format", "Stretch to") |
|
|
if method in {"Fit (reflect edges) in", "Fit within", "Fill (with center crop) in"}: |
|
|
return "Fit (black edges) in" |
|
|
if method not in {"Stretch to", "Fit (black edges) in", "Fit (white edges) in", "Fit (grey edges) in"}: |
|
|
return "Stretch to" |
|
|
return method |
|
|
|
|
|
|
|
|
def preprocess_image(image: Any, preproc: Dict[str, Any], input_hw: Tuple[int, int]) -> Tuple[np.ndarray, Tuple[int, int]]: |
|
|
np_image, is_bgr = load_image(image) |
|
|
processed, img_dims = apply_preproc(np_image, preproc) |
|
|
resize_method = get_resize_method(preproc) |
|
|
h, w = input_hw |
|
|
if resize_method == "Stretch to": |
|
|
resized = cv2.resize(processed, (w, h)) |
|
|
elif resize_method == "Fit (white edges) in": |
|
|
resized = letterbox_image(processed, (w, h), (255, 255, 255)) |
|
|
elif resize_method == "Fit (grey edges) in": |
|
|
resized = letterbox_image(processed, (w, h), (114, 114, 114)) |
|
|
else: |
|
|
resized = letterbox_image(processed, (w, h), (0, 0, 0)) |
|
|
if is_bgr: |
|
|
resized = cv2.cvtColor(resized, cv2.COLOR_BGR2RGB) |
|
|
img_in = resized.astype(np.float32) |
|
|
img_in = np.transpose(img_in, (2, 0, 1)) |
|
|
img_in = np.expand_dims(img_in, axis=0) |
|
|
return img_in, img_dims |
|
|
|
|
|
|
|
|
def sigmoid(x: np.ndarray) -> np.ndarray: |
|
|
return 1.0 / (1.0 + np.exp(-x)) |
|
|
|
|
|
|
|
|
def non_max_suppression_fast(boxes: np.ndarray, overlap_thresh: float) -> List[np.ndarray]: |
|
|
if len(boxes) == 0: |
|
|
return [] |
|
|
if boxes.dtype.kind == "i": |
|
|
boxes = boxes.astype("float") |
|
|
pick = [] |
|
|
x1 = boxes[:, 0] |
|
|
y1 = boxes[:, 1] |
|
|
x2 = boxes[:, 2] |
|
|
y2 = boxes[:, 3] |
|
|
conf = boxes[:, 4] |
|
|
area = (x2 - x1 + 1) * (y2 - y1 + 1) |
|
|
idxs = np.argsort(conf) |
|
|
while len(idxs) > 0: |
|
|
last = len(idxs) - 1 |
|
|
i = idxs[last] |
|
|
pick.append(i) |
|
|
xx1 = np.maximum(x1[i], x1[idxs[:last]]) |
|
|
yy1 = np.maximum(y1[i], y1[idxs[:last]]) |
|
|
xx2 = np.minimum(x2[i], x2[idxs[:last]]) |
|
|
yy2 = np.minimum(y2[i], y2[idxs[:last]]) |
|
|
w = np.maximum(0, xx2 - xx1 + 1) |
|
|
h = np.maximum(0, yy2 - yy1 + 1) |
|
|
overlap = (w * h) / area[idxs[:last]] |
|
|
idxs = np.delete(idxs, np.concatenate(([last], np.where(overlap > overlap_thresh)[0]))) |
|
|
return boxes[pick].astype("float") |
|
|
|
|
|
|
|
|
def w_np_non_max_suppression( |
|
|
prediction: np.ndarray, |
|
|
conf_thresh: float = 0.25, |
|
|
iou_thresh: float = 0.45, |
|
|
class_agnostic: bool = False, |
|
|
max_detections: int = 300, |
|
|
max_candidate_detections: int = 3000, |
|
|
num_masks: int = 0, |
|
|
box_format: str = "xywh", |
|
|
): |
|
|
num_classes = prediction.shape[2] - 5 - num_masks |
|
|
if box_format == "xywh": |
|
|
pred_view = prediction[:, :, :4] |
|
|
x1 = pred_view[:, :, 0] - pred_view[:, :, 2] / 2 |
|
|
y1 = pred_view[:, :, 1] - pred_view[:, :, 3] / 2 |
|
|
x2 = pred_view[:, :, 0] + pred_view[:, :, 2] / 2 |
|
|
y2 = pred_view[:, :, 1] + pred_view[:, :, 3] / 2 |
|
|
pred_view[:, :, 0] = x1 |
|
|
pred_view[:, :, 1] = y1 |
|
|
pred_view[:, :, 2] = x2 |
|
|
pred_view[:, :, 3] = y2 |
|
|
elif box_format != "xyxy": |
|
|
raise ValueError(f"box_format must be 'xywh' or 'xyxy', got {box_format}") |
|
|
|
|
|
batch_predictions = [] |
|
|
for np_image_pred in prediction: |
|
|
np_conf_mask = np_image_pred[:, 4] >= conf_thresh |
|
|
if not np.any(np_conf_mask): |
|
|
batch_predictions.append([]) |
|
|
continue |
|
|
np_image_pred = np_image_pred[np_conf_mask] |
|
|
if np_image_pred.shape[0] == 0: |
|
|
batch_predictions.append([]) |
|
|
continue |
|
|
cls_confs = np_image_pred[:, 5 : num_classes + 5] |
|
|
if cls_confs.shape[1] == 0: |
|
|
batch_predictions.append([]) |
|
|
continue |
|
|
np_class_conf = np.max(cls_confs, axis=1, keepdims=True) |
|
|
np_class_pred = np.argmax(cls_confs, axis=1, keepdims=True) |
|
|
if num_masks > 0: |
|
|
np_mask_pred = np_image_pred[:, 5 + num_classes :] |
|
|
np_detections = np.concatenate( |
|
|
[ |
|
|
np_image_pred[:, :5], |
|
|
np_class_conf, |
|
|
np_class_pred.astype(np.float32), |
|
|
np_mask_pred, |
|
|
], |
|
|
axis=1, |
|
|
) |
|
|
else: |
|
|
np_detections = np.concatenate( |
|
|
[np_image_pred[:, :5], np_class_conf, np_class_pred.astype(np.float32)], |
|
|
axis=1, |
|
|
) |
|
|
filtered_predictions = [] |
|
|
if class_agnostic: |
|
|
sorted_indices = np.argsort(-np_detections[:, 4]) |
|
|
np_detections_sorted = np_detections[sorted_indices] |
|
|
filtered_predictions.extend(non_max_suppression_fast(np_detections_sorted, iou_thresh)) |
|
|
else: |
|
|
np_unique_labels = np.unique(np_class_pred) |
|
|
for c in np_unique_labels: |
|
|
class_mask = np.atleast_1d(np_class_pred.squeeze() == c) |
|
|
np_detections_class = np_detections[class_mask] |
|
|
if np_detections_class.shape[0] == 0: |
|
|
continue |
|
|
sorted_indices = np.argsort(-np_detections_class[:, 4]) |
|
|
np_detections_sorted = np_detections_class[sorted_indices] |
|
|
filtered_predictions.extend(non_max_suppression_fast(np_detections_sorted, iou_thresh)) |
|
|
|
|
|
if filtered_predictions: |
|
|
filtered_np = np.array(filtered_predictions) |
|
|
idx = np.argsort(-filtered_np[:, 4]) |
|
|
filtered_np = filtered_np[idx] |
|
|
if len(filtered_np) > max_detections: |
|
|
filtered_np = filtered_np[:max_detections] |
|
|
batch_predictions.append(list(filtered_np)) |
|
|
else: |
|
|
batch_predictions.append([]) |
|
|
return batch_predictions |
|
|
|
|
|
|
|
|
def get_static_crop_dimensions(orig_shape: Tuple[int, int], preproc: dict) -> Tuple[Tuple[int, int], Tuple[int, int]]: |
|
|
if not static_crop_should_be_applied(preproc): |
|
|
return (0, 0), orig_shape |
|
|
crop = preproc["static-crop"] |
|
|
x_min, y_min, x_max, y_max = (crop[k] / 100.0 for k in ["x_min", "y_min", "x_max", "y_max"]) |
|
|
crop_shift_x, crop_shift_y = (round(x_min * orig_shape[1]), round(y_min * orig_shape[0])) |
|
|
cropped_percent_x = x_max - x_min |
|
|
cropped_percent_y = y_max - y_min |
|
|
new_shape = (round(orig_shape[0] * cropped_percent_y), round(orig_shape[1] * cropped_percent_x)) |
|
|
return (crop_shift_x, crop_shift_y), new_shape |
|
|
|
|
|
|
|
|
def post_process_bboxes( |
|
|
predictions: List[List[List[float]]], |
|
|
infer_shape: Tuple[int, int], |
|
|
img_dims: List[Tuple[int, int]], |
|
|
preproc: dict, |
|
|
resize_method: str, |
|
|
) -> List[List[List[float]]]: |
|
|
scaled_predictions = [] |
|
|
for i, batch_predictions in enumerate(predictions): |
|
|
if len(batch_predictions) == 0: |
|
|
scaled_predictions.append([]) |
|
|
continue |
|
|
np_batch_predictions = np.array(batch_predictions) |
|
|
predicted_bboxes = np_batch_predictions[:, :4] |
|
|
(crop_shift_x, crop_shift_y), origin_shape = get_static_crop_dimensions(img_dims[i], preproc) |
|
|
if resize_method == "Stretch to": |
|
|
scale_height = origin_shape[0] / infer_shape[0] |
|
|
scale_width = origin_shape[1] / infer_shape[1] |
|
|
predicted_bboxes[:, 0] *= scale_width |
|
|
predicted_bboxes[:, 2] *= scale_width |
|
|
predicted_bboxes[:, 1] *= scale_height |
|
|
predicted_bboxes[:, 3] *= scale_height |
|
|
else: |
|
|
scale = min(infer_shape[0] / origin_shape[0], infer_shape[1] / origin_shape[1]) |
|
|
inter_h = round(origin_shape[0] * scale) |
|
|
inter_w = round(origin_shape[1] * scale) |
|
|
pad_x = (infer_shape[1] - inter_w) / 2 |
|
|
pad_y = (infer_shape[0] - inter_h) / 2 |
|
|
predicted_bboxes[:, 0] -= pad_x |
|
|
predicted_bboxes[:, 2] -= pad_x |
|
|
predicted_bboxes[:, 1] -= pad_y |
|
|
predicted_bboxes[:, 3] -= pad_y |
|
|
predicted_bboxes /= scale |
|
|
predicted_bboxes[:, 0] = np.round(np.clip(predicted_bboxes[:, 0], 0, origin_shape[1])) |
|
|
predicted_bboxes[:, 2] = np.round(np.clip(predicted_bboxes[:, 2], 0, origin_shape[1])) |
|
|
predicted_bboxes[:, 1] = np.round(np.clip(predicted_bboxes[:, 1], 0, origin_shape[0])) |
|
|
predicted_bboxes[:, 3] = np.round(np.clip(predicted_bboxes[:, 3], 0, origin_shape[0])) |
|
|
predicted_bboxes[:, 0] += crop_shift_x |
|
|
predicted_bboxes[:, 2] += crop_shift_x |
|
|
predicted_bboxes[:, 1] += crop_shift_y |
|
|
predicted_bboxes[:, 3] += crop_shift_y |
|
|
np_batch_predictions[:, :4] = predicted_bboxes |
|
|
scaled_predictions.append(np_batch_predictions.tolist()) |
|
|
return scaled_predictions |
|
|
|
|
|
|
|
|
def post_process_keypoints( |
|
|
predictions: List[List[List[float]]], |
|
|
keypoints_start_index: int, |
|
|
infer_shape: Tuple[int, int], |
|
|
img_dims: List[Tuple[int, int]], |
|
|
preproc: dict, |
|
|
resize_method: str, |
|
|
) -> List[List[List[float]]]: |
|
|
scaled_predictions = [] |
|
|
for i, batch_predictions in enumerate(predictions): |
|
|
if len(batch_predictions) == 0: |
|
|
scaled_predictions.append([]) |
|
|
continue |
|
|
np_batch_predictions = np.array(batch_predictions) |
|
|
keypoints = np_batch_predictions[:, keypoints_start_index:] |
|
|
(crop_shift_x, crop_shift_y), origin_shape = get_static_crop_dimensions(img_dims[i], preproc) |
|
|
if resize_method == "Stretch to": |
|
|
scale_width = origin_shape[1] / infer_shape[1] |
|
|
scale_height = origin_shape[0] / infer_shape[0] |
|
|
for k in range(keypoints.shape[1] // 3): |
|
|
keypoints[:, k * 3] *= scale_width |
|
|
keypoints[:, k * 3 + 1] *= scale_height |
|
|
else: |
|
|
scale = min(infer_shape[0] / origin_shape[0], infer_shape[1] / origin_shape[1]) |
|
|
inter_w = int(origin_shape[1] * scale) |
|
|
inter_h = int(origin_shape[0] * scale) |
|
|
pad_x = (infer_shape[1] - inter_w) / 2 |
|
|
pad_y = (infer_shape[0] - inter_h) / 2 |
|
|
for k in range(keypoints.shape[1] // 3): |
|
|
keypoints[:, k * 3] -= pad_x |
|
|
keypoints[:, k * 3] /= scale |
|
|
keypoints[:, k * 3 + 1] -= pad_y |
|
|
keypoints[:, k * 3 + 1] /= scale |
|
|
for k in range(keypoints.shape[1] // 3): |
|
|
keypoints[:, k * 3] = np.round(np.clip(keypoints[:, k * 3], 0, origin_shape[1])) |
|
|
keypoints[:, k * 3 + 1] = np.round(np.clip(keypoints[:, k * 3 + 1], 0, origin_shape[0])) |
|
|
keypoints[:, k * 3] += crop_shift_x |
|
|
keypoints[:, k * 3 + 1] += crop_shift_y |
|
|
np_batch_predictions[:, keypoints_start_index:] = keypoints |
|
|
scaled_predictions.append(np_batch_predictions.tolist()) |
|
|
return scaled_predictions |
|
|
|
|
|
|
|
|
def masks2poly(masks: np.ndarray) -> List[np.ndarray]: |
|
|
segments = [] |
|
|
for mask in masks: |
|
|
if mask.dtype == np.bool_: |
|
|
m_uint8 = mask |
|
|
if not m_uint8.flags.c_contiguous: |
|
|
m_uint8 = np.ascontiguousarray(m_uint8) |
|
|
m_uint8 = m_uint8.view(np.uint8) |
|
|
elif mask.dtype == np.uint8: |
|
|
m_uint8 = mask if mask.flags.c_contiguous else np.ascontiguousarray(mask) |
|
|
else: |
|
|
m_bool = mask > 0 |
|
|
if not m_bool.flags.c_contiguous: |
|
|
m_bool = np.ascontiguousarray(m_bool) |
|
|
m_uint8 = m_bool.view(np.uint8) |
|
|
if not np.any(m_uint8): |
|
|
segments.append(np.zeros((0, 2), dtype=np.float32)) |
|
|
continue |
|
|
contours = cv2.findContours(m_uint8, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)[0] |
|
|
if contours: |
|
|
contours = np.array(contours[np.array([len(x) for x in contours]).argmax()]).reshape(-1, 2) |
|
|
else: |
|
|
contours = np.zeros((0, 2)) |
|
|
segments.append(contours.astype("float32")) |
|
|
return segments |
|
|
|
|
|
|
|
|
def post_process_polygons( |
|
|
origin_shape: Tuple[int, int], |
|
|
polys: List[List[Tuple[float, float]]], |
|
|
infer_shape: Tuple[int, int], |
|
|
preproc: dict, |
|
|
resize_method: str, |
|
|
) -> List[List[Tuple[float, float]]]: |
|
|
(crop_shift_x, crop_shift_y), origin_shape = get_static_crop_dimensions(origin_shape, preproc) |
|
|
new_polys = [] |
|
|
if resize_method == "Stretch to": |
|
|
width_ratio = origin_shape[1] / infer_shape[1] |
|
|
height_ratio = origin_shape[0] / infer_shape[0] |
|
|
for poly in polys: |
|
|
new_polys.append([(p[0] * width_ratio, p[1] * height_ratio) for p in poly]) |
|
|
else: |
|
|
scale = min(infer_shape[0] / origin_shape[0], infer_shape[1] / origin_shape[1]) |
|
|
inter_w = int(origin_shape[1] * scale) |
|
|
inter_h = int(origin_shape[0] * scale) |
|
|
pad_x = (infer_shape[1] - inter_w) / 2 |
|
|
pad_y = (infer_shape[0] - inter_h) / 2 |
|
|
for poly in polys: |
|
|
new_polys.append([((p[0] - pad_x) / scale, (p[1] - pad_y) / scale) for p in poly]) |
|
|
shifted_polys = [] |
|
|
for poly in new_polys: |
|
|
shifted_polys.append([(p[0] + crop_shift_x, p[1] + crop_shift_y) for p in poly]) |
|
|
return shifted_polys |
|
|
|
|
|
|
|
|
def preprocess_segmentation_masks(protos: np.ndarray, masks_in: np.ndarray, shape: Tuple[int, int]) -> np.ndarray: |
|
|
c, mh, mw = protos.shape |
|
|
masks = protos.astype(np.float32) |
|
|
masks = masks.reshape((c, -1)) |
|
|
masks = masks_in @ masks |
|
|
masks = sigmoid(masks) |
|
|
masks = masks.reshape((-1, mh, mw)) |
|
|
gain = min(mh / shape[0], mw / shape[1]) |
|
|
pad = (mw - shape[1] * gain) / 2, (mh - shape[0] * gain) / 2 |
|
|
top, left = int(pad[1]), int(pad[0]) |
|
|
bottom, right = int(mh - pad[1]), int(mw - pad[0]) |
|
|
return masks[:, top:bottom, left:right] |
|
|
|
|
|
|
|
|
def crop_mask(masks: np.ndarray, boxes: np.ndarray) -> np.ndarray: |
|
|
n, h, w = masks.shape |
|
|
x1, y1, x2, y2 = np.split(boxes[:, :, None], 4, 1) |
|
|
r = np.arange(w, dtype=x1.dtype)[None, None, :] |
|
|
c = np.arange(h, dtype=x1.dtype)[None, :, None] |
|
|
masks = masks * ((r >= x1) * (r < x2) * (c >= y1) * (c < y2)) |
|
|
return masks |
|
|
|
|
|
|
|
|
def process_mask_accurate(protos: np.ndarray, masks_in: np.ndarray, bboxes: np.ndarray, shape: Tuple[int, int]) -> np.ndarray: |
|
|
masks = preprocess_segmentation_masks(protos, masks_in, shape) |
|
|
if len(masks.shape) == 2: |
|
|
masks = np.expand_dims(masks, axis=0) |
|
|
masks = masks.transpose((1, 2, 0)) |
|
|
masks = cv2.resize(masks, (shape[1], shape[0]), cv2.INTER_LINEAR) |
|
|
if len(masks.shape) == 2: |
|
|
masks = np.expand_dims(masks, axis=2) |
|
|
masks = masks.transpose((2, 0, 1)) |
|
|
masks = crop_mask(masks, bboxes) |
|
|
masks[masks < 0.5] = 0 |
|
|
return masks |
|
|
|
|
|
|
|
|
def process_mask_tradeoff(protos: np.ndarray, masks_in: np.ndarray, bboxes: np.ndarray, shape: Tuple[int, int], tradeoff_factor: float) -> np.ndarray: |
|
|
c, mh, mw = protos.shape |
|
|
masks = preprocess_segmentation_masks(protos, masks_in, shape) |
|
|
if len(masks.shape) == 2: |
|
|
masks = np.expand_dims(masks, axis=0) |
|
|
masks = masks.transpose((1, 2, 0)) |
|
|
ih, iw = shape |
|
|
h = int(mh * (1 - tradeoff_factor) + ih * tradeoff_factor) |
|
|
w = int(mw * (1 - tradeoff_factor) + iw * tradeoff_factor) |
|
|
if tradeoff_factor != 0: |
|
|
masks = cv2.resize(masks, (w, h), cv2.INTER_LINEAR) |
|
|
if len(masks.shape) == 2: |
|
|
masks = np.expand_dims(masks, axis=2) |
|
|
masks = masks.transpose((2, 0, 1)) |
|
|
c, mh, mw = masks.shape |
|
|
scale_x = mw / iw |
|
|
scale_y = mh / ih |
|
|
bboxes = bboxes.copy() |
|
|
bboxes[:, 0] *= scale_x |
|
|
bboxes[:, 2] *= scale_x |
|
|
bboxes[:, 1] *= scale_y |
|
|
bboxes[:, 3] *= scale_y |
|
|
masks = crop_mask(masks, bboxes) |
|
|
masks[masks < 0.5] = 0 |
|
|
return masks |
|
|
|
|
|
|
|
|
def process_mask_fast(protos: np.ndarray, masks_in: np.ndarray, bboxes: np.ndarray, shape: Tuple[int, int]) -> np.ndarray: |
|
|
ih, iw = shape |
|
|
c, mh, mw = protos.shape |
|
|
masks = preprocess_segmentation_masks(protos, masks_in, shape) |
|
|
scale_x = mw / iw |
|
|
scale_y = mh / ih |
|
|
bboxes = bboxes.copy() |
|
|
bboxes[:, 0] *= scale_x |
|
|
bboxes[:, 2] *= scale_x |
|
|
bboxes[:, 1] *= scale_y |
|
|
bboxes[:, 3] *= scale_y |
|
|
masks = crop_mask(masks, bboxes) |
|
|
masks[masks < 0.5] = 0 |
|
|
return masks |
|
|
|
|
|
|
|
|
def load_onnx_session(onnx_path: str, providers: List[str] = None) -> ort.InferenceSession: |
|
|
if providers is None: |
|
|
providers = ["CUDAExecutionProvider", "CPUExecutionProvider"] |
|
|
return ort.InferenceSession(onnx_path, providers=providers) |
|
|
|
|
|
|
|
|
def find_default_onnx(data_dir: str) -> str: |
|
|
candidates = [f for f in os.listdir(data_dir) if f.lower().endswith(".onnx")] |
|
|
candidates.sort() |
|
|
if not candidates: |
|
|
raise FileNotFoundError(f"No .onnx file found in {data_dir}") |
|
|
if len(candidates) > 1: |
|
|
|
|
|
for name in candidates: |
|
|
if name.lower() == "weights.onnx": |
|
|
return os.path.join(data_dir, name) |
|
|
return os.path.join(data_dir, candidates[0]) |
|
|
|
|
|
|
|
|
def get_input_hw(session: ort.InferenceSession, preproc: Dict[str, Any]) -> Tuple[int, int]: |
|
|
inputs = session.get_inputs()[0] |
|
|
shape = inputs.shape |
|
|
h, w = shape[2], shape[3] |
|
|
if isinstance(h, str) or isinstance(w, str) or h is None or w is None: |
|
|
resize = preproc.get("resize") if preproc else None |
|
|
if resize: |
|
|
h = int(resize.get("height", 640)) |
|
|
w = int(resize.get("width", 640)) |
|
|
else: |
|
|
h, w = 640, 640 |
|
|
return int(h), int(w) |
|
|
|
|
|
|
|
|
def build_meta(data_dir: str, session: ort.InferenceSession) -> Dict[str, Any]: |
|
|
environment = load_environment(data_dir) |
|
|
preproc = environment.get("PREPROCESSING") or {} |
|
|
class_names = load_class_names(data_dir, environment) |
|
|
resize_method = get_resize_method(preproc) |
|
|
input_hw = get_input_hw(session, preproc) |
|
|
keypoints_metadata = load_keypoints_metadata(data_dir) |
|
|
return { |
|
|
"environment": environment, |
|
|
"preproc": preproc, |
|
|
"class_names": class_names, |
|
|
"resize_method": resize_method, |
|
|
"input_hw": input_hw, |
|
|
"keypoints_metadata": keypoints_metadata, |
|
|
} |
|
|
|
|
|
|
|
|
def normalize_rgb(img_in: np.ndarray, means: List[float], stds: List[float]) -> np.ndarray: |
|
|
img_in = img_in.astype(np.float32) |
|
|
img_in /= 255.0 |
|
|
img_in[:, 0, :, :] = (img_in[:, 0, :, :] - means[0]) / stds[0] |
|
|
img_in[:, 1, :, :] = (img_in[:, 1, :, :] - means[1]) / stds[1] |
|
|
img_in[:, 2, :, :] = (img_in[:, 2, :, :] - means[2]) / stds[2] |
|
|
return img_in |
|
|
|
|
|
|
|
|
MODEL_TASK_TYPE = "object-detection" |
|
|
|
|
|
|
|
|
def preprocess_for_model(image: Any, meta: Dict[str, Any]) -> Tuple[np.ndarray, Tuple[int, int]]: |
|
|
img_in, img_dims = preprocess_image(image, meta["preproc"], meta["input_hw"]) |
|
|
img_in = img_in.astype(np.float32) |
|
|
img_in /= 255.0 |
|
|
return img_in, img_dims |
|
|
|
|
|
|
|
|
def pack_predictions(predictions: np.ndarray) -> np.ndarray: |
|
|
predictions = predictions.transpose(0, 2, 1) |
|
|
boxes = predictions[:, :, :4] |
|
|
class_confs = predictions[:, :, 4:] |
|
|
confs = np.expand_dims(np.max(class_confs, axis=2), axis=2) |
|
|
return np.concatenate([boxes, confs, class_confs], axis=2) |
|
|
|
|
|
|
|
|
def postprocess_predictions(predictions: np.ndarray, meta: Dict[str, Any], img_dims: List[Tuple[int, int]], |
|
|
confidence: float = 0.4, iou_threshold: float = 0.3, max_detections: int = 300): |
|
|
preds = w_np_non_max_suppression( |
|
|
predictions, |
|
|
conf_thresh=confidence, |
|
|
iou_thresh=iou_threshold, |
|
|
class_agnostic=False, |
|
|
max_detections=max_detections, |
|
|
box_format="xywh", |
|
|
) |
|
|
infer_shape = meta["input_hw"] |
|
|
preds = post_process_bboxes(preds, infer_shape, img_dims, meta["preproc"], meta["resize_method"]) |
|
|
class_names = meta["class_names"] |
|
|
results = [] |
|
|
for batch_preds in preds: |
|
|
batch_out = [] |
|
|
for pred in batch_preds: |
|
|
cls_id = int(pred[6]) |
|
|
batch_out.append({ |
|
|
"x": (pred[0] + pred[2]) / 2, |
|
|
"y": (pred[1] + pred[3]) / 2, |
|
|
"width": pred[2] - pred[0], |
|
|
"height": pred[3] - pred[1], |
|
|
"confidence": float(pred[4]), |
|
|
"class_id": cls_id, |
|
|
"class": class_names[cls_id] if cls_id < len(class_names) else str(cls_id), |
|
|
}) |
|
|
results.append(batch_out) |
|
|
return results |
|
|
|
|
|
|
|
|
def load_model(onnx_path: str | None = None, data_dir: str | None = None): |
|
|
data_dir = data_dir or os.path.dirname(os.path.abspath(__file__)) |
|
|
onnx_path = onnx_path or find_default_onnx(data_dir) |
|
|
session = load_onnx_session(onnx_path) |
|
|
meta = build_meta(data_dir, session) |
|
|
model_type_fn = globals().get("load_model_type") |
|
|
model_type = model_type_fn(data_dir) if callable(model_type_fn) else "unknown" |
|
|
return {"session": session, "meta": meta, "model_type": model_type} |
|
|
|
|
|
|
|
|
def run_model(model: Any, image: Any = None, onnx_path: str | None = None, data_dir: str | None = None): |
|
|
if image is None: |
|
|
image = model |
|
|
model = load_model(onnx_path=onnx_path, data_dir=data_dir) |
|
|
session = model["session"] |
|
|
meta = model["meta"] |
|
|
model_type = model["model_type"] |
|
|
|
|
|
img_in, img_dims = preprocess_for_model(image, meta) |
|
|
input_name = session.get_inputs()[0].name |
|
|
outputs = session.run(None, {input_name: img_in}) |
|
|
predictions = pack_predictions(outputs[0]) |
|
|
return postprocess_predictions(predictions, meta, [img_dims]) |
|
|
|
|
|
|
|
|
def main(): |
|
|
if len(sys.argv) < 2: |
|
|
print("Usage: main.py <image_path> [onnx_path]", file=sys.stderr) |
|
|
sys.exit(1) |
|
|
image_path = sys.argv[1] |
|
|
data_dir = os.path.dirname(os.path.abspath(__file__)) |
|
|
onnx_path = sys.argv[2] if len(sys.argv) > 2 else find_default_onnx(data_dir) |
|
|
results = run_model(image_path, onnx_path=onnx_path, data_dir=data_dir) |
|
|
print(json.dumps(results, indent=2)) |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
main() |
|
|
|
