注释&README.md

README.md

@@ -0,0 +1,131 @@
+# 货物高度测量服务（baoshi-measure-cargo）
+
+本项目基于 **Orbbec 深度相机 + FastAPI**，用于在固定 ROI（感兴趣区域）内计算最近距离，并通过中位数输出稳定的高度测量结果。
+
+## 1. 项目功能
+
+- 提供 HTTP 接口 `/height`，返回毫米级测量结果。
+- 支持深度帧预处理（阈值过滤、中值滤波、形态学开运算、时间滤波）。
+- 支持自动保存本次测量的彩色图与深度标注图。
+- 内置请求日志中间件，记录接口耗时与响应摘要。
+
+## 2. 目录结构
+
+- `api.py`：FastAPI 服务入口与路由定义。
+- `main.py`：命令行启动入口。
+- `api_config.py`：配置对象与环境变量读取。
+- `cargo_service.py`：测高核心服务（采样、统计、图片保存）。
+- `depth_common.py`：深度处理通用逻辑（ROI、滤波、最近点等）。
+- `cargo_height_measure.py`：本地可视化调试脚本。
+- `request_logging.py`：请求日志中间件。
+- `utils.py`：彩色帧格式转换工具。
+- `examples/`：SDK 使用示例脚本。
+
+## 3. 环境准备
+
+### 3.1 Python 依赖
+
+```bash
+pip install -r requirements.txt
+```
+
+> 说明：请确保本机已正确安装并可使用 Orbbec SDK（`pyorbbecsdk`）。
+
+### 3.2 硬件要求
+
+- 已连接并可被系统识别的 Orbbec 深度相机。
+- 建议固定相机姿态，保证测量区域稳定。
+
+## 4. 启动方式
+
+### 4.1 启动 API 服务
+
+```bash
+python main.py
+```
+
+默认监听地址：`127.0.0.1:8080`
+
+### 4.2 本地窗口调试
+
+```bash
+python cargo_height_measure.py
+```
+
+按 `q` 或 `ESC` 退出。
+
+### 4.3 示例脚本
+
+```bash
+python examples/quick_start.py
+python examples/save_image_to_disk.py
+```
+
+## 5. 接口说明
+
+### 5.1 健康检查
+
+- 路径：`GET /health`
+- 返回示例：
+
+```json
+{"status": "ok"}
+```
+
+### 5.2 高度测量
+
+- 路径：`GET /height`
+- 成功返回示例：
+
+```json
+{
+  "height_mm": 1234,
+  "samples": [1231, 1235, 1234, 1232, 1236],
+  "unit": "mm",
+  "sample_count": 5,
+  "image_paths": {
+    "color": "/sample_images/color_640x480_20260227_133000_123.png",
+    "depth": "/sample_images/depth_annotated_640x480_20260227_133000_123.png"
+  }
+}
+```
+
+- 失败返回：当有效样本不足时返回 `503`。
+
+## 6. 关键环境变量
+
+- `SAMPLE_COUNT`：单次测量采样数量（默认 `10`）。
+- `FRAME_TIMEOUT_MS`：每帧等待超时毫秒数（默认 `200`）。
+- `SAMPLE_TIMEOUT_SEC`：单次测量总超时秒数（默认 `8`）。
+- `MAX_SAVED_IMAGES`：最多保留图片数量（默认 `1000`）。
+- `REQUEST_LOG_MAX_LEN`：日志中响应体最大长度（默认 `1000`）。
+- `REQUEST_LOG_MAX_BYTES`：单个日志文件最大字节（默认 `20MB`）。
+- `REQUEST_LOG_BACKUP_COUNT`：日志轮转保留份数（默认 `10`）。
+- `API_HOST`：服务监听地址（默认 `127.0.0.1`）。
+- `API_PORT`：服务端口（默认 `8080`）。
+- `MIN_DEPTH`：有效最小深度（mm，默认 `500`）。
+- `MAX_DEPTH`：有效最大深度（mm，默认 `4000`）。
+- `ROI_WIDTH_CM`：ROI 宽度（cm，默认 `10`）。
+- `ROI_HEIGHT_CM`：ROI 高度（cm，默认 `12`）。
+
+## 7. 日志与输出
+
+- 请求日志目录：`Log/request.log`
+- 测量图片目录：`sample_images/`
+- 示例脚本输出目录：
+  - `color_images/`
+  - `depth_images/`
+
+## 8. 常见问题
+
+- 如果 `/height` 返回 `503`：
+  - 检查相机是否正常连接；
+  - 检查当前场景是否在有效深度区间内；
+  - 适当增大 `SAMPLE_TIMEOUT_SEC` 或降低 `SAMPLE_COUNT`。
+- 如果没有彩色图：
+  - 设备可能不支持彩色传感器，或彩色流未成功启用。
+
+## 9. 开发建议
+
+- 优先在 `cargo_height_measure.py` 中观察 ROI 与最近点标注效果，再调整环境变量。
+- 对线上接口调用，建议同时监控 `request.log` 与保存的 `sample_images` 进行问题定位。

api.py

@@ -1,14 +1,30 @@
-from fastapi import FastAPI, HTTPException
+"""FastAPI 服务入口。
+
+该模块负责：
+1. 从环境变量加载配置；
+2. 初始化测高服务；
+3. 注册生命周期事件与 HTTP 路由；
+4. 启动 Uvicorn 服务。
+"""
+
+import os
+
 import uvicorn
+from fastapi import FastAPI, HTTPException
+from fastapi.staticfiles import StaticFiles
 
 from api_config import ApiConfig
 from cargo_service import CargoHeightService
 from request_logging import setup_request_logging
 
+# 读取配置并创建核心服务实例。
 config = ApiConfig.from_env()
 service = CargoHeightService(config)
 
+# 创建 Web 应用对象。
 app = FastAPI(title="Cargo Height API")
+
+# 安装请求日志中间件，用于记录每次 API 调用的响应与耗时。
 setup_request_logging(
     app,
     max_response_len=config.request_log_max_len,
@@ -16,31 +32,48 @@ setup_request_logging(
     backup_count=config.request_log_backup_count,
 )
 
+# 暴露本地样例图片目录，便于通过 HTTP 直接访问保存的测量图像。
+sample_images_dir = os.path.join(os.getcwd(), "sample_images")
+os.makedirs(sample_images_dir, exist_ok=True)
+app.mount("/sample_images", StaticFiles(directory=sample_images_dir), name="sample_images")
+
 
 @app.on_event("startup")
 def on_startup() -> None:
+    """应用启动时初始化相机管线与滤波器等资源。"""
     service.startup()
 
 
 @app.on_event("shutdown")
 def on_shutdown() -> None:
+    """应用关闭时释放相机资源，避免设备占用。"""
     service.shutdown()
 
 
 @app.get("/height")
 def get_height():
+    """执行一次测高流程并返回结果。
+
+    返回值中包含：
+    - `height_mm`: 估计高度（毫米）；
+    - `samples`: 本次采样的原始距离列表；
+    - `image_paths`: 相关彩色图与深度标注图路径。
+    """
     result = service.measure_height()
     if result is None:
+        # 采样不足时返回 503，表示服务暂时无法给出有效结果。
         raise HTTPException(status_code=503, detail="Insufficient valid samples from depth camera")
     return result
 
 
 @app.get("/health")
 def health():
+    """健康检查接口，用于探活。"""
     return {"status": "ok"}
 
 
 def main() -> None:
+    """启动 Uvicorn 服务器。"""
     uvicorn.run("api:app", host=config.api_host, port=config.api_port, log_level="info")
 
 

api_config.py

@@ -1,3 +1,8 @@
+"""API 配置模型。
+
+集中管理服务运行参数，并提供从环境变量读取配置的能力。
+"""
+
 import os
 from dataclasses import dataclass
 
@@ -6,6 +11,18 @@ from depth_common import Settings
 
 @dataclass(frozen=True)
 class ApiConfig:
+    """服务配置。
+
+    字段说明：
+    - `sample_count`: 单次测量期望采样帧数；
+    - `frame_timeout_ms`: 每次等待帧的超时时间；
+    - `sample_timeout_sec`: 整体采样总超时；
+    - `max_saved_images`: 本地最多保留图片数量；
+    - `request_log_*`: 请求日志相关限制；
+    - `api_host` / `api_port`: API 监听地址；
+    - `settings`: 深度处理参数集合。
+    """
+
     sample_count: int
     frame_timeout_ms: int
     sample_timeout_sec: int
@@ -19,6 +36,7 @@ class ApiConfig:
 
     @classmethod
     def from_env(cls) -> "ApiConfig":
+        """从环境变量构建配置，未设置时使用默认值。"""
         return cls(
             sample_count=int(os.getenv("SAMPLE_COUNT", "10")),
             frame_timeout_ms=int(os.getenv("FRAME_TIMEOUT_MS", "200")),

cargo_height_measure.py

@@ -1,3 +1,9 @@
+"""本地可视化测量脚本。
+
+用于直接打开深度相机窗口，实时查看 ROI、最近点与距离标注，
+便于调试参数与现场观测效果。
+"""
+
 import time
 
 import cv2
@@ -12,45 +18,53 @@ from depth_common import (
     nearest_distance_in_roi,
 )
 
-# 键盘退出键
+# 键盘退出键。
 ESC_KEY = 27
-# 打印间隔（秒）
-PRINT_INTERVAL = 1  # seconds
-# 从环境变量加载测量配置
+
+# 控制终端打印频率（秒）。
+PRINT_INTERVAL = 1
+
+# 从环境变量加载深度处理参数。
 SETTINGS = Settings.from_env()
 
 
 def main():
-    # 初始化时间滤波器，减少抖动
+    """启动实时深度查看窗口。"""
+    # 启用时间滤波以降低深度抖动。
     temporal_filter = TemporalFilter(alpha=0.5)
     try:
-        # 启动深度相机管线
+        # 初始化深度相机并获取深度内参。
         pipeline, depth_intrinsics, depth_profile = init_depth_pipeline()
         print("depth profile: ", depth_profile)
     except Exception as e:
         print(e)
         return
+
     last_print_time = time.time()
     while True:
         try:
-            # 获取一帧深度数据
+            # 等待一帧数据，超时时间 100ms。
             frames = pipeline.wait_for_frames(100)
             if frames is None:
                 continue
+
             depth_frame = frames.get_depth_frame()
             depth_data = extract_depth_data(depth_frame, SETTINGS, temporal_filter)
             if depth_data is None:
                 continue
-            # 计算中心 ROI 区域
+
+            # 计算中心 ROI 范围。
             bounds = compute_roi_bounds(depth_data, depth_intrinsics, SETTINGS)
             if bounds is None:
                 continue
+
             x_start, x_end, y_start, y_end, center_distance = bounds
             roi = depth_data[y_start:y_end, x_start:x_end]
-            # 计算 ROI 内最近距离
+
+            # 计算 ROI 最近距离。
             nearest_distance = nearest_distance_in_roi(roi, SETTINGS) or 0
 
-            # 找出 ROI 内最近点用于可视化
+            # 找到最近点坐标用于绘制圆点标记。
             nearest_point = find_nearest_point(
                 roi,
                 x_start,
@@ -59,9 +73,9 @@ def main():
                 nearest_distance,
             )
 
+            # 限频打印测量值，避免刷屏。
             current_time = time.time()
             if current_time - last_print_time >= PRINT_INTERVAL:
-                # 定期输出最近距离
                 print(
                     "nearest distance in "
                     f"{SETTINGS.roi_width_cm}cm x {SETTINGS.roi_height_cm}cm area: ",
@@ -69,7 +83,7 @@ def main():
                 )
                 last_print_time = current_time
 
-            # 生成彩色深度图并叠加标注
+            # 生成深度伪彩图并叠加标注信息。
             depth_image = cv2.normalize(depth_data, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
             depth_image = cv2.applyColorMap(depth_image, cv2.COLORMAP_JET)
 
@@ -83,11 +97,10 @@ def main():
             if nearest_point is not None:
                 cv2.circle(depth_image, nearest_point, 4, (0, 0, 0), -1)
                 cv2.circle(depth_image, nearest_point, 6, (0, 255, 255), 2)
-            # 文字标注当前测量值
-            label = f"nearest: {nearest_distance} mm"
+
             cv2.putText(
                 depth_image,
-                label,
+                f"nearest: {nearest_distance} mm",
                 (10, 30),
                 cv2.FONT_HERSHEY_SIMPLEX,
                 0.8,
@@ -95,10 +108,9 @@ def main():
                 2,
                 cv2.LINE_AA,
             )
-            center_label = f"center: {int(center_distance)} mm"
             cv2.putText(
                 depth_image,
-                center_label,
+                f"center: {int(center_distance)} mm",
                 (10, 60),
                 cv2.FONT_HERSHEY_SIMPLEX,
                 0.8,
@@ -109,11 +121,12 @@ def main():
 
             cv2.imshow("Depth Viewer", depth_image)
             key = cv2.waitKey(1)
-            if key == ord('q') or key == ESC_KEY:
+            if key == ord("q") or key == ESC_KEY:
                 break
         except KeyboardInterrupt:
             break
-    # 清理窗口与相机资源
+
+    # 退出前释放窗口和相机资源。
     cv2.destroyAllWindows()
     pipeline.stop()
 

cargo_service.py

@@ -1,3 +1,12 @@
+"""货物高度测量服务。
+
+该服务封装一次完整测量流程：
+1. 从深度相机连续采样；
+2. 在 ROI 内计算最近距离；
+3. 通过中位数汇总结果；
+4. 保存对应彩色图与深度标注图。
+"""
+
 import os
 import threading
 import time
@@ -19,14 +28,22 @@ from utils import frame_to_bgr_image
 
 
 class CargoHeightService:
+    """面向 API 的测高服务对象。"""
+
     def __init__(self, config: ApiConfig) -> None:
+        # 配置对象由外部注入，便于统一管理参数。
         self.config = config
+
+        # 以下对象在 startup() 中初始化，在 shutdown() 中释放。
         self._pipeline = None
         self._depth_intrinsics = None
         self._temporal_filter = None
+
+        # 使用互斥锁避免并发请求同时访问同一相机管线。
         self._lock = threading.Lock()
 
     def startup(self) -> None:
+        """初始化相机与滤波器。"""
         if self._pipeline is not None:
             return
         try:
@@ -38,28 +55,41 @@ class CargoHeightService:
         self._temporal_filter = TemporalFilter(alpha=0.5)
 
     def shutdown(self) -> None:
+        """释放相机资源。"""
         if self._pipeline is None:
             return
         self._pipeline.stop()
         self._pipeline = None
 
     def measure_height(self) -> Optional[Dict[str, Any]]:
+        """执行一次测量并返回结果。
+
+        返回 `None` 表示在规定时间内没有采够有效样本。
+        """
         start_time = time.time()
         samples = []
         first_valid_sample = None
         first_color_frame = None
 
+        # 相机读取与状态更新必须在锁内进行，确保线程安全。
         with self._lock:
             while len(samples) < self.config.sample_count and (time.time() - start_time) < self.config.sample_timeout_sec:
                 sample = self._measure_once()
                 if sample is None:
                     continue
+
+                # 记录最近距离样本用于最终统计。
                 samples.append(sample["nearest_distance"])
+
+                # 保留第一份有效样本，用于后续生成标注深度图。
                 if first_valid_sample is None:
                     first_valid_sample = sample
+
+                # 优先复用测量过程中采到的第一帧彩色图。
                 if first_color_frame is None and sample.get("color_frame") is not None:
                     first_color_frame = sample.get("color_frame")
 
+            # 若测量期间没拿到彩色帧，再额外尝试抓取几次。
             if first_color_frame is None:
                 for _ in range(5):
                     frames = self._pipeline.wait_for_frames(self.config.frame_timeout_ms)
@@ -70,39 +100,48 @@ class CargoHeightService:
                         first_color_frame = color_frame
                         break
 
+            image_paths = None
             if first_valid_sample is not None:
                 if first_color_frame is not None:
                     first_valid_sample["color_frame"] = first_color_frame
-                self._save_current_sample_images(first_valid_sample)
+                image_paths = self._save_current_sample_images(first_valid_sample)
 
         if len(samples) < self.config.sample_count:
             return None
 
+        # 用中位数抵抗离群值，比均值更稳健。
         median_value = int(np.median(np.array(samples, dtype=np.int32)))
         return {
             "height_mm": median_value,
             "samples": samples,
             "unit": "mm",
             "sample_count": self.config.sample_count,
+            "image_paths": image_paths,
         }
 
     def _measure_once(self) -> Optional[Dict[str, Any]]:
+        """采集并计算单帧测量结果。"""
         frames = self._pipeline.wait_for_frames(self.config.frame_timeout_ms)
         if frames is None:
             return None
 
         color_frame = frames.get_color_frame()
         depth_frame = frames.get_depth_frame()
+
+        # 深度帧预处理：格式检查、量纲转换、噪声过滤、时间滤波。
         depth_data = extract_depth_data(depth_frame, self.config.settings, self._temporal_filter)
         if depth_data is None:
             return None
 
+        # 根据中心距离动态计算 ROI 像素范围。
         bounds = compute_roi_bounds(depth_data, self._depth_intrinsics, self.config.settings)
         if bounds is None:
             return None
 
         x_start, x_end, y_start, y_end, center_distance = bounds
         roi = depth_data[y_start:y_end, x_start:x_end]
+
+        # 计算 ROI 中最近距离（或分位距离）。
         nearest_distance = nearest_distance_in_roi(roi, self.config.settings)
         if nearest_distance is None:
             return None
@@ -115,26 +154,32 @@ class CargoHeightService:
             "center_distance": center_distance,
         }
 
-    def _save_current_sample_images(self, sample: Dict[str, Any]) -> None:
+    def _save_current_sample_images(self, sample: Dict[str, Any]) -> Dict[str, Optional[str]]:
+        """保存当前样本的彩色图和深度标注图，并返回可访问路径。"""
         save_image_dir = os.path.join(os.getcwd(), "sample_images")
         os.makedirs(save_image_dir, exist_ok=True)
 
+        # 生成包含日期时间和毫秒的文件名，降低重名概率。
         now = time.localtime()
         time_str = time.strftime("%Y%m%d_%H%M%S", now)
         millis = int((time.time() % 1) * 1000)
         timestamp = f"{time_str}_{millis:03d}"
 
+        color_relative_path = None
         color_frame = sample.get("color_frame")
         if color_frame is not None:
             color_image = frame_to_bgr_image(color_frame)
             if color_image is not None:
                 color_height, color_width = color_image.shape[:2]
+                color_name = f"color_{color_width}x{color_height}_{timestamp}.png"
                 color_file = os.path.join(
                     save_image_dir,
-                    f"color_{color_width}x{color_height}_{timestamp}.png",
+                    color_name,
                 )
                 cv2.imwrite(color_file, color_image)
+                color_relative_path = f"/sample_images/{color_name}"
 
+        # 下面构建带标注的深度可视化图。
         depth_data = sample["depth_data"]
         x_start, x_end, y_start, y_end, center_distance = sample["bounds"]
         nearest_distance = sample["nearest_distance"]
@@ -176,15 +221,23 @@ class CargoHeightService:
         )
 
         depth_h, depth_w = depth_image.shape[:2]
+        depth_name = f"depth_annotated_{depth_w}x{depth_h}_{timestamp}.png"
         depth_file = os.path.join(
             save_image_dir,
-            f"depth_annotated_{depth_w}x{depth_h}_{timestamp}.png",
+            depth_name,
         )
         cv2.imwrite(depth_file, depth_image)
+
+        # 控制目录体积，超上限时删除最旧图片。
         self._prune_saved_images(save_image_dir, self.config.max_saved_images)
+        return {
+            "color": color_relative_path,
+            "depth": f"/sample_images/{depth_name}",
+        }
 
     @staticmethod
     def _prune_saved_images(save_dir: str, max_images: int) -> None:
+        """删除超出上限的旧图片文件。"""
         png_files = [
             os.path.join(save_dir, name)
             for name in os.listdir(save_dir)
@@ -192,9 +245,12 @@ class CargoHeightService:
         ]
         if len(png_files) <= max_images:
             return
+
+        # 按修改时间升序排序，优先删除最旧文件。
         png_files.sort(key=os.path.getmtime)
         for file_path in png_files[: len(png_files) - max_images]:
             try:
                 os.remove(file_path)
             except OSError:
+                # 删除失败时忽略，避免影响主流程。
                 pass

depth_common.py

@@ -1,14 +1,23 @@
+"""深度测量通用能力。
+
+该模块封装了深度相机初始化、深度帧预处理、ROI 计算、最近点提取等核心逻辑，
+供 API 服务与本地调试脚本共用。
+"""
+
 import os
 from dataclasses import dataclass
 
 import cv2
 import numpy as np
-
-from pyorbbecsdk import Config, Pipeline, OBSensorType, OBFormat, OBError
+from pyorbbecsdk import Config, OBError, OBFormat, OBSensorType, Pipeline
 
 
 def _get_env_int(name, default):
-    # 从环境变量读取整数，失败时返回默认值
+    """读取整型环境变量。
+
+    - 当变量不存在或为空字符串时返回默认值；
+    - 当变量无法转换为整数时返回默认值。
+    """
     value = os.getenv(name)
     if value is None or value.strip() == "":
         return default
@@ -20,13 +29,21 @@ def _get_env_int(name, default):
 
 @dataclass(frozen=True)
 class Settings:
-    # 深度数据处理参数
+    """深度处理参数集合。"""
+
+    # 有效深度区间（毫米）；超出范围的像素会被置为 0。
     min_depth: int
     max_depth: int
+
+    # 以图像中心为基准，计算物理尺寸为 roi_width_cm x roi_height_cm 的 ROI。
     roi_width_cm: int
     roi_height_cm: int
+
+    # 预处理参数：中值滤波核大小、形态学开运算核大小（建议为奇数）。
     median_blur_ksize: int
     morph_open_ksize: int
+
+    # 最近距离统计分位数；例如 5 表示取前 5% 的分位值，降低偶发噪声影响。
     nearest_percentile: int
 
     @classmethod
@@ -37,7 +54,7 @@ class Settings:
         morph_open_ksize=3,
         nearest_percentile=5,
     ):
-        # 从环境变量构建配置
+        """从环境变量创建参数对象。"""
         return cls(
             min_depth=_get_env_int("MIN_DEPTH", 500),
             max_depth=_get_env_int("MAX_DEPTH", 4000),
@@ -50,14 +67,21 @@ class Settings:
 
 
 class TemporalFilter:
+    """简易时间域平滑滤波器。
+
+    使用指数加权平均（EWMA）降低帧间抖动：
+    current = alpha * new + (1 - alpha) * previous
+    """
+
     def __init__(self, alpha):
-        # alpha 越大越偏向当前帧
+        # alpha 越大，结果越偏向当前帧；越小，结果越平滑。
         self.alpha = alpha
         self.previous_frame = None
 
     def process(self, frame):
-        # 对连续帧做指数加权平滑
+        """对单帧执行时间滤波并返回结果。"""
         if self.previous_frame is None:
+            # 第一帧没有历史值，直接返回原始帧。
             result = frame
         else:
             result = cv2.addWeighted(frame, self.alpha, self.previous_frame, 1 - self.alpha, 0)
@@ -66,9 +90,14 @@ class TemporalFilter:
 
 
 def init_depth_pipeline():
-    # 初始化相机并获取深度内参
+    """初始化深度相机管线并返回关键对象。
+
+    返回：`(pipeline, depth_intrinsics, depth_profile)`。
+    """
     config = Config()
     pipeline = Pipeline()
+
+    # 配置深度流（必须）。
     profile_list = pipeline.get_stream_profile_list(OBSensorType.DEPTH_SENSOR)
     if profile_list is None:
         raise RuntimeError("depth profile list is empty")
@@ -78,7 +107,7 @@ def init_depth_pipeline():
     depth_intrinsics = depth_profile.get_intrinsic()
     config.enable_stream(depth_profile)
 
-    # Optionally enable color stream so callers can access color frames when available.
+    # 尝试配置彩色流（可选）。部分设备不支持彩色，失败时保留深度能力即可。
     try:
         color_profile_list = pipeline.get_stream_profile_list(OBSensorType.COLOR_SENSOR)
         if color_profile_list is not None:
@@ -86,7 +115,6 @@ def init_depth_pipeline():
             if color_profile is not None:
                 config.enable_stream(color_profile)
     except OBError:
-        # Some devices do not provide color sensors; keep depth-only behavior.
         pass
 
     pipeline.start(config)
@@ -94,16 +122,18 @@ def init_depth_pipeline():
 
 
 def extract_depth_data(depth_frame, settings, temporal_filter):
-    # 从原始帧中提取并滤波深度数据
+    """从深度帧中提取并清洗深度矩阵（单位：毫米）。"""
     if depth_frame is None:
         return None
     if depth_frame.get_format() != OBFormat.Y16:
+        # 当前逻辑仅处理 Y16 格式深度帧。
         return None
+
     width = depth_frame.get_width()
     height = depth_frame.get_height()
     scale = depth_frame.get_depth_scale()
 
-    # 读取深度数据并转换单位（毫米）
+    # 1) 将原始缓冲区转为 2D 深度矩阵；2) 应用深度比例尺；3) 过滤无效深度范围。
     depth_data = np.frombuffer(depth_frame.get_data(), dtype=np.uint16)
     depth_data = depth_data.reshape((height, width))
     depth_data = depth_data.astype(np.float32) * scale
@@ -113,9 +143,11 @@ def extract_depth_data(depth_frame, settings, temporal_filter):
         0,
     ).astype(np.uint16)
 
-    # 中值滤波与开运算，去除噪点
+    # 中值滤波：抑制椒盐噪声；核必须是奇数。
     if settings.median_blur_ksize and settings.median_blur_ksize % 2 == 1:
         depth_data = cv2.medianBlur(depth_data, settings.median_blur_ksize)
+
+    # 开运算：移除孤立噪点，保留连通区域。
     if settings.morph_open_ksize and settings.morph_open_ksize % 2 == 1:
         kernel = cv2.getStructuringElement(
             cv2.MORPH_ELLIPSE,
@@ -125,7 +157,7 @@ def extract_depth_data(depth_frame, settings, temporal_filter):
         valid_mask = cv2.morphologyEx(valid_mask, cv2.MORPH_OPEN, kernel)
         depth_data = np.where(valid_mask > 0, depth_data, 0).astype(np.uint16)
 
-    # 可选的时间滤波
+    # 时间滤波：进一步减小帧间波动。
     if temporal_filter is not None:
         depth_data = temporal_filter.process(depth_data)
 
@@ -133,26 +165,40 @@ def extract_depth_data(depth_frame, settings, temporal_filter):
 
 
 def compute_roi_bounds(depth_data, depth_intrinsics, settings):
-    # 根据中心点距离动态计算 ROI 的像素范围
+    """计算中心 ROI 的像素边界。
+
+    逻辑说明：
+    1. 读取图像中心点深度作为当前距离参考；
+    2. 将目标物理尺寸（厘米）换算为该距离下的像素尺寸；
+    3. 将 ROI 裁剪到图像有效范围内。
+    """
     height, width = depth_data.shape
     center_y = height // 2
     center_x = width // 2
+
     center_distance = int(depth_data[center_y, center_x])
     if center_distance <= 0:
         return None
+
     center_distance_m = center_distance / 1000.0
     if center_distance_m <= 0:
         return None
+
     half_width_m = (settings.roi_width_cm / 100) / 2
     half_height_m = (settings.roi_height_cm / 100) / 2
+
+    # 使用相机内参把真实长度投影到像素平面。
     half_width_px = int(depth_intrinsics.fx * half_width_m / center_distance_m)
     half_height_px = int(depth_intrinsics.fy * half_height_m / center_distance_m)
     if half_width_px <= 0 or half_height_px <= 0:
         return None
+
+    # 防止 ROI 超出图像边界。
     half_width_px = min(half_width_px, center_x, width - center_x - 1)
     half_height_px = min(half_height_px, center_y, height - center_y - 1)
     if half_width_px <= 0 or half_height_px <= 0:
         return None
+
     x_start = center_x - half_width_px
     x_end = center_x + half_width_px + 1
     y_start = center_y - half_height_px
@@ -161,7 +207,10 @@ def compute_roi_bounds(depth_data, depth_intrinsics, settings):
 
 
 def nearest_distance_in_roi(roi, settings):
-    # 计算 ROI 内的最近距离（或按百分位）
+    """计算 ROI 内最近距离。
+
+    默认返回有效像素最小值；若设置了分位数则返回对应分位值。
+    """
     valid_values = roi[(roi >= settings.min_depth) & (roi <= settings.max_depth)]
     if valid_values.size == 0:
         return None
@@ -171,20 +220,26 @@ def nearest_distance_in_roi(roi, settings):
 
 
 def find_nearest_point(roi, x_start, y_start, settings, nearest_distance):
-    # 返回最近点在整图中的坐标
+    """在 ROI 中定位最近点，并返回其在整幅图中的坐标。"""
     if nearest_distance is None or nearest_distance <= 0:
         return None
+
+    # 先把无效深度置为最大值，确保 argmin 不会选到无效像素。
     roi_mask = (roi >= settings.min_depth) & (roi <= settings.max_depth)
     roi_candidate = np.where(roi_mask, roi, np.iinfo(np.uint16).max)
+
+    # 使用分位数策略时，仅保留不大于 nearest_distance 的候选点。
     if settings.nearest_percentile and 0 < settings.nearest_percentile < 100:
         roi_candidate = np.where(
             roi_candidate <= nearest_distance,
             roi_candidate,
             np.iinfo(np.uint16).max,
         )
+
     min_idx = np.argmin(roi_candidate)
     min_val = roi_candidate.flat[min_idx]
     if min_val == np.iinfo(np.uint16).max:
         return None
+
     min_y, min_x = np.unravel_index(min_idx, roi_candidate.shape)
     return x_start + min_x, y_start + min_y

examples/__init__.py

@@ -0,0 +1,6 @@
+"""示例脚本包。
+
+该目录包含 Orbbec SDK 的简单演示：
+- quick_start.py：实时显示彩色图和深度图；
+- save_image_to_disk.py：抓取并保存若干帧到本地。
+"""

examples/quick_start.py

@@ -1,8 +1,14 @@
-import cv2
-import numpy as np
+"""快速预览示例。
+
+同时显示彩色图与深度伪彩图，便于验证相机连接和图像质量。
+"""
+
 import time
 
+import cv2
+import numpy as np
 from pyorbbecsdk import *
+
 from utils import frame_to_bgr_image
 
 ESC_KEY = 27
@@ -11,12 +17,13 @@ MAX_DEPTH = 10000  # 10000mm
 
 
 def main():
+    """启动实时预览窗口。"""
     pipeline = Pipeline()
 
     pipeline.start()
     print("Pipeline started successfully. Press 'q' or ESC to exit.")
 
-    # Set window size
+    # 设置窗口大小，左右并排展示彩色图和深度图。
     window_width = 1280
     window_height = 720
     cv2.namedWindow("QuickStart Viewer", cv2.WINDOW_NORMAL)
@@ -28,13 +35,13 @@ def main():
             if frames is None:
                 continue
 
-            # Get color frame
+            # 读取彩色帧并转换为 OpenCV BGR 图像。
             color_frame = frames.get_color_frame()
             if color_frame is None:
                 continue
             color_image = frame_to_bgr_image(color_frame)
 
-            # Get depth frame
+            # 读取深度帧并校验格式。
             depth_frame = frames.get_depth_frame()
             if depth_frame is None:
                 continue
@@ -42,7 +49,7 @@ def main():
                 print("Depth format is not Y16")
                 continue
 
-            # Process depth data
+            # 深度数据预处理：重塑、按 scale 转毫米、过滤区间外像素。
             width = depth_frame.get_width()
             height = depth_frame.get_height()
             scale = depth_frame.get_depth_scale()
@@ -51,18 +58,18 @@ def main():
             depth_data = depth_data.astype(np.float32) * scale
             depth_data = np.where((depth_data > MIN_DEPTH) & (depth_data < MAX_DEPTH), depth_data, 0).astype(np.uint16)
 
-            # Create depth visualization
+            # 深度伪彩可视化。
             depth_image = cv2.normalize(depth_data, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
             depth_image = cv2.applyColorMap(depth_image, cv2.COLORMAP_JET)
 
-            # Resize and combine images
+            # 缩放并拼接：左彩色、右深度。
             color_image_resized = cv2.resize(color_image, (window_width // 2, window_height))
             depth_image_resized = cv2.resize(depth_image, (window_width // 2, window_height))
             combined_image = np.hstack((color_image_resized, depth_image_resized))
 
             cv2.imshow("QuickStart Viewer", combined_image)
 
-            if cv2.waitKey(1) in [ord('q'), ESC_KEY]:
+            if cv2.waitKey(1) in [ord("q"), ESC_KEY]:
                 break
         except KeyboardInterrupt:
             break

examples/save_image_to_disk.py

@@ -1,15 +1,22 @@
+"""保存图像示例。
+
+从相机采集少量彩色帧和深度帧并落盘，便于离线分析。
+"""
+
 import os
 
 import cv2
 import numpy as np
-
 from pyorbbecsdk import *
+
 from utils import frame_to_bgr_image
 
 
 def save_depth_frame(frame: DepthFrame, index):
+    """保存单帧深度数据为原始 .raw 文件。"""
     if frame is None:
         return
+
     width = frame.get_width()
     height = frame.get_height()
     timestamp = frame.get_timestamp()
@@ -18,26 +25,34 @@ def save_depth_frame(frame: DepthFrame, index):
     if depth_format != OBFormat.Y16:
         print("depth format is not Y16")
         return
+
+    # 深度数据转毫米后按 uint16 存储。
     data = np.frombuffer(frame.get_data(), dtype=np.uint16)
     data = data.reshape((height, width))
     data = data.astype(np.float32) * scale
     data = data.astype(np.uint16)
+
     save_image_dir = os.path.join(os.getcwd(), "depth_images")
     if not os.path.exists(save_image_dir):
         os.mkdir(save_image_dir)
+
     raw_filename = save_image_dir + "/depth_{}x{}_{}_{}.raw".format(width, height, index, timestamp)
     data.tofile(raw_filename)
 
 
 def save_color_frame(frame: ColorFrame, index):
+    """保存单帧彩色图为 PNG 文件。"""
     if frame is None:
         return
+
     width = frame.get_width()
     height = frame.get_height()
     timestamp = frame.get_timestamp()
+
     save_image_dir = os.path.join(os.getcwd(), "color_images")
     if not os.path.exists(save_image_dir):
         os.mkdir(save_image_dir)
+
     filename = save_image_dir + "/color_{}x{}_{}_{}.png".format(width, height, index, timestamp)
     image = frame_to_bgr_image(frame)
     if image is None:
@@ -47,12 +62,15 @@ def save_color_frame(frame: ColorFrame, index):
 
 
 def main():
+    """采集并保存若干帧样例图片。"""
     pipeline = Pipeline()
     config = Config()
     saved_color_cnt: int = 0
     saved_depth_cnt: int = 0
     has_color_sensor = False
+
     try:
+        # 优先尝试启用彩色流。
         profile_list = pipeline.get_stream_profile_list(OBSensorType.COLOR_SENSOR)
         if profile_list is not None:
             color_profile: VideoStreamProfile = profile_list.get_default_video_stream_profile()
@@ -60,25 +78,32 @@ def main():
             has_color_sensor = True
     except OBError as e:
         print(e)
+
+    # 启用深度流。
     depth_profile_list = pipeline.get_stream_profile_list(OBSensorType.DEPTH_SENSOR)
     if depth_profile_list is not None:
         depth_profile = depth_profile_list.get_default_video_stream_profile()
         config.enable_stream(depth_profile)
+
     pipeline.start(config)
     while True:
         try:
             frames = pipeline.wait_for_frames(100)
             if frames is None:
                 continue
+
+            # 已达到目标帧数则结束采集。
             if has_color_sensor:
                 if saved_color_cnt >= 5 and saved_depth_cnt >= 5:
                     break
             elif saved_depth_cnt >= 5:
                 break
+
             color_frame = frames.get_color_frame()
             if color_frame is not None and saved_color_cnt < 5:
                 save_color_frame(color_frame, saved_color_cnt)
                 saved_color_cnt += 1
+
             depth_frame = frames.get_depth_frame()
             if depth_frame is not None and saved_depth_cnt < 5:
                 save_depth_frame(depth_frame, saved_depth_cnt)

main.py

@@ -1,6 +1,8 @@
+"""命令行启动入口。"""
+
 from api import main
 
 
 if __name__ == "__main__":
-    # 命令行入口
+    # 直接复用 API 模块中的主函数，保持单一启动逻辑。
     main()

request_logging.py

@@ -1,3 +1,8 @@
+"""请求日志中间件。
+
+记录 FastAPI 每个请求的路径、状态码、耗时和响应摘要，支持日志轮转。
+"""
+
 import json
 import logging
 import os
@@ -10,16 +15,19 @@ from starlette.concurrency import iterate_in_threadpool
 
 
 def setup_request_logging(app: FastAPI, max_response_len: int, max_bytes: int, backup_count: int) -> None:
+    """为应用安装请求日志中间件。"""
     logger = logging.getLogger("cargo_height.request")
     _setup_request_logger(logger, max_bytes=max_bytes, backup_count=backup_count)
 
     @app.middleware("http")
     async def request_log_middleware(request: Request, call_next):
+        # 记录请求开始时间，用于后续耗时统计。
         request_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
         start = time.perf_counter()
         try:
             response = await call_next(request)
         except Exception:
+            # 异常请求按 500 记录并保留堆栈。
             elapsed_ms = (time.perf_counter() - start) * 1000
             logger.exception(
                 "request_time=%s method=%s path=%s status=%s duration_ms=%.2f response=%s",
@@ -32,6 +40,7 @@ def setup_request_logging(app: FastAPI, max_response_len: int, max_bytes: int, b
             )
             raise
 
+        # 读取响应体用于日志输出，然后恢复迭代器，避免影响客户端接收数据。
         body = b""
         async for chunk in response.body_iterator:
             body += chunk
@@ -55,6 +64,7 @@ def setup_request_logging(app: FastAPI, max_response_len: int, max_bytes: int, b
 
 
 def _setup_request_logger(logger: logging.Logger, max_bytes: int, backup_count: int) -> None:
+    """初始化请求日志记录器（文件 + 控制台）。"""
     log_dir = os.path.join(os.getcwd(), "Log")
     os.makedirs(log_dir, exist_ok=True)
     log_file = os.path.join(log_dir, "request.log")
@@ -87,10 +97,12 @@ def _setup_request_logger(logger: logging.Logger, max_bytes: int, backup_count:
 
 
 def _parse_response_text(body: bytes, content_type: str) -> str:
+    """根据响应类型提取可读日志文本。"""
     if not body:
         return ""
     if "application/json" in content_type:
         try:
+            # 统一 JSON 格式，便于检索。
             return json.dumps(json.loads(body), ensure_ascii=False)
         except Exception:
             return body.decode("utf-8", errors="replace")

utils.py

@@ -1,25 +1,31 @@
-from typing import Union, Any, Optional
+"""图像格式转换工具。
+
+该模块负责把 Orbbec SDK 的彩色帧转换为 OpenCV 常用 BGR 图像。
+"""
+
+from typing import Any, Optional, Union
 
 import cv2
 import numpy as np
-
-from pyorbbecsdk import FormatConvertFilter, VideoFrame
-from pyorbbecsdk import OBFormat, OBConvertFormat
+from pyorbbecsdk import FormatConvertFilter, OBConvertFormat, OBFormat, VideoFrame
 
 
 def yuyv_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
+    """将 YUYV 原始数据转换为 BGR 图像。"""
     yuyv = frame.reshape((height, width, 2))
     bgr_image = cv2.cvtColor(yuyv, cv2.COLOR_YUV2BGR_YUY2)
     return bgr_image
 
 
 def uyvy_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
+    """将 UYVY 原始数据转换为 BGR 图像。"""
     uyvy = frame.reshape((height, width, 2))
     bgr_image = cv2.cvtColor(uyvy, cv2.COLOR_YUV2BGR_UYVY)
     return bgr_image
 
 
 def i420_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
+    """将 I420 原始数据转换为 BGR 图像。"""
     y = frame[0:height, :]
     u = frame[height:height + height // 4].reshape(height // 2, width // 2)
     v = frame[height + height // 4:].reshape(height // 2, width // 2)
@@ -29,6 +35,7 @@ def i420_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
 
 
 def nv21_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
+    """将 NV21 原始数据转换为 BGR 图像。"""
     y = frame[0:height, :]
     uv = frame[height:height + height // 2].reshape(height // 2, width)
     yuv_image = cv2.merge([y, uv])
@@ -37,6 +44,7 @@ def nv21_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
 
 
 def nv12_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
+    """将 NV12 原始数据转换为 BGR 图像。"""
     y = frame[0:height, :]
     uv = frame[height:height + height // 2].reshape(height // 2, width)
     yuv_image = cv2.merge([y, uv])
@@ -45,6 +53,7 @@ def nv12_to_bgr(frame: np.ndarray, width: int, height: int) -> np.ndarray:
 
 
 def determine_convert_format(frame: VideoFrame):
+    """根据帧格式选择 SDK 可用的转换策略。"""
     if frame.get_format() == OBFormat.I420:
         return OBConvertFormat.I420_TO_RGB888
     elif frame.get_format() == OBFormat.MJPG:
@@ -62,12 +71,15 @@ def determine_convert_format(frame: VideoFrame):
 
 
 def frame_to_rgb_frame(frame: VideoFrame) -> Union[Optional[VideoFrame], Any]:
+    """将任意支持格式帧转换为 RGB 帧。"""
     if frame.get_format() == OBFormat.RGB:
         return frame
+
     convert_format = determine_convert_format(frame)
     if convert_format is None:
         print("Unsupported format")
         return None
+
     print("covert format: {}".format(convert_format))
     convert_filter = FormatConvertFilter()
     convert_filter.set_format_convert_format(convert_format)
@@ -78,10 +90,13 @@ def frame_to_rgb_frame(frame: VideoFrame) -> Union[Optional[VideoFrame], Any]:
 
 
 def frame_to_bgr_image(frame: VideoFrame) -> Union[Optional[np.array], Any]:
+    """将 SDK 视频帧转换为 OpenCV BGR 图像。"""
     width = frame.get_width()
     height = frame.get_height()
     color_format = frame.get_format()
     data = np.asanyarray(frame.get_data())
+
+    # 默认初始化为全黑图，后续按格式填充。
     image = np.zeros((height, width, 3), dtype=np.uint8)
     if color_format == OBFormat.RGB:
         image = np.resize(data, (height, width, 3))