关于具体实现效果，在本文末尾会有效果演示，需要相关产品型号等信息可以私聊我获取，欢迎关注。
1. 背景说明
我们一个项目已经在实施阶段，客户提出了一个新的要求：即在一个特定紧急情况下，在他们(客户)的电脑设备上连接一个报警器发出报警信号（报警灯闪光+发出蜂鸣警告）。
关于硬件的开发已经可以追溯到我的大学生活了，那时的我们刚开始时使用汇编和 C 做单片机开发，再往后一点使用 C++进行 ARM 开发，最后使用 Z-Stack 做物联网开发。可以距今已经太遥远了，当初掌握的知识早已献祭给时间了。
这个小小的要求摆在我的面前，蛮大的压力，我必须攻克它，但是我也并不准备为了它付出很大的代价。所以我最后的决定是这样的：硬件设备部分，直接采购成型产品(暴露 API 或者类似串口之类的调用方式)；软件部分，和服务端建立 websocket 连接，接收到服务端推送的紧急信息时，调用硬件发出报警信息。
2.环境物料准备
当我在某宝搜索关键词【报警灯 二次开发】、【报警灯 api】这样的关键词，找到的适合的产品寥寥，仅有的几个外形、价格都不让人满意，最后当把关键词换为【报警灯 串口 开发】之后，我找到了我想要的东西(售价 60 米)。内建 USB 转串口，提供相应的串口指令集。
火速网购，物流，快！硬件到手，为了开发简单，我准备使用 Python 进行串口通信和程序开发。
鉴于此次开发使用 Python，特在此列出开发环境：
Windows7 64
Python3.6
python 依赖模块包括：websocket-client(websocket)、pyserial(串口)、pyinstaller(打包 exe)
3.前期调试测试
3.1 硬件产品测试
首先需要测试一下收到的产品是否正常运行，并且可以使用串口发送指令控制。
第一步、将硬件设备的 usb 插口插入电脑
第二步、安装驱动：卖家提供了该驱动程序
第三步、找到具体串口号（右键计算机->管理->设备管理器->端口），我的串口是 COM10（注意，此串口号不同设备会有不同，同设备不同插口也会有所不同）
第四步、使用产品附带控制软件进行测试，打开串口之后，测试无异常
3.2 Python 串口调试
接下来进行使用 Python 控制该硬件的调试。
01 05 00 00 ff 00 8C 3A 打开红灯(常亮) 01 05 00 00 00 00 CD CA 关闭红色01 05 00 03 ff 00 7C 3A 打开蜂鸣器(一直响) 01 05 00 03 00 00 3D CA 关闭蜂鸣器
01 05 00 00 f0 00 89 CA 红灯闪烁(2Hz)01 05 00 00 f1 00 88 5A 红灯闪烁(1Hz)01 05 00 00 f2 00 88 AA 红灯闪烁(0.5Hz)01 05 00 00 f3 00 89 3A 红灯闪烁(0.25Hz)
01 05 00 03 f0 00 79 CA 蜂鸣器模式1(2Hz)01 05 00 03 f1 00 78 5A 蜂鸣器模式2(1Hz)01 05 00 03 f2 00 78 AA 蜂鸣器模式2(0.5Hz)01 05 00 03 f3 00 79 3A 蜂鸣器模式2(0.25Hz)
复制代码
该设备提供了上述指令集，并且串口和指令要求如下
串口设置： 9600,8,n,1。控制命令，HEX格式发送(不能以ASCII格式发送)
复制代码
Python 实现 Demo 如下
import serialimport time
#### 串口号，右键计算机->点击设备->点击端口->查看获取SERIAL_PORT = 'COM10'ser=serial.Serial(SERIAL_PORT,9600,timeout=0.5)
######## 指令声明## 红灯常亮RED_LIGHT_CMD_ALWAYS = b'\x01\x05\x00\x00\xff\x00\x8C\x3A'## 关闭红灯RED_LIGHT_CMD_CLOSE = b'\x01\x05\x00\x00\x00\x00\xCD\xCA'## 红灯闪烁2HZRED_LIGHT_CMD_2HZ = b'\x01\x05\x00\x00\xf0\x00\x89\xCA'## 红灯闪烁1HZRED_LIGHT_CMD_1HZ = b'\x01\x05\x00\x00\xf1\x00\x88\x5A'## 红灯闪烁0.5HZRED_LIGHT_CMD_0_5HZ = b'\x01\x05\x00\x00\xf2\x00\x88\xAA'## 红灯闪烁0.25HZRED_LIGHT_CMD_0_25HZ = b'\x01\x05\x00\x00\xf3\x00\x89\x3A'## 蜂鸣器常响BUZZ_CMD_ALWAYS = b'\x01\x05\x00\x03\xff\x00\x7C\x3A'## 关闭蜂鸣器BUZZ_CMD_CLOSE = b'\x01\x05\x00\x03\x00\x00\x3D\xCA'## 蜂鸣器模式2HZBUZZ_CMD_2HZ = b'\x01\x05\x00\x03\xf0\x00\x79\xCA'## 蜂鸣器模式1HZBUZZ_CMD_1HZ = b'\x01\x05\x00\x03\xf1\x00\x78\x5A'## 蜂鸣器模式0.5HZBUZZ_CMD_0_5HZ = b'\x01\x05\x00\x03\xf2\x00\x78\xAA'## 蜂鸣器模式0.25HZBUZZ_CMD_0_25HZ = b'\x01\x05\x00\x03\xf3\x00\x79\x3A'
##### 打开端口####def openSerial(): if not ser.isOpen(): ser.close() ser.open() print('成功连接到串口！')
##### 向设备发送指令####def sendCmdToDevice(cmd, t=0.1): openSerial() ser.write(cmd) ##默认延时0.1秒，防止指令叠加发送导致指令失效，也可以控制指令持续时间，如希望红灯亮10秒，可传t=10 time.sleep(t) ser.close()
if __name__ == '__main__': #sendCmdToDevice(RED_LIGHT_CMD_ALWAYS) sendCmdToDevice(RED_LIGHT_CMD_CLOSE) #sendCmdToDevice(RED_LIGHT_CMD_2HZ) #sendCmdToDevice(RED_LIGHT_CMD_1HZ) #sendCmdToDevice(RED_LIGHT_CMD_0_5HZ) #sendCmdToDevice(RED_LIGHT_CMD_0_25HZ) #sendCmdToDevice(BUZZ_CMD_ALWAYS) #sendCmdToDevice(BUZZ_CMD_CLOSE) #sendCmdToDevice(BUZZ_CMD_2HZ) #sendCmdToDevice(BUZZ_CMD_1HZ) #sendCmdToDevice(BUZZ_CMD_0_5HZ) #sendCmdToDevice(BUZZ_CMD_0_25HZ)
复制代码
如上，对所有的指令进行 16 进制转换后，使用 open 函数打开端口，使用 write 函数向端口写指令，发送指令后 close 掉。对所有操作都测试无误后进入下阶段的开发。
4.源码开发
websocket 的服务端使用 Java 开发，不再赘述。
接下来使用 Python 开发 websocket 客户端，监听到报警信息，调用设备发出报警。源码如下：
config.json
{ "USER_ID": "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX", "WARN_TIME": "3", "WEB_SOCKET_URL": "wss://{{DOMAIN}}/websocket/websocket/", "SERIAL_PORT": "COM10", "IF_BUZZ": "True"}
复制代码
USER_ID：用于区分 socket 的中的不同用户
WARN_TIME：警报持续时间（单位秒）
WEB_SOCKET_URL：websocket 的服务端地址
SERIAL_PORT：串口号
IF_BUZZ：是否开启蜂鸣警报（True/False）
daemon.py
import websocketimport threadingimport timeimport serialimport json
#### 读取配置文件jsonConfigStr = ''with open("config.json", "r") as f: jsonConfigStr = f.read()configData = json.loads(jsonConfigStr)
#### !!!!用户IDDEPT_ID = configData['USER_ID']#### !!!!警报持续时长(秒)WARN_TIME = int(configData['WARN_TIME'])WEB_SOCKET_URL = configData['WEB_SOCKET_URL']WS_URL = WEB_SOCKET_URL + DEPT_ID + '-' + str(int(time.time()))
#### !!!!串口号，右键计算机->点击设备->点击端口->查看获取SERIAL_PORT = configData['SERIAL_PORT']ser=serial.Serial(SERIAL_PORT,9600,timeout=0.5)
#### !!!!是否开启蜂鸣IF_BUZZ = configData['IF_BUZZ'] == 'True'
######## 指令声明## 红灯常亮RED_LIGHT_CMD_ALWAYS = b'\x01\x05\x00\x00\xff\x00\x8C\x3A'## 关闭红灯RED_LIGHT_CMD_CLOSE = b'\x01\x05\x00\x00\x00\x00\xCD\xCA'## 红灯闪烁2HZRED_LIGHT_CMD_2HZ = b'\x01\x05\x00\x00\xf0\x00\x89\xCA'## 红灯闪烁1HZRED_LIGHT_CMD_1HZ = b'\x01\x05\x00\x00\xf1\x00\x88\x5A'## 红灯闪烁0.5HZRED_LIGHT_CMD_0_5HZ = b'\x01\x05\x00\x00\xf2\x00\x88\xAA'## 红灯闪烁0.25HZRED_LIGHT_CMD_0_25HZ = b'\x01\x05\x00\x00\xf3\x00\x89\x3A'## 蜂鸣器常响BUZZ_CMD_ALWAYS = b'\x01\x05\x00\x03\xff\x00\x7C\x3A'## 关闭蜂鸣器BUZZ_CMD_CLOSE = b'\x01\x05\x00\x03\x00\x00\x3D\xCA'## 蜂鸣器模式2HZBUZZ_CMD_2HZ = b'\x01\x05\x00\x03\xf0\x00\x79\xCA'## 蜂鸣器模式1HZBUZZ_CMD_1HZ = b'\x01\x05\x00\x03\xf1\x00\x78\x5A'## 蜂鸣器模式0.5HZBUZZ_CMD_0_5HZ = b'\x01\x05\x00\x03\xf2\x00\x78\xAA'## 蜂鸣器模式0.25HZBUZZ_CMD_0_25HZ = b'\x01\x05\x00\x03\xf3\x00\x79\x3A'
##### 延时####def delay(s): try: time.sleep(s) except: delay(s)
##### 打开端口####def openSerial(): if not ser.isOpen(): ser.close() ser.open() print('成功连接到串口！')
##### 向设备发送指令####def sendCmdToDevice(cmd, t=0.1): openSerial() ser.write(cmd) delay(t) ser.close()
###### 心跳线程#####class heartBeatThread (threading.Thread): def __init__(self, ws): threading.Thread.__init__(self) self.ws = ws def run(self): while True: ws.send('ping') print('ping!') delay(3)
###### 警报线程:红灯2HZ闪烁#####class warnThread (threading.Thread): def __init__(self, ws): threading.Thread.__init__(self) self.ws = ws def run(self): sendCmdToDevice(RED_LIGHT_CMD_2HZ) if IF_BUZZ: sendCmdToDevice(BUZZ_CMD_2HZ) delay(WARN_TIME) sendCmdToDevice(RED_LIGHT_CMD_CLOSE) if IF_BUZZ: sendCmdToDevice(BUZZ_CMD_CLOSE)
#### websocket-client:接受消息回调###def on_message(ws, message): if 'warninfo' in message: print('检测到报警信息') warnThread(ws).start() else: print(message)
#### websocket-client:发生错误回调###def on_error(ws, error): print("链接出现错误")
#### websocket-client:链接断开回调###def on_close(ws): print("链接已经断开")
#### websocket-client:链接建立回调###def on_open(ws): print('检测到链接建立') heartBeatThread(ws).start()
if __name__ == '__main__': websocket.enableTrace(True) ws = websocket.WebSocketApp(WS_URL, on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close) try: ws.run_forever() except: ws.close()
复制代码
截止到此，开发全部完成
以上涉及到的源码其实还是有不足和缺陷的，如果你发现的话，不妨在讨论区交流哦～
5.打包
pyinstaller -D -i i.ico daemon.py
复制代码
在 dist 下生成了可执行文件和相关依赖
6.效果演示
收到报警信息后红灯闪烁并蜂鸣，点击查看视频效果
评论