Vehicle-to-Cloud Architecture

"""Vehicle telemetry MQTT client (AWS IoT Core)."""
import json
import time
import ssl
import paho.mqtt.client as mqtt

VEHICLE_ID     = "VIN_1HGBH41JXMN109186"
AWS_IOT_ENDPOINT = "xxxxx.iot.eu-central-1.amazonaws.com"
AWS_IOT_PORT   = 8883
CERT_PATH      = "/etc/vehicle/certs"

# MQTT topics follow IoT FleetWise convention
TELEMETRY_TOPIC = f"vehicles/{VEHICLE_ID}/telemetry"
COMMAND_TOPIC   = f"vehicles/{VEHICLE_ID}/commands"
SHADOW_TOPIC    = f"$aws/things/{VEHICLE_ID}/shadow/update"

class VehicleMqttClient:
    def __init__(self):
        self.client = mqtt.Client(client_id=VEHICLE_ID, protocol=mqtt.MQTTv5)
        # Mutual TLS authentication (x.509 certificate per vehicle)
        self.client.tls_set(
            ca_certs   = f"{CERT_PATH}/AmazonRootCA1.pem",
            certfile   = f"{CERT_PATH}/vehicle_cert.pem",
            keyfile    = f"{CERT_PATH}/vehicle_private_key.pem",
            tls_version= ssl.PROTOCOL_TLS_CLIENT
        )
        self.client.on_connect    = self._on_connect
        self.client.on_message    = self._on_message

    def _on_connect(self, client, userdata, flags, rc, props):
        print(f"Connected to AWS IoT Core (rc={rc})")
        client.subscribe(COMMAND_TOPIC, qos=1)

    def _on_message(self, client, userdata, msg):
        command = json.loads(msg.payload)
        print(f"Command received: {command}")
        # Handle OTA trigger, remote diagnostics, config update

    def publish_telemetry(self, signals: dict):
        payload = {
            "vehicleId": VEHICLE_ID,
            "timestamp": int(time.time() * 1000),
            "signals": signals
        }
        self.client.publish(
            TELEMETRY_TOPIC,
            json.dumps(payload),
            qos=1
        )

    def connect_and_run(self):
        self.client.connect(AWS_IOT_ENDPOINT, AWS_IOT_PORT)
        self.client.loop_start()