dsh/dual_camera_system/app.py

"""FastAPI 应用工厂与全局服务初始化."""
import logging
import os
import tempfile
import threading
import time
from contextlib import asynccontextmanager

import cv2
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse

from config import CAMERA_GROUPS, SDK_PATH, SYSTEM_CONFIG, STORAGE_CONFIG
from dahua_sdk import DahuaSDK
from ptz_camera import PTZCamera
from third_party_pusher import get_third_party_pusher

from core.stream_manager import StreamManager
from core.scan_point_store import ScanPointStore
from core.spatial_scanner import SpatialScanner
from core.polling_scheduler import PollingScheduler
from core.capture_uploader import CaptureUploader
from core.detector_service import DetectorService
from core.group_state import group_state
from core.oss_uploader import OSSUploader
from core.file_cleanup import make_cleanup_workers, CleanupWorker
from web.routes import router
from web.state import WebState
import web.state as _web_state_module

logger = logging.getLogger(__name__)


def build_rtsp_url(camera_config: dict) -> str:
    if camera_config.get("rtsp_url"):
        return camera_config["rtsp_url"]
    ip = camera_config["ip"]
    port = camera_config.get("rtsp_port", 554)
    username = camera_config["username"]
    password = camera_config["password"]
    channel = camera_config.get("rtsp_channel") or camera_config.get("channel", 1)
    subtype = camera_config.get("subtype", 0)
    return f"rtsp://{username}:{password}@{ip}:{port}/cam/realmonitor?channel={channel}&subtype={subtype}"


def create_app(test_mode: bool = False) -> FastAPI:
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        # 重置全局状态，允许同一进程内反复创建应用（测试场景）
        group_state.reset()

        stream_manager = StreamManager()
        store_path = os.path.join(tempfile.mkdtemp(), "scan_models.json") if test_mode else "data/scan_models.json"
        scan_store = ScanPointStore(store_path)
        scanners: dict = {}
        schedulers: dict = {}
        ptz_cameras: dict = {}
        threads: list = []
        stop_event = threading.Event()

        pusher = None
        sdk = None
        detector_service = None
        oss_uploader = None
        cleanup_workers: list = []

        if not test_mode:
            if SYSTEM_CONFIG.get("enable_detection", True):
                detector_service = DetectorService()

            if SYSTEM_CONFIG.get("enable_ptz_camera", True):
                try:
                    sdk_path = os.path.join(SDK_PATH["lib_path"], SDK_PATH.get("netsdk", "libdhnetsdk.so"))
                    sdk = DahuaSDK(sdk_path)
                    sdk.init()
                except Exception as exc:
                    print(f"[lifespan] SDK init failed: {exc}")
                    sdk = None

            if SYSTEM_CONFIG.get("enable_event_push", False):
                try:
                    oss_uploader = OSSUploader()
                except Exception as exc:
                    print(f"[lifespan] OSS uploader init failed: {exc}")
                    oss_uploader = None

                try:
                    pusher = get_third_party_pusher()
                    if pusher and not pusher.running:
                        pusher.start()
                except Exception as exc:
                    print(f"[lifespan] Pusher start failed: {exc}")
                    pusher = None

            # 上报成功后，根据配置删除本地抓拍图片
            if pusher is not None:
                def _delete_captures_on_success(report):
                    if STORAGE_CONFIG.get("captures", {}).get("keep_local_copy", False):
                        return
                    for path in report.batch_info.get("image_paths") or []:
                        try:
                            if path and os.path.exists(path):
                                os.remove(path)
                                logger.info("[cleanup] 上报成功后删除本地图片: %s", path)
                        except Exception as exc:
                            logger.warning("[cleanup] 删除本地图片失败: %s, %s", path, exc)

                try:
                    pusher.set_callbacks(on_success=_delete_captures_on_success)
                except Exception as exc:
                    print(f"[lifespan] set pusher success callback failed: {exc}")

            # 启动 captures / previews 定期清理 worker
            try:
                group_ids = [
                    g.get("group_id", g.get("id"))
                    for g in CAMERA_GROUPS
                    if g.get("enabled", True)
                ]
                cleanup_workers = make_cleanup_workers(STORAGE_CONFIG, group_ids)
                for w in cleanup_workers:
                    w.start()
            except Exception as exc:
                print(f"[lifespan] cleanup worker start failed: {exc}")

        for group in CAMERA_GROUPS:
            if not group.get("enabled", True):
                continue
            gid = group.get("group_id", group.get("id"))

            try:
                ptz_cfg = group["ptz"]
                pano_cfg = group["panorama"]

                pano_url = build_rtsp_url(pano_cfg)
                ptz_url = build_rtsp_url(ptz_cfg)

                group_state.init_group(gid, pano_url, ptz_url, ptz_cfg)
                group_state.update(gid, "ptz_connected", False)
                scan_store.ensure_group(gid, {
                    "ptz_name": ptz_cfg.get("name", gid),
                    "panorama_name": pano_cfg.get("name", gid),
                })

                if test_mode:
                    continue

                # 注册枪机 RTSP 流
                if SYSTEM_CONFIG.get("enable_panorama_camera", True):
                    stream_manager.register(f"{gid}_panorama", pano_url)

                ptz = None
                if SYSTEM_CONFIG.get("enable_ptz_camera", True) and sdk is not None:
                    # 连接球机，并把组级别的校准文件路径透传给球机
                    ptz_cfg_with_calib = dict(ptz_cfg)
                    ptz_cfg_with_calib['calibration_file'] = group.get('calibration_file')
                    ptz = PTZCamera(sdk, ptz_cfg_with_calib)
                    try:
                        ptz_connected = ptz.connect()
                    except Exception as exc:
                        print(f"[lifespan] PTZ connect raised for group {gid}: {exc}")
                        ptz_connected = False

                    if ptz_connected:
                        ptz_cameras[gid] = ptz
                        group_state.update(gid, "ptz_connected", True)

                        # 注册球机 RTSP 流
                        stream_manager.register(f"{gid}_ptz", ptz_url)

                        scanners[gid] = SpatialScanner(
                            gid, ptz, lambda g=gid: stream_manager.get(f"{g}_ptz").get_frame()
                        )

                        def on_arrived(point, g=gid, ptz=ptz):
                            group_state.update(g, "ptz_position", {
                                "pan": point["pan"], "tilt": point["tilt"], "zoom": point.get("zoom", 1)
                            })

                        schedulers[gid] = PollingScheduler(
                            gid, ptz,
                            get_points=lambda g=gid: scan_store.list_enabled_points(g),
                            on_arrived=on_arrived,
                            default_dwell=3.0,
                        )
                    else:
                        print(f"[lifespan] PTZ connect failed for group {gid}; skipping PTZ features")

                # 创建上传器（即使球机关闭，枪机检测仍可能使用）
                uploader = CaptureUploader(
                    gid,
                    upload_callback=pusher.report_batch if pusher else None,
                    oss_uploader=oss_uploader,
                )

                if SYSTEM_CONFIG.get("enable_detection", True) and detector_service is not None:
                    def panorama_detect_loop(g=gid, uploader=uploader, detector=detector_service):
                        interval = 0.5
                        while not stop_event.is_set():
                            try:
                                stream = stream_manager.get(f"{g}_panorama")
                                frame = stream.get_frame() if stream else None
                                if frame is not None:
                                    dets = detector.detect(frame)
                                    marked = frame.copy()
                                    if dets:
                                        det_dicts = [{
                                            "bbox": [d.bbox[0], d.bbox[1], d.bbox[0]+d.bbox[2], d.bbox[1]+d.bbox[3]],
                                            "confidence": d.confidence,
                                        } for d in dets]
                                        uploader.handle_detection("panorama", frame, det_dicts)
                                        for d in det_dicts:
                                            x1, y1, x2, y2 = d["bbox"]
                                            cv2.rectangle(marked, (x1, y1), (x2, y2), (0, 255, 0), 2)
                                    stream.set_marked_frame(marked)
                            except Exception as exc:
                                print(f"[panorama_detect_loop {g}] error: {exc}")
                            time.sleep(interval)

                    t_panorama = threading.Thread(target=panorama_detect_loop, daemon=True)
                    t_panorama.start()
                    threads.append(t_panorama)

                    if ptz is not None:
                        def ptz_detect_loop(g=gid, uploader=uploader, detector=detector_service):
                            interval = 0.2
                            while not stop_event.is_set():
                                try:
                                    stream = stream_manager.get(f"{g}_ptz")
                                    frame = stream.get_frame() if stream else None
                                    if frame is not None:
                                        dets = detector.detect(frame)
                                        marked = frame.copy()
                                        if dets:
                                            det_dicts = [{
                                                "bbox": [d.bbox[0], d.bbox[1], d.bbox[0]+d.bbox[2], d.bbox[1]+d.bbox[3]],
                                                "confidence": d.confidence,
                                            } for d in dets]
                                            pos = group_state.get(g).get("ptz_position")
                                            uploader.handle_detection("ptz", frame, det_dicts, pos)
                                            for d in det_dicts:
                                                x1, y1, x2, y2 = d["bbox"]
                                                cv2.rectangle(marked, (x1, y1), (x2, y2), (0, 255, 0), 2)
                                        stream.set_marked_frame(marked)
                                except Exception as exc:
                                    print(f"[ptz_detect_loop {g}] error: {exc}")
                                time.sleep(interval)

                        t_ptz = threading.Thread(target=ptz_detect_loop, daemon=True)
                        t_ptz.start()
                        threads.append(t_ptz)

            except Exception as exc:
                print(f"[lifespan] Group {gid} setup failed: {exc}")
                continue

        _web_state_module.web_state = WebState(
            group_state, stream_manager, scan_store, scanners, schedulers, ptz_cameras
        )

        yield

        # 清理
        stop_event.set()
        for t in threads:
            try:
                t.join(timeout=2.0)
            except Exception as exc:
                print(f"[lifespan] Thread join error: {exc}")

        for s in schedulers.values():
            try:
                s.stop()
            except Exception as exc:
                print(f"[lifespan] Scheduler stop error: {exc}")

        for gid, ptz in ptz_cameras.items():
            try:
                ptz.disconnect()
                stream = stream_manager.get(f"{gid}_ptz")
                if stream:
                    stream.stop()
            except Exception as exc:
                print(f"[lifespan] PTZ cleanup error for {gid}: {exc}")

        stream_manager.stop_all()

        if pusher and getattr(pusher, "running", False):
            try:
                pusher.stop()
            except Exception as exc:
                print(f"[lifespan] Pusher stop error: {exc}")

        if sdk:
            try:
                sdk.cleanup()
            except Exception as exc:
                print(f"[lifespan] SDK cleanup error: {exc}")

        for w in cleanup_workers:
            try:
                w.stop()
            except Exception as exc:
                print(f"[lifespan] cleanup worker stop error: {exc}")

    if test_mode:
        # 在测试模式下预先初始化共享状态（不依赖 lifespan，兼容非上下文管理器的 TestClient）
        group_state.reset()
        test_stream_manager = StreamManager()
        test_store_path = os.path.join(tempfile.mkdtemp(), "scan_models.json")
        test_scan_store = ScanPointStore(test_store_path)
        for group in CAMERA_GROUPS:
            if not group.get("enabled", True):
                continue
            gid = group.get("group_id", group.get("id"))
            try:
                ptz_cfg = group["ptz"]
                pano_cfg = group["panorama"]
                pano_url = build_rtsp_url(pano_cfg)
                ptz_url = build_rtsp_url(ptz_cfg)
                group_state.init_group(gid, pano_url, ptz_url, ptz_cfg)
                group_state.update(gid, "ptz_connected", False)
                test_scan_store.ensure_group(gid, {
                    "ptz_name": ptz_cfg.get("name", gid),
                    "panorama_name": pano_cfg.get("name", gid),
                })
            except Exception as exc:
                print(f"[create_app] Test group {gid} setup failed: {exc}")
        _web_state_module.web_state = WebState(
            group_state, test_stream_manager, test_scan_store, {}, {}, {}
        )

    app = FastAPI(lifespan=lifespan)
    static_dir = os.path.join(os.path.dirname(__file__), "web_static")
    if os.path.isdir(static_dir):
        app.mount("/static", StaticFiles(directory=static_dir), name="static")
    app.include_router(router)

    @app.get("/")
    async def root():
        index_path = os.path.join(static_dir, "index.html")
        if os.path.isfile(index_path):
            return FileResponse(index_path)
        return {"message": "PTZ 360 scan + panorama polling system"}

    return app