1137 lines
49 KiB
Python
1137 lines
49 KiB
Python
import cv2
|
||
import numpy as np
|
||
import yaml
|
||
import torch
|
||
from ultralytics import YOLO
|
||
import time
|
||
import datetime
|
||
import threading
|
||
import queue
|
||
import sys
|
||
import argparse
|
||
import base64
|
||
import os
|
||
from openai import OpenAI
|
||
from io import BytesIO
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
|
||
|
||
def save_alert_image(frame, cam_id, roi_name, alert_type, alert_info=""):
|
||
"""保存告警图片
|
||
|
||
Args:
|
||
frame: OpenCV图像
|
||
cam_id: 摄像头ID
|
||
roi_name: ROI区域名称
|
||
alert_type: 告警类型 ("离岗" 或 "入侵")
|
||
alert_info: 告警信息（可选）
|
||
"""
|
||
try:
|
||
# 创建文件夹结构
|
||
data_dir = "data"
|
||
alert_dir = os.path.join(data_dir, alert_type)
|
||
|
||
os.makedirs(alert_dir, exist_ok=True)
|
||
|
||
# 生成文件名：根据告警类型使用不同的命名方式
|
||
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
# 清理文件名中的特殊字符
|
||
safe_roi_name = roi_name.replace("/", "_").replace("\\", "_").replace(":", "_")
|
||
|
||
# 对于入侵告警，使用告警类型+ROI名称；对于离岗告警，使用ROI名称
|
||
if alert_type == "入侵":
|
||
# 周界入侵：使用"入侵_区域名称"格式
|
||
filename = f"{cam_id}_入侵_{safe_roi_name}_{timestamp}.jpg"
|
||
else:
|
||
# 离岗：使用原有格式
|
||
filename = f"{cam_id}_{safe_roi_name}_{timestamp}.jpg"
|
||
|
||
filepath = os.path.join(alert_dir, filename)
|
||
|
||
# 保存图片
|
||
cv2.imwrite(filepath, frame)
|
||
print(f"[{cam_id}] 💾 告警图片已保存: {filepath}")
|
||
|
||
# 如果有告警信息，保存到文本文件
|
||
if alert_info:
|
||
info_filepath = filepath.replace(".jpg", ".txt")
|
||
with open(info_filepath, 'w', encoding='utf-8') as f:
|
||
f.write(f"告警时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
|
||
f.write(f"摄像头ID: {cam_id}\n")
|
||
f.write(f"ROI区域: {roi_name}\n")
|
||
f.write(f"告警类型: {alert_type}\n")
|
||
f.write(f"告警信息:\n{alert_info}\n")
|
||
|
||
return filepath
|
||
except Exception as e:
|
||
print(f"[{cam_id}] 保存告警图片失败: {e}")
|
||
return None
|
||
|
||
|
||
def put_chinese_text(img, text, position, font_size=20, color=(255, 255, 255), thickness=1):
|
||
"""在OpenCV图像上绘制中文文本
|
||
|
||
Args:
|
||
img: OpenCV图像 (BGR格式)
|
||
text: 要显示的文本（支持中文）
|
||
position: 文本位置 (x, y)
|
||
font_size: 字体大小
|
||
color: 颜色 (BGR格式，会被转换为RGB)
|
||
thickness: 线条粗细（PIL不支持，保留参数以兼容）
|
||
"""
|
||
try:
|
||
# 将OpenCV图像转换为PIL图像
|
||
img_pil = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
|
||
draw = ImageDraw.Draw(img_pil)
|
||
|
||
# 转换颜色格式：BGR -> RGB
|
||
color_rgb = (color[2], color[1], color[0])
|
||
|
||
# 尝试使用系统字体
|
||
font = None
|
||
font_paths = [
|
||
"C:/Windows/Fonts/simhei.ttf", # 黑体
|
||
"C:/Windows/Fonts/msyh.ttc", # 微软雅黑
|
||
"C:/Windows/Fonts/simsun.ttc", # 宋体
|
||
"C:/Windows/Fonts/msyhbd.ttc", # 微软雅黑 Bold
|
||
]
|
||
|
||
for font_path in font_paths:
|
||
if os.path.exists(font_path):
|
||
try:
|
||
font = ImageFont.truetype(font_path, font_size)
|
||
break
|
||
except:
|
||
continue
|
||
|
||
# 如果找不到字体，使用默认字体
|
||
if font is None:
|
||
font = ImageFont.load_default()
|
||
|
||
# 绘制文本
|
||
draw.text(position, text, font=font, fill=color_rgb)
|
||
|
||
# 转换回OpenCV格式
|
||
img = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)
|
||
return img
|
||
except Exception as e:
|
||
# 如果绘制失败，使用英文替代或直接返回原图
|
||
print(f"中文文本绘制失败: {e}，使用OpenCV默认字体")
|
||
# 降级方案：使用OpenCV绘制（可能显示为问号，但至少不会崩溃）
|
||
cv2.putText(img, text.encode('utf-8').decode('latin-1', 'ignore'), position,
|
||
cv2.FONT_HERSHEY_SIMPLEX, font_size/40, color, thickness)
|
||
return img
|
||
|
||
|
||
class LLMClient:
|
||
"""大模型客户端，用于人员判断和离岗分析"""
|
||
def __init__(self, api_key, base_url, model_name):
|
||
self.client = OpenAI(
|
||
api_key=api_key,
|
||
base_url=base_url,
|
||
)
|
||
self.model_name = model_name
|
||
|
||
def frame_to_base64(self, frame):
|
||
"""将OpenCV帧转换为base64编码"""
|
||
_, buffer = cv2.imencode('.jpg', frame)
|
||
img_base64 = base64.b64encode(buffer).decode('utf-8')
|
||
return img_base64
|
||
|
||
def check_if_staff(self, frame, cam_id, roi_name):
|
||
"""判断ROI中的人员是否为工作人员"""
|
||
try:
|
||
img_base64 = self.frame_to_base64(frame)
|
||
prompt = f"""你是一个智能安防辅助系统，负责对监控画面中指定敏感区域（如高配间门口、天台、禁行通道）的人员活动进行分析。
|
||
|
||
请根据以下规则生成结构化响应：
|
||
|
||
### 【判定标准】
|
||
✅ **本单位物业员工**需满足下列条件之一：
|
||
1. **清晰可见的正式工牌**（胸前佩戴）
|
||
2. **穿着标准制服**（如：带有白色反光条的深色工程服、黄蓝工程服、白衬衫+黑领带、蓝色清洁装、浅色客服装等）
|
||
3. **行为符合岗位规范**（如巡检、维修、清洁，无徘徊、张望、翻越）
|
||
|
||
> 注意： 满足部分关键条件（如戴有安全帽、穿有工作人员服饰、带有工牌）→ 视为员工，不生成告警。
|
||
|
||
### 【输出规则】
|
||
#### 情况1：ROI区域内**无人**
|
||
→ 输出：
|
||
🟢无异常：敏感区域当前无人员活动。
|
||
[客观描述：画面整体状态]
|
||
|
||
#### 情况2：ROI区域内**有本单位员工**
|
||
→ 输出：
|
||
🟢无异常：检测到本单位工作人员正常作业。
|
||
[客观描述：人数+制服类型+工牌状态+行为]
|
||
|
||
#### 情况3：ROI区域内**有非员工或身份不明人员**
|
||
→ 输出：
|
||
🚨[区域类型]入侵告警：检测到疑似非工作人员，请立即核查。
|
||
[客观描述：人数+衣着+工牌状态+位置+行为]
|
||
|
||
### 【描述要求】
|
||
- 所有描述必须**≤30字**
|
||
- 仅陈述**可观察事实**，禁止主观推测（如"意图破坏""形迹可疑"）
|
||
- 使用简洁、标准化语言
|
||
|
||
### 【示例】
|
||
▶ 示例1（无人）：
|
||
🟢无异常：敏感区域当前无人员活动。
|
||
高配间门口区域空旷，无人员进入。
|
||
|
||
▶ 示例2（员工）：
|
||
🟢无异常：检测到本单位工作人员正常作业。
|
||
1名工程人员穿带有反光条的深蓝色工服在高配间巡检。
|
||
|
||
▶ 示例3（非员工）：
|
||
🚨天台区域入侵告警：检测到疑似非工作人员，请立即核查。
|
||
1人穿绿色外套未佩戴工牌进入天台区域。
|
||
|
||
---
|
||
请分析摄像头{cam_id}的{roi_name}区域，按照上述格式输出结果。"""
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.model_name,
|
||
messages=[
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{
|
||
"type": "image_url",
|
||
"image_url": {
|
||
"url": f"data:image/jpeg;base64,{img_base64}"
|
||
}
|
||
},
|
||
{"type": "text", "text": prompt}
|
||
]
|
||
}
|
||
]
|
||
)
|
||
result_text = response.choices[0].message.content
|
||
|
||
# 判断是否为工作人员（根据新的输出格式）
|
||
# 如果包含"🟢无异常"且包含"工作人员"，则为员工
|
||
# 如果包含"🚨"或"入侵告警"或"非工作人员"，则为非员工
|
||
is_staff = False
|
||
if "🟢无异常" in result_text and "工作人员" in result_text:
|
||
is_staff = True
|
||
elif "🚨" in result_text or "入侵告警" in result_text or "非工作人员" in result_text:
|
||
is_staff = False
|
||
elif "无人员活动" in result_text or "无人" in result_text:
|
||
is_staff = None # 无人情况
|
||
|
||
return is_staff, result_text
|
||
except Exception as e:
|
||
print(f"[{cam_id}] 大模型调用失败: {e}")
|
||
return None, str(e)
|
||
|
||
def analyze_off_duty_duration(self, key_frames_info, cam_id):
|
||
"""分析离岗时长并判断是否为同一人"""
|
||
try:
|
||
frames = key_frames_info.get('frames', [])
|
||
if not frames:
|
||
return False, False, "无关键帧"
|
||
|
||
off_duty_duration = key_frames_info.get('off_duty_duration', 0)
|
||
duration_minutes = int(off_duty_duration / 60)
|
||
duration_seconds = int(off_duty_duration % 60)
|
||
|
||
# 构建消息内容
|
||
content_parts = [
|
||
{
|
||
"type": "text",
|
||
"text": f"""请分析以下关键帧图像，判断人员离岗情况。请按照以下格式简洁回答：
|
||
|
||
【输出格式】
|
||
1. 是否告警：[是/否]
|
||
2. 离岗时间：{duration_minutes}分{duration_seconds}秒
|
||
3. 是否为同一人：[是/否/无法确定]
|
||
4. 简要分析：[一句话概括，不超过30字]
|
||
|
||
要求：
|
||
- 如果离岗时间超过6分钟且确认为同一人，则告警
|
||
- 简要分析需客观描述关键帧中人员的特征和行为变化
|
||
- 回答要简洁明了，避免冗余描述
|
||
|
||
关键帧信息："""
|
||
}
|
||
]
|
||
|
||
# 添加图像和说明
|
||
for i, frame_info in enumerate(frames):
|
||
img_base64 = self.frame_to_base64(frame_info['frame'])
|
||
content_parts.append({
|
||
"type": "image_url",
|
||
"image_url": {
|
||
"url": f"data:image/jpeg;base64,{img_base64}"
|
||
}
|
||
})
|
||
content_parts.append({
|
||
"type": "text",
|
||
"text": f"关键帧{i+1} - 时间: {frame_info['time']}, 事件: {frame_info['event']}"
|
||
})
|
||
|
||
response = self.client.chat.completions.create(
|
||
model=self.model_name,
|
||
messages=[
|
||
{
|
||
"role": "user",
|
||
"content": content_parts
|
||
}
|
||
]
|
||
)
|
||
result = response.choices[0].message.content
|
||
|
||
# 解析结果 - 更灵活的解析逻辑
|
||
# 判断是否告警
|
||
exceeds_6min = False
|
||
if duration_minutes >= 6:
|
||
# 如果时间已经超过6分钟，检查大模型是否确认告警
|
||
if any(keyword in result for keyword in ["是否告警：是", "是否告警:是", "告警：是", "告警:是", "需要告警", "应告警"]):
|
||
exceeds_6min = True
|
||
elif "是否告警：否" not in result and "是否告警:否" not in result:
|
||
# 如果没有明确说否，且时间超过6分钟，默认告警
|
||
exceeds_6min = True
|
||
else:
|
||
# 时间未超过6分钟，即使大模型说告警也不告警
|
||
exceeds_6min = False
|
||
|
||
# 判断是否为同一人
|
||
is_same_person = False
|
||
if any(keyword in result for keyword in ["是否为同一人：是", "是否为同一人:是", "同一人：是", "同一人:是", "是同一人", "确认为同一人"]):
|
||
is_same_person = True
|
||
elif any(keyword in result for keyword in ["是否为同一人：否", "是否为同一人:否", "同一人：否", "同一人:否", "不是同一人", "非同一人"]):
|
||
is_same_person = False
|
||
elif "无法确定" in result or "不确定" in result:
|
||
is_same_person = False # 无法确定时，不告警
|
||
|
||
return exceeds_6min, is_same_person, result
|
||
except Exception as e:
|
||
print(f"[{cam_id}] 离岗分析失败: {e}")
|
||
return None, None, str(e)
|
||
|
||
|
||
class ThreadedFrameReader:
|
||
def __init__(self, cam_id, rtsp_url):
|
||
self.cam_id = cam_id
|
||
self.rtsp_url = rtsp_url
|
||
self._lock = threading.Lock() # 添加锁保护VideoCapture访问
|
||
self.cap = None
|
||
try:
|
||
self.cap = cv2.VideoCapture(rtsp_url, cv2.CAP_FFMPEG)
|
||
if self.cap.isOpened():
|
||
self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
|
||
else:
|
||
print(f"[{cam_id}] 警告：无法打开视频流: {rtsp_url}")
|
||
except Exception as e:
|
||
print(f"[{cam_id}] 初始化VideoCapture失败: {e}")
|
||
self.q = queue.Queue(maxsize=2)
|
||
self.running = True
|
||
self.thread = threading.Thread(target=self._reader, daemon=True)
|
||
self.thread.start()
|
||
|
||
def _reader(self):
|
||
"""读取帧的线程函数"""
|
||
try:
|
||
while self.running:
|
||
with self._lock:
|
||
if self.cap is None or not self.cap.isOpened():
|
||
break
|
||
ret, frame = self.cap.read()
|
||
|
||
if not ret:
|
||
time.sleep(0.1)
|
||
continue
|
||
|
||
if self.q.full():
|
||
try:
|
||
self.q.get_nowait()
|
||
except queue.Empty:
|
||
pass
|
||
self.q.put(frame)
|
||
except Exception as e:
|
||
print(f"[{self.cam_id}] 读取帧线程异常: {e}")
|
||
finally:
|
||
# 确保资源释放（使用锁保护）
|
||
with self._lock:
|
||
if self.cap is not None:
|
||
try:
|
||
if self.cap.isOpened():
|
||
self.cap.release()
|
||
except Exception as e:
|
||
print(f"[{self.cam_id}] 释放VideoCapture时出错: {e}")
|
||
finally:
|
||
self.cap = None
|
||
|
||
def read(self):
|
||
if not self.q.empty():
|
||
return True, self.q.get()
|
||
return False, None
|
||
|
||
def release(self):
|
||
"""释放资源，等待线程结束"""
|
||
if not self.running:
|
||
return # 已经释放过了
|
||
|
||
self.running = False
|
||
|
||
# 等待线程结束，最多等待3秒
|
||
if self.thread.is_alive():
|
||
self.thread.join(timeout=3.0)
|
||
if self.thread.is_alive():
|
||
print(f"[{self.cam_id}] 警告：读取线程未能在3秒内结束")
|
||
|
||
# 清空队列
|
||
while not self.q.empty():
|
||
try:
|
||
self.q.get_nowait()
|
||
except queue.Empty:
|
||
break
|
||
|
||
# VideoCapture的释放由_reader线程的finally块处理，这里不再重复释放
|
||
|
||
|
||
class MultiCameraMonitor:
|
||
def __init__(self, config_path):
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
self.cfg = yaml.safe_load(f)
|
||
|
||
# === 全局模型（只加载一次）===
|
||
model_cfg = self.cfg['model']
|
||
self.device = model_cfg.get('device', 'auto')
|
||
if self.device == 'auto' or not self.device:
|
||
self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
|
||
print(f"🚀 全局加载模型到 {self.device}...")
|
||
self.model = YOLO(model_cfg['path'])
|
||
self.model.to(self.device)
|
||
self.use_half = (self.device == 'cuda')
|
||
if self.use_half:
|
||
print("✅ 启用 FP16 推理")
|
||
|
||
self.imgsz = model_cfg['imgsz']
|
||
self.conf_thresh = model_cfg['conf_threshold']
|
||
|
||
# === 初始化大模型客户端 ===
|
||
llm_cfg = self.cfg.get('llm', {})
|
||
if llm_cfg.get('api_key'):
|
||
self.llm_client = LLMClient(
|
||
llm_cfg['api_key'],
|
||
llm_cfg['base_url'],
|
||
llm_cfg.get('model_name', 'qwen-vl-max')
|
||
)
|
||
print("✅ 大模型客户端已初始化")
|
||
else:
|
||
self.llm_client = None
|
||
print("⚠️ 未配置大模型API密钥，大模型功能将不可用")
|
||
|
||
# === 初始化所有摄像头 ===
|
||
self.common = self.cfg['common']
|
||
self.cameras = {}
|
||
self.frame_readers = {}
|
||
self.queues = {} # cam_id -> queue for detection results
|
||
self.perimeter_queues = {} # cam_id -> queue for perimeter detection (每秒抽帧)
|
||
|
||
for cam_cfg in self.cfg['cameras']:
|
||
cam_id = cam_cfg['id']
|
||
self.cameras[cam_id] = CameraLogic(cam_id, cam_cfg, self.common, self.llm_client)
|
||
self.frame_readers[cam_id] = ThreadedFrameReader(cam_id, cam_cfg['rtsp_url'])
|
||
self.queues[cam_id] = queue.Queue(maxsize=1) # 存放检测结果（人员离岗）
|
||
self.perimeter_queues[cam_id] = queue.Queue(maxsize=1) # 存放检测结果（周界入侵）
|
||
|
||
# === 控制信号 ===
|
||
self.running = True
|
||
self.inference_thread = threading.Thread(target=self._inference_loop, daemon=True)
|
||
self.perimeter_thread = threading.Thread(target=self._perimeter_inference_loop, daemon=True)
|
||
self.inference_thread.start()
|
||
self.perimeter_thread.start()
|
||
|
||
def _inference_loop(self):
|
||
"""统一推理线程：轮询各摄像头最新帧，逐个推理（用于人员离岗）"""
|
||
while self.running:
|
||
processed = False
|
||
for cam_id, reader in self.frame_readers.items():
|
||
ret, frame = reader.read()
|
||
if not ret:
|
||
continue
|
||
|
||
cam_logic = self.cameras[cam_id]
|
||
if cam_logic.should_skip_frame():
|
||
continue
|
||
|
||
# 检查是否有ROI启用了人员离岗算法
|
||
if not cam_logic.has_enabled_algorithm('人员离岗'):
|
||
continue
|
||
|
||
results = self.model(
|
||
frame,
|
||
imgsz=self.imgsz,
|
||
conf=self.conf_thresh,
|
||
verbose=False,
|
||
device=self.device,
|
||
half=self.use_half,
|
||
classes=[0] # person only
|
||
)
|
||
|
||
if not self.queues[cam_id].full():
|
||
self.queues[cam_id].put((frame.copy(), results[0]))
|
||
processed = True
|
||
|
||
if not processed:
|
||
time.sleep(0.01)
|
||
|
||
def _perimeter_inference_loop(self):
|
||
"""周界入侵推理线程：每秒抽一帧进行检测"""
|
||
while self.running:
|
||
processed = False
|
||
for cam_id, reader in self.frame_readers.items():
|
||
cam_logic = self.cameras[cam_id]
|
||
# 检查是否有ROI启用了周界入侵算法
|
||
if not cam_logic.has_enabled_algorithm('周界入侵'):
|
||
continue
|
||
|
||
ret, frame = reader.read()
|
||
if not ret:
|
||
continue
|
||
|
||
# 每秒抽一帧
|
||
current_time = time.time()
|
||
if not hasattr(cam_logic, 'last_perimeter_check_time'):
|
||
cam_logic.last_perimeter_check_time = {}
|
||
if cam_id not in cam_logic.last_perimeter_check_time:
|
||
cam_logic.last_perimeter_check_time[cam_id] = 0
|
||
|
||
if current_time - cam_logic.last_perimeter_check_time[cam_id] < 1.0:
|
||
continue
|
||
|
||
cam_logic.last_perimeter_check_time[cam_id] = current_time
|
||
|
||
results = self.model(
|
||
frame,
|
||
imgsz=self.imgsz,
|
||
conf=self.conf_thresh,
|
||
verbose=False,
|
||
device=self.device,
|
||
half=self.use_half,
|
||
classes=[0] # person only
|
||
)
|
||
|
||
if not self.perimeter_queues[cam_id].full():
|
||
self.perimeter_queues[cam_id].put((frame.copy(), results[0]))
|
||
processed = True
|
||
|
||
if not processed:
|
||
time.sleep(0.1)
|
||
|
||
def run(self):
|
||
"""启动所有摄像头的显示和告警逻辑（主线程）"""
|
||
try:
|
||
while self.running:
|
||
for cam_id, cam_logic in self.cameras.items():
|
||
# 处理人员离岗检测结果
|
||
if not self.queues[cam_id].empty():
|
||
frame, results = self.queues[cam_id].get()
|
||
cam_logic.process_off_duty(frame, results)
|
||
|
||
# 处理周界入侵检测结果
|
||
if not self.perimeter_queues[cam_id].empty():
|
||
frame, results = self.perimeter_queues[cam_id].get()
|
||
cam_logic.process_perimeter(frame, results)
|
||
|
||
# 更新显示
|
||
cam_logic.update_display()
|
||
|
||
key = cv2.waitKey(1) & 0xFF
|
||
if key == ord('q'):
|
||
break
|
||
time.sleep(0.01)
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
self.stop()
|
||
|
||
def stop(self):
|
||
"""停止监控，清理所有资源"""
|
||
print("正在停止监控系统...")
|
||
self.running = False
|
||
|
||
# 等待推理线程结束
|
||
if hasattr(self, 'inference_thread') and self.inference_thread.is_alive():
|
||
self.inference_thread.join(timeout=2.0)
|
||
|
||
if hasattr(self, 'perimeter_thread') and self.perimeter_thread.is_alive():
|
||
self.perimeter_thread.join(timeout=2.0)
|
||
|
||
# 释放所有摄像头资源
|
||
for cam_id, reader in self.frame_readers.items():
|
||
try:
|
||
print(f"正在释放摄像头 {cam_id}...")
|
||
reader.release()
|
||
except Exception as e:
|
||
print(f"释放摄像头 {cam_id} 时出错: {e}")
|
||
|
||
# 关闭所有窗口
|
||
try:
|
||
cv2.destroyAllWindows()
|
||
except:
|
||
pass
|
||
|
||
# 强制清理（如果还有线程在运行）
|
||
import sys
|
||
import os
|
||
if sys.platform == 'win32':
|
||
# Windows下可能需要额外等待
|
||
time.sleep(0.5)
|
||
|
||
print("监控系统已停止")
|
||
|
||
|
||
class ROILogic:
|
||
"""单个ROI区域的逻辑处理"""
|
||
def __init__(self, roi_cfg, cam_id, common_cfg, llm_client):
|
||
self.cam_id = cam_id
|
||
self.roi_name = roi_cfg.get('name', '未命名区域')
|
||
self.llm_client = llm_client
|
||
|
||
# 处理points：如果不存在或为空，设置为None（表示使用整张画面）
|
||
if 'points' in roi_cfg and roi_cfg['points']:
|
||
self.roi_points = np.array(roi_cfg['points'], dtype=np.int32)
|
||
self.use_full_frame = False
|
||
else:
|
||
# 对于周界入侵算法，如果没有points，使用整张画面
|
||
self.roi_points = None
|
||
self.use_full_frame = True
|
||
|
||
# 算法配置
|
||
self.algorithms = {}
|
||
for alg_cfg in roi_cfg.get('algorithms', []):
|
||
alg_name = alg_cfg['name']
|
||
if alg_cfg.get('enabled', False):
|
||
self.algorithms[alg_name] = alg_cfg
|
||
|
||
# 人员离岗相关状态（需要ROI，如果没有points则不能启用）
|
||
if '人员离岗' in self.algorithms:
|
||
if self.roi_points is None:
|
||
print(f"[{cam_id}] 警告：{self.roi_name} 启用了人员离岗算法但没有配置points，已禁用")
|
||
del self.algorithms['人员离岗']
|
||
else:
|
||
alg_cfg = self.algorithms['人员离岗']
|
||
self.off_duty_threshold_sec = alg_cfg.get('off_duty_threshold_sec', 300)
|
||
self.on_duty_confirm_sec = alg_cfg.get('on_duty_confirm_sec', 5)
|
||
self.off_duty_confirm_sec = alg_cfg.get('off_duty_confirm_sec', 30)
|
||
|
||
self.is_on_duty = False
|
||
self.is_off_duty = True
|
||
self.on_duty_start_time = None
|
||
self.last_no_person_time = None
|
||
self.off_duty_timer_start = None
|
||
self.last_alert_time = 0
|
||
self.last_person_seen_time = None
|
||
|
||
# 关键时间记录
|
||
self.on_duty_confirm_time = None # 上岗确认时间
|
||
self.off_duty_confirm_time = None # 离岗确认时间
|
||
self.key_frames = [] # 关键帧存储
|
||
|
||
# 初始化状态跟踪
|
||
self.initial_state_start_time = None # 初始化状态开始时间（进入工作时间时）
|
||
self.has_ever_seen_person = False # 是否曾经检测到过人员
|
||
self.initial_state_frame = None # 初始化状态时的帧（用于大模型分析）
|
||
|
||
# 周界入侵相关状态（如果没有points，使用整张画面）
|
||
if '周界入侵' in self.algorithms:
|
||
self.perimeter_last_check_time = 0
|
||
self.perimeter_alert_cooldown = 120 # 周界入侵告警冷却120秒（2分钟）
|
||
if self.use_full_frame:
|
||
print(f"[{cam_id}] 提示：{self.roi_name} 周界入侵算法将使用整张画面进行检测")
|
||
|
||
def is_point_in_roi(self, x, y):
|
||
"""判断点是否在ROI内，如果没有ROI（use_full_frame=True），总是返回True"""
|
||
if self.use_full_frame or self.roi_points is None:
|
||
return True
|
||
return cv2.pointPolygonTest(self.roi_points, (int(x), int(y)), False) >= 0
|
||
|
||
|
||
class CameraLogic:
|
||
def __init__(self, cam_id, cam_cfg, common_cfg, llm_client):
|
||
self.cam_id = cam_id
|
||
self.llm_client = llm_client
|
||
|
||
# 工作时间段配置
|
||
self.working_hours = common_cfg.get('working_hours', [[8, 30, 11, 0], [12, 0, 17, 30]])
|
||
self.process_every_n = cam_cfg.get('process_every_n_frames', common_cfg['process_every_n_frames'])
|
||
self.alert_cooldown_sec = common_cfg.get('alert_cooldown_sec', 300)
|
||
self.off_duty_alert_threshold_sec = common_cfg.get('off_duty_alert_threshold_sec', 360) # 6分钟
|
||
|
||
# 初始化所有ROI
|
||
self.rois = []
|
||
for roi_cfg in cam_cfg.get('rois', []):
|
||
self.rois.append(ROILogic(roi_cfg, cam_id, common_cfg, llm_client))
|
||
|
||
# 兼容旧配置格式
|
||
if 'roi_points' in cam_cfg:
|
||
# 创建默认ROI用于人员离岗
|
||
default_roi = {
|
||
'name': '离岗检测区域',
|
||
'points': cam_cfg['roi_points'],
|
||
'algorithms': [{
|
||
'name': '人员离岗',
|
||
'enabled': True,
|
||
'off_duty_threshold_sec': cam_cfg.get('off_duty_threshold_sec', 300),
|
||
'on_duty_confirm_sec': cam_cfg.get('on_duty_confirm_sec', 5),
|
||
'off_duty_confirm_sec': cam_cfg.get('off_duty_confirm_sec', 30)
|
||
}]
|
||
}
|
||
self.rois.append(ROILogic(default_roi, cam_id, common_cfg, llm_client))
|
||
|
||
self.frame_count = 0
|
||
self.display_frame = None # 用于显示的帧
|
||
self.display_results = None # 用于显示的检测结果（YOLO results）
|
||
|
||
def should_skip_frame(self):
|
||
self.frame_count += 1
|
||
return self.frame_count % self.process_every_n != 0
|
||
|
||
def has_enabled_algorithm(self, alg_name):
|
||
"""检查是否有ROI启用了指定算法"""
|
||
return any(alg_name in roi.algorithms for roi in self.rois)
|
||
|
||
def in_working_hours(self):
|
||
"""判断是否在工作时间段内"""
|
||
now = datetime.datetime.now()
|
||
h, m = now.hour, now.minute
|
||
current_minutes = h * 60 + m
|
||
|
||
for period in self.working_hours:
|
||
start_h, start_m, end_h, end_m = period
|
||
start_minutes = start_h * 60 + start_m
|
||
end_minutes = end_h * 60 + end_m
|
||
if start_minutes <= current_minutes < end_minutes:
|
||
return True
|
||
return False
|
||
|
||
def is_edge_time(self):
|
||
"""判断是否为边缘时间段（8:30-9:00, 11:00-12:00, 17:30-18:00）"""
|
||
now = datetime.datetime.now()
|
||
h, m = now.hour, now.minute
|
||
current_minutes = h * 60 + m
|
||
|
||
edge_periods = [
|
||
(8 * 60 + 30, 9 * 60), # 8:30-9:00
|
||
(11 * 60, 12 * 60), # 11:00-12:00
|
||
(17 * 60 + 30, 18 * 60) # 17:30-18:00
|
||
]
|
||
|
||
for start, end in edge_periods:
|
||
if start <= current_minutes < end:
|
||
return True
|
||
return False
|
||
|
||
def get_end_of_work_time(self):
|
||
"""获取当天工作结束时间（17:30）"""
|
||
now = datetime.datetime.now()
|
||
end_time = now.replace(hour=17, minute=30, second=0, microsecond=0)
|
||
if now > end_time:
|
||
# 如果已经过了17:30，返回明天的17:30
|
||
end_time += datetime.timedelta(days=1)
|
||
return end_time
|
||
|
||
def process_off_duty(self, frame, results):
|
||
"""处理人员离岗检测"""
|
||
current_time = time.time()
|
||
now = datetime.datetime.now()
|
||
boxes = results.boxes
|
||
|
||
for roi in self.rois:
|
||
if '人员离岗' not in roi.algorithms:
|
||
continue
|
||
|
||
# 检查ROI中是否有人
|
||
roi_has_person = any(
|
||
roi.is_point_in_roi((b.xyxy[0][0] + b.xyxy[0][2]) / 2,
|
||
(b.xyxy[0][1] + b.xyxy[0][3]) / 2)
|
||
for b in boxes
|
||
)
|
||
|
||
in_work = self.in_working_hours()
|
||
is_edge = self.is_edge_time()
|
||
|
||
if in_work:
|
||
# 初始化状态跟踪：如果刚进入工作时间，记录开始时间
|
||
if roi.initial_state_start_time is None:
|
||
roi.initial_state_start_time = current_time
|
||
roi.has_ever_seen_person = False
|
||
roi.initial_state_frame = frame.copy() # 保存初始化状态时的帧
|
||
|
||
if roi_has_person:
|
||
roi.last_person_seen_time = current_time
|
||
roi.has_ever_seen_person = True
|
||
# 如果检测到人员，清除初始化状态
|
||
if roi.initial_state_start_time is not None:
|
||
roi.initial_state_start_time = None
|
||
roi.initial_state_frame = None
|
||
|
||
effective = (
|
||
roi.last_person_seen_time is not None and
|
||
(current_time - roi.last_person_seen_time) < 1.0
|
||
)
|
||
|
||
# 处理初始化状态：如果系统启动时没有人，且超过离岗确认时间
|
||
if not roi.has_ever_seen_person and roi.initial_state_start_time is not None:
|
||
elapsed_since_start = current_time - roi.initial_state_start_time
|
||
if elapsed_since_start >= roi.off_duty_confirm_sec:
|
||
# 超过离岗确认时间，触发离岗确认逻辑
|
||
roi.is_off_duty, roi.is_on_duty = True, False
|
||
roi.off_duty_confirm_time = roi.initial_state_start_time + roi.off_duty_confirm_sec # 使用离岗确认时间点
|
||
roi.off_duty_timer_start = current_time
|
||
|
||
# 保存关键帧（使用初始化状态时的帧作为离岗确认帧）
|
||
if roi.initial_state_frame is not None:
|
||
roi.key_frames.append({
|
||
'frame': roi.initial_state_frame.copy(),
|
||
'time': datetime.datetime.fromtimestamp(roi.off_duty_confirm_time).strftime('%Y-%m-%d %H:%M:%S'),
|
||
'event': '离岗确认（初始化状态）'
|
||
})
|
||
# 也保存当前帧
|
||
roi.key_frames.append({
|
||
'frame': frame.copy(),
|
||
'time': now.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'event': '当前状态'
|
||
})
|
||
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] 🚪 初始化状态：超过离岗确认时间，进入离岗计时 ({now.strftime('%H:%M:%S')})")
|
||
roi.initial_state_start_time = None # 清除初始化状态标记
|
||
roi.initial_state_frame = None
|
||
|
||
if effective:
|
||
roi.last_no_person_time = None
|
||
if roi.is_off_duty:
|
||
if roi.on_duty_start_time is None:
|
||
roi.on_duty_start_time = current_time
|
||
elif current_time - roi.on_duty_start_time >= roi.on_duty_confirm_sec:
|
||
roi.is_on_duty, roi.is_off_duty = True, False
|
||
roi.on_duty_confirm_time = current_time
|
||
roi.on_duty_start_time = None
|
||
|
||
# 保存关键帧
|
||
roi.key_frames.append({
|
||
'frame': frame.copy(),
|
||
'time': now.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'event': '上岗确认'
|
||
})
|
||
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] ✅ 上岗确认成功 ({now.strftime('%H:%M:%S')})")
|
||
else:
|
||
roi.on_duty_start_time = None
|
||
roi.last_person_seen_time = None
|
||
if not roi.is_off_duty:
|
||
if roi.last_no_person_time is None:
|
||
roi.last_no_person_time = current_time
|
||
elif current_time - roi.last_no_person_time >= roi.off_duty_confirm_sec:
|
||
roi.is_off_duty, roi.is_on_duty = True, False
|
||
roi.off_duty_confirm_time = current_time
|
||
roi.last_no_person_time = None
|
||
roi.off_duty_timer_start = current_time
|
||
|
||
# 保存关键帧
|
||
roi.key_frames.append({
|
||
'frame': frame.copy(),
|
||
'time': now.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'event': '离岗确认'
|
||
})
|
||
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] 🚪 进入离岗计时 ({now.strftime('%H:%M:%S')})")
|
||
|
||
# 离岗告警逻辑（边缘时间只记录，不告警）
|
||
if roi.is_off_duty and roi.off_duty_timer_start:
|
||
elapsed = current_time - roi.off_duty_timer_start
|
||
off_duty_duration = elapsed
|
||
|
||
# 如果到了下班时间还没回来，计算到下班时间的离岗时长
|
||
end_time = self.get_end_of_work_time()
|
||
if now >= end_time and roi.off_duty_confirm_time:
|
||
# 计算离岗时长：下班时间 - 离岗确认时间
|
||
off_duty_duration = (end_time.timestamp() - roi.off_duty_confirm_time)
|
||
|
||
# 超过6分钟且不在边缘时间，使用大模型判断
|
||
if off_duty_duration >= self.off_duty_alert_threshold_sec and not is_edge:
|
||
# 对于初始化状态，即使只有1帧也要进行分析
|
||
is_initial_state = any('初始化状态' in f.get('event', '') for f in roi.key_frames)
|
||
min_frames_required = 1 if is_initial_state else 2
|
||
|
||
if self.llm_client and len(roi.key_frames) >= min_frames_required:
|
||
# 限制关键帧数量，只保留最近10帧
|
||
if len(roi.key_frames) > 10:
|
||
roi.key_frames = roi.key_frames[-10:]
|
||
|
||
# 准备关键帧信息
|
||
key_frames_info = {
|
||
'frames': roi.key_frames[-5:] if len(roi.key_frames) >= 2 else roi.key_frames, # 如果有足够帧，取最近5帧；否则全部使用
|
||
'off_duty_duration': off_duty_duration
|
||
}
|
||
|
||
# 调用大模型分析
|
||
exceeds_6min, is_same_person, analysis_result = self.llm_client.analyze_off_duty_duration(
|
||
key_frames_info, self.cam_id
|
||
)
|
||
|
||
# 对于初始化状态，只要超过6分钟就告警（因为无法判断是否为同一人）
|
||
if is_initial_state:
|
||
should_alert = exceeds_6min if exceeds_6min is not None else (off_duty_duration >= self.off_duty_alert_threshold_sec)
|
||
else:
|
||
should_alert = exceeds_6min and is_same_person
|
||
|
||
if should_alert:
|
||
if (current_time - roi.last_alert_time) >= self.alert_cooldown_sec:
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] 🚨 离岗告警！离岗时长: {int(off_duty_duration)}秒 ({int(off_duty_duration/60)}分钟)")
|
||
print(f"大模型分析结果: {analysis_result}")
|
||
# 保存告警图片
|
||
save_alert_image(
|
||
frame.copy(),
|
||
self.cam_id,
|
||
roi.roi_name,
|
||
"离岗",
|
||
f"离岗时长: {int(off_duty_duration)}秒 ({int(off_duty_duration/60)}分钟)\n大模型分析结果:\n{analysis_result}"
|
||
)
|
||
roi.last_alert_time = current_time
|
||
elif not is_edge:
|
||
# 如果没有大模型，直接告警
|
||
if (current_time - roi.last_alert_time) >= self.alert_cooldown_sec:
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] 🚨 离岗告警！离岗时长: {int(off_duty_duration)}秒 ({int(off_duty_duration/60)}分钟)")
|
||
# 保存告警图片
|
||
save_alert_image(
|
||
frame.copy(),
|
||
self.cam_id,
|
||
roi.roi_name,
|
||
"离岗",
|
||
f"离岗时长: {int(off_duty_duration)}秒 ({int(off_duty_duration/60)}分钟)"
|
||
)
|
||
roi.last_alert_time = current_time
|
||
elif is_edge and roi.off_duty_confirm_time:
|
||
# 边缘时间只记录，不告警
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] ℹ️ 边缘时间段，记录离岗时长: {int(off_duty_duration)}秒")
|
||
|
||
self.display_frame = frame.copy()
|
||
self.display_results = results # 保存检测结果用于显示
|
||
|
||
def crop_roi(self, frame, roi_points):
|
||
"""裁剪ROI区域，如果roi_points为None，返回整张画面"""
|
||
if roi_points is None:
|
||
return frame.copy()
|
||
|
||
x_coords = roi_points[:, 0]
|
||
y_coords = roi_points[:, 1]
|
||
x_min, x_max = int(x_coords.min()), int(x_coords.max())
|
||
y_min, y_max = int(y_coords.min()), int(y_coords.max())
|
||
|
||
# 确保坐标在图像范围内
|
||
h, w = frame.shape[:2]
|
||
x_min = max(0, x_min)
|
||
y_min = max(0, y_min)
|
||
x_max = min(w, x_max)
|
||
y_max = min(h, y_max)
|
||
|
||
roi_frame = frame[y_min:y_max, x_min:x_max]
|
||
|
||
# 创建掩码
|
||
mask = np.zeros(frame.shape[:2], dtype=np.uint8)
|
||
cv2.fillPoly(mask, [roi_points], 255)
|
||
mask_roi = mask[y_min:y_max, x_min:x_max]
|
||
|
||
# 应用掩码
|
||
if len(roi_frame.shape) == 3:
|
||
mask_roi = cv2.cvtColor(mask_roi, cv2.COLOR_GRAY2BGR)
|
||
roi_frame = cv2.bitwise_and(roi_frame, mask_roi)
|
||
|
||
return roi_frame
|
||
|
||
def process_perimeter(self, frame, results):
|
||
"""处理周界入侵检测"""
|
||
current_time = time.time()
|
||
boxes = results.boxes
|
||
|
||
for roi in self.rois:
|
||
if '周界入侵' not in roi.algorithms:
|
||
continue
|
||
|
||
# 检查ROI中是否有人（如果没有ROI，检查整张画面是否有人）
|
||
if roi.use_full_frame:
|
||
# 使用整张画面，只要检测到人就触发
|
||
roi_has_person = len(boxes) > 0
|
||
else:
|
||
# 检查ROI中是否有人
|
||
roi_has_person = any(
|
||
roi.is_point_in_roi((b.xyxy[0][0] + b.xyxy[0][2]) / 2,
|
||
(b.xyxy[0][1] + b.xyxy[0][3]) / 2)
|
||
for b in boxes
|
||
)
|
||
|
||
if roi_has_person:
|
||
# 冷却时间检查
|
||
if current_time - roi.perimeter_last_check_time >= roi.perimeter_alert_cooldown:
|
||
roi.perimeter_last_check_time = current_time
|
||
|
||
# 裁剪ROI区域（如果没有ROI，使用整张画面）
|
||
roi_frame = self.crop_roi(frame, roi.roi_points)
|
||
|
||
# 调用大模型判断是否为工作人员
|
||
if self.llm_client:
|
||
is_staff, result = self.llm_client.check_if_staff(roi_frame, self.cam_id, roi.roi_name)
|
||
area_desc = "整张画面" if roi.use_full_frame else roi.roi_name
|
||
|
||
if is_staff is None:
|
||
# 无人情况
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] ℹ️ 大模型判断：{result}")
|
||
elif not is_staff:
|
||
# 非工作人员
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] 🚨 周界入侵告警！检测到非工作人员（检测区域：{area_desc}）")
|
||
print(f"大模型判断结果: {result}")
|
||
# 保存告警图片（使用区域描述作为名称，更清晰）
|
||
save_alert_image(
|
||
frame.copy(),
|
||
self.cam_id,
|
||
area_desc, # 使用area_desc而不是roi.roi_name
|
||
"入侵",
|
||
f"检测区域: {area_desc}\nROI名称: {roi.roi_name}\n大模型判断结果:\n{result}"
|
||
)
|
||
else:
|
||
# 工作人员
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] ℹ️ 检测到工作人员，无需告警")
|
||
print(f"大模型判断结果: {result}")
|
||
else:
|
||
# 没有大模型时，直接告警
|
||
area_desc = "整张画面" if roi.use_full_frame else roi.roi_name
|
||
print(f"[{self.cam_id}] [{roi.roi_name}] 🚨 周界入侵告警！检测到人员进入（检测区域：{area_desc}）")
|
||
# 保存告警图片（使用区域描述作为名称，更清晰）
|
||
save_alert_image(
|
||
frame.copy(),
|
||
self.cam_id,
|
||
area_desc, # 使用area_desc而不是roi.roi_name
|
||
"入侵",
|
||
f"检测区域: {area_desc}\nROI名称: {roi.roi_name}\n检测到人员进入"
|
||
)
|
||
|
||
self.display_frame = frame.copy()
|
||
self.display_results = results # 保存检测结果用于显示
|
||
|
||
def update_display(self):
|
||
"""更新显示"""
|
||
if self.display_frame is None:
|
||
return
|
||
|
||
vis = self.display_frame.copy()
|
||
now = datetime.datetime.now()
|
||
in_work = self.in_working_hours()
|
||
|
||
# 如果有检测结果，先绘制YOLO识别框
|
||
if self.display_results is not None:
|
||
vis = self.display_results.plot() # 使用YOLO的plot方法绘制识别框
|
||
|
||
# 检查是否有启用人员离岗算法的ROI
|
||
has_off_duty_algorithm = any('人员离岗' in roi.algorithms for roi in self.rois)
|
||
|
||
# 绘制所有ROI
|
||
full_frame_roi_count = 0 # 用于跟踪使用整张画面的ROI数量，避免文本重叠
|
||
for roi in self.rois:
|
||
color = (0, 255, 0) # 默认绿色
|
||
thickness = 2
|
||
|
||
# 根据算法状态设置颜色
|
||
if '人员离岗' in roi.algorithms:
|
||
if roi.is_on_duty:
|
||
color = (0, 255, 0) # 绿色：在岗
|
||
elif roi.is_off_duty and roi.off_duty_timer_start:
|
||
elapsed = time.time() - roi.off_duty_timer_start
|
||
if elapsed >= roi.off_duty_threshold_sec:
|
||
color = (0, 0, 255) # 红色：离岗告警
|
||
else:
|
||
color = (0, 255, 255) # 黄色：离岗中
|
||
else:
|
||
color = (255, 0, 0) # 蓝色：未在岗
|
||
|
||
if '周界入侵' in roi.algorithms:
|
||
color = (255, 255, 0) # 青色：周界入侵区域
|
||
|
||
# 如果有ROI，绘制ROI框
|
||
if roi.roi_points is not None:
|
||
cv2.polylines(vis, [roi.roi_points], True, color, thickness)
|
||
# 创建半透明覆盖层
|
||
overlay = vis.copy()
|
||
cv2.fillPoly(overlay, [roi.roi_points], color)
|
||
cv2.addWeighted(overlay, 0.2, vis, 0.8, 0, vis)
|
||
|
||
# 显示ROI名称（使用中文文本绘制函数）
|
||
text_pos = tuple(roi.roi_points[0])
|
||
vis = put_chinese_text(vis, roi.roi_name, text_pos, font_size=20, color=color, thickness=1)
|
||
else:
|
||
# 如果没有ROI（使用整张画面），在左上角显示提示，避免重叠
|
||
display_text = f"{roi.roi_name} (整张画面)"
|
||
text_y = 30 + full_frame_roi_count * 25 # 每个ROI文本向下偏移25像素
|
||
vis = put_chinese_text(vis, display_text, (10, text_y), font_size=18, color=color, thickness=1)
|
||
full_frame_roi_count += 1
|
||
|
||
# 只在启用人员离岗算法时显示岗位状态
|
||
if has_off_duty_algorithm:
|
||
status = "OUT OF HOURS"
|
||
status_color = (128, 128, 128)
|
||
if in_work:
|
||
# 检查所有ROI的状态
|
||
has_on_duty = any(roi.is_on_duty for roi in self.rois if '人员离岗' in roi.algorithms)
|
||
has_off_duty = any(roi.is_off_duty and roi.off_duty_timer_start
|
||
for roi in self.rois if '人员离岗' in roi.algorithms)
|
||
|
||
if has_on_duty:
|
||
status, status_color = "ON DUTY", (0, 255, 0)
|
||
elif has_off_duty:
|
||
status, status_color = "OFF DUTY", (0, 255, 255)
|
||
else:
|
||
status, status_color = "OFF DUTY", (255, 0, 0)
|
||
|
||
cv2.putText(vis, f"[{self.cam_id}] {status}", (20, 50),
|
||
cv2.FONT_HERSHEY_SIMPLEX, 1, status_color, 2)
|
||
|
||
# 显示时间戳（所有摄像头都显示）
|
||
cv2.putText(vis, now.strftime('%Y-%m-%d %H:%M:%S'), (20, 90),
|
||
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
|
||
cv2.imshow(f"Monitor - {self.cam_id}", vis)
|
||
|
||
|
||
def main():
|
||
import signal
|
||
import sys
|
||
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--config", default="config.yaml", help="配置文件路径")
|
||
args = parser.parse_args()
|
||
|
||
monitor = None
|
||
try:
|
||
monitor = MultiCameraMonitor(args.config)
|
||
|
||
# 注册信号处理，确保优雅退出
|
||
def signal_handler(sig, frame):
|
||
print("\n收到退出信号，正在关闭...")
|
||
if monitor:
|
||
monitor.stop()
|
||
sys.exit(0)
|
||
|
||
signal.signal(signal.SIGINT, signal_handler)
|
||
if sys.platform != 'win32':
|
||
signal.signal(signal.SIGTERM, signal_handler)
|
||
|
||
monitor.run()
|
||
except KeyboardInterrupt:
|
||
print("\n收到键盘中断，正在关闭...")
|
||
except Exception as e:
|
||
print(f"程序异常: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
finally:
|
||
if monitor:
|
||
monitor.stop()
|
||
# 确保进程退出
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |