dsh/dual_camera_system/camera_group.py

"""
摄像头组封装类
封装一组全景+球机摄像头的所有组件
"""

import os
import time
import logging
import threading
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass
import numpy as np

from panorama_camera import PanoramaCamera, ObjectDetector, DetectedObject
from ptz_camera import PTZCamera
from ocr_recognizer import NumberDetector, PersonInfo
from coordinator import SequentialCoordinator
from calibration import CameraCalibrator, CalibrationManager
from paired_image_saver import PairedImageSaver, get_paired_saver
from oss_uploader import get_oss_uploader
from third_party_pusher import get_third_party_pusher
from config import DEVICE_CONFIG, S3_COMPATIBLE_CONFIG, THIRD_PARTY_CONFIG

logger = logging.getLogger(__name__)


@dataclass
class GroupConfig:
    """摄像头组配置"""
    group_id: str
    name: str
    panorama_config: Dict[str, Any]
    ptz_config: Dict[str, Any]
    calibration_file: str
    paired_image_dir: str


class CameraGroup:
    """
    摄像头组封装类
    
    封装一组全景+球机摄像头的所有组件，包括：
    - 全景摄像头实例
    - 球机实例
    - 校准器
    - 联动控制器
    - 配对图片保存器
    """
    
    def __init__(self, 
                 group_config: Dict[str, Any],
                 sdk,
                 detector: ObjectDetector,
                 number_detector: Optional[NumberDetector] = None,
                 shared_config: Optional[Dict[str, Any]] = None):
        """
        初始化摄像头组
        
        Args:
            group_config: 组配置字典
            sdk: 大华SDK实例（共享）
            detector: 检测器实例（共享）
            number_detector: 编号检测器实例（可选，共享）
            shared_config: 共享配置（校准配置、联动配置等）
        """
        self.group_id = group_config.get('group_id', 'unknown')
        self.name = group_config.get('name', self.group_id)
        self.config = group_config
        self.sdk = sdk
        self.detector = detector
        self.number_detector = number_detector
        self.shared_config = shared_config or {}
        
        # 组件实例
        self.panorama_camera: Optional[PanoramaCamera] = None
        self.ptz_camera: Optional[PTZCamera] = None
        self.calibrator: Optional[CameraCalibrator] = None
        self.calibration_manager: Optional[CalibrationManager] = None
        self.coordinator: Optional[SequentialCoordinator] = None
        self.paired_saver: Optional[PairedImageSaver] = None
        
        # 校准数据文件路径
        self.calibration_file = group_config.get(
            'calibration_file', 
            f'/home/admin/dsh/calibration_{self.group_id}.json'
        )
        
        # 配对图片目录
        self.paired_image_dir = group_config.get(
            'paired_image_dir',
            f'/home/admin/dsh/paired_images_{self.group_id}'
        )
        
        # 运行状态
        self.running = False
        self.initialized = False
        
        # 定时校准
        self.calibration_thread = None
        self.calibration_running = False
        
        logger.info(f"[{self.group_id}] 摄像头组实例创建: {self.name}")
    
    def initialize(self, skip_calibration: bool = False) -> bool:
        """
        初始化组内所有组件
        
        Args:
            skip_calibration: 是否跳过校准
            
        Returns:
            是否成功
        """
        logger.info(f"[{self.group_id}] 开始初始化摄像头组...")
        
        # 1. 初始化全景摄像头
        panorama_config = self.config.get('panorama', {})
        panorama_config['group_id'] = self.group_id
        self.panorama_camera = PanoramaCamera(self.sdk, panorama_config)
        
        # 2. 初始化球机
        ptz_config = self.config.get('ptz', {})
        ptz_config['group_id'] = self.group_id
        self.ptz_camera = PTZCamera(self.sdk, ptz_config)
        
        # 3. 初始化配对图片保存器（使用单例模式，支持 OSS）
        try:
            # 获取 OSS 上传器
            oss_uploader = None
            enable_oss = S3_COMPATIBLE_CONFIG.get('enabled', False)
            if enable_oss:
                oss_uploader = get_oss_uploader()
                if not oss_uploader.running:
                    oss_uploader.start()
                logger.info(f"[{self.group_id}] OSS 上传已启用")
            
            # 创建设备配置（包含 group_id）
            device_config = DEVICE_CONFIG.copy()
            device_config['group_id'] = self.group_id
            
            # 使用单例获取配对保存器
            self.paired_saver = get_paired_saver(
                base_dir=self.paired_image_dir,
                time_window=5.0,
                enable_oss=enable_oss,
                oss_uploader=oss_uploader,
                device_config=device_config
            )
            
            # 设置第三方平台推送回调
            if THIRD_PARTY_CONFIG.get('enabled', False):
                pusher = get_third_party_pusher()
                if not pusher.running:
                    pusher.start()
                self.paired_saver.set_upload_callback(pusher.report_batch)
                logger.info(f"[{self.group_id}] 第三方平台推送已启用")
            
            logger.info(f"[{self.group_id}] 配对图片保存器初始化成功: {self.paired_image_dir}")
        except Exception as e:
            logger.warning(f"[{self.group_id}] 配对图片保存器初始化失败: {e}")
        
        # 4. 初始化联动控制器
        self.coordinator = SequentialCoordinator(
            self.panorama_camera,
            self.ptz_camera,
            self.detector,
            self.number_detector
        )
        
        # 应用顺序模式配置
        from config import COORDINATOR_CONFIG
        seq_config = COORDINATOR_CONFIG.get('sequential_mode', {})
        self.coordinator.set_capture_config(
            ptz_stabilize_time=seq_config.get('ptz_stabilize_time', 2.5),
            capture_wait_time=seq_config.get('capture_wait_time', 0.5),
            return_to_panorama=seq_config.get('return_to_panorama', True),
            default_pan=seq_config.get('default_pan', 0.0),
            default_tilt=seq_config.get('default_tilt', 0.0),
            default_zoom=seq_config.get('default_zoom', 1)
        )
        
        # 设置配对保存器
        if self.paired_saver:
            self.coordinator._paired_saver = self.paired_saver
            self.coordinator._enable_paired_saving = True
        
        # 5. 设置回调
        self._setup_callbacks()
        
        logger.info(f"[{self.group_id}] 组件初始化完成")
        
        # 6. 执行校准
        from config import SYSTEM_CONFIG
        should_calibrate = SYSTEM_CONFIG.get('enable_calibration', True)
        if not should_calibrate:
            logger.info(f"[{self.group_id}] 自动校准已禁用")
        elif skip_calibration:
            logger.info(f"[{self.group_id}] 自动校准已跳过")
        else:
            if not self._auto_calibrate():
                logger.error(f"[{self.group_id}] 自动校准失败!")
                return False
        
        self.initialized = True
        return True
    
    def _auto_calibrate(self, force: bool = False) -> bool:
        """
        执行自动校准
        
        Args:
            force: 是否强制重新校准
            
        Returns:
            是否成功
        """
        logger.info(f"[{self.group_id}] 开始自动校准...")
        
        # 连接摄像头
        if not self.panorama_camera.connect():
            logger.error(f"[{self.group_id}] 连接全景摄像头失败")
            return False
        
        if not self.ptz_camera.connect():
            logger.error(f"[{self.group_id}] 连接球机失败")
            self.panorama_camera.disconnect()
            return False
        
        # 启动视频流
        if not self.panorama_camera.start_stream_rtsp():
            logger.warning(f"[{self.group_id}] 全景RTSP启动失败，尝试SDK方式...")
            self.panorama_camera.start_stream()
        
        if not self.ptz_camera.start_stream_rtsp():
            logger.warning(f"[{self.group_id}] 球机RTSP启动失败")
        
        # 等待视频流稳定
        logger.info(f"[{self.group_id}] 等待视频流稳定...")
        time.sleep(3)
        
        # 创建校准器
        self.calibrator = CameraCalibrator(
            ptz_camera=self.ptz_camera,
            get_frame_func=self.panorama_camera.get_frame,
            detect_marker_func=None,
            ptz_capture_func=self._capture_ptz_frame,
            calibration_file=self.calibration_file
        )
        
        # 配置校准参数
        from config import CALIBRATION_CONFIG
        overlap_cfg = CALIBRATION_CONFIG.get('overlap_discovery', {})
        self.calibrator.overlap_pan_range = overlap_cfg.get('pan_range', (0, 360))
        self.calibrator.overlap_tilt_range = overlap_cfg.get('tilt_range', (-30, 30))
        self.calibrator.overlap_pan_step = overlap_cfg.get('pan_step', 20)
        self.calibrator.overlap_tilt_step = overlap_cfg.get('tilt_step', 15)
        self.calibrator.stabilize_time = overlap_cfg.get('stabilize_time', 2.0)
        
        # 创建校准管理器
        self.calibration_manager = CalibrationManager(self.calibrator)
        
        # 执行校准
        result = self.calibration_manager.auto_calibrate(
            force=force,
            fallback_on_failure=True
        )
        
        if not result.success:
            logger.error(f"[{self.group_id}] 校准失败: {result.error_message}")
            return False
        
        logger.info(f"[{self.group_id}] 校准成功! RMS误差: {result.rms_error:.4f}")
        
        # 设置校准器到联动控制器
        if self.coordinator and self.calibrator.is_calibrated():
            self.coordinator.set_calibrator(self.calibrator)
        
        return True
    
    def _capture_ptz_frame(self) -> Optional[np.ndarray]:
        """从球机抓拍一帧"""
        if self.ptz_camera is None:
            return None
        try:
            return self.ptz_camera.get_frame()
        except Exception as e:
            logger.error(f"[{self.group_id}] 球机抓拍失败: {e}")
            return None
    
    def _setup_callbacks(self):
        """设置回调函数"""
        def on_person_detected(person: DetectedObject, frame: np.ndarray):
            logger.info(f"[{self.group_id}] 检测到人体: 位置={person.center}, 置信度={person.confidence:.2f}")
        
        def on_number_recognized(person_info: PersonInfo):
            logger.info(f"[{self.group_id}] 识别到编号: {person_info.number_text}")
        
        if self.coordinator:
            self.coordinator.on_person_detected = on_person_detected
            self.coordinator.on_number_recognized = on_number_recognized
    
    def start(self) -> bool:
        """
        启动组内联动
        
        Returns:
            是否成功
        """
        if not self.initialized:
            logger.error(f"[{self.group_id}] 组件未初始化，无法启动")
            return False
        
        logger.info(f"[{self.group_id}] 启动摄像头组...")
        
        # 启动联动控制器
        if self.coordinator:
            if not self.coordinator.start():
                logger.error(f"[{self.group_id}] 联动控制器启动失败")
                return False
        
        self.running = True
        logger.info(f"[{self.group_id}] 摄像头组启动成功")
        return True
    
    def stop(self):
        """停止组内所有组件"""
        logger.info(f"[{self.group_id}] 停止摄像头组...")
        
        self.running = False
        self.calibration_running = False
        
        # 停止联动控制器
        if self.coordinator:
            self.coordinator.stop()
        
        # 停止视频流
        if self.panorama_camera:
            self.panorama_camera.stop_stream()
            self.panorama_camera.disconnect()
        
        if self.ptz_camera:
            self.ptz_camera.stop_stream()
            self.ptz_camera.disconnect()
        
        # 关闭配对保存器
        if self.paired_saver:
            self.paired_saver.close()
        
        # 停止 OSS 上传器和第三方推送器（只在最后一个组停止时执行）
        try:
            from oss_uploader import reset_oss_uploader
            from third_party_pusher import reset_third_party_pusher
            reset_oss_uploader()
            reset_third_party_pusher()
        except Exception as e:
            logger.debug(f"[{self.group_id}] 清理全局上传器: {e}")
        
        logger.info(f"[{self.group_id}] 摄像头组已停止")
    
    def get_status(self) -> Dict[str, Any]:
        """获取组状态"""
        return {
            'group_id': self.group_id,
            'name': self.name,
            'running': self.running,
            'initialized': self.initialized,
            'calibrated': self.calibrator.is_calibrated() if self.calibrator else False,
            'panorama_connected': self.panorama_camera.is_connected() if self.panorama_camera else False,
            'ptz_connected': self.ptz_camera.is_connected() if self.ptz_camera else False,
        }
    
    def start_scheduled_calibration(self, interval_hours: int = 24, daily_time: str = '08:00'):
        """
        启动定时校准
        
        Args:
            interval_hours: 校准间隔（小时）
            daily_time: 每日校准时间 (HH:MM)
        """
        if self.calibration_thread and self.calibration_thread.is_alive():
            logger.warning(f"[{self.group_id}] 定时校准已在运行")
            return
        
        self.calibration_running = True
        self.calibration_thread = threading.Thread(
            target=self._calibration_scheduler,
            args=(interval_hours, daily_time),
            name=f"calibration-{self.group_id}",
            daemon=True
        )
        self.calibration_thread.start()
        logger.info(f"[{self.group_id}] 定时校准已启动，每日 {daily_time} 执行")
    
    def _calibration_scheduler(self, interval_hours: int, daily_time: str):
        """定时校准调度器"""
        from datetime import datetime, timedelta
        import time
        
        while self.calibration_running and self.running:
            try:
                # 计算下次校准时间
                now = datetime.now()
                target_time = datetime.strptime(daily_time, '%H:%M')
                next_calibration = now.replace(
                    hour=target_time.hour,
                    minute=target_time.minute,
                    second=0,
                    microsecond=0
                )
                
                if next_calibration <= now:
                    next_calibration += timedelta(days=1)
                
                wait_seconds = (next_calibration - now).total_seconds()
                logger.info(f"[{self.group_id}] 下次校准时间: {next_calibration} (等待 {wait_seconds/3600:.1f} 小时)")
                
                # 等待
                time.sleep(min(wait_seconds, 60))  # 每分钟检查一次
                
                if not self.calibration_running or not self.running:
                    break
                
                # 检查是否到达校准时间
                if datetime.now() >= next_calibration:
                    logger.info(f"[{self.group_id}] 开始定时校准...")
                    self._auto_calibrate(force=True)
                    
            except Exception as e:
                logger.error(f"[{self.group_id}] 定时校准错误: {e}")
                time.sleep(60)