154 lines
4.6 KiB
Python
154 lines
4.6 KiB
Python
"""
|
||
摄像头管理器
|
||
处理摄像头配置和URL生成
|
||
"""
|
||
import logging
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import requests
|
||
from requests.adapters import HTTPAdapter
|
||
from urllib3.util.retry import Retry
|
||
|
||
from .config import BASE_URL, CAMERA_URL, CAMERAS
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class CameraManager:
|
||
"""摄像头管理器"""
|
||
|
||
def __init__(self, cameras: Optional[List[Dict[str, Any]]] = None) -> None:
|
||
"""
|
||
初始化摄像头管理器
|
||
|
||
Args:
|
||
cameras: 摄像头配置列表，如果为None则使用全局配置
|
||
"""
|
||
self.base_url: str = BASE_URL
|
||
self.camera_url: str = CAMERA_URL
|
||
self.cameras: List[Dict[str, Any]] = cameras if cameras is not None else CAMERAS.copy()
|
||
|
||
# 初始化请求会话，配置连接池和重试策略
|
||
self.session: requests.Session = requests.Session()
|
||
|
||
# 配置重试策略
|
||
retry_strategy = Retry(
|
||
total=3,
|
||
backoff_factor=1,
|
||
status_forcelist=[429, 500, 502, 503, 504],
|
||
method_whitelist=["HEAD", "GET", "OPTIONS"]
|
||
)
|
||
|
||
# 配置适配器
|
||
adapter = HTTPAdapter(
|
||
max_retries=retry_strategy,
|
||
pool_connections=10,
|
||
pool_maxsize=10
|
||
)
|
||
self.session.mount("http://", adapter)
|
||
self.session.mount("https://", adapter)
|
||
|
||
# 配置请求头
|
||
self.session.headers.update({
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
|
||
'Accept': 'application/json, text/plain, */*',
|
||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||
'Content-Type': 'application/json',
|
||
'Connection': 'keep-alive',
|
||
})
|
||
|
||
def get_camera_url(self, camera_id: int, camera_number: str = 'mixed') -> str:
|
||
"""
|
||
根据摄像头ID和编号生成URL
|
||
|
||
Args:
|
||
camera_id: 摄像头ID
|
||
camera_number: 摄像头编号，默认为'mixed'
|
||
|
||
Returns:
|
||
摄像头URL
|
||
|
||
Raises:
|
||
ValueError: 摄像头ID不存在
|
||
"""
|
||
camera = next((c for c in self.cameras if c['id'] == camera_id), None)
|
||
if not camera:
|
||
raise ValueError(f"摄像头ID {camera_id} 不存在")
|
||
|
||
room = camera['room']
|
||
if camera_number == 'mixed':
|
||
return f"{self.camera_url}?room={room}&camera=mixed"
|
||
else:
|
||
return f"{self.camera_url}?room={room}&camera=camera-{camera_number}"
|
||
|
||
def refresh_camera(self, camera_id: int) -> bool:
|
||
"""
|
||
刷新指定摄像头（模拟操作）
|
||
|
||
Args:
|
||
camera_id: 摄像头ID
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
logger.info(f"刷新摄像头 {camera_id}")
|
||
return True
|
||
|
||
def get_all_cameras(self) -> List[Dict[str, Any]]:
|
||
"""
|
||
返回所有摄像头配置
|
||
|
||
Returns:
|
||
摄像头配置列表
|
||
"""
|
||
return self.cameras
|
||
|
||
def check_connection(self, timeout: int = 5) -> bool:
|
||
"""
|
||
检查连接状态
|
||
|
||
Args:
|
||
timeout: 超时时间（秒）
|
||
|
||
Returns:
|
||
连接是否正常
|
||
"""
|
||
try:
|
||
response = self.session.get(
|
||
self.base_url,
|
||
timeout=timeout,
|
||
allow_redirects=True
|
||
)
|
||
response.raise_for_status()
|
||
return response.status_code == 200
|
||
except requests.exceptions.Timeout as e:
|
||
logger.warning(f"连接超时: {e}")
|
||
return False
|
||
except requests.exceptions.ConnectionError as e:
|
||
logger.warning(f"连接错误: {e}")
|
||
return False
|
||
except requests.exceptions.HTTPError as e:
|
||
logger.warning(f"HTTP错误: {e}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f"请求异常: {e}")
|
||
return False
|
||
|
||
def get_camera_by_id(self, camera_id: int) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
根据ID获取摄像头配置
|
||
|
||
Args:
|
||
camera_id: 摄像头ID
|
||
|
||
Returns:
|
||
摄像头配置，如果不存在返回None
|
||
"""
|
||
return next((c for c in self.cameras if c['id'] == camera_id), None)
|
||
|
||
def close(self) -> None:
|
||
"""
|
||
关闭会话
|
||
"""
|
||
self.session.close()
|