需求
实现通过微信公众号对树莓派上面连接的外设进行控制和信息采集，温湿度获取、LED灯开关、步进电机控制、人员检测、试试监控等。
web采用Flask框架，微信消息处理采用wechatpy模块
Flask简介
Flask是一个基于python开发的一个强大的web框架，能够实现更加简单、灵活且细致的控制，能够由用户自主决定实现方式，实现更多的定制，适用于中小型网站的开发。基本可以说是接受用户请求然后反馈给用户，开发时可以很方便的对功能进行扩展和集成。使用flask进行web应用开发时,不需要关注http请求是如何发出和接收的,flask框架为我们实现好了相关的网络报的处理以及相应的网络相应，我们只需要关注接受到用户请求如何处理相关的逻辑然后将结果返回给用户即可。
wechatpy简介
wechatpy 是一个微信 (WeChat) 的第三方 Python SDK, 实现了微信公众号、企业微信和微信支付等 API。
环境配置
这里采用的python3版本，使用pycharm创建的flask项目，安装wechatpy
pip install wechatpy
系统实现的主要代码
微信公众号这里直接使用的微信测试号，微信扫码即可使用。
菜单设置
通过菜单实现一部分操作，这里只用到了点击和页面跳转操作。*.proxys.nknow.top是使用ngrok自己搭建的内网穿透配置的域名。
from wechatpy import WeChatClient
client = WeChatClient('wxfb561a9*********', '19078e4488932d02938**********')
client.menu.delete()
status = client.menu.create(
{
"button": [
{
"name": "控制",
"sub_button": [
{
"type": "click",
"name": "开灯",
"key": "on_light"
},
{
"type": "click",
"name": "关灯",
"key": "off_light"
},
{
"type": "click",
"name": "开窗帘",
"key": "on_curtain"
},
{
"type": "click",
"name": "关窗帘",
"key": "off_curtain"
},
{
"type": "click",
"name": "实况拍摄",
"key": "monitor"
},
]
},
{
"name": "实时",
"sub_button": [
{
"type": "view",
"name": "设备状况",
"url": "http://dashboard.proxys.nknow.top/"
},
{
"type": "view",
"name": "室内监控",
"url": "http://sv.proxys.nknow.top/"
},
{
"type": "click",
"name": "房间温度",
"key": "temperature"
},
{
"type": "click",
"name": "房间湿度",
"key": "humidity"
},
{
"type": "click",
"name": "people",
"key": "people"
}
]
},
{
"name": "其他",
"sub_button": [
{
"type": "view",
"name": "智能助手",
"url": "http://robot.proxys.nknow.top/"
},
{
"type": "view",
"name": "cloud",
"url": "http://cloud.proxys.nknow.top/"
},
{
"type": "view",
"name": "关于",
"url": "http://about.proxys.nknow.top/"
}
]
}
]
}
)
print(status)
直接运行该代码即可，返回{'errcode': 0, 'errmsg': 'ok'}菜单设置成功。
微信公众号认证
接入指南参考微信公众号接入指南
接口认证时微信服务器发起get请求，对参数进行加密校验，flask视图增加get请求进行结果认证，代码如下：
def get(self):
signature = request.args.get('signature', '')
timestamp = request.args.get('timestamp', '')
nonce = request.args.get('nonce', '')
echo_str = request.args.get('echostr', '')
try:
check_signature('my_token'],
signature, timestamp, nonce)
except InvalidSignatureException:
echo_str = '错误的请求'
return echo_str
然后在微信公众号管理页面配置该链接即可。
处理用户消息
这里只处理了用户点击事件的消息，其他类型的消息没有处理，可以根据自己的情况二次开发扩展，在flask视图中的关键代码如下：
def post(self):
msg = parse_message(request.data)
# 是文本消息
if msg.type == 'text':
reply = TextReply(message=msg)
reply.content = 'text reply'
xml = reply.render()
return xml
# 事件类型消息
elif msg.type == 'event':
# 处理菜单点击事件
if msg.event == 'click':
# MenuClick是自定义类，处理菜单点击事件
click = MenuClick(msg)
return click.menu_click()
# 其他类型的事件消息不处理
else:
reply = TextReply(message=msg)
reply.content = 'event未知'
xml = reply.render()
return xml
# 其他消息
else:
reply = TextReply(message=msg)
reply.content = '未知操作'
xml = reply.render()
return xml
菜单点击事件处理类
为了方便菜单点击事件的处理，之间将菜单的key设置为对应的函数名字，这样不用判断就可以直接调用对应的方法进行处理了：
class MenuClick(object):
def __init__(self, msg, ):
self.msg = msg
self.path = current_app.config['CODE_PATH'],
redis_client = Redis.from_url(current_app.config['REDIS_STORAGE'])
session_interface = RedisStorage(
redis_client,
prefix="ACCESS_TOKEN"
)
self.client = WeChatClient(
current_app.config['APP_ID'],
current_app.config['SECRET'],
session=session_interface
)
# 根据点击的按钮实际完成相应的操作
def menu_click(self):
try:
return getattr(self, self.msg.key)()
except Exception as e:
print('发生错误的文件：', e.__traceback__.tb_frame.f_globals['__file__'])
print('错误所在的行号：', e.__traceback__.tb_lineno)
print(e)
return self.return_text("未知操作")
# 开灯
def on_light(self):
text = PiSysCall(self.path[0]).on_light()
# 开灯相关操作
return self.return_text(text)
# 关灯
def off_light(self):
text = PiSysCall(self.path[0]).off_light()
# 开灯相关操作
return self.return_text(text)
# 温度
def temperature(self):
celsius = PiSysCall(self.path[0]).temperature()
return self.return_text(celsius)
# 湿度
def humidity(self):
celsius = PiSysCall(self.path[0]).humidity()
return self.return_text(celsius)
# 监控[当前状况拍摄照片返回]
def monitor(self):
self.send_img()
return self.return_text("请点击查看实况拍摄照片")
# 关闭窗帘
def off_curtain(self):
text = PiSysCall(self.path[0]).off_curtain()
# text = "窗帘已关闭"
return self.return_text(text)
# 打开窗帘
def on_curtain(self):
text = PiSysCall(self.path[0]).on_curtain()
# text = "窗帘已打开"
return self.return_text(text)
# 反馈是否有人
def people(self):
text = PiSysCall(self.path[0]).people()
if "有人" in text:
self.send_img()
return self.return_text(text)
# 描述性文字
def return_text(self, text):
reply = TextReply(message=self.msg)
reply.content = text
# 转换成 XML
xml = reply.render()
return xml
# 发送照片
def send_img(self):
# 调用系统函数
# 拍照截图 图片为类型+时间戳.jpg
# img_file = "类型+时间戳.jpg"
img_file = PiSysCall(self.path[0]).monitor()
img = open(img_file, 'rb')
img_upload = self.client.media.upload('image', img)
self.client.message.send_image(self.msg.source, img_upload['media_id'])
PiSysCall类
上述代码中使用了PiSysCall类进行了事件处理，代码内容如下：
class PiSysCall(object):
def __init__(self, path_info):
self.path = path_info + '/'
# 系统调用
def sys_call(self, command):
# command 命令字符串
# ["echo '123'"]
# 执行结果列表数据
tmp = subprocess.check_output(command, shell=True)
return tmp.decode('utf8').strip().split('\n')
# 开灯
def on_light(self):
# 开灯相关操作
try:
command = ['python {} {}'.format(self.path + 'light.py', 'on')]
self.sys_call(command)
return "灯已打开"
except Exception as e:
return '操作失败'
# 关灯
def off_light(self):
# 关灯相关操作
try:
command = ['python {} {}'.format(self.path + 'light.py', 'off')]
self.sys_call(command)
return "灯已关闭"
except Exception as e:
return '操作失败'
# 温度
def temperature(self):
try:
command = ['python {}'.format(self.path + 'temperature.py')]
celsius = self.sys_call(command)
return "当前温度" + str(celsius[0]) + "℃"
except Exception as e:
return '操作失败'
# 湿度
def humidity(self):
try:
command = ['python {}'.format(self.path + 'humidity.py')]
dampness = self.sys_call(command)
return "当前湿度" + str(dampness[0]) + "%"
except Exception as e:
return '操作失败'
# 监控[当前状况拍摄照片返回]
def monitor(self):
tz_sh = tz.gettz('Asia/Shanghai')
name = 'vs-{}.jpg'.format(datetime.datetime.now(tz=tz_sh).strftime('%Y-%m-%d-%H:%M:%S.%f'))
img_file = self.path + 'images/' + name
try:
command = ['fswebcam --no-banner -r 640x480 {}'.format(img_file)]
self.sys_call(command)
return img_file
except Exception as e:
print(e)
return '操作失败'
# 调用步进电机开窗帘
def on_curtain(self):
try:
command = ['python {}'.format(self.path + 'tasks.py on')]
dampness = self.sys_call(command)
return dampness[0]
except Exception as e:
return '操作失败'
# 调用步进电机关窗帘
def off_curtain(self):
try:
command = ['python {}'.format(self.path + 'tasks.py off')]
dampness = self.sys_call(command)
return dampness[0]
except Exception as e:
return '操作失败'
# 判断房间是否有人
def people(self):
try:
command = ['python {}'.format(self.path + 'people.py')]
dampness = self.sys_call(command)
if dampness[0] == 'yes':
return "房间有人"
else:
return "房间无人"
except Exception as e:
return '操作失败'
其中图片消息，实现的思路是先发起系统调用进行图片拍摄，然后读取文件名字，打开图片文件上传到微信公众平台素材中心，根据返回的素材id发送给用户。
效果图
其中控制和实时下的子菜单都是点击事件类型的菜单，执行效果如下：
