Code for DDS Feeder and containerization

dds2val/src/Dockerfile

@@ -0,0 +1,69 @@
+# /********************************************************************************
+# * Copyright (c) 2023 Contributors to the Eclipse Foundation
+# *
+# * See the NOTICE file(s) distributed with this work for additional
+# * information regarding copyright ownership.
+# *
+# * This program and the accompanying materials are made available under the
+# * terms of the Apache License 2.0 which is available at
+# * http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * SPDX-License-Identifier: Apache-2.0
+# ********************************************************************************/
+
+
+# Build stage, to create a Virtual Environent
+ARG TARGETPLATFORM
+ARG BUILDPLATFORM
+
+FROM --platform=$TARGETPLATFORM python:3.9-slim-bullseye as builder
+
+RUN echo "-- Running on $BUILDPLATFORM, building for $TARGETPLATFORM"
+
+RUN apt-get update -qqy && apt-get upgrade -qqy && apt-get install -qqy binutils g++ git
+
+COPY . /
+
+RUN python3 -m venv /opt/venv
+
+ENV PATH="/opt/venv/bin:$PATH"
+
+RUN /opt/venv/bin/python3 -m pip install --upgrade pip \
+    && pip3 install --no-cache-dir -r requirements.txt
+
+RUN pip3 install wheel scons && pip3 install pyinstaller patchelf==0.17.0.0 staticx
+WORKDIR /idls
+RUN ./generate_py_dataclass.sh
+
+WORKDIR /
+ENV PATH="/idls:$PATH"
+# The cyclonedds module uses a dynamic library which is not automatically discovered (--add-binary)
+RUN pyinstaller --clean --paths=. --add-binary="/opt/venv/lib/python3.9/site-packages/cyclonedds.libs/libddsc-62ba48e6.so.0.11.0:cyclonedds.libs/" --add-data="mapping.yml:." -F -s ddsfeeder.py 
+
+WORKDIR /dist
+RUN staticx ddsfeeder ddsfeeder-exe
+
+# Runner stage, to copy in the virtual environment and the app
+FROM scratch
+
+LABEL org.opencontainers.image.source="https://github.com/eclipse/kuksa.val.feeders"
+
+LABEL org.opencontainers.image.licenses="APACHE-2.0"
+
+# needed as /dist/binary unpacks and runs from /tmp
+WORKDIR /tmp
+# optional volume mapping
+WORKDIR /conf
+
+WORKDIR /dist
+
+COPY --from=builder /dist/ddsfeeder-exe .
+
+ENV PATH="/dist:$PATH"
+
+# useful dumps about feeding values
+ENV LOG_LEVEL="info"
+
+ENV PYTHONUNBUFFERED=yes
+
+ENTRYPOINT ["./ddsfeeder-exe"]

dds2val/src/OverallSequence.puml

@@ -0,0 +1,23 @@
+@startuml
+
+title DDS on SDV
+box "Container-1"
+participant databroker
+end box
+databroker <-- ddsfeeder : grpc_connect
+alt on connection
+ddsfeeder --> databroker : register_datapoints
+ddsfeeder --> DDS_network : subscribe to topics
+box "Container-2"
+participant ddspublisher1
+end box
+box "Container-3"
+participant ddspublisher2
+end box
+ddspublisher1 --> DDS_network : publish dds message
+ddspublisher2 --> DDS_network : publish dds message
+alt on data reception
+ddsfeeder --> databroker : update_datapoint
+end note
+end
+@enduml

dds2val/src/README.md

@@ -0,0 +1,24 @@
+# DDS Feeder
+
+## How to build
+### local build
+
+1. ```python3 -m venv env && source env/bin/activate```
+2. ```pip install -r requirements.txt```
+3. ```./idls/generate_py_dataclass.sh```
+
+### build image (suggested)
+
+1. ```docker build -f Dockerfile --progress=plain --build-arg TARGETPLATFORM=linux/amd64 -t ddsfeeder:latest .```
+
+### KML replay
+
+1. ```python3 -m venv env && source env/bin/activate```
+2. ```pip install -r requirements-kml.txt```
+
+## How to run
+Choose from local build or contanerization via docker.
+These steps are necessary:
+1. Run an instance of databroker aka: ```docker run -it --rm --net=host ghcr.io/eclipse/kuksa.val/databroker:master```
+2. Start the KML replay with an active local python virtual environment: ```python3 dds_kmlreplay.py directions.kml```
+3. Start the DDS feeder with either: ```docker run --rm -it --net=host ddsfeeder:latest``` or with an active local python virtual environment: ```python3 ddsfeeder.py```

dds2val/src/databroker.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+
+#################################################################################
+# Copyright (c) 2022 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License 2.0 which is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+#################################################################################
+
+import logging
+from typing import Any, Optional, Union
+
+import grpc.aio
+import kuksa_client.grpc
+from kuksa_client.grpc import (
+    DataEntry,
+    Datapoint,
+    DataType,
+    EntryUpdate,
+    Field,
+    Metadata,
+    VSSClient,
+)
+
+log = logging.getLogger(__name__)
+
+
+class Provider:
+    def __init__(
+        self, vss_client: VSSClient, grpc_metadata: Optional[grpc.aio.Metadata] = None
+    ):
+        self._name_to_type: dict[str, dict] = {}
+        self._rpc_kwargs = {"metadata": grpc_metadata}
+        log.info("Using %s", self._rpc_kwargs)
+        self._vss_client = vss_client
+
+    def register(
+        self, name: str, data_type: Union[str, DataType], description: str
+    ) -> bool:
+        if isinstance(data_type, str):
+            data_type = getattr(DataType, data_type)
+        try:
+            log.debug(
+                "register(%s, data_type: %s, '%s')",
+                name,
+                data_type.name,
+                description,
+            )
+            metadata = self._vss_client.get_metadata((name,), **self._rpc_kwargs)
+            if len(metadata) == 1:
+                self._name_to_type[name] = metadata[name].data_type
+                log.info(
+                    "%s was already registered with type %s",
+                    name,
+                    metadata[name].data_type.name,
+                )
+                return True
+
+            return self._register(name, data_type, description)
+        except kuksa_client.grpc.VSSClientError:
+            log.debug("Failed to get metadata")
+            return False
+
+    def _register(self, name: str, data_type: DataType, description: str) -> bool:
+        try:
+            self._vss_client.set_metadata(
+                updates={name: Metadata(data_type=data_type, description=description)},
+                **self._rpc_kwargs,
+            )
+            # Store datapoint IDs
+            self._name_to_type[name] = data_type
+            log.info(
+                "%s was registered with type %s",
+                name,
+                data_type.name,
+            )
+            return True
+        except kuksa_client.grpc.VSSClientError:
+            log.warning("Failed to register datapoint %s", name, exc_info=True)
+            return False
+
+    def update_datapoint(self, name: str, value: Any) -> bool:
+        updates = (
+            EntryUpdate(
+                DataEntry(
+                    name,
+                    value=Datapoint(value=value),
+                    # Specifying data_type removes the need for the client to query data_type from the server before
+                    # issuing every set() call.
+                    metadata=Metadata(data_type=self._name_to_type[name]),
+                ),
+                (Field.VALUE,),
+            ),
+        )
+        try:
+            self._vss_client.set(updates=updates, **self._rpc_kwargs)
+        except kuksa_client.grpc.VSSClientError:
+            log.error("Failed to update data point %s", name)
+            return False
+        log.debug("%s => %s", name, value)
+        return True

dds2val/src/dds_kmlreplay.py

@@ -0,0 +1,105 @@
+#!/bin/env python3
+
+#################################################################################
+# Copyright (c) 2022 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License 2.0 which is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+#################################################################################
+
+import logging
+import sys
+import time
+
+import sensor_msgs
+import std_msgs
+from cyclonedds.domain import DomainParticipant
+from cyclonedds.pub import DataWriter
+from cyclonedds.topic import Topic
+from pykml import parser
+
+logger = logging.getLogger(__name__)
+
+
+def read_kml_file(kml_filename):
+    """Read kml and return array of coordinates."""
+    logger.info("Reading KML file %s", kml_filename)
+    with open(kml_filename, "r", encoding="utf8") as kml_file:
+        return (
+            parser.parse(kml_file)
+            .getroot()
+            .Document.Placemark.LineString.coordinates.text.split()
+        )
+
+
+class DDSPublisher:  # pylint: disable=too-few-public-methods
+    """
+    Sample DDS Publisher.
+
+    Sends out "NavStatFix" message with values read from KML file every 1 second
+    """
+
+    def __init__(self, data):
+        self.data = data
+        participant = DomainParticipant()
+        """
+        OMG specification supports only '-' as a separator in topic names
+        Currently cyclonedds does not support symbols '.' or '-' in the dds topic names
+        Hence '_' is used. https://github.com/eclipse-cyclonedds/cyclonedds/issues/1393
+        """
+        topic = Topic(participant, "Nav_Sat_Fix", sensor_msgs.msg.NavSatFix)
+
+        self.writer = DataWriter(participant, topic)
+
+    def _publish(self, message):
+        self.writer.write(message)
+
+    def run(self):
+        ctr = 0
+        while 1:
+            sec = int(time.time())
+            nanosec = 2  # time.time_ns()
+            print(f"sec : {sec}")
+            print(f"nanosec : {nanosec}")
+            header = std_msgs.msg.Header(
+                stamp=std_msgs.msg.Time(sec=sec, nanosec=nanosec),
+                frame_id=str(ctr),
+            )
+            latitude, longitutde, altitude = self.data[ctr].split(",")
+            status = sensor_msgs.msg.NavSatStatus(
+                status=sensor_msgs.msg.NavSatStatus__STATUS_FIX,
+                service=sensor_msgs.msg.NavSatStatus__SERVICE_GPS,
+            )
+            message = sensor_msgs.msg.NavSatFix(
+                header=header,
+                status=status,
+                latitude=float(latitude),
+                longitude=float(longitutde),
+                altitude=float(altitude),
+                position_covariance=[1.1, 2.2, 3.3, 4.4, 5.5, 6.6, 7.7, 8.8, 9.9],
+                # pylint: disable=C0301
+                position_covariance_type=sensor_msgs.msg.NavSatFix__COVARIANCE_TYPE_KNOWN,  # noqa: E501
+            )
+            self._publish(message)
+            print("Published Message", message)
+            time.sleep(1)
+            ctr = (ctr + 1) % len(self.data)
+
+
+def main():
+    """Start the KML replay."""
+    logger.info("Starting KML Replay...")
+    kml_file = sys.argv[1:][0]
+    data = read_kml_file(kml_file)
+    pub_obj = DDSPublisher(data)
+    pub_obj.run()
+
+
+if __name__ == "__main__":
+    main()