# -*- 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
import subprocess
from http.server import HTTPServer, BaseHTTPRequestHandler
from scipy.special import softmax as scipy_softmax
# ── Model config ──
MODEL_PATH = “/home/sunrise/yolov5_best_quant(kl.bin”
INPUT_SIZE = 640
REG_MAX = 16
MASK_COEFFS = 32
CONF_THRES = 0.25
NMS_THRES = 0.30
FRAME_W, FRAME_H = 640, 640 # VSE dns0 output (square crop)
FRAME_CX = FRAME_W // 2
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")
\# ── MIPI CSI IMX219 via hobot_vio ──
from hobot_vio import libsrcampy as srcampy
cam = srcampy.Camera()
cam.open_cam(0, -1, 30, \[640, 640\], \[640, 640\])
print("MIPI camera opened OK")
threading.Thread(target=\_web_thread, daemon=True).start()
print(f" Web: http://<x5-ip>:{WEB_PORT}")
web_lock = threading.Lock()
latest_nv12, 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:
nv12_bytes = latest_nv12
dets = list(latest_dets); fv = fps_val
if nv12_bytes is not None:
frame = cv2.cvtColor(
np.frombuffer(nv12_bytes, dtype=np.uint8).reshape(FRAME_H \* 3 // 2, FRAME_W),
cv2.COLOR_YUV2BGR_NV12)
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):
nv12_raw = cam.get_img(2, FRAME_W, FRAME_H)
if nv12_raw is not None:
nv12 = np.frombuffer(nv12_raw, dtype=np.uint8)
nv12 = nv12.reshape(1, FRAME_H \* 3 // 2, FRAME_W, 1)
model.run({mname: {iname: nv12}})
print("Ready.")
fps_t0 = time.time(); fps_count = 0
t_all = \[0.0\]\*4 # read, infer, postproc, lock
try:
while True:
t0 = time.time()
nv12_raw = cam.get_img(2, FRAME_W, FRAME_H)
t_all\[0\] += time.time() - t0
if nv12_raw is None: time.sleep(0.01); continue
t0 = time.time()
nv12 = np.frombuffer(nv12_raw, dtype=np.uint8)
nv12 = nv12.reshape(1, FRAME_H \* 3 // 2, FRAME_W, 1)
raw = model.run({mname: {iname: nv12}})\[mname\]
t_all\[1\] += time.time() - t0
t0 = time.time()
dets = process_output_seg(raw, onames)
t_all\[2\] += time.time() - t0
t0 = time.time()
with web_lock:
latest_nv12 = nv12_raw # raw bytes, convert in web thread
latest_dets = dets
t_all\[3\] += 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} infer:{ms\[1\]:.0f} post:{ms\[2\]:.0f} lock:{ms\[3\]:.0f}ms")
t_all = \[0.0\]\*4
except KeyboardInterrupt: pass
finally:
cam.close_cam(); print("Done.")
if _name_ == “_main_”: main()帧数怎么再提高
核心结论
正确排查步骤
常见原因
下一步行动
相关资源
正常部分
失败：报错 
关键提醒
请补充以下信息
当前瓶颈分析
最大瓶颈
快速优化版本（推荐先试）
相关资源
验证帧率