HTTP API 参考

shimeta_camera 和 shimeta_svp 启动后自动在 8080 端口 提供 HTTP API。

1. shimeta_camera

GET /api/status

返回实时推理状态和检测结果。

{
  "total_frames": 150,
  "fps": 18.5,
  "infer_ms": 12.3,
  "detections": [
    {
      "class_id": 0,
      "label": "person",
      "conf": 0.92,
      "x1": 100,
      "y1": 200,
      "x2": 300,
      "y2": 400
    }
  ]
}

字段	类型	说明
`total_frames`	int	累计帧数
`fps`	float	实时帧率
`infer_ms`	float	推理耗时（毫秒）
`detections`	array	检测结果
`detections[].class_id`	int	类别 ID（0=person, 1=bicycle, ...）
`detections[].label`	string	类别名称
`detections[].conf`	float	置信度 (0-1)
`detections[].x1,y1,x2,y2`	int	边界框坐标

GET /api/snapshot

返回当前帧 JPEG 快照。

Content-Type: image/jpeg

GET /stream

MJPEG 实时视频流。

Content-Type: multipart/x-mixed-replace; boundary=frame

2. shimeta_svp

GET /

返回当前 SVP 任务的检测结果。根据任务类型返回不同的 JSON 结构。

person / head / car / bike / fireworks

{
  "task": "head",
  "fps": 22.5,
  "total": 300,
  "dets": [
    {
      "class": 4,
      "conf": 88,
      "x1": 100,
      "y1": 200,
      "x2": 150,
      "y2": 280
    }
  ]
}

字段	类型	说明
`task`	string	任务名称
`fps`	float	实时帧率
`total`	int	累计帧数
`dets[].class`	int	类别 ID（4=head, 0=person, ...）
`dets[].conf`	int	置信度百分比（0-100）
`dets[].x1,y1,x2,y2`	int	边界框坐标（640×360 空间）

person_kp

在 dets 基础上增加 kp 字段，17 点人体关键点：

{
  "task": "person_kp",
  "fps": 22.5,
  "total": 300,
  "dets": [
    {
      "class": 8,
      "conf": 73,
      "x1": 100,
      "y1": 200,
      "x2": 250,
      "y2": 350,
      "kp": [
        [320, 50, 0.9],
        [310, 45, 0.8]
      ]
    }
  ]
}

字段	说明
`kp[][0]`	关键点 X 坐标
`kp[][1]`	关键点 Y 坐标
`kp[][2]`	关键点置信度 (0-1)

17 点顺序：nose, left_eye, right_eye, left_ear, right_ear, left_shoulder, right_shoulder, left_elbow, right_elbow, left_wrist, right_wrist, left_hip, right_hip, left_knee, right_knee, left_ankle, right_ankle

face_emo

{
  "task": "face_emo",
  "fps": 22.5,
  "total": 300,
  "faces": [
    {
      "emotion": "SMILE",
      "conf": 88,
      "x1": 200,
      "y1": 100,
      "x2": 280,
      "y2": 200
    }
  ]
}

emotion 值	说明
`SMILE`	微笑
`NORMAL`	正常
`UNKNOWN`	未知

dms

{
  "task": "dms",
  "fps": 22.5,
  "total": 300,
  "dms": [
    {
      "fatigue": 1,
      "phone": 0,
      "cigar": 1,
      "x1": 150,
      "y1": 80,
      "x2": 350,
      "y2": 320
    }
  ],
  "phone_boxes": [{ "x1": 260, "y1": 180, "x2": 290, "y2": 210 }],
  "cigar_boxes": [{ "x1": 270, "y1": 160, "x2": 285, "y2": 175 }]
}

字段	说明
`dms[].fatigue`	疲劳标志（0/1，闭眼或打哈欠）
`dms[].phone`	手机检测数量
`dms[].cigar`	香烟检测数量
`phone_boxes`	手机位置框
`cigar_boxes`	香烟位置框

3. Python SDK

shimeta_camera 客户端

"""
ShiMeta Camera 客户端 — 单文件，copy 即用
用法:
    from camera_client import Camera
    cam = Camera("192.168.49.10:8080")
    for d in cam.detect(): print(d.label, d.conf)
"""
import requests, time
from collections import namedtuple

Detection = namedtuple("Detection",
    ["class_id","label","conf","x1","y1","x2","y2","center"])

class Camera:
    def __init__(self, host, timeout=2.0):
        if ":" not in host: host = host + ":8080"
        self._base = f"http://{host}"
        self._timeout = timeout

    def status(self):
        """完整状态: frames, fps, infer_ms, det_count"""
        return self._get("/api/status")

    def detect(self, class_name=None, min_conf=0.0):
        """检测结果，支持按类别名和最低置信度过滤"""
        j = self._get("/api/status")
        result = []
        for d in j.get("detections", []):
            label = d.get("label") or f"cls_{d.get('class_id',0)}"
            conf = float(d.get("conf", 0))
            if class_name and label != class_name: continue
            if conf < min_conf: continue
            x1,y1,x2,y2 = int(d["x1"]),int(d["y1"]),int(d["x2"]),int(d["y2"])
            result.append(Detection(d["class_id"],label,conf,x1,y1,x2,y2,
                         ((x1+x2)/2,(y1+y2)/2)))
        return result

    def snapshot(self, save_path):
        """保存当前帧 JPEG 快照"""
        r = requests.get(f"{self._base}/api/snapshot", timeout=self._timeout)
        with open(save_path, "wb") as f: f.write(r.content)
        return save_path

    def stream_url(self):  return f"{self._base}/stream"
    def snapshot_url(self): return f"{self._base}/api/snapshot"

    def wait_online(self, timeout=30):
        deadline = time.time() + timeout
        while time.time() < deadline:
            try: self.status(); return True
            except: time.sleep(0.5)
        return False

    def _get(self, path):
        r = requests.get(f"{self._base}{path}", timeout=self._timeout)
        r.raise_for_status()
        return r.json()

shimeta_svp 客户端

class SVP:
    def __init__(self, host, timeout=2.0):
        if ":" not in host: host = host + ":8080"
        self._base = f"http://{host}"
        self._timeout = timeout

    def status(self):
        """获取 SVP 任务状态和检测结果"""
        return self._get("/")

    def detect(self):
        """返回标准化检测列表"""
        j = self._get("/")
        result = []

        # 通用检测格式
        for d in j.get("dets", []):
            result.append({
                "class": d["class"],
                "conf": d["conf"] / 100.0,   # 转为 0-1
                "bbox": (d["x1"], d["y1"], d["x2"], d["y2"])
            })

        # 人脸识别
        for f in j.get("faces", []):
            result.append({
                "name": f.get("name", ""),
                "sim": f.get("sim", 0),
                "emotion": f.get("emotion", ""),
                "bbox": (f["x1"], f["y1"], f["x2"], f["y2"])
            })

        # 驾驶员监控
        for d in j.get("dms", []):
            result.append({
                "fatigue": d.get("fatigue", 0),
                "phone": d.get("phone", 0),
                "cigar": d.get("cigar", 0)
            })

        return result

    def _get(self, path):
        r = requests.get(f"{self._base}{path}", timeout=self._timeout)
        r.raise_for_status()
        return r.json()

使用示例

# ---- shimeta_camera ----
cam = Camera("192.168.49.10")
cam.wait_online()

# 按类别过滤
people = cam.detect(class_name="person", min_conf=0.5)
for p in people:
    print(f"人 @ ({p.center[0]:.0f}, {p.center[1]:.0f}) 置信度 {p.conf:.0%}")

# 全部检测（label 由模型 labels.txt 决定，不硬编码 COCO）
all = cam.detect(min_conf=0.3)
for d in all:
    print(f"{d.label} conf={d.conf:.2f} box=({d.x1},{d.y1})-({d.x2},{d.y2})")

# 状态监控
st = cam.status()
print(f"FPS:{st.get('fps',0):.1f} 推理:{st.get('infer_ms',0):.1f}ms")

# 保存快照
cam.snapshot("frame.jpg")

# 流地址
print(cam.stream_url())    # http://IP:8080/stream


# ---- shimeta_svp ----
svp = SVP("192.168.49.10")
j = svp.status()
task = j.get("task", "")

if "dets" in j:       # person/head/car/pet 等
    for d in j["dets"]:
        print(f"class={d['class']} conf={d['conf']}% box=({d['x1']},{d['y1']})-({d['x2']},{d['y2']})")
elif "faces" in j:    # face_recog / face_emo
    for f in j["faces"]:
        print(f"{f.get('name','?')}{f.get('emotion','')} sim={f.get('sim',0)}% box=({f['x1']},{f['y1']})-({f['x2']},{f['y2']})")
elif "dms" in j:      # dms
    for d in j["dms"]:
        print(f"fatigue={d['fatigue']} phone={d['phone']} cigar={d['cigar']}")

抓取快照

import requests
from PIL import Image
from io import BytesIO

r = requests.get("http://192.168.49.10:8080/api/snapshot")
img = Image.open(BytesIO(r.content))
img.save("frame.jpg")

OpenCV 拉流

import cv2

# MJPEG 流（shimeta_camera）
cap = cv2.VideoCapture("http://192.168.49.10:8080/stream")

# RTSP 流（shimeta_svp）
cap = cv2.VideoCapture("rtsp://192.168.49.10:554/livestream/0")

while True:
    ret, frame = cap.read()
    if ret: cv2.imshow("Camera", frame)
    if cv2.waitKey(1) & 0xFF == ord('q'): break

4. 通用说明

标签解析：label 字段由模型 zip 包内的 labels.txt 提供；服务端不发 label 时 SDK 兜底用 cls_{id} 格式。不硬编码 COCO 类名。
坐标空间：SVP 坐标为 640×360（输出分辨率），需要按显示比例缩放
更新频率：每帧更新一次（~22-25fps）
跨域：响应头已设置 Access-Control-Allow-Origin: *
端口：默认 8080，shimeta_camera 和 shimeta_svp 共用，不能同时启动