分享一个不用node-red,靠sh脚本以及pyscript来做企业微信推送....

animals

补充一下前提, 需要本地图像识别系统才能使用, 具体可以参照
i. 免费的本地图像识别系统（人体、物体）并接入HA - 『HomeAssistant』综合讨论区 - 『瀚思彼岸』» 智能家居技术论坛 - Powered by Discuz! (hassbian.com)
ii. 搬运2个摄像头人体插件和一个人脸识别插件（都是本地） - 『HomeAssistant』综合讨论区 - 『瀚思彼岸』» 智能家居技术论坛 - Powered by Discuz! (hassbian.com) 的第二款
iii. 或者其他关于deepstack的教程

pyscript的教程 我彻底弃用HA模版Jinja了，因为Python更好更容易 [Pyscript插件] - 『HomeAssistant』综合讨论区 - 『瀚思彼岸』» 智能家居技术论坛 - Powered by Discuz! (hassbian.com)

1. 编写sh脚本,用于调用ffmpeg命令生成gif

#! /bin/bash

today=$(date -d $4 +"%Y-%m-%d")
time=$(date -d $5 +"%H-%M-%S")
root_path="/config/www/video"
user=$1
password=$2
ip=$3

# echo $4
# echo ${today}
# echo ${time}

mkdir -p ${root_path}/${today}

/usr/bin/ffmpeg -rtsp_transport tcp -i rtsp://$1:$2@$3:554/h265/ch1/main/av_stream -max_muxing_queue_size 9999 -t 10 -vf scale=1024:768 ${root_path}/${today}/${time}.avi > /dev/null
/usr/bin/ffmpeg -i ${root_path}/${today}/${time}.avi ${root_path}/${today}/${time}.gif > /dev/null
rm -f ${root_path}/${today}/${time}.avi

复制代码

补充一个指定分辨率来生成avi文件,适合硬盘不够大的同学使用, 分辨率根据实际情况来 参数是"-vf scale=1024:768" 上面的代码中已经补上了

2. 编写shell_command, 用来调用sh脚本

shell_command:
  #服务名称根据自己喜好来
  save_video: sh /config/ffmpeg_ten_second_video.sh {{user}} {{pwd}} {{ip}} {{date}} {{time}}
  #{{}}中的参数按顺序对应sh脚本中的$1-$5

复制代码

<span style="background-color: rgb(255, 255, 255);">3. 编写pyscript脚本,用来监听事件</span>

复制代码

import os
from datetime import datetime

def findAllVideo(base):
    files = []
    for root, ds, fs in os.walk(base):
        for f in fs:
            if f.endswith('.avi'):
                videos.append({'name': f, 'time': os.path.getmtime(f)})

#监听之前定义好的人物识别事件, 并当对象是人时
@event_trigger('deepstack.object_detected', "name == 'person'")
def push_video(**kwargs):
    # log.warning(f'WARNING: print kwargs: {kwargs}')
    # print(kwargs)
    
    #根据事件获取实体
    entity = kwargs["entity_id"]
    #这是我这里的相机实体
    cam_entity = f'camera.{entity}'
    
    # log.warning(f'WARNING:  cam_entity: {cam_entity}')
    ip = None #摄像头ip
    pwd = '' #摄像头rtsp密码
    place = None #摄像头的名字 例如 大门
    user = '' #摄像头的rtsp用户名
    
    if entity == 'camera_1':
        ip = 'ip1'
        place = "大门入口"
        user = 'admin'
        pwd = '123456'
    elif entity == 'camera_1':
        ip = 'ip2'
        place = "花园"
        user = 'haha'
        pwd = '567890'
    #...可以继续
    
    if ip:
        now = datetime.now()
        png_name = now.strftime('%Y-%m-%d-%H:%M')
        today = str(now.date())
        now_time = now.strftime("%H:%M:%S")#camera.snapshot
        #调用截图事件,生成封面图
        service.call("camera", "snapshot", entity_id=cam_entity, filename=f"截图存放路径")
        # print(pwd,ip,today,now_time)
        # log.info(f'INFO: pwd:{pwd}, ip:{ip}, today:{today}, time: {now_time}')
        # log.warning(f'WARNING:  pwd:{pwd}, ip:{ip}, today:{today}, time: {now_time}')
        
        #调用shell_command命令,使用ffmepg生成gif文件
        #问题: 产生文件效率较慢,有其他更好的截取视频的方案吗?
        service.call("shell_command", "save_video", user=user,pwd=pwd,ip=ip,date=today, time=now_time)
        
        path = f'视频及gif的存放路径'
        if os.path.exists(path):
            _try_times = 0
            
            #重试20次
            gif_path = f"{path}/{now_time.replace(':','-')}.gif" #gif文件路径
            avi_path = f"{path}/{now_time.replace(':','-')}.avi" #avi文件路径
            for i in range(20):
                # log.warning(f'WARNING: gif path check{str(i+1)}')
                # if os.path.exists(avi_path):
                #     log.warning(f'WARNING: avi:{avi_path} size: {os.path.getsize(avi_path)}')
                # if os.path.exists(gif_path):
                #     log.warning(f'WARNING: gif:{gif_path} size: {os.path.getsize(gif_path)}')
                #如果avi删除并且存在gif文件,表示gif已经成功生成了
                if os.path.exists(gif_path) and not os.path.exists(avi_path):
                    # log.warning(f'WARNING: gif exists')
                    service.call("notify", "wework", data={"type":"news","picurl":f"{hasUrl}/local/{封面路径}", "url":f"{hasUrl}/local/{gif路径}", title=f"{place}发现有人活动", message=f"##可信度:{kwargs['confidence']}，位置为:{kwargs['centroid']['x']},{kwargs['centroid']['y']} 点击查看动图")
                    return
                task.sleep(3)
                #使用task.sleep, 如果使用time包的sleep会造成线程阻塞
                # else:
                #     _time.sleep(3) #5秒重试一次

复制代码

4. 大概效果

ucomeito

不错，有空试一下。

sorrypqa

太牛了，高手就是可以为所欲为

hyq

高手可以为所欲为，小白就只能眼馋！

ilongjiang

没看明白怎么用

ahmengkai

虽然没看懂 还是先收藏起来