Hands-On: Cloud-Connected Vehicle App

#!/usr/bin/env python3
"""Realistic vehicle telemetry simulator for cloud connectivity lab."""
import asyncio
import json
import math
import random
import time
import paho.mqtt.client as mqtt

BROKER = "localhost"
PORT   = 1883
VIN    = "SIM_VEHICLE_001"

class VehicleSimulator:
    def __init__(self, vin: str):
        self.vin = vin
        self.speed   = 0.0
        self.soc     = 85.0
        self.lat     = 48.137154  # Munich
        self.lon     = 11.576124
        self.odometer = 12543.0
        self.dtcs    = []
        self.client  = mqtt.Client()
        self.client.connect(BROKER, PORT)
        self.client.loop_start()

    def simulate_step(self):
        """Update vehicle state for one 1-second step."""
        # Simple city driving profile: accelerate/cruise/brake
        t = time.time() % 120  # 2-minute drive cycle
        if t < 30:
            self.speed = min(50, self.speed + 2.0)  # accelerate
        elif t < 80:
            self.speed += random.gauss(0, 1)  # cruise with noise
            self.speed = max(30, min(60, self.speed))
        else:
            self.speed = max(0, self.speed - 3.0)  # brake

        # SOC decreases with speed
        self.soc -= self.speed * 0.0001
        self.soc = max(10, self.soc)

        # Move location
        self.lat += self.speed * 0.00001
        self.odometer += self.speed / 3600

        # Occasionally inject a DTC
        if random.random() < 0.001:
            self.dtcs.append("P0300")  # random misfire

    def publish_telemetry(self):
        payload = {
            "vehicleId": self.vin,
            "timestamp": int(time.time() * 1000),
            "signals": {
                "Vehicle.Speed": round(self.speed, 1),
                "Vehicle.Powertrain.TractionBattery.StateOfCharge.Current": round(self.soc, 1),
                "Vehicle.CurrentLocation.Latitude": self.lat,
                "Vehicle.CurrentLocation.Longitude": self.lon,
                "Vehicle.TraveledDistance": round(self.odometer, 2)
            }
        }
        self.client.publish(f"vehicles/{self.vin}/telemetry",
                            json.dumps(payload), qos=0)

if __name__ == "__main__":
    sim = VehicleSimulator(VIN)
    print(f"Simulating vehicle {VIN}...")
    while True:
        sim.simulate_step()
        sim.publish_telemetry()
        time.sleep(1)

"""Remote diagnostics REST API (FastAPI)."""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import json
import paho.mqtt.client as mqtt
from typing import List

app = FastAPI(title="SDV Remote Diagnostics API")

# In-memory store for demo (use database in production)
vehicle_dtcs: dict = {}  # VIN -> [DTC codes]
vehicle_signals: dict = {}  # VIN -> {signal: value}

# MQTT subscriber to receive vehicle telemetry
def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    vid = data.get("vehicleId")
    if vid:
        vehicle_signals[vid] = data.get("signals", {})

mqtt_client = mqtt.Client()
mqtt_client.on_message = on_message
mqtt_client.connect("localhost", 1883)
mqtt_client.subscribe("vehicles/+/telemetry")
mqtt_client.loop_start()

@app.get("/vehicles/{vin}/signals")
async def get_vehicle_signals(vin: str):
    if vin not in vehicle_signals:
        raise HTTPException(404, f"Vehicle {vin} not found or offline")
    return {"vin": vin, "signals": vehicle_signals[vin]}

@app.post("/vehicles/{vin}/ota-trigger")
async def trigger_ota(vin: str, version: str):
    """Send OTA update command to vehicle via MQTT."""
    command = {"action": "ota_update", "target_version": version}
    mqtt_client.publish(f"vehicles/{vin}/commands",
                        json.dumps(command), qos=1)
    return {"status": "command_sent", "vin": vin, "version": version}