picam/picam_cv.py

import os
import re
import json
import subprocess
import time
import numpy as np
import cv2
import threading
import requests
from setting import WEBHOOK_URL
from setting import RTMP_URL
from setting import CAM_NAME

# --- 配置参数 ---
FONT_PATH = "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc"
cam_name = CAM_NAME
push_url = RTMP_URL
notify_interval = 30  # 告警间隔，设为60秒防止轰炸
WIDTH, HEIGHT = 960, 540 # 沿用你跑通的分辨率
FPS = 25
y_size = WIDTH * HEIGHT
frame_size = int(y_size * 1.5)
notify = False


def send_ding_msg(detect_time):
    if not notify:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] notify 参数为 False, 不发送通知")
        return

    current_hour = time.localtime().tm_hour
    if current_hour >= 20 or current_hour < 8:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 夜间免打扰模式, 不发送 webhook 通知")
        return

    """
    异步发送钉钉通知
    """
    headers = {'Content-Type': 'application/json'}
    
    # 构造钉钉要求的 JSON 结构
    data = {
        "msgtype": "text",
        "text": {
            "content": (
                f"【cam-alert】\n"
                f"机位：{cam_name}\n"
                f"时间：{detect_time}\n"
                f"状态：检测到物体移动\n"
                f"请查看回放对齐画面 OSD 时间。"
            )
        },
        "at": {
            "isAtAll": False  # 如果需要提醒所有人，设为 True
        }
    }

    try:
        response = requests.post(
            WEBHOOK_URL, 
            data=json.dumps(data), 
            headers=headers, 
            timeout=5
        )
        res_json = response.json()
        if res_json.get("errcode") == 0:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 钉钉通知发送成功")
        else:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 钉钉发送失败: {res_json.get('errmsg')}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 网络异常(钉钉): {e}")


def monitor_ffmpeg_log(stderr_pipe):
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 日志监控线程已启动...")
    """
    专门解析 FFmpeg stderr 日志的线程函数
    """
    # 正则表达式匹配：帧数、FPS、码率、速度
    # 示例: frame=  150 fps= 25 q=28.0 size=    1024kB time=00:00:06.00 bitrate=1365.3kbits/s speed=1.01x
    regex = re.compile(r"frame=\s*(\d+)\s+fps=\s*([\d.]+)\s+.*?bitrate=\s*([\d.]+kbits/s)\s+speed=\s*([\d.]+x)")

    while True:
        # FFmpeg 的进度信息是用 \r 更新的
        # 我们一次读一部分字节，并按 \r 或 \n 切分
        try:
            # 这里的 read(128) 能保证及时拿到数据
            chunk = stderr_pipe.read(128).decode('utf-8', errors='ignore')
            if not chunk:
                break
            
            # 只要包含 frame=，就说明是进度行
            if "frame=" in chunk:
                # 使用正则或简单的字符串查找提取 fps 和 speed
                # 简单打印出来看看是否有输出
                line_str = chunk.strip().replace('\r', '\n').split('\n')[-1]
                #print(f">>> FFmpeg 状态: {line_str}")
                # 查找统计信息行
                match = regex.search(line_str)
                if match:
                    frame, fps, bitrate, speed = match.groups()
                    # 这里你可以根据数值做逻辑处理
                    # 例如：如果 speed < 0.9x，说明 CPU 跑不动了或者网络堵塞
                    #print(f"--- 推流状态: FPS={fps}, 码率={bitrate}, 速度={speed} ---")
            
                    # 如果速度太慢，可以触发一个标志位让主程序知道
                    # if float(speed[:-1]) < 0.8: print("警告：推流速度过慢！")

                # 也可以捕捉特定的错误信息
                if "Broken pipe" in line_str or "Connection refused" in line_str:
                    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 🚨 关键错误：推流连接断开！")
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 日志读取异常: {e}")
            break


# --- FFmpeg 命令：整合了你的稳定参数 + 我们的管道输入 ---
FFMPEG_CMD = [
    'ffmpeg', '-y',
    '-hide_banner',
    '-f', 'rawvideo', '-pixel_format', 'yuv420p',
    '-video_size', f'{WIDTH}x{HEIGHT}',
    '-framerate', str(FPS),
    '-i', '-', # 从 Python 管道接收数据
    '-thread_queue_size', '8192',
    '-f', 'alsa', '-channels', '1', '-i', 'hw:1,0', # 加上音频防止断流
    '-vf', (
        # 将 cam_name 和 localtime 拼在一起，中间用 | 分隔
        f"drawtext=fontfile={FONT_PATH}:"
        f"text='{cam_name} | %{{localtime\\:%Y-%m-%d %H\\\\\\:%M\\\\\\:%S}}':"
        "x=20:y=20:fontsize=24:fontcolor=yellow:shadowcolor=black:shadowx=2:shadowy=2"
    ),
    '-vcodec', 'libx264', # 使用你验证成功的软编
    '-acodec', 'aac',
    '-b:v', '1M', '-b:a', '64K',
    '-preset:v', 'ultrafast',
    '-tune:v', 'zerolatency',
    '-g', '25',                # 将 GOP 缩短到 25（即每秒一个关键帧），方便服务器切片
    '-keyint_min', '25',
    '-sc_threshold', '0',      # 禁用场景切换检测，保证 GOP 长度固定
    '-f', 'flv',
    push_url
]


def run_smart_cam():
    while True:
        # 清理残余进程，防止端口占用
        os.system("pkill -9 ffmpeg > /dev/null 2>&1")
        os.system("pkill -9 libcamera-vid > /dev/null 2>&1")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 休眠 5s 等待残余进程被彻底清理...")
        time.sleep(5)
        
        # 1. 启动 libcamera 采集原始数据（为了 OpenCV 必须拿原始流）
        # 如果你一定要用 v4l2，可以改用 ffmpeg 读取再转给 python
        input_cmd = [
            'rpicam-vid', '-t', '0', '--width', str(WIDTH), '--height', str(HEIGHT),
            '--framerate', str(FPS), '--codec', 'yuv420', '--nopreview', '--flush', '-o', '-'
        ]
        
        in_proc = subprocess.Popen(input_cmd, stdout=subprocess.PIPE, bufsize=frame_size*3)
        #out_proc = subprocess.Popen(FFMPEG_CMD, stdin=subprocess.PIPE)
        out_proc = subprocess.Popen(
                FFMPEG_CMD,
                stdin=subprocess.PIPE,
                stderr=None,  # 必须开启 stderr 捕获日志
                bufsize=0                # 禁用缓冲，确保实时获取日志
        )

        # 启动 ffmpeg 日志监控线程
        log_thread = threading.Thread(target=monitor_ffmpeg_log, args=(out_proc.stderr,))
        log_thread.daemon = True
        log_thread.start()
        
        fgbg = cv2.createBackgroundSubtractorMOG2(history=100, varThreshold=50)
        
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 智能推流启动...")
        try:
            last_log_time = 0
            # --- 循环外定义变量 ---
            last_notify_time = 0
            while True:
                raw_data = in_proc.stdout.read(frame_size)
                if len(raw_data) < frame_size: break
                
                # --- OpenCV 移动检测 (抽样检测节省 CPU) ---
                # 直接在 Y 通道操作，不影响最终推流
                y_buf = np.frombuffer(raw_data, dtype=np.uint8, count=y_size).reshape((HEIGHT, WIDTH))
                
                # 每 5 帧做一次检测
                small_y = cv2.resize(y_buf, (212, 120)) 
                fgmask = fgbg.apply(small_y)
                is_moving = np.count_nonzero(fgmask > 200) > 50
                if is_moving:
                    current_time_val = time.time()
                    # 获取精确到秒的时间戳
                    readable_time = time.strftime('%Y-%m-%d %Y-%m-%d %H:%M:%S')
                    # 冷却时间 2 秒，避免频繁触发
                    if current_time_val - last_log_time > 2:
                        #print(f"视频回放定位点: 请寻找画面 OSD 显示为 {readable_time} 附近的片段")
                        # 如果你有关联的抓拍，文件名也用这个时间
                        # filename = f"{save_dir}/motion_{time.strftime('%H%M%S')}.jpg"
                        # cv2.imwrite(filename, y_buf)
                        last_log_time = current_time_val

                    if current_time_val - last_notify_time > notify_interval:
                        # 开启线程异步发送
                        t = threading.Thread(target=send_ding_msg, args=(readable_time,))

                        t.daemon = True
                        t.start()
                        last_notify_time = current_time_val

                # 将处理后的数据写入 FFmpeg
                out_proc.stdin.write(raw_data)

                if out_proc.poll() is not None: break

        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 运行异常: {e}")
        finally:
            in_proc.terminate()
            out_proc.terminate()
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] FFmpeg 进程已结束，5秒后重启...")
            time.sleep(5)


if __name__ == "__main__":
    run_smart_cam()