能源电力解决方案概览

市场背景

能源电力是移远通信的重要战略市场，随着"双碳"目标推进和能源互联网建设，IoT 在能源领域的应用快速增长。

主要应用场景

场景	通信需求	推荐模组	市场规模
智能电表 AMI	低速低功耗	BC660K（NB-IoT）	数亿只
光伏储能监控	中速稳定	EC25（4G）	快速增长
充电桩管理	中速实时	EC200U（Cat.1）	千万级
输电线路监测	低速低功耗	BC660K（NB-IoT）	百万级
变电站自动化	高速可靠	EC25/RG520N	稳定增长
微电网控制	低时延	5G/4G	新兴市场

能源 IoT 特殊要求

安全合规

国内能源行业安全要求：
  - 等保 2.0 三级（关键信息基础设施）
  - 数据不出境（国内云平台）
  - 设备认证（国网/南网入网认证）
  - 通信加密（国密算法 SM2/SM4）

国密算法支持：
  移远模组支持国密 TLS（TLCP）
  AT+QSSLCFG="ciphersuite",0,"0xE011"  # 国密套件

电磁兼容

能源设备 EMC 要求：
  IEC 61000-4 系列（工业 EMC）
  GB/T 17626 系列（国内等效标准）
  
  特别注意：
  - 变电站强电磁环境
  - 开关操作产生的瞬态干扰
  - 雷击浪涌保护

充电桩管理方案

OCPP 协议

OCPP（Open Charge Point Protocol）是充电桩行业标准协议：

OCPP 1.6（主流）：
  传输层：WebSocket over HTTP
  消息格式：JSON
  
  核心功能：
  - BootNotification：充电桩上线注册
  - Heartbeat：心跳保活
  - StartTransaction：开始充电
  - StopTransaction：停止充电
  - MeterValues：电量数据上报
  - RemoteStartTransaction：远程启动充电
  - RemoteStopTransaction：远程停止充电

充电桩通信方案

import websocket
import json
import time
import uuid

class OCPPChargePoint:
    """OCPP 1.6 充电桩客户端"""
    
    def __init__(self, cp_id, server_url):
        self.cp_id = cp_id
        self.server_url = f"{server_url}/{cp_id}"
        self.ws = None
        self.pending_calls = {}
    
    def connect(self):
        self.ws = websocket.WebSocketApp(
            self.server_url,
            subprotocols=["ocpp1.6"],
            on_message=self.on_message,
            on_open=self.on_open
        )
        self.ws.run_forever()
    
    def on_open(self, ws):
        print(f"Connected to OCPP server: {self.server_url}")
        self.boot_notification()
    
    def on_message(self, ws, message):
        msg = json.loads(message)
        msg_type = msg[0]
        
        if msg_type == 3:  # CallResult
            call_id = msg[1]
            result = msg[2]
            if call_id in self.pending_calls:
                self.pending_calls[call_id](result)
                del self.pending_calls[call_id]
        
        elif msg_type == 2:  # Call（来自服务器的命令）
            call_id = msg[1]
            action = msg[2]
            payload = msg[3]
            self.handle_server_command(call_id, action, payload)
    
    def send_call(self, action, payload, callback=None):
        """发送 OCPP Call 消息"""
        call_id = str(uuid.uuid4())
        message = json.dumps([2, call_id, action, payload])
        
        if callback:
            self.pending_calls[call_id] = callback
        
        self.ws.send(message)
        return call_id
    
    def boot_notification(self):
        """充电桩上线注册"""
        payload = {
            "chargePointVendor": "Quectel",
            "chargePointModel": "QCP-7kW",
            "chargePointSerialNumber": self.cp_id,
            "firmwareVersion": "1.0.0"
        }
        
        def on_result(result):
            if result['status'] == 'Accepted':
                print("Boot notification accepted")
                self.start_heartbeat(result['interval'])
        
        self.send_call("BootNotification", payload, on_result)
    
    def start_heartbeat(self, interval):
        """定期发送心跳"""
        def heartbeat():
            while True:
                self.send_call("Heartbeat", {})
                time.sleep(interval)
        
        import threading
        threading.Thread(target=heartbeat, daemon=True).start()
    
    def meter_values(self, connector_id, transaction_id, energy_kwh):
        """上报电量数据"""
        payload = {
            "connectorId": connector_id,
            "transactionId": transaction_id,
            "meterValue": [{
                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                "sampledValue": [{
                    "value": str(energy_kwh * 1000),  # Wh
                    "measurand": "Energy.Active.Import.Register",
                    "unit": "Wh"
                }]
            }]
        }
        self.send_call("MeterValues", payload)
    
    def handle_server_command(self, call_id, action, payload):
        """处理服务器下发的命令"""
        if action == "RemoteStartTransaction":
            # 远程启动充电
            connector_id = payload.get('connectorId', 1)
            id_tag = payload['idTag']
            
            # 执行开始充电逻辑
            success = self.start_charging(connector_id, id_tag)
            
            response = json.dumps([3, call_id, {
                "status": "Accepted" if success else "Rejected"
            }])
            self.ws.send(response)
        
        elif action == "RemoteStopTransaction":
            transaction_id = payload['transactionId']
            success = self.stop_charging(transaction_id)
            
            response = json.dumps([3, call_id, {
                "status": "Accepted" if success else "Rejected"
            }])
            self.ws.send(response)

输电线路监测

线路状态监测

监测参数：
  导线温度（红外测温）
  导线弧垂（图像识别）
  微气象（风速/风向/温湿度/雨量）
  覆冰厚度（称重法）
  导线振动（加速度传感器）
  杆塔倾斜（倾角传感器）

通信方案：
  NB-IoT（BC660K）：低频数据，超低功耗
  4G（EC25）：视频图像传输
  
供电方案：
  感应取电（从导线感应电流取电）
  太阳能 + 蓄电池