Просмотр исходного кода

refactor(config, calibration, pusher): 更新配置与校准逻辑,简化第三方推送

1. 调整人体检测置信度阈值与人体跟踪阈值至0.45
2. 更新相机俯仰轴配置,改用手动标定查找表替代线性映射,新增tilt偏移补偿
3. 为校准器添加pan/tilt翻转支持与全局偏移补偿
4. 简化第三方推送的认证头与上报负载逻辑
wenhongquan 2 дней назад
Родитель
Сommit
6ef8d72d46

+ 14 - 0
dual_camera_system/calibration.py

@@ -570,6 +570,9 @@ class CameraCalibrator:
         self.tilt_y0 = ptz_config.get('tilt_y0', 0)
         self.tilt_y1 = ptz_config.get('tilt_y1', 45)
         self.tilt_curve_power = ptz_config.get('tilt_curve_power', 1.0)
+        # 安装方向翻转(与 ptz_camera.calculate_ptz_position 保持一致)
+        self.tilt_flip = ptz_config.get('tilt_flip', False)
+        self.pan_flip = ptz_config.get('pan_flip', False)
 
         # 校准配置
         self.stabilize_time = 1.0
@@ -1176,6 +1179,10 @@ class CameraCalibrator:
             pan_correction = self.pan_edge_offset * y_scale * math.copysign(abs(dx) ** self.pan_curve_power, dx)
             pan += pan_correction
 
+        # pan 方向翻转(与 ptz_camera.calculate_ptz_position 保持一致)
+        if self.pan_flip:
+            pan = -pan
+
         # 将pan归一化到[0, 360),便于发送给球机
         pan = pan % 360
 
@@ -1187,6 +1194,13 @@ class CameraCalibrator:
         else:
             tilt = self.tilt_offset + self.tilt_scale_x * x_ratio + self.tilt_scale_y * y_ratio
 
+        # tilt 方向翻转(与 ptz_camera.calculate_ptz_position 保持一致)
+        if self.tilt_flip:
+            tilt = -tilt
+
+        # 应用全局 tilt 偏移补偿(配置中正值=向下补偿)
+        tilt += self.tilt_offset_deg
+
         return (pan, tilt)
 
     def _interp_lookup(self, lookup: List[Tuple[float, float]], ratio: float) -> float:

+ 5 - 2
dual_camera_system/config/camera.py

@@ -81,6 +81,7 @@ CAMERA_GROUPS = [
             # 测试环境球机竖装且设备端已做倒影,与枪机同向,无需翻转 pan
             'pan_flip': False,
             # tilt_flip: 俯仰方向相反时设为 True(设备端 ceiling 模式已处理 tilt,代码层不翻转)
+            # 2026-06-15 实测:使用手动标定 lookup 表,不在 transform 中额外翻转
             'tilt_flip': False,
             # 视野映射:全景 x=0 -> pan=-90, x=1 -> pan=+90
             'pan_range': (-90, 90),
@@ -89,8 +90,10 @@ CAMERA_GROUPS = [
             # 人要被显示器挡住,需要适当向上看(tilt 正值)
             'tilt_range': (-5, 20),
             'tilt_center': 7,
-            'tilt_linear_enabled': True,
-            'tilt_y0': 13,                   # 经 RK3588 实测微调:y=0(画面上方/远处)略向上,使人头/上半身不被切出画面
+            # tilt 全局偏移补偿(度):正值向下,负值向上;用于现场微调
+            'tilt_offset': 5,
+            'tilt_linear_enabled': False,    # 使用手动标定的 tilt_lookup,不用线性映射
+            'tilt_y0': 13,                   # 备用:y=0(画面上方/远处)略向上,使人头/上半身不被切出画面
             'tilt_y1': -5,
             'tilt_curve_power': 0.8,
             # 自动校准扫描范围:完整 360° 粗扫 + 每步由下朝上细扫,建立全景→PTZ 映射

+ 2 - 2
dual_camera_system/config/detection.py

@@ -29,7 +29,7 @@ _MODEL_PATH, _MODEL_TYPE = _default_model()
 # 人体检测配置(用于 panorama_camera.ObjectDetector,多组模式)
 DETECTION_CONFIG = {
     'target_classes': ['person'],   # 检测目标类别 (支持中英文)
-    'confidence_threshold': 0.3,     # 置信度阈值(yolo26n end2end 内置 NMS)
+    'confidence_threshold': 0.45,     # 置信度阈值(yolo26n end2end 内置 NMS)
     'detection_fps': 2,              # 检测帧率(每秒检测帧数),替代原来的detection_interval
     'detection_interval': 4,       # 兼容保留:检测间隔(秒),当detection_fps=2时间隔为0.5秒
 
@@ -48,7 +48,7 @@ DETECTION_CONFIG = {
     'use_gpu': True,                # 是否使用 GPU / NPU
 
     # 人体检测后处理阈值
-    'person_threshold': 0.3,    # 进入联动跟踪的人体置信度阈值
+    'person_threshold': 0.45,    # 进入联动跟踪的人体置信度阈值
 
     # RKNN/ONNX 模型类别映射(yolo26n COCO80:0=person)
     'class_map': {0: 'person'},

+ 9 - 40
dual_camera_system/third_party_pusher.py

@@ -136,28 +136,8 @@ class ThirdPartyPusher:
                 logger.error(f"[第三方平台] 处理上报错误: {e}")
     
     def _get_auth_headers(self) -> Dict[str, str]:
-        """获取认证请求头"""
-        headers = {
-            'Content-Type': 'application/json',
-            'Accept': 'application/json',
-        }
-        
-        if self.auth_type == 'api_key':
-            headers['X-API-Key'] = self.api_key
-            if self.api_secret:
-                headers['X-API-Secret'] = self.api_secret
-        
-        elif self.auth_type == 'oauth2':
-            token = self._get_oauth2_token()
-            if token:
-                headers['Authorization'] = f'Bearer {token}'
-        
-        elif self.auth_type == 'basic':
-            import base64
-            credentials = base64.b64encode(f"{self.api_key}:{self.api_secret}".encode()).decode()
-            headers['Authorization'] = f'Basic {credentials}'
-        
-        return headers
+        """获取认证请求头(当前第三方接口不需要自定义 header,返回空避免 422)"""
+        return {}
     
     def _get_oauth2_token(self) -> Optional[str]:
         """获取 OAuth2 Token"""
@@ -300,29 +280,15 @@ class ThirdPartyPusher:
     def _build_payload(self, report: BatchReport) -> Dict[str, Any]:
         """
         构建上报请求体
-        
+
         Args:
             report: 批次上报数据
-            
+
         Returns:
             Dict: 请求体字典
         """
         batch_info = report.batch_info
-        
-        # 标准上报格式
-        payload = {
-            'deviceId': report.device_id,
-            'projectId': report.project_id,
-            'batchId': report.batch_id,
-            'timestamp': report.timestamp,
-            'datetime': datetime.fromtimestamp(report.timestamp).isoformat(),
-            'totalPersons': batch_info.get('total_persons', 0),
-            'ptzImagesCount': batch_info.get('ptz_images_count', 0),
-            'panorama': batch_info.get('panorama', {}),
-            'persons': batch_info.get('persons', []),
-            'uploadStatus': batch_info.get('upload_status', {}),
-        }
-        
+
         # 根据平台类型调整格式
         if self.platform_type == 'jtjai':
             # jtjai 平台特定格式
@@ -339,7 +305,10 @@ class ThirdPartyPusher:
                     'persons': batch_info.get('persons', []),
                 })
             }
-        
+        else:
+            # custom / 其他平台:原样发送 batch_info(snake_case)
+            payload = dict(batch_info)
+
         return payload
     
     def report_batch(self, batch_info: Dict[str, Any], local_path: Optional[str] = None):