Lovingh
2021/09/18 阅读：52 主题：默认主题
树莓派
CPU温度实时预警并将信息发送到钉钉
前言：树莓派扮演了服务器的角色，随着访问量的增加或者调试时出现的bug，难免会遇到超载，满载的可能，这是如果有个cpu温度预警，并且可以及时发送到终端，那是再好不过了。
参考了这个脚本，GitHub - 544672716/raspberry: 树莓派学习笔记，并在此基础上修改了温度报警的逻辑
获取钉钉Token
-
在钉钉开发者后台创办一个组织
-
创建应用
创建应用 -
创建好后，进入开发管理，填入服务器出口IP，也就是你树莓派的IP地址，消息接受地址可以不填
-
授权，在权限管理中开启企业内机器人发送消息权限
-
进入版本管理和发布，上线机器人
-
登陆客户端，创建一个组织的内部群
-
设置，进入智能群助手，创建一个自定义机器人
-
创建完成就有可以获取到webhook，其中就有token的信息，保存下来，接下来有用
参照这里自定义机器人接入 - 钉钉开放平台 (dingtalk.com)
python脚本编辑
实现功能：每五秒获取一次温度，当连续10次采集温度都在阀值以上，则通过接口报警，在钉钉实时推送出来
send_temp.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import ssl
from urllib.request import urlopen, Request
import os
import sys
import time
import socket
import subprocess
#钉钉机器人接口函数
class DingDing(object):
def __init__(self, token):
self.url = 'https://oapi.dingtalk.com/robot/send?access_token=%s' % token
self.headers = {'Content-Type': 'application/json'}
def send_text(self, text, at_mobiles=[], at_all=False):
"""
例子: send_text('天气不错', ['13333333333'])
:param text: 消息类型，此时固定为:text
:param at_mobiles: 被@人的手机号 ['13333333333', ]
:param at_all: @所有人时:true,否则为:false
:return:
"""
return self._send_text(text, at_mobiles, at_all)
def send_link(self, title, text, message_url='', pic_url=''):
return self._send_link(title, text, message_url, pic_url)
def send_markdown(self, title, text, at_mobiles=[], at_all=False):
"""发送markdown格式
:param title: 首屏会话透出的展示内容
:param text: markdown格式的消息
:param at_mobiles: 被@人的手机号(在text内容里要有@手机号)
:param at_all: @所有人时:true,否则为:false
:return:
"""
return self._send_markdown(title, text, at_mobiles, at_all)
def send_single_action_card(self, title, text, single_title, single_url, btn_orientation='0', hide_avatar='0'):
"""整体跳转ActionCard类型
:param title: 首屏会话透出的展示内容
:param text: markdown格式的消息
:param single_title: 单个按钮的方案。(设置此项和singleURL后btns无效。)
:param single_url: 点击singleTitle按钮触发的URL
:param btn_orientation: 0-按钮竖直排列，1-按钮横向排列
:param hide_avatar: 0-正常发消息者头像,1-隐藏发消息者头像
:return:
"""
return self._send_single_action_card(title, text, single_title, single_url, btn_orientation, hide_avatar)
def send_action_card(self, title, text, btns, btn_orientation='0', hide_avatar='0'):
"""独立跳转ActionCard类型
:param title: 首屏会话透出的展示内容
:param text: markdown格式的消息
:param btns: 按钮的信息：title-按钮方案，actionURL-点击按钮触发的URL
:param btn_orientation: 0-按钮竖直排列，1-按钮横向排列
:param hide_avatar: 0-正常发消息者头像,1-隐藏发消息者头像
:return:
"""
return self._send_action_card(title, text, btns, btn_orientation, hide_avatar)
def send_feed_card(self, rows):
"""FeedCard类型
例子: send_feed_card([('学vue','https://cn.vuejs.org/','https://cn.vuejs.org/images/logo.png'),
('哪家强', 'https://cn.vuejs.org/', 'https://cn.vuejs.org/images/logo.png')])
:param rows: [(title, messageURL, picURL), (...)]
:return:
"""
return self._send_feed_card(rows)
def _send_feed_card(self, rows):
rows = [{'title': row[0], 'messageURL': row[1], 'picURL': row[2]} for row in rows]
data = {
'feedCard': {
'links': rows
},
'msgtype': 'feedCard'
}
return self._post(data)
def _send_action_card(self, title, text, btns, btn_orientation, hide_avatar):
btns = [{'title': btn[0], 'actionURL': btn[1]} for btn in btns]
data = {
"actionCard": {
"title": title,
"text": text,
"hideAvatar": hide_avatar,
"btnOrientation": btn_orientation,
"btns": btns
},
"msgtype": "actionCard"
}
return self._post(data)
def _send_single_action_card(self, title, text, single_title, single_url, btn_orientation, hide_avatar):
data = {
"actionCard": {
"title": title,
"text": text,
"hideAvatar": hide_avatar,
"btnOrientation": btn_orientation,
"singleTitle": single_title,
"singleURL": single_url
},
"msgtype": "actionCard"
}
return self._post(data)
def _send_markdown(self, title, text, at_mobiles, at_all):
data = {
"msgtype": "markdown",
"markdown": {
"title": title,
"text": text
},
"at": {
"atMobiles": at_mobiles,
"isAtAll": at_all
}
}
return self._post(data)
def _send_text(self, text, at_mobiles, at_all):
data = {
"msgtype": "text",
"text": {
"content": text
},
"at": {
"atMobiles": at_mobiles,
"isAtAll": at_all
}
}
return self._post(data)
def _send_link(self, title, text, message_url, pic_url):
data = {
"msgtype": "link",
"link": {
"text": text,
"title": title,
"picUrl": pic_url,
"messageUrl": message_url
}
}
return self._post(data)
def _post(self, data):
data = json.dumps(data)
req = Request(self.url, data=data.encode('utf-8'), headers=self.headers)
_context = ssl._create_unverified_context()
response = urlopen(req, context=_context)
the_page = response.read()
return json.loads(the_page.decode('utf-8'))
#获取树莓派ip地址
# def get_ip_address():
# s =socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
# s.connect(("1.1.1.1",80))
# ipaddr=s.getsockname()[0]
# s.close()
# return ipaddr
#获取树莓派cpu温度
def get_pi_temperature():
file = open("/sys/class/thermal/thermal_zone0/temp")
temp = float(file.read()) / 1000
return temp
#处理温度数据
def send_temp():
time_str = time.strftime("[%Y-%m-%d %H:%M:%S]",time.location(time.time()))
#ip = get_ip_address()
max_temp = 50 #温度阀值
cnt = 0 #记录温度超过阀值的次数
while True:
now_temp = get_pi_temperature()
#print("打印cpu温度：", now_temp) #调试的时候用到
if now_temp > max_temp:
message = time_str + "请注意！当前设备温度较高，请给它浇点水" + str(now_temp)
cnt += 1
else:
cnt = 0
message = "当前温度:" + str(now_temp)
if(cnt > 10):
# 下面的DingDing里面的参数为ding_token，在创建自定义机器人的时候就能获得
ding = DingDing("3245354254.......")
ding.send_text(message,"这里填入手机号码")
cnt = 0
time.sleep(5)
if __name__ == "__main__":
time.sleep(30) #这个延时一定要加！等待树莓派开机连接网络！
send_message()
这个脚本有很大的扩展性，可以根据需要改写程序增加功能。接下来就是测试程序能否正常运行。
调试脚本
-
将脚本被打印函数的注释去掉，方便查看运行状态 -
将阀值调低，根据实际场景，尽可能让设备程序容易触碰到阀值，一般我的pi工作温度在39-40度，我设置成了39°去测试
回到命令行，输入
python3 send_temp.py #根据自己实际路径调整
正确的场景下，可以看到类似的信息
确认程序没问之后，把打印的注释关掉，并且调整好温度阀值。
让脚本自动开机运行
cd /etc
sudo vim rc.local
在exit0之前，加入
python3 /home/pi/send_temp.py
sudo reboot
重启即可。
--END--
Lovingh
2021/09/18 阅读：52 主题：默认主题
作者介绍