分享一个不用node-red,靠sh脚本以及pyscript来做企业微信推送.... - 『瀚思彼岸』» 智能家居技术论坛

#! /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

复制代码

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秒重试一次

复制代码