Hands-On: Container-Based Vehicle Application

# Vehicle speed averaging service
FROM python:3.11-slim

WORKDIR /app

# Install Kuksa.val Python SDK
RUN pip install kuksa-client==0.4.2

COPY speed_avg_service.py .

# Non-root user (security best practice for vehicle apps)
RUN useradd -r -s /bin/false vehicle_app
USER vehicle_app

CMD ["python", "speed_avg_service.py"]

#!/usr/bin/env python3
"""Vehicle speed averaging service using COVESA Kuksa.val."""

import asyncio
import statistics
from kuksa_client.grpc import VSSClient

KUKSA_ADDRESS = "127.0.0.1"
KUKSA_PORT    = 55555
SPEED_VSS     = "Vehicle.Speed"  # COVESA VSS signal path
WINDOW_SIZE   = 10  # 10-sample rolling average

async def main():
    speed_history = []

    async with VSSClient(KUKSA_ADDRESS, KUKSA_PORT) as client:
        print(f"Connected to Kuksa.val at {KUKSA_ADDRESS}:{KUKSA_PORT}")

        async for update in client.subscribe_current_values(
            [SPEED_VSS]
        ):
            speed_val = update[SPEED_VSS]
            if speed_val is not None:
                speed_kmh = speed_val.value
                speed_history.append(speed_kmh)

                # Rolling window average
                if len(speed_history) > WINDOW_SIZE:
                    speed_history.pop(0)

                avg_speed = statistics.mean(speed_history)
                print(f"Speed: {speed_kmh:.1f} km/h  "
                      f"Avg({WINDOW_SIZE}): {avg_speed:.1f} km/h")

                # Publish averaged speed back to Kuksa
                await client.set_current_values({
                    "Vehicle.OBD.Speed": speed_kmh  # OBD mirror
                })

if __name__ == "__main__":
    asyncio.run(main())