Initial commit

Author: pmai

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "open-simulation-interface"]
+	path = open-simulation-interface
+	url = https://github.com/OpenSimulationInterface/open-simulation-interface.git

LICENSE

@@ -0,0 +1,13 @@
+Open Simulation Interface (OSI) Python Bindings
+===============================================
+
+This python package provides python bindings and utility modules for the [ASAM Open Simulation Interface](https://github.com/OpenSimulationInterface/open-simulation-interface).
+
+For more information on OSI see the [official documentation](https://opensimulationinterface.github.io/osi-antora-generator/asamosi/latest/specification/index.html) or the [class list](https://opensimulationinterface.github.io/osi-antora-generator/asamosi/latest/gen/annotated.html) for defined protobuf messages.
+
+## Usage
+For usage examples, please refer to the official documentation:
+- [Trace file generation with python](https://opensimulationinterface.github.io/osi-antora-generator/asamosi/latest/interface/architecture/trace_file_example.html)
+
+## Installation
+

README.md

@@ -0,0 +1,94 @@
+import os
+import subprocess
+import sys
+import re
+import glob
+import pathlib
+from typing import Any
+
+from poetry.core.masonry.api import get_requires_for_build_wheel, get_requires_for_build_sdist, prepare_metadata_for_build_wheel, build_wheel as build_wheel_orig, build_sdist as build_sdist_orig, build_editable, get_requires_for_build_editable, prepare_metadata_for_build_editable
+
+import poetry_dynamic_versioning.patch as patch
+
+from protoc import PROTOC_EXE
+
+
+# Activate Versioning Plugin
+patch.activate()
+
+
+def _generate_python_files(package_name):
+    package_path = pathlib.Path(os.getcwd()) / package_name
+    try:
+        os.mkdir(package_path)
+    except Exception:
+        pass
+
+    # configure the version number
+    VERSION_MAJOR = None
+    VERSION_MINOR = None
+    VERSION_PATCH = None
+    VERSION_SUFFIX = None
+    with open(pathlib.Path("open-simulation-interface")/"VERSION", "rt") as versionin:
+        for line in versionin:
+            if line.startswith("VERSION_MAJOR"):
+                VERSION_MAJOR = int(line.split("=")[1].strip())
+            if line.startswith("VERSION_MINOR"):
+                VERSION_MINOR = int(line.split("=")[1].strip())
+            if line.startswith("VERSION_PATCH"):
+                VERSION_PATCH = int(line.split("=")[1].strip())
+            if line.startswith("VERSION_SUFFIX"):
+                VERSION_SUFFIX = line.split("=")[1].strip()
+
+    # Generate osi_version.proto
+    with open("open-simulation-interface/osi_version.proto.in", "rt") as fin:
+        with open("open-simulation-interface/osi_version.proto", "wt") as fout:
+            for line in fin:
+                lineConfigured = line.replace("@VERSION_MAJOR@", str(VERSION_MAJOR))
+                lineConfigured = lineConfigured.replace(
+                    "@VERSION_MINOR@", str(VERSION_MINOR)
+                )
+                lineConfigured = lineConfigured.replace(
+                    "@VERSION_PATCH@", str(VERSION_PATCH)
+                )
+                fout.write(lineConfigured)
+
+    # Copy and adjust imports
+    pattern = re.compile('^import "osi_')
+    for source in glob.glob("open-simulation-interface/*.proto"):
+        with open(source) as src_file:
+            with open(package_path / pathlib.Path(source).name, "w") as dst_file:
+                for line in src_file:
+                    dst_file.write(
+                        pattern.sub('import "' + package_name + "/osi_", line)
+                    )
+
+    # Run protoc
+    subprocess.check_call(
+        [PROTOC_EXE, "--python_out=.", "--pyi_out=.", package_name + "/*.proto"]
+    )
+
+    # Write __init__.py
+    with open(package_path / "__init__.py", "wt") as init_file:
+        init_file.write(
+            f"__version__ = '{VERSION_MAJOR}.{VERSION_MINOR}.{VERSION_PATCH}{VERSION_SUFFIX or ''}'\n"
+        )
+
+# Override build actions
+
+def build_wheel(
+    wheel_directory: str,
+    config_settings: dict[str, Any] | None = None,
+    metadata_directory: str | None = None,
+) -> str:
+    """Builds a wheel, places it in wheel_directory"""
+    _generate_python_files("osi3")
+    return build_wheel_orig(wheel_directory, config_settings, metadata_directory)
+
+def build_sdist(
+    sdist_directory: str, config_settings: dict[str, Any] | None = None
+) -> str:
+    """Builds an sdist, places it in sdist_directory"""
+    _generate_python_files("osi3")
+    return build_sdist_orig(sdist_directory, config_settings)
+

build-backend/__pycache__/protoc_backend.cpython-311.pyc

@@ -0,0 +1,66 @@
+"""
+This program converts serialized osi trace files into a human readable txth file. 
+
+Example usage:
+    python3 osi2read.py -d trace.osi -o myreadableosifile
+"""
+
+from osi3trace.osi_trace import OSITrace
+import argparse
+import pathlib
+
+
+def command_line_arguments():
+    """Define and handle command line interface"""
+
+    parser = argparse.ArgumentParser(
+        description="Convert a serialized osi trace file to a readable txth output.",
+        prog="osi2read",
+    )
+    parser.add_argument(
+        "--data",
+        "-d",
+        help="Path to the file with serialized data.",
+        type=str,
+        required=True,
+    )
+    parser.add_argument(
+        "--type",
+        "-t",
+        help="Name of the type used to serialize data.",
+        choices=OSITrace.message_types(),
+        default="SensorView",
+        type=str,
+        required=False,
+    )
+    parser.add_argument(
+        "--output",
+        "-o",
+        help="Output name of the file.",
+        type=str,
+        required=False,
+    )
+
+    return parser.parse_args()
+
+
+def main():
+    # Handling of command line arguments
+    args = command_line_arguments()
+
+    # Initialize the OSI trace class
+    trace = OSITrace(args.data, args.type)
+
+    if not args.output:
+        path = pathlib.Path(args.data).with_suffix(".txth")
+        args.output = str(path)
+
+    with open(args.output, "wt") as f:
+        for message in trace:
+            f.write(str(message))
+
+    trace.close()
+
+
+if __name__ == "__main__":
+    main()

build-backend/protoc_backend.py

@@ -0,0 +1,188 @@
+"""
+Module to handle and manage OSI trace files.
+"""
+
+import lzma
+import struct
+
+from osi3.osi_sensorview_pb2 import SensorView
+from osi3.osi_sensorviewconfiguration_pb2 import SensorViewConfiguration
+from osi3.osi_groundtruth_pb2 import GroundTruth
+from osi3.osi_hostvehicledata_pb2 import HostVehicleData
+from osi3.osi_sensordata_pb2 import SensorData
+from osi3.osi_trafficcommand_pb2 import TrafficCommand
+from osi3.osi_trafficcommandupdate_pb2 import TrafficCommandUpdate
+from osi3.osi_trafficupdate_pb2 import TrafficUpdate
+from osi3.osi_motionrequest_pb2 import MotionRequest
+from osi3.osi_streamingupdate_pb2 import StreamingUpdate
+
+
+MESSAGES_TYPE = {
+    "SensorView": SensorView,
+    "SensorViewConfiguration": SensorViewConfiguration,
+    "GroundTruth": GroundTruth,
+    "HostVehicleData": HostVehicleData,
+    "SensorData": SensorData,
+    "TrafficCommand": TrafficCommand,
+    "TrafficCommandUpdate": TrafficCommandUpdate,
+    "TrafficUpdate": TrafficUpdate,
+    "MotionRequest": MotionRequest,
+    "StreamingUpdate": StreamingUpdate,
+}
+
+
+class OSITrace:
+    """This class can import and decode OSI trace files."""
+
+    @staticmethod
+    def map_message_type(type_name):
+        """Map the type name to the protobuf message type."""
+        return MESSAGES_TYPE[type_name]
+
+    @staticmethod
+    def message_types():
+        """Message types that OSITrace supports."""
+        return list(MESSAGES_TYPE.keys())
+
+    def __init__(self, path=None, type_name="SensorView", cache_messages=False):
+        self.type = self.map_message_type(type_name)
+        self.file = None
+        self.current_index = None
+        self.message_offsets = None
+        self.read_complete = False
+        self.message_cache = {} if cache_messages else None
+        self._header_length = 4
+        if path:
+            self.from_file(path, type_name, cache_messages)
+
+    def from_file(self, path, type_name="SensorView", cache_messages=False):
+        """Import a trace from a file"""
+        self.type = self.map_message_type(type_name)
+
+        if path.lower().endswith((".lzma", ".xz")):
+            self.file = lzma.open(path, "rb")
+        else:
+            self.file = open(path, "rb")
+
+        self.read_complete = False
+        self.current_index = 0
+        self.message_offsets = [0]
+        self.message_cache = {} if cache_messages else None
+
+    def retrieve_offsets(self, limit=None):
+        """Retrieve the offsets of the messages from the file."""
+        if not self.read_complete:
+            self.current_index = len(self.message_offsets) - 1
+            self.file.seek(self.message_offsets[-1], 0)
+        while not self.read_complete and (
+            not limit or len(self.message_offsets) <= limit
+        ):
+            self.retrieve_message(skip=True)
+        return self.message_offsets
+
+    def retrieve_message(self, index=None, skip=False):
+        """Retrieve the next message from the file at the current position or given index, or skip it if skip is true."""
+        if index is not None:
+            self.current_index = index
+            self.file.seek(self.message_offsets[index], 0)
+        if self.message_cache is not None and self.current_index in self.message_cache:
+            message = self.message_cache[self.current_index]
+            self.current_index += 1
+            if self.current_index == len(self.message_offsets):
+                self.file.seek(0, 2)
+            else:
+                self.file.seek(self.message_offsets[self.current_index], 0)
+            if skip:
+                return self.message_offsets[self.current_index]
+            else:
+                return message
+        start = self.file.tell()
+        header = self.file.read(self._header_length)
+        if len(header) < self._header_length:
+            if start == self.message_offsets[-1]:
+                self.message_offsets.pop()
+                self.read_complete = True
+            self.file.seek(start, 0)
+            return None
+        message_length = struct.unpack("<L", header)[0]
+        if skip:
+            new_pos = self.file.seek(message_length, 1)
+            if new_pos - start < message_length + self._header_length:
+                if start == self.message_offsets[-1]:
+                    self.message_offsets.pop()
+                    self.read_complete = True
+                self.file.seek(start, 0)
+                return None
+            self.current_index += 1
+            if start == self.message_offsets[-1]:
+                self.message_offsets.append(new_pos)
+            return new_pos
+        message_data = self.file.read(message_length)
+        if len(message_data) < message_length:
+            if start == self.message_offsets[-1]:
+                self.message_offsets.pop()
+                self.read_complete = True
+            self.file.seek(start, 0)
+            return None
+        self.current_index += 1
+        message = self.type()
+        message.ParseFromString(message_data)
+        if start == self.message_offsets[-1]:
+            if self.message_cache is not None:
+                self.message_cache[len(self.message_offsets) - 1] = message
+            self.message_offsets.append(self.file.tell())
+        return message
+
+    def restart(self, index=None):
+        """Restart the reading of the file from the beginning or from a given index."""
+        self.current_index = index if index else 0
+        self.file.seek(self.message_offsets[self.current_index], 0)
+
+    def __iter__(self):
+        while message := self.retrieve_message():
+            yield message
+
+    def get_message_by_index(self, index):
+        """
+        Get a message by its index.
+        """
+        if index >= len(self.message_offsets):
+            self.retrieve_offsets(index)
+        if self.message_cache is not None and index in self.message_cache:
+            return self.message_cache[index]
+        return self.retrieve_message(index=index)
+
+    def get_messages(self):
+        """
+        Yield an iterator over all messages in the file.
+        """
+        return self.get_messages_in_index_range(0, None)
+
+    def get_messages_in_index_range(self, begin, end):
+        """
+        Yield an iterator over messages of indexes between begin and end included.
+        """
+        if begin >= len(self.message_offsets):
+            self.retrieve_offsets(begin)
+        self.restart(begin)
+        current = begin
+        while end is None or current < end:
+            if self.message_cache is not None and current in self.message_cache:
+                yield self.message_cache[current]
+            else:
+                message = self.retrieve_message()
+                if message is None:
+                    break
+                yield message
+            current += 1
+
+    def close(self):
+        if self.file:
+            self.file.close()
+        self.file = None
+        self.current_index = None
+        self.message_cache = None
+        self.message_offsets = None
+        self.read_complete = False
+        self.read_limit = None
+        self.type = None