|
@@ -0,0 +1,84 @@ |
|
|
|
|
|
# // /Users/zhangzhiqi/Library/Mobile Documents/M6HJR9W95L~com~textasticapp~textastic/Documents/PycharmProjects/PythonAll/自己的/面试相关题目/Answer_0813.py |
|
|
|
|
|
# // Copyright (c) 2023. |
|
|
|
|
|
# // @Time : 2023/8/13 9:54:11 |
|
|
|
|
|
# // @Author : 张稚琦 |
|
|
|
|
|
# // @Address: 广东省深圳市科瑞智造产业园瑞明大厦A座 18楼 |
|
|
|
|
|
# // @Email : zhang@zhang.mba / zhangzhiqi828@gmail.com / zhangzhiqi@lh83.onmicrosoft.com / 2272358828@qq.com / zhangzhiqi@innorev.com.cn |
|
|
|
|
|
# // @Personal Blog: https://zhang.mba |
|
|
|
|
|
# // @File : Answer_0813.py |
|
|
|
|
|
# // @LastModified: 2023/8/13 上午9:54 |
|
|
|
|
|
# // @ProjectName : PythonAll |
|
|
|
|
|
# |
|
|
|
|
|
import json # 导入json库,用于处理JSON数据 |
|
|
|
|
|
from flask import Flask, request, jsonify # 导入Flask、request和jsonify模块 |
|
|
|
|
|
import paho.mqtt.client as mqtt # 导入mqtt模块 |
|
|
|
|
|
|
|
|
|
|
|
app = Flask(__name__) # 创建Flask应用 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
mqtt_client = mqtt.Client() # 创建MQTT客户端实例 |
|
|
|
|
|
mqtt_client.connect("公司设备MQTT代理服务器地址", 1883) # 连接到MQTT代理服务器 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route('/motor_control', methods=['POST']) |
|
|
|
|
|
def motor_control(): |
|
|
|
|
|
""" |
|
|
|
|
|
控制电机的接口 |
|
|
|
|
|
""" |
|
|
|
|
|
data = request.get_json() # 从请求中获取JSON数据 |
|
|
|
|
|
topic = '/motor/{}/{}'.format(data['model'], data['id']) # 生成电机控制主题 |
|
|
|
|
|
payload = generate_motor_payload(data) # 生成电机控制消息负载 |
|
|
|
|
|
mqtt_client.publish(topic, json.dumps(payload)) # 发布消息到MQTT代理 |
|
|
|
|
|
|
|
|
|
|
|
return jsonify({'message': 'Motor control request sent'}) # 返回JSON响应 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.route('/claw_control', methods=['POST']) |
|
|
|
|
|
def claw_control(): |
|
|
|
|
|
""" |
|
|
|
|
|
控制爪子的接口 |
|
|
|
|
|
""" |
|
|
|
|
|
data = request.get_json() # 从请求中获取JSON数据 |
|
|
|
|
|
topic = '/claw/{}/{}'.format(data['model'], data['id']) # 生成爪子控制主题 |
|
|
|
|
|
payload = generate_claw_payload(data) # 生成爪子控制消息负载 |
|
|
|
|
|
mqtt_client.publish(topic, json.dumps(payload)) # 发布消息到MQTT代理 |
|
|
|
|
|
|
|
|
|
|
|
return jsonify({'message': 'Claw control request sent'}) # 返回JSON响应 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def generate_motor_payload(data): |
|
|
|
|
|
""" |
|
|
|
|
|
生成控制电机的消息负载 |
|
|
|
|
|
""" |
|
|
|
|
|
payload = { |
|
|
|
|
|
"MesType": "Request", |
|
|
|
|
|
"ClientId": "1", |
|
|
|
|
|
"DeviceID": data['id'], |
|
|
|
|
|
"Function": data['function'], |
|
|
|
|
|
"Set": { |
|
|
|
|
|
"P_pos": data['position'], |
|
|
|
|
|
"P_vel": data['velocity'], |
|
|
|
|
|
"P_mode": data['mode'] |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
return payload |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def generate_claw_payload(data): |
|
|
|
|
|
""" |
|
|
|
|
|
生成控制爪子的消息负载 |
|
|
|
|
|
""" |
|
|
|
|
|
payload = { |
|
|
|
|
|
"MesType": "Request", |
|
|
|
|
|
"ClientId": "2", |
|
|
|
|
|
"DeviceID": data['id'], |
|
|
|
|
|
"Function": data['function'], |
|
|
|
|
|
"Set": { |
|
|
|
|
|
"GripPos": data['grip_pos'], |
|
|
|
|
|
"GripI": data['grip_current'], |
|
|
|
|
|
"GripVel": data['grip_velocity'] |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
return payload |