Created Action to extract ULog file data into Topics

This commit adds an Action that extracts ULog file data into Topics for downstream visualization and search.

- Convert ULog files to per-topic MCAP files using JSONSchema.
- Create topic records
- Create message path records
- Set default topic representations

Tested locally and on the beta environment.

README.md

@@ -10,6 +10,7 @@ The package currently includes the following Actions:
 - `ulog_info`: Extract metadata from a .ulg file.
 - `ulog_to_csv`: Convert .ulg files to .csv files.
 - `px4_flight_review`: Generate interactive plots to analyze PX4 flights.
+- `ulog_ingestion`: Ingest data from .ulg files into topics for visualization and search.
 
 # Prerequisites
 

actions/ulog_ingestion/.gitignore

@@ -0,0 +1,11 @@
+.venv
+.mypy_cache
+**/*.egg-info
+**/.mypy_cache/
+**/.venv/
+**/__pycache__/
+**/.pytest_cache
+**/dist/
+*.swp
+*.pyc
+output

actions/ulog_ingestion/.python-version

@@ -0,0 +1,4 @@
+# This file is a directive to pyenv (https://github.com/pyenv/pyenv) to set matching version of Python in this directory.
+# If you don't use pyenv, you can safely delete this file.
+# The roboto CLI requires Python 3.9 or higher.
+3.10

actions/ulog_ingestion/Dockerfile

@@ -0,0 +1,11 @@
+ARG PYTHON_MAJOR=3
+ARG PYTHON_MINOR=10
+ARG OS_VARIANT=slim-bookworm
+FROM --platform=linux/amd64 public.ecr.aws/docker/library/python:${PYTHON_MAJOR}.${PYTHON_MINOR}-${OS_VARIANT}
+
+COPY requirements.runtime.txt ./
+RUN python -m pip install --upgrade pip setuptools && python -m pip install -r requirements.runtime.txt
+
+COPY src/ulog_ingestion/ ./ulog_ingestion
+
+ENTRYPOINT [ "python", "-m", "ulog_ingestion" ]

actions/ulog_ingestion/README.md

@@ -0,0 +1,14 @@
+# ulog_ingestion
+
+This Action processes ULog files for visualization within the Roboto platform.
+
+## Getting started
+
+1. Setup a virtual environment specific to this project and install development dependencies, including the `roboto` CLI: `./scripts/setup.sh`
+2. Build Docker image: `./scripts/build.sh`
+3. Run Action image locally: `./scripts/run.sh <path-to-input-data-directory>`
+4. Deploy to Roboto Platform: `./scripts/deploy.sh`
+
+## Action configuration file
+
+This Roboto Action is configured in `action.json`. Refer to Roboto's latest documentation for the expected structure.

actions/ulog_ingestion/action.json

@@ -0,0 +1,23 @@
+{
+    "name": "ulog_ingestion",
+    "description": "Ingest data from ULog files into topics for visualization and search.",
+    "short_description": "Ingest data from ULog files into topics for visualization and search.",
+    "parameters": [
+        {
+            "name": "TOPICS",
+            "required": false,
+            "description": "Comma-separated list of topics to extract. For example: battery_status,actuator_armed"
+        }
+    ],
+    "compute_requirements": {
+        "vCPU": 4096,
+        "memory": 8192,
+        "storage": 21
+    },
+    "tags": [
+        "px4"
+    ],
+    "metadata": {
+        "github_url": "https://github.com/roboto-ai/robologs-px4-actions/tree/main/actions/ulog_ingestion"
+    }
+}

actions/ulog_ingestion/requirements.dev.txt

@@ -0,0 +1,7 @@
+# Python packages to install into the this directory's virtual environment
+# for the purpose of development, testing, and deployment.
+
+# Install all required runtime dependencies in local virtual environment.
+-r requirements.runtime.txt
+
+# Add additional Python packages to install here.

actions/ulog_ingestion/requirements.runtime.txt

@@ -0,0 +1,5 @@
+# Python packages to install within the Docker image associated with this Action.
+roboto==0.2.11
+pyulog==1.0.2
+mcap==1.1.1
+jsonschema>=4.21.1

actions/ulog_ingestion/scripts/build.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPTS_ROOT=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd)
+PACKAGE_ROOT=$(dirname "${SCRIPTS_ROOT}")
+
+build_subcommand=(build)
+# if buildx is installed, use it
+if docker buildx version &> /dev/null; then
+    build_subcommand=(buildx build --platform linux/amd64 --output type=image)
+fi
+
+docker "${build_subcommand[@]}" -f $PACKAGE_ROOT/Dockerfile -t ulog_ingestion:latest $PACKAGE_ROOT

actions/ulog_ingestion/scripts/deploy.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPTS_ROOT=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd)
+PACKAGE_ROOT=$(dirname "${SCRIPTS_ROOT}")
+
+# Early exit if virtual environment does not exist and/or roboto is not yet installed
+if [ ! -f "$PACKAGE_ROOT/.venv/bin/roboto" ]; then
+    echo "Virtual environment with roboto CLI does not exist. Please run ./scripts/setup.sh first."
+    exit 1
+fi
+
+# Set org_id to $ROBOTO_ORG_ID if defined, else the first argument passed to this script
+org_id=${ROBOTO_ORG_ID:-}
+if [ $# -gt 0 ]; then
+    org_id=$1  
+fi
+
+roboto_exe="$PACKAGE_ROOT/.venv/bin/roboto"
+
+echo "Pushing ulog_ingestion:latest to Roboto's private registry"
+image_push_args=(
+    --suppress-upgrade-check
+    images push
+    --quiet
+)
+if [[ -n $org_id ]]; then
+    image_push_args+=(--org $org_id)
+fi
+image_push_args+=(ulog_ingestion:latest)
+image_push_ret_code=0
+image_uri=$($roboto_exe "${image_push_args[@]}")
+image_push_ret_code=$?
+
+if [ $image_push_ret_code -ne 0 ]; then
+    echo "Failed to push ulog_ingestion:latest to Roboto's private registry"
+    exit 1
+fi
+
+echo "Creating ulog_ingestion action"
+create_args=(
+  --from-file $PACKAGE_ROOT/action.json
+  --image $image_uri
+  --yes
+)
+if [[ -n $org_id ]]; then
+    create_args+=(--org $org_id)
+fi
+$roboto_exe actions create "${create_args[@]}"

actions/ulog_ingestion/scripts/run.sh

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPTS_ROOT=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd)
+PACKAGE_ROOT=$(dirname "${SCRIPTS_ROOT}")
+
+# Set input_dir to $ROBOTO_INPUT_DIR if defined, else the first argument passed to this script
+input_dir=${ROBOTO_INPUT_DIR:-}
+if [ $# -gt 0 ]; then
+    input_dir=$1  
+fi
+
+# Fail if input_dir is not an existing directory
+if [ ! -d "$input_dir" ]; then
+    echo "Specify an existing input directory as the first argument to this script, or set the ROBOTO_INPUT_DIR environment variable"
+    exit 1
+fi
+
+# Set output_dir variable to $ROBOTO_OUTPUT_DIR if defined, else set it to "output/" in the package root (creating if necessary)
+output_dir=${ROBOTO_OUTPUT_DIR:-$PACKAGE_ROOT/output}
+mkdir -p $output_dir
+
+# Assert both directories are absolute paths
+if [[ ! "$input_dir" = /* ]]; then
+    echo "Input directory '$input_dir' must be specified as an absolute path"
+    exit 1
+fi
+
+if [[ ! "$output_dir" = /* ]]; then
+    echo "Output directory '$output_dir' must be specified as an absolute path"
+    exit 1
+fi
+
+docker run --rm -it \
+    -v $input_dir:/input \
+    -v $output_dir:/output \
+    -e ROBOTO_INPUT_DIR=/input \
+    -e ROBOTO_OUTPUT_DIR=/output \
+    ulog_ingestion:latest

actions/ulog_ingestion/scripts/setup.sh

@@ -0,0 +1,15 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPTS_ROOT=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd)
+PACKAGE_ROOT=$(dirname "${SCRIPTS_ROOT}")
+
+venv_dir="$PACKAGE_ROOT/.venv"
+
+# Create a virtual environment
+python -m venv --upgrade-deps $venv_dir
+
+# Install roboto
+pip_exe="$venv_dir/bin/pip"
+$pip_exe install --upgrade -r $PACKAGE_ROOT/requirements.dev.txt

actions/ulog_ingestion/scripts/test.sh

@@ -0,0 +1,80 @@
+#!/bin/bash
+
+SCRIPTS_ROOT=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd)
+PACKAGE_ROOT=$(dirname "${SCRIPTS_ROOT}")
+
+# Define constants for directories and file paths
+
+INPUT_DIR=${PACKAGE_ROOT}/test/input
+ACTUAL_OUTPUT_DIR=${PACKAGE_ROOT}/test/actual_output
+EXPECTED_OUTPUT_DIR=${PACKAGE_ROOT}/test/expected_output
+
+if [ ! -d "$ACTUAL_OUTPUT_DIR" ]; then
+    mkdir -p "$ACTUAL_OUTPUT_DIR"
+fi
+
+# Remove previous outputs
+clean_actual_output() {
+    rm -rf $ACTUAL_OUTPUT_DIR/
+}
+
+# Check if file exists
+file_exists_or_error() {
+    local file_path="$1"
+
+    if [ ! -f "$file_path" ]; then
+        echo "Error: File '$file_path' does not exist."
+        exit 1
+    fi
+    echo "Test passed!"
+
+}
+
+# Run the docker command with the given parameters
+run_docker_test() {
+    local additional_args="$1"
+
+    docker run \
+        -v $INPUT_DIR:/input \
+        -v $ACTUAL_OUTPUT_DIR:/output \
+        -e ROBOTO_INPUT_DIR=/input \
+        -e ROBOTO_OUTPUT_DIR=/output \
+        $additional_args \
+        ulog_ingestion:latest
+}
+
+function check_file_does_not_exist() {
+    local file_path="$1"
+    if [[ ! -e "$file_path" ]]; then
+        echo "Test passed!"
+    else
+        echo "Test failed: $1 exists!"
+        exit 1
+    fi
+}
+
+# Compare the actual output to the expected output
+compare_outputs() {
+    local actual_file="$1"
+    local expected_file="$2"
+
+    diff $ACTUAL_OUTPUT_DIR/$actual_file $EXPECTED_OUTPUT_DIR/$expected_file
+
+    if [ $? -eq 0 ]; then
+        echo "Test passed!"
+    else
+        echo "Test failed!"
+	exit 1
+    fi
+}
+
+# Main test execution
+main() {
+
+    # Test 1
+        echo "No-op test passed!"
+}
+
+# Run the main test execution
+main
+clean_actual_output

actions/ulog_ingestion/src/tests/test_utils.py

@@ -0,0 +1,47 @@
+import os.path
+import shutil
+import ulog_ingestion.utils as utils
+from pyulog.core import ULog
+
+
+def test_create_mcap_file_from_ulog(tmp_path):
+    ulog_file_path = "./tests/test.ulg"
+
+    test_topic_name = "vehicle_acceleration"
+
+    output_path_per_topic_mcap = tmp_path / f"{test_topic_name}.mcap"
+
+    ulog = ULog(ulog_file_path, [test_topic_name], True)
+
+    schema_registry_dict = {}
+
+    for key in ulog.message_formats:
+        json_schema_topic = utils.create_json_schema(ulog.message_formats[key].fields)
+        schema_registry_dict[key] = json_schema_topic
+
+    for data_object in sorted(ulog.data_list, key=lambda obj: obj.name):
+        print(data_object.name)
+        utils.create_per_topic_mcap_from_ulog(output_path_per_topic_mcap,
+                                              data_object,
+                                              schema_registry_dict)
+
+    assert output_path_per_topic_mcap.exists()
+
+
+def test_setup_output_folder_structure():
+
+    ulog_file_path = "/workspace/abc/test.ulg"
+    input_dir = "/workspace/"
+
+    output_folder_path, temp_dir = utils.setup_output_folder_structure(ulog_file_path, input_dir)
+
+    assert output_folder_path == f"{temp_dir}/.VISUALIZATION_ASSETS/abc/test"
+    assert os.path.exists(output_folder_path)
+    shutil.rmtree(output_folder_path)
+
+
+def test_is_valid_ulog():
+
+    ulog_file_path = "./tests/test.ulg"
+    is_valid = utils.is_valid_ulog(ulog_file_path)
+    assert is_valid is True