### Run Vehicle App SDK Examples

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/README.md

This script executes various examples provided with the Vehicle App Python SDK. Navigate to the 'examples' directory and run the script with the desired example folder name. It defaults to using native middleware.

```bash
cd examples
./run-app.sh -a <example-folder-name>
```

--------------------------------

### Build Docker Image for Seat-Adjuster Example

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/examples/seat-adjuster/README.md

Builds a Docker image for the Seat-Adjuster example application. This command assumes the Dockerfile is located in the 'app' directory and the build context is the repository's root folder. It also includes build arguments for proxy configuration.

```bash
docker build -f app/Dockerfile .

```

```bash
docker build -f app/Dockerfile . --build-arg http_proxy --build-arg HTTP_PROXY --build-arg https_proxy --build-arg HTTPS_PROXY --build-arg no_proxy --build-arg NO_PROXY

```

--------------------------------

### Install Python Dependencies

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/CONTRIBUTING.md

Installs project dependencies using pip from a requirements file. This is a common step for setting up a Python project for development.

```bash
pip3 install -r requirements.txt
```

--------------------------------

### Install Vehicle App Python SDK using pip

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/README.md

This command installs the Vehicle App Python SDK directly from its GitHub repository using pip. Ensure you replace '<version>' with the desired tag or branch. Python 3.10 or later is required.

```bash
pip install git+https://github.com/eclipse-velocitas/vehicle-app-python-sdk.git@<version>
```

--------------------------------

### Install and Run Pre-commit Hooks

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/CONTRIBUTING.md

Installs pre-commit hooks to automatically check code style and formatting before committing. It also runs all pre-commit checks on the entire codebase.

```bash
pre-commit install
pre-commit run --all-files
```

--------------------------------

### VehicleApp Base Class Example (Python)

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Demonstrates the basic structure of a vehicle application using the VehicleApp base class. It shows how to initialize the app, subscribe to data point changes (like vehicle speed), and define a callback function to handle these changes. This example requires the 'velocitas_sdk' and a generated 'sdv_model'.

```python
import asyncio
import json
import logging
import signal

from velocitas_sdk.vehicle_app import VehicleApp
from velocitas_sdk.vdb.subscriptions import DataPointReply
from sdv_model import vehicle  # Generated vehicle model

logger = logging.getLogger(__name__)

class MyVehicleApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    async def on_start(self):
        """Called when the app starts - set up subscriptions here"""
        logger.info("Vehicle App started")
        # Subscribe to data point changes
        await self.Vehicle.Speed.subscribe(self.on_speed_changed)

    async def on_speed_changed(self, data: DataPointReply):
        """Callback when vehicle speed changes"""
        speed = data.get(self.Vehicle.Speed).value
        logger.info(f"Current speed: {speed} km/h")

async def main():
    app = MyVehicleApp(vehicle)
    await app.run()

LOOP = asyncio.get_event_loop()
LOOP.add_signal_handler(signal.SIGTERM, LOOP.stop)
LOOP.run_until_complete(main())
LOOP.close()
```

--------------------------------

### Manage Subscriptions Dynamically - Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Illustrates how to dynamically manage subscriptions using `subscribe()` and `unsubscribe()`. This example shows how to create an initial subscription, and then update or replace it in response to an external event (like an MQTT message). The `rule` attribute stores the current subscription object.

```python
import json
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_topic
from velocitas_sdk.vdb.subscriptions import DataPointReply

class DynamicRuleApp(VehicleApp):
    def __init__(self, vehicle_client, initial_limit: float = 130.0):
        super().__init__()
        self.Vehicle = vehicle_client
        self.speed_limit = initial_limit
        self.rule = None

    async def on_start(self):
        """Create initial subscription with default speed limit"""
        condition = f"Vehicle.Speed > {self.speed_limit}"
        self.rule = await (
            self.Vehicle.Speed
            .join(self.Vehicle.ADAS.ABS.IsEngaged)
            .where(condition)
            .subscribe(self.on_speed_exceeded)
        )

    @subscribe_topic("speedlimitwarner/setLimit/request")
    async def on_change_limit_request(self, data: str) -> None:
        """Handle dynamic speed limit change via MQTT"""
        request = json.loads(data)
        self.speed_limit = request["speed"]

        # Unsubscribe from old rule
        await self.rule.unsubscribe()

        # Create new subscription with updated condition
        condition = f"Vehicle.Speed > {self.speed_limit}"
        self.rule = await (
            self.Vehicle.Speed
            .join(self.Vehicle.ADAS.ABS.IsEngaged)
            .where(condition)
            .subscribe(self.on_speed_exceeded)
        )

        print(f"Speed limit updated to: {self.speed_limit}")

    def on_speed_exceeded(self, data: DataPointReply):
        speed = data.get(self.Vehicle.Speed).value
        abs_engaged = data.get(self.Vehicle.ADAS.ABS.IsEngaged).value
        print(f"Warning: Speed {speed} exceeds limit {self.speed_limit}")

```

--------------------------------

### Setup Test Data with IntTestHelper in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Demonstrates how to use the IntTestHelper class to set various data point types (float, int32, bool, double, string, arrays) and register custom data points for integration testing. It connects to the Vehicle Data Broker and cleans up resources afterward.

```python
import asyncio
from velocitas_sdk.test.inttesthelper import IntTestHelper

async def setup_test_data():
    # Create helper - uses localhost:55555 by default or VDB_PORT env var
    helper = IntTestHelper(port=55555)

    try:
        # Set various data point types for testing
        await helper.set_float_datapoint("Vehicle.Speed", 65.5)
        await helper.set_int32_datapoint(
            "Vehicle.Cabin.Seat.Row1.Pos1.Position", 250
        )
        await helper.set_bool_datapoint("Vehicle.ADAS.ABS.IsEngaged", True)
        await helper.set_double_datapoint(
            "Vehicle.Cabin.AmbientAirTemperature", 22.5
        )
        await helper.set_string_datapoint(
            "Vehicle.VehicleIdentification.VIN", "WDB1234567890"
        )

        # Set array data points
        await helper.set_floatArray_datapoint(
            "Vehicle.SomeArraySignal", [1.1, 2.2, 3.3]
        )
        await helper.set_boolArray_datapoint(
            "Vehicle.SomeBoolArray", [True, False, True]
        )

        # Register a new custom data point
        from velocitas_sdk.proto.types_pb2 import DataType
        await helper.register_datapoint("Custom.DataPoint", DataType.FLOAT)
        await helper.set_float_datapoint("Custom.DataPoint", 42.0)

        print("Test data setup complete")
    finally:
        await helper.close()

asyncio.run(setup_test_data())
```

--------------------------------

### Install Python Package in Editable Mode

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/CONTRIBUTING.md

Installs the current Python package in editable mode, allowing changes to the source code to be reflected immediately without reinstallation. This is useful during development.

```bash
pip3 install -e .
```

--------------------------------

### Update Dependencies with pip-compile

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/README.md

This snippet demonstrates how to update Python dependencies using `pip-compile`. It involves updating the main requirements, extra development dependencies, and specific example requirements. Ensure you navigate to the correct directory before running the commands.

```bash
pip-compile -U --extra=dev
cd examples/seat-adjuster
pip-compile -U
cd ../../.project-creation/.skeleton/
pip-compile -U
```

--------------------------------

### Subscribe Data Points Decorator Examples (Python)

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Illustrates the use of the @subscribe_data_points decorator for declarative subscriptions to vehicle data points. It shows how to subscribe to a single data point with a condition (speed limit) and multiple data points simultaneously. This requires the 'velocitas_sdk' and a generated 'sdv_model'.

```python
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_data_points
from velocitas_sdk.vdb.subscriptions import DataPointReply

class SpeedWarnerApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    @subscribe_data_points("Vehicle.Speed", "Vehicle.Speed > 130.0")
    def on_speed_limit_exceeded(self, data: DataPointReply):
        """Called when vehicle speed exceeds 130 km/h"""
        speed = data.get(self.Vehicle.Speed).value
        print(f"Warning: Speed limit exceeded! Current speed: {speed}")

    @subscribe_data_points(
        "Vehicle.Cabin.DogModeTemperature, Vehicle.Cabin.DogMode,
        Vehicle.Powertrain.Battery.StateOfCharge.Current"
    )
    async def on_dog_mode_data_changed(self, data: DataPointReply):
        """Called when any of the subscribed data points change"""
        temperature = data.get(self.Vehicle.Cabin.DogModeTemperature).value
        dog_mode = data.get(self.Vehicle.Cabin.DogMode).value
        soc = data.get(self.Vehicle.Powertrain.Battery.StateOfCharge.Current).value
        print(f"DogMode: {dog_mode}, Temp: {temperature}, Battery: {soc}%")
```

--------------------------------

### Direct Vehicle Data Broker Access with Python SDK

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Demonstrates direct access to the Vehicle Data Broker gRPC API using the VehicleDataBrokerClient. It shows how to retrieve multiple data points, get metadata for data points, and subscribe to data point updates. This client is intended for advanced use cases requiring low-level interaction with the VDB.

```python
import asyncio
from velocitas_sdk.vdb.client import VehicleDataBrokerClient

async def direct_vdb_access():
    client = VehicleDataBrokerClient()

    try:
        # Get multiple data points in one call
        response = await client.GetDatapoints([
            "Vehicle.Speed",
            "Vehicle.Cabin.Seat.Row1.Pos1.Position",
            "Vehicle.Powertrain.Battery.StateOfCharge.Current"
        ])

        for path, datapoint in response.datapoints.items():
            print(f"{path}: {datapoint}")

        # Get metadata for data points
        metadata = await client.GetMetadata([
            "Vehicle.Speed",
            "Vehicle.Cabin.Seat.Row1.Pos1.Position"
        ])

        for item in metadata.list:
            print(f"Name: {item.name}, ID: {item.id}, Type: {item.data_type}")

        # Subscribe to data points (returns async iterator)
        async for reply in client.Subscribe("SELECT Vehicle.Speed WHERE Vehicle.Speed > 50"):
            for path, datapoint in reply.fields.items():
                print(f"Subscription update - {path}: {datapoint.float_value}")
            break  # Exit after first update for demo

    finally:
        await client.close()

asyncio.run(direct_vdb_access())
```

--------------------------------

### Subscribe to MQTT Topic with @subscribe_topic Decorator in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

The `@subscribe_topic` decorator in Python allows methods to subscribe to specific MQTT topics. The decorated method receives the message payload as a string and can process it. This example demonstrates handling a seat position request, checking vehicle speed, and publishing a response.

```python
import json
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_topic

class SeatAdjusterApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    @subscribe_topic("seatadjuster/setPosition/request")
    async def on_set_position_request(self, data_str: str) -> None:
        """Handle seat position request from MQTT topic"""
        data = json.loads(data_str)
        position = data["position"]
        request_id = data["requestId"]

        # Check if vehicle is stationary
        vehicle_speed = (await self.Vehicle.Speed.get()).value

        response = {"requestId": request_id, "result": {}}

        if vehicle_speed == 0:
            try:
                await self.Vehicle.Cabin.Seat.Row1.DriverSide.Position.set(position)
                response["result"] = {
                    "status": 0,
                    "message": f"Seat position set to: {position}"
                }
            except Exception as error:
                response["result"] = {
                    "status": 1,
                    "message": f"Failed to set position: {error}"
                }
        else:
            response["result"] = {
                "status": 1,
                "message": f"Cannot move seat while driving at {vehicle_speed} km/h"
            }

        await self.publish_event(
            "seatadjuster/setPosition/response",
            json.dumps(response)
        )
```

--------------------------------

### Read Vehicle Data with DataPoint.get() in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

The `DataPoint.get()` method in Python retrieves the current value of a vehicle data point. It returns a `TypedDataPointResult` object containing the value, path, and timestamp. This example shows how to read vehicle speed, seat position, battery state, and cabin temperature.

```python
from velocitas_sdk.vehicle_app import VehicleApp
from sdv_model import vehicle

class DataReaderApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    async def on_start(self):
        # Read current vehicle speed
        speed_result = await self.Vehicle.Speed.get()
        print(f"Speed: {speed_result.value} km/h")
        print(f"Path: {speed_result.path}")
        print(f"Timestamp: {speed_result.timestamp}")

        # Read seat position using Row/Pos accessors
        seat_pos = await self.Vehicle.Cabin.Seat.Row(1).Pos(1).Position.get()
        print(f"Seat position: {seat_pos.value}")

        # Read battery state of charge
        soc = await self.Vehicle.Powertrain.Battery.StateOfCharge.Current.get()
        print(f"Battery: {soc.value}%")

        # Read cabin temperature
        temp = await self.Vehicle.Cabin.AmbientAirTemperature.get()
        print(f"Cabin temperature: {temp.value}C")
```

--------------------------------

### Write Actuator Values with DataPoint.set() in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

The `DataPoint.set()` method in Python writes a value to an actuator data point. Only actuator data points can be set; attempting to set a sensor will raise a `TypeError`. This example demonstrates setting a seat position and shows an attempt to set a sensor value, which results in an error.

```python
import json
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_topic

class ActuatorControlApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    @subscribe_topic("vehicleapp/setValue/request")
    async def on_set_value_request(self, data_str: str) -> None:
        data = json.loads(data_str)
        position = data["position"]

        try:
            # Set seat position (actuator - allowed)
            await self.Vehicle.Cabin.Seat.Row1.Pos1.Position.set(position)
            await self.publish_event(
                "vehicleapp/setValue/response",
                json.dumps({"status": "success", "position": position})
            )
        except TypeError as error:
            # Raised if trying to set a non-actuator data point
            await self.publish_event(
                "vehicleapp/setValue/response",
                json.dumps({"status": "error", "message": str(error)})
            )

    @subscribe_topic("vehicleapp/setSensor/request")
    async def on_set_sensor_request(self, data_str: str) -> None:
        """Example of invalid set operation on a sensor"""
        data = json.loads(data_str)
        value = data["sensor"]

        try:
            # This will fail - IsBelted is a sensor, not an actuator
            await self.Vehicle.Cabin.Seat.Row1.Pos1.IsBelted.set(value)
        except TypeError as error:
            # Error: "set target value for non-actuator ... is not allowed!"
            print(f"Cannot set sensor value: {error}")
```

--------------------------------

### Configure Service Endpoints with NativeServiceLocator in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Illustrates how to configure service endpoints using environment variables with the NativeServiceLocator. It demonstrates setting custom addresses for services like MQTT, Vehicle Data Broker, HVAC, and Seat services.

```python
import os
from velocitas_sdk.native.locator import NativeServiceLocator

# Default service addresses (used if env vars not set):
# - mqtt: mqtt://localhost:1883
# - vehicledatabroker: grpc://localhost:55555

# Configure via environment variables:
os.environ["SDV_MQTT_ADDRESS"] = "mqtt://mqtt-broker.local:1883"
os.environ["SDV_VEHICLEDATABROKER_ADDRESS"] = "grpc://vdb.local:55555"

# Custom service addresses
os.environ["SDV_HVACSERVICE_ADDRESS"] = "grpc://hvac-service.local:50051"
os.environ["SDV_SEATSERVICE_ADDRESS"] = "grpc://seat-service.local:50052"

# The locator resolves addresses in order:
# 1. Environment variable: SDV_<SERVICE_NAME>_ADDRESS
# 2. Default addresses for known services
locator = NativeServiceLocator()
mqtt_addr = locator.get_service_location("mqtt")
vdb_addr = locator.get_service_location("vehicledatabroker")
hvac_addr = locator.get_service_location("hvacservice")

print(f"MQTT: {mqtt_addr}")
print(f"VDB: {vdb_addr}")
print(f"HVAC: {hvac_addr}")
```

--------------------------------

### Run Python Tests with Tox

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/CONTRIBUTING.md

Executes the unit and integration test suites for Python 3.8 using Tox. This ensures that changes do not break existing functionality.

```bash
tox -e py38
```

--------------------------------

### Configure Logging Utilities in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Shows how to configure logging using the Velocitas SDK, supporting both OpenTelemetry for distributed tracing and standard logging formats. It allows setting log levels via environment variables.

```python
import logging
from velocitas_sdk.util.log import (
    get_opentelemetry_log_factory,
    get_opentelemetry_log_format,
    get_default_log_format,
    get_default_date_format,
    get_log_level,
)

# OpenTelemetry logging setup (for distributed tracing)
logging.setLogRecordFactory(get_opentelemetry_log_factory())
logging.basicConfig(format=get_opentelemetry_log_format())
logging.getLogger().setLevel("DEBUG")
# Log format includes: trace_id, span_id, service_name

# Standard logging setup
logging.basicConfig(
    format=get_default_log_format(),  # "% (asctime)s - %(name)s - %(levelname)s - %(message)s"
    datefmt=get_default_date_format()  # "%m/%d/%Y %I:%M:%S %p"
)

# Get log level from LOGLEVEL environment variable
# Defaults to ERROR, accepts: ERROR, DEBUG, INFO, WARNING, CRITICAL
log_level = get_log_level()
logging.getLogger().setLevel(log_level)

logger = logging.getLogger(__name__)
logger.info("Application started")
```

--------------------------------

### Publish MQTT Events with publish_event() in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Illustrates the use of the `publish_event()` method to send messages to MQTT topics, facilitating communication with external services or UI applications. This method is crucial for broadcasting vehicle data or status updates.

```python
import json
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_data_points
from velocitas_sdk.vdb.subscriptions import DataPointReply

class TelemetryPublisherApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    @subscribe_data_points(
        "Vehicle.Cabin.AmbientAirTemperature, Vehicle.Powertrain.Battery.StateOfCharge.Current"
    )
    async def on_telemetry_update(self, data: DataPointReply):
        temperature = data.get(self.Vehicle.Cabin.AmbientAirTemperature).value
        soc = data.get(self.Vehicle.Powertrain.Battery.StateOfCharge.Current).value

        # Publish temperature data
        temp_data = {"temperature": temperature}
        await self.publish_event(
            "dogmode/ambientAirTemperature",
            json.dumps(temp_data)
        )

        # Publish battery state if low
        if soc < 20:
            await self.publish_event(
                "dogmode/stateOfCharge",
                json.dumps({"soc": soc, "warning": "Low battery"})
            )

        # Publish combined display data
        display_data = {"Temperature": temperature, "StateOfCharge": soc}
        await self.publish_event("dogmode/display", json.dumps(display_data))
```

--------------------------------

### Subscribe to Data Points with Dynamic Conditions - Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Demonstrates how to subscribe to vehicle data points using dynamic queries. This includes simple subscriptions, subscriptions with WHERE clauses for filtering, and joining multiple data points to create complex conditions. The `subscribe()` method takes a callback function to process incoming data.

```python
from velocitas_sdk.vehicle_app import VehicleApp
from velocitas_sdk.vdb.subscriptions import DataPointReply

class DynamicSubscriptionApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client
        self.speed_limit = 130.0

    async def on_start(self):
        # Simple subscription to a single data point
        await self.Vehicle.Speed.subscribe(self.on_speed_update)

        # Subscription with condition (WHERE clause)
        self.speed_rule = await self.Vehicle.Speed.where(
            "Vehicle.Cabin.Seat.Row1.Pos1.Position > 100"
        ).subscribe(self.on_position_exceeded)

        # Join multiple data points with condition
        await (
            self.Vehicle.Cabin.Seat.Row(1).Pos(1).Position
            .join(self.Vehicle.Speed)
            .where("Vehicle.Cabin.Seat.Row1.Pos1.Position < 100")
            .subscribe(self.on_seat_and_speed_update)
        )

        # Join multiple data points without condition
        await (
            self.Vehicle.Speed
            .join(self.Vehicle.ADAS.ABS.IsEngaged)
            .where(f"Vehicle.Speed > {self.speed_limit}")
            .subscribe(self.on_speed_limit_with_abs)
        )

    async def on_speed_update(self, data: DataPointReply):
        speed = data.get(self.Vehicle.Speed).value
        print(f"Speed: {speed}")

    async def on_position_exceeded(self, data: DataPointReply):
        speed = data.get(self.Vehicle.Speed).value
        print(f"Position exceeded 100 at speed: {speed}")

    async def on_seat_and_speed_update(self, data: DataPointReply):
        speed = data.get(self.Vehicle.Speed).value
        position = data.get(self.Vehicle.Cabin.Seat.Row1.Pos1.Position).value
        print(f"Speed: {speed}, Seat position: {position}")

    async def on_speed_limit_with_abs(self, data: DataPointReply):
        speed = data.get(self.Vehicle.Speed).value
        abs_engaged = data.get(self.Vehicle.ADAS.ABS.IsEngaged).value
        print(f"Speed limit exceeded: {speed}, ABS engaged: {abs_engaged}")

```

--------------------------------

### Schedule Periodic Display of Vehicle Data using AsyncIOScheduler

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/examples/dog-mode/README.md

This Python code snippet demonstrates how to use AsyncIOScheduler to periodically execute a function within an AsyncIO event loop. It schedules the `display_values` function to run every `TEMP_REPORT_TIMEOUT` seconds, which is intended to publish the current temperature and state of charge. This requires the `apscheduler` library.

```python
    # This is the on_start() method of the VehicleApp
    async def on_start(self):
        logger.info("VehicleApp Started ...")
        try:
            await self.display_values()

            # Create an instance of the AsyncIOScheduler
            self.scheduler = AsyncIOScheduler()

            # Add the Job that needs to be scheduled periodically with a reference to the target function.
            self.scheduler.add_job(
                self.display_values, "interval", seconds=TEMP_REPORT_TIMEOUT
            )

            # Start the AsyncIOScheduler
            self.scheduler.start()
        except Exception as ex:
            logger.error(ex)

    # This is the target function of the AsyncIOScheduler
    async def display_values(self):
        logger.info("Publish Current Temperature and StateOfCharge")
        # Add your code here ...

```

--------------------------------

### Regenerate Protocol Buffer Descriptors

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/CONTRIBUTING.md

Executes a script to regenerate Python descriptors for protocol buffer files. This is necessary when changes are made to existing .proto files.

```bash
./generate-grpc-stubs.sh
```

--------------------------------

### Define Signal Subscriptions in JSON

Source: https://github.com/eclipse-velocitas/vehicle-app-python-sdk/blob/main/examples/performance-subscribe/README.md

This JSON structure defines the signals to subscribe to from the vehicle. Each signal is identified by its 'path'. Signals not found will be reported as warnings.

```json
{
  "signals": [
    {
      "path": "Vehicle.LowVoltageBattery.CurrentVoltage"
    },
    {
      "path": "Vehicle.LowVoltageBattery.CurrentCurrent"
    },
    {
      "path": "Vehicle.Speed"
    },
    ...
  ]
}
```

--------------------------------

### Set Multiple Data Points Atomically with set_many() in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Demonstrates how to use the `set_many()` method for atomic batch updates of multiple vehicle data points. This method utilizes a fluent builder pattern and ensures that all specified updates are applied together or none are. It's useful for maintaining data consistency across related vehicle signals.

```python
import json
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_topic
from velocitas_sdk.vdb.subscriptions import DataPointReply

class AtomicSetApp(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    async def on_start(self):
        await self.Vehicle.Cabin.Seat.Row1.Pos1.Position.subscribe(
            self.on_position_update
        )

    async def on_position_update(self, data: DataPointReply):
        position = data.get(self.Vehicle.Cabin.Seat.Row1.Pos1.Position).value
        print(f"Position updated: {position}")

    @subscribe_topic("vehicleapp/setValue/request")
    async def on_set_request(self, data_str: str) -> None:
        data = json.loads(data_str)
        position = data["position"]

        try:
            # Set multiple seat positions atomically
            await (
                self.Vehicle.set_many()
                .add(self.Vehicle.Cabin.Seat.Row1.Pos1.Position, position)
                .add(self.Vehicle.Cabin.Seat.Row1.Pos2.Position, position)
                .apply()
            )

            await self.publish_event(
                "vehicleapp/setValue/response",
                json.dumps({
                    "status": "success",
                    "message": f"Both seats set to position {position}"
                })
            )
        except TypeError as error:
            await self.publish_event(
                "vehicleapp/setValue/response",
                json.dumps({"status": "error", "message": str(error)})
            )
```

--------------------------------

### Extract Data Point Values with DataPointReply.get() in Python

Source: https://context7.com/eclipse-velocitas/vehicle-app-python-sdk/llms.txt

Shows how to retrieve specific data point values from subscription responses using the `DataPointReply.get()` method. This method provides type-safe access to the data and includes metadata like the data point path and timestamp.

```python
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_data_points
from velocitas_sdk.vdb.subscriptions import DataPointReply

class DataPointReplyExample(VehicleApp):
    def __init__(self, vehicle_client):
        super().__init__()
        self.Vehicle = vehicle_client

    @subscribe_data_points(
        "Vehicle.Speed, Vehicle.Cabin.Seat.Row1.Pos1.Position, Vehicle.ADAS.ABS.IsEngaged"
    )
    async def on_data_update(self, data: DataPointReply):
        # Get typed values from the reply
        speed_result = data.get(self.Vehicle.Speed)
        print(f"Speed: {speed_result.value} (path: {speed_result.path})")

        position_result = data.get(self.Vehicle.Cabin.Seat.Row1.Pos1.Position)
        print(f"Seat Position: {position_result.value}")

        abs_result = data.get(self.Vehicle.ADAS.ABS.IsEngaged)
        print(f"ABS Engaged: {abs_result.value}")

        # Access timestamp
        print(f"Timestamp: {speed_result.timestamp}")
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.