deploy_pickup/core/human_detection_reader.py

"""
人体检测 SQLite 读取模块

读取 YOLO 人体检测结果，判断是否在冷却期内
用于抑制异常报警：5 分钟内检测到人 → 不报警
"""
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional
import logging

logger = logging.getLogger('PickupMonitor')


class HumanDetectionReader:
    """
    人体检测结果读取器
    
    只要任意摄像头在指定时间窗口内检测到人，就返回 True（抑制报警）
    """
    
    def __init__(self, db_path: str, cooldown_minutes: int = 5):
        """
        初始化读取器
        
        参数:
            db_path: SQLite 数据库路径
            cooldown_minutes: 冷却时间（分钟），检测到人后多久内抑制报警
        """
        self.db_path = Path(db_path) if db_path else None
        self.cooldown_minutes = cooldown_minutes
        
        # 缓存：避免频繁查询数据库
        self._cache_result: Optional[bool] = None
        self._cache_time: Optional[datetime] = None
        self._cache_ttl_seconds = 10  # 缓存有效期（秒）
    
    def _get_connection(self) -> Optional[sqlite3.Connection]:
        """
        获取数据库连接
        
        返回:
            连接对象，数据库不存在时返回 None
        """
        if self.db_path is None or not self.db_path.exists():
            return None
        
        try:
            conn = sqlite3.connect(str(self.db_path), timeout=5)
            return conn
        except Exception as e:
            logger.warning(f"人体检测数据库连接失败: {e}")
            return None
    
    def get_last_person_time(self) -> Optional[datetime]:
        """
        获取最后一次检测到人的时间（任意摄像头）
        
        返回:
            最后检测到人的时间，无数据返回 None
        """
        conn = self._get_connection()
        if conn is None:
            return None
        
        try:
            row = conn.execute("""
                SELECT datetime FROM detections 
                WHERE detected = 1 
                ORDER BY datetime DESC 
                LIMIT 1
            """).fetchone()
            conn.close()
            
            if row:
                # 解析时间字符串
                return datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S")
            return None
            
        except Exception as e:
            logger.warning(f"查询人体检测数据失败: {e}")
            if conn:
                conn.close()
            return None
    
    def is_in_cooldown(self) -> bool:
        """
        判断是否在冷却期内（任意摄像头检测到人后的 N 分钟内）
        
        返回:
            True: 在冷却期内，应抑制报警
            False: 不在冷却期，可正常报警
        """
        now = datetime.now()
        
        # 检查缓存是否有效
        if (self._cache_result is not None and 
            self._cache_time is not None and
            (now - self._cache_time).total_seconds() < self._cache_ttl_seconds):
            return self._cache_result
        
        # 查询数据库
        last_person_time = self.get_last_person_time()
        
        if last_person_time is None:
            result = False  # 从未检测到人，不抑制
        else:
            # 判断是否在冷却期内
            cooldown_end = last_person_time + timedelta(minutes=self.cooldown_minutes)
            result = now < cooldown_end
        
        # 更新缓存
        self._cache_result = result
        self._cache_time = now
        
        return result
    
    def get_status_info(self) -> str:
        """
        获取状态信息（用于日志）
        
        返回:
            状态描述字符串
        """
        last_time = self.get_last_person_time()
        if last_time is None:
            return "无人体检测数据"
        
        elapsed = (datetime.now() - last_time).total_seconds() / 60
        if elapsed < self.cooldown_minutes:
            remaining = self.cooldown_minutes - elapsed
            return f"冷却中(剩余{remaining:.1f}分钟)"
        else:
            return f"已超时({elapsed:.1f}分钟前有人)"