# -*- coding: utf-8 -*-
“”"
Vision Pipeline V3 — Official RDK postProcess + hbm_runtime + MJPEG Web
Uses scipy softmax + flat reshape (official RDK approach) for max CPU speed.
“”"
import cv2
import numpy as np
import hbm_runtime
import time
import threading
import json
from http.server import HTTPServer, BaseHTTPRequestHandler
from scipy.special import softmax as scipy_softmax
# ── Model config ──
MODEL_PATH = “/home/sunrise/yolov5_best_quant（320）.bin”
INPUT_SIZE = 320
REG_MAX = 16
MASK_COEFFS = 32
CONF_THRES = 0.25
NMS_THRES = 0.30
FRAME_W, FRAME_H = 640, 480
FRAME_CX = 285
CLASSES = [‘blackball’, ‘blueball’, ‘bluezone’,
'redball', 'redzone', 'yellowball'\]
NUM_CLS = len(CLASSES)
BLACK_BALL = 0; BLUE_BALL = 1; BLUE_ZONE = 2
RED_BALL = 3; RED_ZONE = 4; YELLOW_BALL = 5
OWN_ZONE = RED_ZONE
OPPONENT_ZONE = BLUE_ZONE
CLASS_CONF_BIAS = np.zeros(NUM_CLS, dtype=np.float32)
CLASS_CONF_BIAS[BLUE_BALL] = -0.05
CLASS_CONF_BIAS[OWN_ZONE] = 0.10
CLASS_CONF_BIAS[OPPONENT_ZONE] = 0.05
CLASS_CONF_BIAS[YELLOW_BALL] = 0.10
CLASS_CONF_BIAS[BLACK_BALL] = 0.10
ZONE_MASK_EXPAND = 30
ZONE_BOTTOM_SHRINK = 10
OPPONENT_ZONE_EXPAND = 20
CONF_THRES_RAW = -np.log(1.0 / CONF_THRES - 1.0)
# ── 320×320 SCALES ──
STRIDES = [8, 16, 32]
GRID_H = [INPUT_SIZE // s for s in STRIDES] # [40, 20, 10]
# ── Precomputed: DFL weights + grid offsets ──
_DFL_WEIGHTS = np.arange(REG_MAX, dtype=np.float32)
_GRIDS = []
for stride, gh in zip(STRIDES, GRID_H):
x = np.tile(np.linspace(0.5, gh - 0.5, gh), gh)
y = np.repeat(np.arange(0.5, gh, 1.0), gh)
\_GRIDS.append(np.stack(\[x, y\], axis=1).astype(np.float32))
PROTO_H = PROTO_W = INPUT_SIZE // 4 # 80
X_SCALE_CROP = PROTO_W / INPUT_SIZE
Y_SCALE_CROP = PROTO_H / INPUT_SIZE
# ── Web globals ──
WEB_PORT = 8080
g_web_frame = None
g_web_state = {}
COLORS = {
0: (0, 0, 0), 1: (255, 0, 0), 2: (255, 128, 0),
3: (0, 0, 255), 4: (255, 0, 255), 5: (0, 255, 255),
}
MASK_COLORS = {
0: (100, 100, 100), 1: (255, 100, 100), 2: (255, 180, 120),
3: (80, 60, 255), 4: (255, 120, 255), 5: (80, 255, 220),
}
def bgr2nv12(image):
"""BGR (H,W,3) uint8 -> NV12 flat uint8"""
h, w = image.shape\[:2\]
area = h \* w
yuv = cv2.cvtColor(image, cv2.COLOR_BGR2YUV_I420).reshape(area \* 3 // 2)
y = yuv\[:area\]
uv_planar = yuv\[area:\].reshape(2, area // 4)
uv = uv_planar.transpose(1, 0).reshape(area // 2)
nv12 = np.zeros(area \* 3 // 2, dtype=np.uint8)
nv12\[:area\] = y; nv12\[area:\] = uv
return nv12
def _crop_mask(mask_full, x1, y1, x2, y2, img_w, img_h):
scale = INPUT_SIZE / PROTO_H
px1 = max(0, int(x1 / scale)); py1 = max(0, int(y1 / scale))
px2 = min(PROTO_W, int(x2 / scale) + 1); py2 = min(PROTO_H, int(y2 / scale) + 1)
crop = mask_full\[py1:py2, px1:px2\]
if crop.size == 0: return None
dx1 = int(px1 \* img_w / PROTO_W); dy1 = int(py1 \* img_h / PROTO_H)
dx2 = int(px2 \* img_w / PROTO_W); dy2 = int(py2 \* img_h / PROTO_H)
bw = max(1, dx2 - dx1); bh = max(1, dy2 - dy1)
crop = cv2.resize(crop, (bw, bh)); crop = (crop > 0.5).astype(np.uint8)
full = np.zeros((img_h, img_w), dtype=np.uint8)
sy1 = max(0, -dy1); sx1 = max(0, -dx1)
sy2 = min(bh, img_h - dy1); sx2 = min(bw, img_w - dx1)
if sy2 > sy1 and sx2 > sx1:
full\[dy1 + sy1:dy1 + sy2, dx1 + sx1:dx1 + sx2\] = crop\[sy1:sy2, sx1:sx2\]
return full
def process_output_seg(raw, onames):
"""
Official RDK postProcess approach:
reshape → flatnonzero → scipy softmax → per-class NMS → mask decode.
"""
rw = FRAME_W / INPUT_SIZE
rh = FRAME_H / INPUT_SIZE
\# Proto
proto = raw\[onames\[9\]\].squeeze()
if proto.shape\[0\] != MASK_COEFFS:
proto = proto.reshape(MASK_COEFFS, PROTO_H, PROTO_W)
\# Reshape all scales at once (official approach)
clses = \[raw\[onames\[0\]\].reshape(-1, NUM_CLS),
raw\[onames\[3\]\].reshape(-1, NUM_CLS),
raw\[onames\[6\]\].reshape(-1, NUM_CLS)\]
bboxes = \[raw\[onames\[1\]\].reshape(-1, REG_MAX \* 4),
raw\[onames\[4\]\].reshape(-1, REG_MAX \* 4),
raw\[onames\[7\]\].reshape(-1, REG_MAX \* 4)\]
mces_l = \[raw\[onames\[2\]\].reshape(-1, MASK_COEFFS),
raw\[onames\[5\]\].reshape(-1, MASK_COEFFS),
raw\[onames\[8\]\].reshape(-1, MASK_COEFFS)\]
all_boxes = \[\]; all_scores = \[\]; all_ids = \[\]; all_mces = \[\]
for cls, bbox, mc, stride, grid in zip(clses, bboxes, mces_l, STRIDES, \_GRIDS):
\# Per-class bias threshold
cls_raw = cls # keep raw for sigmoid later
max_scores = np.max(cls_raw, axis=1)
bbox_idx = np.flatnonzero(max_scores >= CONF_THRES_RAW)
if len(bbox_idx) == 0: continue
best_cls = np.argmax(cls_raw\[bbox_idx\], axis=1)
scores = 1.0 / (1.0 + np.exp(-max_scores\[bbox_idx\]))
\# Apply per-class bias
keep = np.ones(len(bbox_idx), dtype=bool)
for c in range(NUM_CLS):
cmask = best_cls == c
keep\[cmask\] = scores\[cmask\] > (CONF_THRES - CLASS_CONF_BIAS\[c\])
if not keep.any(): continue
bbox_idx = bbox_idx\[keep\]; best_cls = best_cls\[keep\]; scores = scores\[keep\]
\# DFL decode (scipy softmax — vectorized)
ltrb = np.sum(scipy_softmax(bbox\[bbox_idx\].reshape(-1, 4, REG_MAX), axis=2) \* \_DFL_WEIGHTS, axis=2)
g = grid\[bbox_idx\]
x1y1 = (g - ltrb\[:, 0:2\]) \* stride
x2y2 = (g + ltrb\[:, 2:4\]) \* stride
xyxy = np.hstack(\[x1y1, x2y2\])
xyxy\[:, 0\] = np.clip(xyxy\[:, 0\], 0, INPUT_SIZE); xyxy\[:, 1\] = np.clip(xyxy\[:, 1\], 0, INPUT_SIZE)
xyxy\[:, 2\] = np.clip(xyxy\[:, 2\], 0, INPUT_SIZE); xyxy\[:, 3\] = np.clip(xyxy\[:, 3\], 0, INPUT_SIZE)
all_boxes.append(xyxy); all_scores.append(scores)
all_ids.append(best_cls); all_mces.append(mc\[bbox_idx\])
if len(all_boxes) == 0: return \[\]
boxes = np.vstack(all_boxes); scores = np.hstack(all_scores)
ids = np.hstack(all_ids); mces = np.vstack(all_mces)
\# Per-class NMS
xywh = np.hstack(\[boxes\[:, 0:2\], boxes\[:, 2:4\] - boxes\[:, 0:2\]\])
keep_all = \[\]
for c in range(NUM_CLS):
cmask = ids == c
if not cmask.any(): continue
idx = cv2.dnn.NMSBoxes(xywh\[cmask\].tolist(), scores\[cmask\].tolist(), CONF_THRES, NMS_THRES)
if len(idx) == 0: continue
keep_all.append(np.where(cmask)\[0\]\[idx.flatten()\])
if len(keep_all) == 0: return \[\]
keep = np.concatenate(keep_all)
boxes = boxes\[keep\]; scores = scores\[keep\]
cls_arr = ids\[keep\]; mces_nms = mces\[keep\]
\# Scale to image
boxes\[:, 0\] \*= rw; boxes\[:, 1\] \*= rh
boxes\[:, 2\] \*= rw; boxes\[:, 3\] \*= rh
\# Batch mask decode
zone_idx = \[i for i, k in enumerate(keep) if int(cls_arr\[i\]) in (OWN_ZONE, OPPONENT_ZONE)\]
batch_masks = None
if zone_idx:
try:
z_mces = mces_nms\[zone_idx\].astype(np.float32)
proto_flat = proto.reshape(MASK_COEFFS, -1)
batch_masks = (z_mces @ proto_flat).reshape(len(zone_idx), PROTO_H, PROTO_W)
batch_masks = 1.0 / (1.0 + np.exp(-batch_masks))
batch_masks = (batch_masks > 0.5).astype(np.uint8)
except: pass
final_dets = \[\]
for i in range(len(keep)):
cid = int(cls_arr\[i\]); b = boxes\[i\]
det = {
'cls': cid,
'cx': float((b\[0\] + b\[2\]) \* 0.5), 'cy': float((b\[1\] + b\[3\]) \* 0.5),
'w': float(b\[2\] - b\[0\]), 'h': float(b\[3\] - b\[1\]),
'conf': float(scores\[i\]), 'mask': None,
}
if cid in (OWN_ZONE, OPPONENT_ZONE) and batch_masks is not None:
zi = zone_idx.index(i) if i in zone_idx else -1
if zi >= 0:
try:
x1m = boxes\[keep\[i\], 0\] / rw; y1m = boxes\[keep\[i\], 1\] / rh
x2m = boxes\[keep\[i\], 2\] / rw; y2m = boxes\[keep\[i\], 3\] / rh
det\['mask'\] = \_crop_mask(batch_masks\[zi\], x1m, y1m, x2m, y2m, FRAME_W, FRAME_H)
if det\['mask'\] is not None and det\['mask'\].any():
contours, \_ = cv2.findContours(det\['mask'\], cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
if contours:
cnt = max(contours, key=cv2.contourArea)
M = cv2.moments(cnt)
if M\['m00'\] > 0:
det\['cx'\] = M\['m10'\] / M\['m00'\]; det\['cy'\] = M\['m01'\] / M\['m00'\]
pts = cv2.approxPolyDP(cnt, 0.005 \* cv2.arcLength(cnt, True), True).reshape(-1, 2)
det\['mask_x1'\] = int(pts\[:, 0\].min()); det\['mask_y1'\] = int(pts\[:, 1\].min())
det\['mask_x2'\] = int(pts\[:, 0\].max()); det\['mask_y2'\] = int(pts\[:, 1\].max())
dm = det\['mask'\].copy()
if cid == OPPONENT_ZONE and OPPONENT_ZONE_EXPAND > 0:
n = OPPONENT_ZONE_EXPAND
u = np.roll(dm, -n, axis=0); u\[-n:, :\] = 0
d = np.roll(dm, n, axis=0); d\[:n, :\] = 0
l = np.roll(dm, -n, axis=1); l\[:, -n:\] = 0
r = np.roll(dm, n, axis=1); r\[:, :n\] = 0
dm = np.maximum.reduce(\[dm, u, d, l, r\])
elif cid == OWN_ZONE:
if ZONE_MASK_EXPAND > 0:
u = np.roll(dm, -ZONE_MASK_EXPAND, axis=0); u\[-ZONE_MASK_EXPAND:, :\] = 0
dm = np.maximum(dm, u)
if ZONE_BOTTOM_SHRINK > 0:
d = np.roll(dm, ZONE_BOTTOM_SHRINK, axis=0); d\[:ZONE_BOTTOM_SHRINK, :\] = 0
dm = np.minimum(dm, d)
det\['mask_dilated'\] = dm
dc, \_ = cv2.findContours(dm, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
if dc: det\['filter_contour'\] = max(dc, key=cv2.contourArea)
except: pass
final_dets.append(det)
return final_dets
# ── Web draw ──
def draw_dets(frame, dets):
img = frame.copy()
mask_overlay = np.zeros_like(img)
for d in dets:
mask = d.get('mask_dilated') if d\['cls'\] in (OWN_ZONE, OPPONENT_ZONE) else d.get('mask')
if mask is not None and mask.any():
mask_overlay\[mask > 0\] = MASK_COLORS.get(d\['cls'\], (0, 255, 255))
if mask_overlay.any():
alpha = 0.45
m = mask_overlay.any(axis=2)
for c in range(3):
img\[:, :, c\] = np.where(m, (img\[:, :, c\] \* (1 - alpha) + mask_overlay\[:, :, c\] \* alpha).astype(np.uint8), img\[:, :, c\])
for d in dets:
cx, cy, w, h = d\['cx'\], d\['cy'\], d\['w'\], d\['h'\]
cid = d\['cls'\]
x1, y1 = int(cx - w / 2), int(cy - h / 2)
x2, y2 = int(cx + w / 2), int(cy + h / 2)
cv2.rectangle(img, (x1, y1), (x2, y2), COLORS.get(cid, (0, 255, 0)), 2)
cv2.putText(img, f"{CLASSES\[cid\]} {d\['conf'\]:.2f}", (x1, y1 - 8),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, COLORS.get(cid, (0, 255, 0)), 2)
cv2.line(img, (FRAME_CX, 0), (FRAME_CX, FRAME_H), (0, 255, 0), 2)
cv2.putText(img, f"Dets:{len(dets)}", (10, FRAME_H - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 2)
return img
# ── Web handler ──
class WebHandler(BaseHTTPRequestHandler):
def log_message(self, \*a): pass
def do_GET(self):
if self.path == "/": self.\_html()
elif self.path == "/video": self.\_stream()
elif self.path == "/status.json": self.\_json()
else: self.send_error(404)
def \_html(self):
h = """<!DOCTYPE html><html><head><meta charset="utf-8"><title>V3 Seg</title>
body{margin:0;background:#111;color:#fff;font-family:monospace}
#s{padding:10px;background:#222;font-size:14px}img{width:100%;max-width:900px;display:block}
Loading...
"""
self.send_response(200); self.send_header("Content-Type", "text/html; charset=utf-8"); self.end_headers()
self.wfile.write(h.encode())
def \_stream(self):
self.send_response(200); self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=frame"); self.end_headers()
while True:
if g_web_frame is not None:
try: self.wfile.write(b"--frame\\r\\nContent-Type: image/jpeg\\r\\n\\r\\n" + g_web_frame + b"\\r\\n")
except: break
time.sleep(0.05)
def \_json(self):
self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
self.wfile.write(json.dumps(g_web_state).encode())
def _web_thread():
HTTPServer(("0.0.0.0", WEB_PORT), WebHandler).serve_forever()
# ── Main loop ──
def main():
global g_web_frame, g_web_state
print(f"Loading {MODEL_PATH} ...")
model = hbm_runtime.HB_HBMRuntime(MODEL_PATH)
mname = model.model_names\[0\]
iname = model.input_names\[mname\]\[0\]
onames = model.output_names\[mname\]
print(f" Model: {mname}, {len(onames)} outputs")
cap = cv2.VideoCapture("/dev/video1", cv2.CAP_V4L2)
if not cap.isOpened():
cap = cv2.VideoCapture("/dev/video0", cv2.CAP_V4L2)
if not cap.isOpened():
print("ERROR: Cannot open camera"); return
cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(\*'MJPG'))
cap.set(3, FRAME_W); cap.set(4, FRAME_H)
cap.set(cv2.CAP_PROP_FPS, 30)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
threading.Thread(target=\_web_thread, daemon=True).start()
print(f" Web: http://<x5-ip>:{WEB_PORT}")
web_lock = threading.Lock()
latest_frame, latest_dets, fps_val = None, \[\], 0.0
def web_encode_loop():
nonlocal fps_val
global g_web_frame, g_web_state
while True:
with web_lock:
frame = latest_frame.copy() if latest_frame is not None else None
dets = list(latest_dets); fv = fps_val
if frame is not None:
annotated = draw_dets(frame, dets)
\_, buf = cv2.imencode(".jpg", annotated, \[cv2.IMWRITE_JPEG_QUALITY, 55\])
g_web_frame = buf.tobytes()
g_web_state = {"fps": fv, "dets": len(dets)}
else: time.sleep(0.01)
threading.Thread(target=web_encode_loop, daemon=True).start()
\# Warmup
for \_ in range(5):
ok, frame = cap.read()
if ok:
img = cv2.resize(frame, (INPUT_SIZE, INPUT_SIZE))
nv12 = bgr2nv12(img).reshape(1, INPUT_SIZE \* 3 // 2, INPUT_SIZE, 1)
model.run({mname: {iname: nv12}})
print("Ready.")
fps_t0 = time.time(); fps_count = 0
t_all = \[0.0\]\*6 # read, resize, nv12, inference, postproc, lock
try:
while True:
t0 = time.time()
ok, frame = cap.read()
t_all\[0\] += time.time() - t0
if not ok: time.sleep(0.01); continue
t0 = time.time()
img = cv2.resize(frame, (INPUT_SIZE, INPUT_SIZE))
t_all\[1\] += time.time() - t0
t0 = time.time()
nv12 = bgr2nv12(img).reshape(1, INPUT_SIZE \* 3 // 2, INPUT_SIZE, 1)
t_all\[2\] += time.time() - t0
t0 = time.time()
raw = model.run({mname: {iname: nv12}})\[mname\]
t_all\[3\] += time.time() - t0
t0 = time.time()
dets = process_output_seg(raw, onames)
t_all\[4\] += time.time() - t0
t0 = time.time()
with web_lock:
latest_frame, latest_dets = frame, dets
t_all\[5\] += time.time() - t0
fps_count += 1
if fps_count >= 30:
fps_val = 30 / (time.time() - fps_t0 + 1e-7)
fps_t0 = time.time(); fps_count = 0
ms = \[v/30\*1000 for v in t_all\]
print(f"FPS:{fps_val:.1f} | read:{ms\[0\]:.0f} resize:{ms\[1\]:.0f} nv12:{ms\[2\]:.0f} infer:{ms\[3\]:.0f} post:{ms\[4\]:.0f} lock:{ms\[5\]:.0f}ms")
t_all = \[0.0\]\*6
except KeyboardInterrupt: pass
finally:
cap.release(); print("Done.")
if _name_ == “_main_”: main()，
FPS:14.9 | read:52 resize:3 nv12:2 infer:9 post:1 lock:0ms
FPS:14.8 | read:53 resize:3 nv12:1 infer:9 post:1 lock:0ms瓶颈是不是摄像头
代码评估：320×320 优化版
已实现的关键优化
剩余瓶颈与修正建议
下一步建议
参考资源
预期输出参考
如果 FPS 仍<40
问题分析
好消息！hobot_dnn 版本 3.0.9
两条 C++ 路径
编译步骤
推荐方案