Linux操作系统:智能工业电表开发

智能工业电表开发文档（基于Linux操作系统） 1. 项目概述

智能工业电表是一款用于实时监测工业设备电流和电压的系统。它通过嵌入式Linux设备采集数据，并将数据上传到云端服务器。用户可以通过微信小程序查询设备的实时数据和历史数据，并进行充值操作。

2. 系统架构 2.1 系统架构图 +-------------------+ +------------------+ +-------------------+ | | | | | | | 智能工业电表 |----->| 云服务器 |----->| 微信小程序 | | (嵌入式Linux设备)| | (Linux服务器) | | (用户交互界面) | | | | (数据存储与处理)| | | +-------------------+ +------------------+ +-------------------+ 2.2 技术栈

硬件：嵌入式Linux设备（如树莓派、BeagleBone等）、电流电压传感器模块

通信协议：MQTT协议（设备与服务器通信）

后端：Python + Flask（运行在Linux服务器上）

数据库：MySQL（运行在Linux服务器上）

前端：微信小程序（基于微信开发者工具）

3. 功能模块 3.1 功能模块表 模块名称功能描述数据采集通过电流电压传感器实时采集工业设备的电流和电压数据数据上传将采集到的数据通过MQTT协议上传到云服务器数据存储将上传的数据存储到MySQL数据库数据查询用户通过微信小程序查询设备的实时数据和历史数据充值功能用户通过微信小程序进行电表充值，并更新余额用户管理管理用户账号、设备绑定和权限控制 4. 数据库设计 4.1 数据库表结构 4.1.1 用户表（users） 字段名数据类型描述idINT (Primary Key)用户唯一标识usernameVARCHAR(50)用户名passwordVARCHAR(100)密码（加密存储）devicesJSON绑定的设备列表 4.1.2 设备表（devices） 字段名数据类型描述idINT (Primary Key)设备唯一标识device_idVARCHAR(50)设备编号owner_idINT所属用户IDbalanceFLOAT当前余额dataJSON历史数据记录 4.1.3 数据记录表（data_records） 字段名数据类型描述idINT (Primary Key)记录唯一标识device_idVARCHAR(50)设备编号timestampDATETIME数据采集时间voltageFLOAT电压值currentFLOAT电流值 5. 接口设计 5.1 后端接口 5.1.1 用户登录

URL：/api/login

方法：POST

请求参数：

{ "username": "user123", "password": "password123" }

返回值：

{ "status": "success", "token": "JWT_TOKEN" } 5.1.2 设备数据上传

URL：/api/device/data

方法：POST

请求参数：

{ "device_id": "device123", "voltage": 220.5, "current": 10.2 }

返回值：

{ "status": "success" } 5.1.3 查询设备数据

URL：/api/device/data/:device_id

方法：GET

返回值：

[ { "timestamp": "2024-01-01T12:00:00Z", "voltage": 220.5, "current": 10.2 }, ... ] 5.1.4 充值接口

URL：/api/device/recharge/:device_id

方法：POST

请求参数：

{ "amount": 100 }

返回值：

{ "status": "success", "new_balance": 200 } 6. 示例代码 6.1 嵌入式Linux设备端代码（Python + MQTT） import paho.mqtt.client as mqtt import time import random MQTT_BROKER = "mqtt.example " MQTT_PORT = 1883 DEVICE_ID = "device123" def on_connect(client, userdata, flags, rc): print("Connected to MQTT Broker") def read_voltage(): return random.uniform(220.0, 230.0) # 模拟电压读数 def read_current(): return random.uniform(10.0, 15.0) # 模拟电流读数 client = mqtt.Client() client.on_connect = on_connect client.connect(MQTT_BROKER, MQTT_PORT) while True: voltage = read_voltage() current = read_current() data = f"{{\"device_id\": \"{DEVICE_ID}\", \"voltage\": {voltage}, \"current\": {current}}}" client.publish("device/data", data) print(f"Published: {data}") time.sleep(10) # 每10秒发送一次数据 6.2 后端代码（Python + Flask + MySQL） from flask import Flask, request, jsonify from flask_mysqldb import MySQL import jwt import datetime app = Flask(__name__) app.config["MYSQL_HOST"] = "localhost" app.config["MYSQL_USER"] = "root" app.config["MYSQL_PASSWORD"] = "password" app.config["MYSQL_DB"] = "smartmeter" mysql = MySQL(app) @app.route("/api/login", methods=["POST"]) def login(): data = request.json username = data["username"] password = data["password"] cursor = mysql.connection.cursor() cursor.execute("SELECT id, password FROM users WHERE username = %s", (username,)) user = cursor.fetchone() if user and user[1] == password: token = jwt.encode({"user_id": user[0], "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)}, "secret_key") return jsonify({"status": "success", "token": token}) return jsonify({"status": "error", "message": "Invalid credentials"}), 401 @app.route("/api/device/data", methods=["POST"]) def upload_data(): data = request.json device_id = data["device_id"] voltage = data["voltage"] current = data["current"] cursor = mysql.connection.cursor() cursor.execute("INSERT INTO data_records (device_id, timestamp, voltage, current) VALUES (%s, NOW(), %s, %s)", (device_id, voltage, current)) mysql.connection mit() return jsonify({"status": "success"}) @app.route("/api/device/data/<device_id>", methods=["GET"]) def get_data(device_id): cursor = mysql.connection.cursor() cursor.execute("SELECT timestamp, voltage, current FROM data_records WHERE device_id = %s ORDER BY timestamp DESC", (device_id,)) data = cursor.fetchall() result = [{"timestamp": row[0].isoformat(), "voltage": row[1], "current": row[2]} for row in data] return jsonify(result) @app.route("/api/device/recharge/<device_id>", methods=["POST"]) def recharge(device_id): amount = request.json["amount"] cursor = mysql.connection.cursor() cursor.execute("UPDATE devices SET balance = balance + %s WHERE device_id = %s", (amount, device_id)) mysql.connection mit() cursor.execute("SELECT balance FROM devices WHERE device_id = %s", (device_id,)) new_balance = cursor.fetchone()[0] return jsonify({"status": "success", "new_balance": new_balance}) if __name__ == "__main__": app.run(debug=True) 6.3 微信小程序代码 // 查询设备数据 wx.request({ url: " api.example /api/device/data/device123", method: "GET", success(res) { console.log(res.data); // 数据数组 } }); // 充值 wx.request({ url: " api.example /api/device/recharge/device123", method: "POST", data: { amount: 100 }, success(res) { console.log(res.data); // 新余额 } });