墨滴

Lovingh

V1

2021/09/18阅读：396主题：默认主题

树莓派

CPU温度实时预警并将信息发送到钉钉

前言：树莓派扮演了服务器的角色，随着访问量的增加或者调试时出现的bug，难免会遇到超载，满载的可能，这是如果有个cpu温度预警，并且可以及时发送到终端，那是再好不过了。

参考了这个脚本，GitHub - 544672716/raspberry: 树莓派学习笔记，并在此基础上修改了温度报警的逻辑

获取钉钉Token

  1. 在钉钉开发者后台创办一个组织

  2. 创建应用

    创建应用
    创建应用

  3. 创建好后，进入开发管理，填入服务器出口IP，也就是你树莓派的IP地址，消息接受地址可以不填

  4. 授权，在权限管理中开启企业内机器人发送消息权限

  5. 进入版本管理和发布，上线机器人

  6. 登陆客户端，创建一个组织的内部群

  7. 设置，进入智能群助手，创建一个自定义机器人

  8. 创建完成就有可以获取到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--

分类：

后端

标签：

Python

作者介绍

Lovingh
V1