网球场景计算机视觉系统设计：从论文调研到工业实现

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

import cv2
import numpy as np
from typing import List, Tuple
import json


class MultiCameraCalibration:
    """多相机标定 + DLT 三角测量。"""

    def __init__(self, num_cameras: int = 8):
        self.num_cameras = num_cameras
        self.camera_matrices: List[np.ndarray] = []   # K, 3x3
        self.dist_coeffs: List[np.ndarray] = []       # (k1, k2, p1, p2, k3)
        self.R_matrices: List[np.ndarray] = []        # 相对世界坐标系
        self.t_vectors: List[np.ndarray] = []

    def calibrate_single_camera(
        self,
        images: List[np.ndarray],
        pattern_size: Tuple[int, int] = (9, 6),
        square_size: float = 0.025,
    ) -> Tuple[np.ndarray, np.ndarray]:
        """Zhang's method: 用棋盘格求单相机内参。"""
        objp = np.zeros((pattern_size[0] * pattern_size[1], 3), np.float32)
        objp[:, :2] = np.mgrid[0:pattern_size[0], 0:pattern_size[1]].T.reshape(-1, 2)
        objp *= square_size

        objpoints, imgpoints = [], []
        for img in images:
            gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
            ret, corners = cv2.findChessboardCorners(gray, pattern_size, None)
            if not ret:
                continue
            corners = cv2.cornerSubPix(
                gray, corners, (11, 11), (-1, -1),
                (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 30, 1e-3),
            )
            objpoints.append(objp)
            imgpoints.append(corners)

        if len(objpoints) < 10:
            raise ValueError(f"标定图像不足，仅找到 {len(objpoints)} 张有效图像（需 ≥10）")

        ret, K, dist, rvecs, tvecs = cv2.calibrateCamera(
            objpoints, imgpoints, gray.shape[::-1], None, None
        )

        # 重投影误差 - 必须 < 1 px 才算合格
        err = 0.0
        for i in range(len(objpoints)):
            proj, _ = cv2.projectPoints(objpoints[i], rvecs[i], tvecs[i], K, dist)
            err += cv2.norm(imgpoints[i], proj, cv2.NORM_L2) / len(proj)
        err /= len(objpoints)
        print(f"重投影误差: {err:.4f} px {'OK' if err < 1.0 else '需重拍'}")
        return K, dist

    def triangulate_point(
        self,
        points_2d: List[Tuple[float, float]],
        camera_ids: List[int],
    ) -> np.ndarray:
        """DLT 三角测量：≥2 视角的 2D 观测 → 3D。"""
        if len(points_2d) < 2:
            raise ValueError("至少需要 2 个视角")

        A = []
        for (x, y), cid in zip(points_2d, camera_ids):
            P = self.camera_matrices[cid] @ np.hstack(
                [self.R_matrices[cid], self.t_vectors[cid]]
            )
            A.append(x * P[2] - P[0])
            A.append(y * P[2] - P[1])
        A = np.asarray(A)

        _, _, Vt = np.linalg.svd(A)
        X = Vt[-1]
        return (X / X[3])[:3]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

from ultralytics import YOLO


class TennisBallDetector:
    """实时网球检测：YOLOv8 + 物理先验后处理。"""

    def __init__(self, model_path: str = "yolov8n.pt", device: str = "cuda"):
        self.model = YOLO(model_path)
        self.device = device
        self.img_size = 1280       # 大输入是小物体检测的"免费午餐"
        self.conf_thr = 0.25       # 适当低 → 减漏检，由后处理压误检
        self.iou_thr = 0.45
        self.ball_cls = 32         # COCO 'sports ball'

    def detect(self, frame, expected_size_px: Tuple[int, int] = (5, 60)) -> list:
        """单帧检测。expected_size_px 用相机焦距 + 球距估算，是关键先验。"""
        results = self.model.predict(
            frame, imgsz=self.img_size, conf=self.conf_thr, iou=self.iou_thr,
            classes=[self.ball_cls], device=self.device, verbose=False,
        )

        out = []
        s_min, s_max = expected_size_px
        for box in results[0].boxes:
            x1, y1, x2, y2 = box.xyxy[0].cpu().numpy()
            w, h = x2 - x1, y2 - y1
            # 物理先验 1：球径在合理范围内（按焦距和距离估算）
            if not (s_min <= max(w, h) <= s_max):
                continue
            # 物理先验 2：网球近似圆形，aspect ratio 接近 1
            if not 0.6 <= w / max(h, 1e-6) <= 1.6:
                continue
            out.append({
                "bbox": [float(x1), float(y1), float(x2), float(y2)],
                "center": [float((x1 + x2) / 2), float((y1 + y2) / 2)],
                "confidence": float(box.conf[0]),
            })
        out.sort(key=lambda d: d["confidence"], reverse=True)
        return out

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

from collections import deque


class TennisBallTracker:
    """9 维状态 (x,y,z, vx,vy,vz, ax,ay,az) Kalman tracker。"""

    def __init__(self, fps: int = 60):
        self.dt = 1.0 / fps
        self.kf = cv2.KalmanFilter(9, 3)

        dt = self.dt
        # 匀加速运动模型：x_{t+1} = x_t + v dt + 0.5 a dt^2
        F = np.eye(9, dtype=np.float32)
        F[0, 3] = F[1, 4] = F[2, 5] = dt
        F[0, 6] = F[1, 7] = F[2, 8] = 0.5 * dt ** 2
        F[3, 6] = F[4, 7] = F[5, 8] = dt
        self.kf.transitionMatrix = F

        H = np.zeros((3, 9), dtype=np.float32)
        H[0, 0] = H[1, 1] = H[2, 2] = 1
        self.kf.measurementMatrix = H

        Q = np.eye(9, dtype=np.float32) * 0.03
        Q[6:, 6:] *= 5            # 加速度更不确定（旋转、风、阻力非线性）
        self.kf.processNoiseCov = Q
        self.kf.measurementNoiseCov = np.eye(3, dtype=np.float32) * 0.1
        self.kf.errorCovPost = np.eye(9, dtype=np.float32) * 1000

        self.history = deque(maxlen=300)   # 5 秒 @ 60 fps
        self.lost = 0
        self.initialized = False
        self.max_lost = 30                  # 0.5 s 没看到 → 复位

    def update(self, measurement: np.ndarray = None):
        prediction = self.kf.predict()
        if measurement is not None:
            m = measurement.reshape(3, 1).astype(np.float32)
            if not self.initialized:
                self.kf.statePost = np.array(
                    [m[0, 0], m[1, 0], m[2, 0], 0, 0, 0, 0, 0, -9.8],
                    dtype=np.float32,
                ).reshape(9, 1)
                self.initialized = True
            self.kf.correct(m)
            self.lost = 0
        else:
            self.lost += 1
            if self.lost > self.max_lost:
                self.initialized = False

        s = self.kf.statePost if measurement is not None else prediction
        pos, vel, acc = s[0:3].flatten(), s[3:6].flatten(), s[6:9].flatten()
        self.history.append({"position": pos.copy(), "velocity": vel.copy()})
        return pos, vel, acc

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

from scipy.integrate import odeint


class TennisTrajectoryPredictor:
    """物理模型轨迹预测：drag + Magnus + gravity。"""

    def __init__(self):
        self.g = 9.81
        self.rho = 1.225          # kg/m^3
        self.m = 0.0585           # kg, ITF 标准
        self.r = 0.0335           # m
        self.Cd = 0.55            # 球形物体阻力系数
        self.Cm = 0.00029         # Magnus 系数（实测拟合）
        self.A = np.pi * self.r ** 2

    def predict(self, p0, v0, spin=None, dt=0.01, duration=2.0):
        if spin is None:
            spin = np.zeros(3)
        t = np.arange(0, duration, dt)
        traj = odeint(self._dyn, np.concatenate([p0, v0]), t, args=(spin,))
        # 截到落地
        idx = np.where(traj[:, 2] <= 0)[0]
        return traj[: idx[0] + 1, :3] if len(idx) else traj[:, :3]

    def _dyn(self, state, t, spin):
        v = state[3:6]
        speed = np.linalg.norm(v) + 1e-9
        drag = -0.5 * self.rho * self.Cd * self.A * speed * v
        magnus = self.Cm * np.cross(spin, v)
        gravity = np.array([0, 0, -self.g * self.m])
        a = (drag + magnus + gravity) / self.m
        return np.concatenate([v, a])

    def landing(self, p0, v0, spin=None):
        traj = self.predict(p0, v0, spin, dt=0.005, duration=5.0)
        if len(traj) == 0:
            return None
        return traj[-1]

1
2
3
4
5
6
7

import cv2

gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
edges = cv2.Canny(gray, 60, 180)
lines = cv2.HoughLinesP(edges, 1, np.pi / 180, threshold=80,
                        minLineLength=50, maxLineGap=10)
# lines: shape (N, 1, 4) -> (x1, y1, x2, y2)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

class TennisPoseClassifier:
    """关键点规则模板，单帧 < 1 ms。"""

    SKELETON_NAMES = {  # COCO subset
        5: "L_shoulder", 6: "R_shoulder",
        7: "L_elbow",    8: "R_elbow",
        9: "L_wrist",    10: "R_wrist",
        11: "L_hip",     12: "R_hip",
        15: "L_ankle",   16: "R_ankle",
    }

    def classify(self, kp: np.ndarray) -> Tuple[str, float]:
        scores = {
            "serve":    self._serve(kp),
            "forehand": self._forehand(kp),
            "backhand": self._backhand(kp),
            "volley":   self._volley(kp),
            "ready":    self._ready(kp),
        }
        action = max(scores, key=scores.get)
        return action, scores[action]

    def _serve(self, kp):
        feats = [
            (0.30, kp[10, 1] < kp[6, 1] - 50),     # 右手腕高过右肩
            (0.20, kp[9, 1]  < kp[5, 1]),          # 左手腕高过左肩（抛球）
            (0.15, abs(kp[15, 0] - kp[16, 0]) > 100),  # 双脚分开
            (0.20, self._arm_angle(kp, "R") >= 120),
            (0.15, kp[5, 0] - kp[11, 0] < -10),    # 身体后倾
        ]
        return self._weighted(feats)

    def _forehand(self, kp):
        feats = [
            (0.30, kp[10, 0] < kp[6, 0]),           # 右手腕到了身体左侧
            (0.25, self._shoulder_rot(kp) > 10),
            (0.20, kp[16, 0] > kp[15, 0]),          # 右脚在前
            (0.15, kp[8, 1] > kp[6, 1]),
            (0.10, kp[10, 0] > kp[5, 0]),           # follow-through
        ]
        return self._weighted(feats)

    # _backhand / _volley / _ready 同理
    @staticmethod
    def _weighted(feats):
        s = sum(w for w, ok in feats if ok)
        z = sum(w for w, _ in feats)
        return s / z if z else 0.0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

from queue import Queue


class FrameSynchronizer:
    """选取时间戳相互最接近的一组帧，超出 max_diff 的丢弃。"""

    def __init__(self, num_cameras: int, max_diff_ms: float = 5.0):
        self.n = num_cameras
        self.max_diff = max_diff_ms / 1000.0
        self.buffers = [Queue(maxsize=10) for _ in range(num_cameras)]

    def add(self, cam_id: int, frame, ts: float):
        if self.buffers[cam_id].full():
            self.buffers[cam_id].get_nowait()
        self.buffers[cam_id].put((frame, ts))

    def pop_synced(self):
        if any(b.empty() for b in self.buffers):
            return None, None
        # 取每个相机队列头的时间戳
        heads = [b.queue[0] for b in self.buffers]
        ts_list = [t for _, t in heads]
        if max(ts_list) - min(ts_list) > self.max_diff:
            # 丢掉最老的那一帧再试
            oldest = int(np.argmin(ts_list))
            self.buffers[oldest].get_nowait()
            return None, None
        # 全部 pop
        frames, ts = [], []
        for b in self.buffers:
            f, t = b.get_nowait()
            frames.append(f); ts.append(t)
        return frames, ts

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class TennisAnalysisSystem:
    def __init__(self, num_cameras=8, calib_path="calib.json"):
        self.calib = MultiCameraCalibration(num_cameras)
        self.calib.load(calib_path)
        self.detector = TennisBallDetector()
        self.tracker = TennisBallTracker()
        self.predictor = TennisTrajectoryPredictor()
        self.pose_clf = TennisPoseClassifier()
        self.sync = FrameSynchronizer(num_cameras)

    def step(self, frames, timestamps):
        # 1. 球检测（每相机一次）
        ball_2d = []
        for cid, f in enumerate(frames):
            dets = self.detector.detect(f)
            if dets:
                ball_2d.append((cid, dets[0]["center"]))

        # 2. 三角测量
        ball_3d = None
        if len(ball_2d) >= 2:
            ball_3d = self.calib.triangulate_point(
                [p for _, p in ball_2d], [c for c, _ in ball_2d]
            )

        # 3. 卡尔曼更新
        pos, vel, acc = self.tracker.update(ball_3d)

        # 4. 落点预测
        landing = self.predictor.landing(pos, vel) if np.linalg.norm(vel) > 1 else None

        # 5. 姿态（仅主相机）
        # ... 调用 HRNet + classify ...

        return {"position": pos, "velocity": vel, "landing": landing}

1
2
3

# YOLOv8 → TensorRT，FP16 速度 ×2，INT8 ×4
yolo export model=yolov8l.pt format=engine half=true device=0
yolo export model=yolov8l.pt format=engine int8=true data=tennis.yaml