在上一篇博客中，我们完成了MQTT服务器的搭建以及服务器和客户端SSL证书的签发，并成功用MQTT.fx软件作为客户端连接发送信息。在实际应用中，我们需要根据需求来开发自己的客户端应用程序，在本博文中，我将使用Python来开发一个模拟车载终端发送MQTT消息的应用。

在V2X领域，有很多标准化组织(ETSI/ISO/DSRC)制定了车辆V2X消息规范，例如我国的V2X标准中定义了以下5种消息规范：

1. BSM，即Basic Safety Message，基础安全消息，包括速度，转向，刹车，双闪，位置等等，多被用在V2V场景即变道预警，盲区预警，交叉路口碰撞预警等等；
2. RSI，即Road Side Information，路侧信息，用于事件的下方，路侧RSU集成，平台下发，多被用于V2I场景即道路施工，限速标志，超速预警，公交车道预警等等；
3. RSM，即Road Safety Message，路侧安全消息，也是V2I，主要对接路侧的边缘设备，用于事件的识别，比如，车辆发生事故，车辆异常，异物闯入等等；
4. SPAT，即Signal phase timing message，交通灯相位与时序消息，也是V2I，路侧RSU集成信号机，或者信号机通过UU方式传入到平台，用于车速引导，绿波推送场景等等；
5. MAP，地图消息，地图消息和SPAT消息一起使用，MAP消息可以描述一个路口，和该路口的红绿灯也存在对应关系；

欧洲的ETSI也定义了类似的消息规范：

1. CAM，Cooperative Awareness Message，合作感知信息，这是时间触发信息，提供车辆的速度、位置、方向灯以及交通信号系统如交通信号灯的状态，天气提醒等信息；
2. DENM，Decentralized Environmental Notification Message，分散环境通知信息，主要用于道路危险预警应用，是时间触发型信息，一旦通过车载设备检测到了安全隐患事件（例如前方车辆紧急刹车、道路施工警告等），车载ITS的相关应用就立即发射DENM信息，接收车辆可对比车辆自身位置与行车路线，判断事件对自车的关联性并预测可能的碰撞风险，以及提前通知驾驶员采取有效的措施，根据事件地点和类型，可能要求收到DENM信息的车辆向外转发；

本文将按照欧洲标准规范，采用CAM来上报车辆的位置信息（每秒2次的频率），采用DENM来上报车辆的紧急状态事件（以紧急刹车为例）。这2个消息协议的规范可以在ETSI的网站中找到，我选择的是CAM ETSI EN 302 637-2 V1.4.1 以及 DENM ETSI EN 302 637-3 V1.3.1这2个最新版本的规范，下载对应的规范的PDF文档后，可以在文档中找到规范的具体的ASN格式的定义，注意这2个规范中都引用ETSI TS 102 894-2 V1.3.1的定义，因此在生成CAM和DENM的ASN文件时，需要把TS 102894这份规范里面的字段定义也拷进去才能构成一个完整的ASN文件的定义。

有了CAM和DENM的ASN定义文件之后，我们就可以构建消息，采用UPER进行编码，再把二进制数据编码为BASE64字符串，然后发送MQTT消息。

以下是模拟两辆车在一段道路上行驶，这2辆车以每秒2条的频率发送CAM消息报告位置，其中一辆车会随机上报DENM消息。道路数据我是先用Google Map来进行GPS点的抓取，先在map上找一条道路，然后在其中以大致均匀的间隔取几个点，然后调用Google的roads API，在path参数里传入这几个点的坐标，设置interpolate为true，这样Google就会自动帮我进行点的插值。我选了一条大概1.6公里的道路，传入8个点，最终返回了54个点的坐标。把返回值进行解析后保存这些点的坐标在一个CSV文件中，作为车辆的模拟行驶路线。

代码如下：

import paho.mqtt.client as mqtt
import ssl
import base64
import asn1tools
import json
import time
import threading
import random
import datetimestart_timestamp = int(datetime.datetime(2004,1,1).timestamp()*1000)
def on_connect(client, userdata, flags, rc):print("Connected with result code "+str(rc))def on_message(client, userdata, msg):print(msg.topic+" "+str(msg.payload))client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.tls_set(ca_certs="cert/root.crt", certfile="cert/vehicle1.crt", keyfile="cert/vehicle1_key.pem", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2, ciphers=None)client.connect("broker", 8883, 60)# Define the message to publish
cam_asn = asn1tools.compile_files('CAM.asn', 'uper')
cam_json = {'header': {'protocolVersion': 1,'messageID': 2,'stationID': 123},'cam': {'generationDeltaTime': 123,'camParameters': {'basicContainer': {'stationType': 0,'referencePosition': {'latitude': 12345678,'longitude': 87654321,'positionConfidenceEllipse': {'semiMajorConfidence': 4095,'semiMinorConfidence': 4095,'semiMajorOrientation': 3601},'altitude': {'altitudeValue': 800001,'altitudeConfidence': 'unavailable'}}},'highFrequencyContainer': ('basicVehicleContainerHighFrequency', {'heading': {'headingValue': 3601,'headingConfidence': 127},'speed': {'speedValue': 30,'speedConfidence': 127},'driveDirection': 'unavailable','vehicleLength': {'vehicleLengthValue': 1023,'vehicleLengthConfidenceIndication': 'unavailable'},'vehicleWidth': 62,'longitudinalAcceleration': {'longitudinalAccelerationValue': 161,'longitudinalAccelerationConfidence': 102},'curvature': {'curvatureValue': 1023,'curvatureConfidence': 'unavailable'},'curvatureCalculationMode': 'unavailable','yawRate': {'yawRateValue': 32767,'yawRateConfidence': 'unavailable'}})}}
}
denm_asn = asn1tools.compile_files('DENM.asn', 'uper')
denm_json = {'header': {'protocolVersion': 1,'messageID': 2,'stationID': 123},'denm': {'management': {'actionID': {'originatingStationID': 123,'sequenceNumber': 1},'detectionTime': 1,'referenceTime': 1,'eventPosition': {'latitude': 123,'longitude': 321,'positionConfidenceEllipse': {'semiMajorConfidence': 4095,'semiMinorConfidence': 4095,'semiMajorOrientation': 3601},'altitude': {'altitudeValue': 800001,'altitudeConfidence': 'unavailable'}},'validityDuration': 20,'stationType': 0},'situation': {'informationQuality': 0,'eventType': {'causeCode': 99,'subCauseCode': 1 },'linkedCause': {'causeCode': 99,'subCauseCode': 1}}}
}# Read the simulation location list
roads_csv = open('roads.csv', 'r')
lines = roads_csv.readlines()
locations = []
for i in range(1, len(lines)):record = lines[i].split(',')lat = int(float(record[1])*10000000)lon = int(float(record[2])*10000000)locations.append((lat, lon))def publish_data(VIN, denm_flag=False):for i in range(1):rand_denm = random.randint(0, len(locations))for j in range(len(locations)):cam_json['cam']['camParameters']['basicContainer']['referencePosition']['latitude'] = locations[j][0]cam_json['cam']['camParameters']['basicContainer']['referencePosition']['longitude'] = locations[j][1]cam_encoded = cam_asn.encode('CAM', cam_json)cam_b64 = str(base64.b64encode(cam_encoded),'utf-8')client.publish("vehicle/cam/"+VIN,cam_b64)if denm_flag:if j == rand_denm:denm_json['denm']['management']['actionID']['sequenceNumber'] = idenm_json['denm']['management']['eventPosition']['latitude'] = locations[j][0]denm_json['denm']['management']['eventPosition']['longitude'] = locations[j][1]denm_json['denm']['situation']['eventType']['causeCode'] = 99denm_json['denm']['situation']['eventType']['subCauseCode'] = 1detect_timestamp = int(datetime.datetime.utcnow().timestamp()*1000) - start_timestampdenm_json['denm']['management']['detectionTime'] = detect_timestampdenm_json['denm']['management']['referenceTime'] = detect_timestampdenm_encoded = denm_asn.encode('DENM', denm_json)denm_b64 = str(base64.b64encode(denm_encoded),'utf-8')client.publish("vehicle/denm/"+VIN,denm_b64)time.sleep(0.5)time.sleep(10)class myThread (threading.Thread):def __init__(self, VIN, denm_flag):threading.Thread.__init__(self)self.VIN = VINself.denm_flag = denm_flagdef run(self):publish_data(self.VIN, self.denm_flag)thread1 = myThread("Vehicle1", True)
thread2 = myThread("Vehicle2", False)thread1.start()
time.sleep(1)
thread2.start()
thread1.join()
thread2.join()client.disconnect()

服务器端订阅CAM和DENM主题，并打印接收到的信息，代码如下：

import paho.mqtt.client as mqtt
import ssl
import base64
import asn1tools
import json
import time
import threading
import random
import datetimestart_timestamp = int(datetime.datetime(2004,1,1).timestamp()*1000)
def on_connect(client, userdata, flags, rc):print("Connected with result code "+str(rc))def on_message(client, userdata, msg):print(msg.topic+" "+str(msg.payload))client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.tls_set(ca_certs="cert/root.crt", certfile="cert/v2xapp.crt", keyfile="cert/v2xapp_key.pem", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2, ciphers=None)client.connect("broker", 8883, 60)# Define the message to publish
cam_asn = asn1tools.compile_files('CAM.asn', 'uper')
cam_json = {'header': {'protocolVersion': 1,'messageID': 2,'stationID': 123},'cam': {'generationDeltaTime': 123,'camParameters': {'basicContainer': {'stationType': 0,'referencePosition': {'latitude': 12345678,'longitude': 87654321,'positionConfidenceEllipse': {'semiMajorConfidence': 4095,'semiMinorConfidence': 4095,'semiMajorOrientation': 3601},'altitude': {'altitudeValue': 800001,'altitudeConfidence': 'unavailable'}}},'highFrequencyContainer': ('basicVehicleContainerHighFrequency', {'heading': {'headingValue': 3601,'headingConfidence': 127},'speed': {'speedValue': 30,'speedConfidence': 127},'driveDirection': 'unavailable','vehicleLength': {'vehicleLengthValue': 1023,'vehicleLengthConfidenceIndication': 'unavailable'},'vehicleWidth': 62,'longitudinalAcceleration': {'longitudinalAccelerationValue': 161,'longitudinalAccelerationConfidence': 102},'curvature': {'curvatureValue': 1023,'curvatureConfidence': 'unavailable'},'curvatureCalculationMode': 'unavailable','yawRate': {'yawRateValue': 32767,'yawRateConfidence': 'unavailable'}})}}
}
denm_asn = asn1tools.compile_files('DENM.asn', 'uper')
denm_json = {'header': {'protocolVersion': 1,'messageID': 2,'stationID': 123},'denm': {'management': {'actionID': {'originatingStationID': 123,'sequenceNumber': 1},'detectionTime': 1,'referenceTime': 1,'eventPosition': {'latitude': 123,'longitude': 321,'positionConfidenceEllipse': {'semiMajorConfidence': 4095,'semiMinorConfidence': 4095,'semiMajorOrientation': 3601},'altitude': {'altitudeValue': 800001,'altitudeConfidence': 'unavailable'}},'validityDuration': 20,'stationType': 0},'situation': {'informationQuality': 0,'eventType': {'causeCode': 99,'subCauseCode': 1 },'linkedCause': {'causeCode': 99,'subCauseCode': 1}}}
}client.subscribe('vehicle/#', 0)
client.loop_start()start = time.time()
while True:time.sleep(1)end = time.time()if int(end - start) > 30:client.loop_stop()break
client.disconnect()