注意：
实际部署时需替换为充电桩的硬件API。
---
模拟代码（仅供参考）
```python
from flask import Flask, request, jsonify
import time
import threading
app = Flask(__name__)
# 模拟充电桩状态数据库
charging_piles = {
"CP001": {"status": "charging", "power": 7.2, "session_id": "S123"},
"CP002": {"status": "idle", "power": 0, "session_id": None}
}
# 模拟硬件通信（实际需替换为Modbus/OCPP协议）
def hardware_power_off(pile_id):
"""模拟向充电桩发送TCP断电指令"""
print(f"[硬件] 向 {pile_id} 发送断电信号...")
time.sleep(1) # 模拟网络延迟
# 模拟硬件成功响应
if pile_id in charging_piles:
return {"code": 0, "msg": "断路器已跳闸"}
return {"code": -1, "msg": "设备离线"}
# 一键断电核心接口
@app.route('/api/power_off/<pile_id>', methods=['POST'])
def remote_power_off(pile_id):
# 1. 身份验证（生产环境需JWT/OAuth2）
token = request.headers.get('Authorization')
if token != "Bearer admin_token":
return jsonify({"code": 401, "msg": "未授权"}), 401
# 2. 检查桩是否存在
pile = charging_piles.get(pile_id)
if not pile:
return jsonify({"code": 404, "msg": "充电桩不存在"}), 404
# 3. 安全检查（防止误操作）
if pile["status"] == "idle":
return jsonify({"code": 400, "msg": "设备已空闲，无需断电"}), 400
# 4. 调用硬件接口断电
result = hardware_power_off(pile_id)
if result["code"] == 0:
# 5. 更新数据库状态
pile["status"] = "offline"
pile["power"] = 0
# 记录操作日志（生产需写入数据库）
print(f"[日志] {time.ctime()} 管理员远程断开了 {pile_id}")
return jsonify({"code": 200, "msg": "断电成功", "data": pile})
else:
return jsonify({"code": 500, "msg": result["msg"]}), 500
# 模拟前端调用（演示用）
if __name__ == '__main__':
print("模拟服务器启动，请用POST请求测试：")
print("curl -X POST -H 'Authorization: Bearer admin_token' http://127.0.0.1:5000/api/power_off/CP001")
app.run(debug=True, host='0.0.0.0', port=5000)
```
---
真正项目中 4 个关键点
模块 关键要求
1物理安全 必须使用工业级继电器，支持急停硬件回路，软件故障时硬件仍可切断。
2通信协议 实际使用 OCPP 1.6/2.0（国际标准）或 Modbus TCP，需实现心跳保活和重连机制。
3权限控制 必须绑定操作员ID + 双因素认证，所有操作留痕（不可篡改日志）。
4异常处理 需处理断电失败回滚、断网重试、状态不一致时的人工确认流程。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2602_95242675/article/details/162794928
