Python implementation of NCodec.

dse/ncodec/examples/CMakeLists.txt

@@ -5,6 +5,7 @@
 
 # Example Models.
 add_subdirectory(codec)
+add_subdirectory(fmpy)
 
 # Code examples for documentation.
 #if(UNIX)

dse/ncodec/examples/fmpy/CMakeLists.txt

@@ -0,0 +1,40 @@
+# Copyright 2025 Robert Bosch GmbH
+#
+# SPDX-License-Identifier: Apache-2.0
+
+cmake_minimum_required(VERSION 3.21)
+project(python_ncodec)
+
+set(EXAMPLE_PATH "examples/fmpy")
+set(FLATCC_SOURCE_DIR  ${DSE_NCODEC_SOURCE_DIR}/schema/abs/flatcc/src)
+set(FLATCC_INCLUDE_DIR ${DSE_NCODEC_SOURCE_DIR}/schema/abs/flatcc/include)
+
+add_library(ncodec SHARED
+    ncodec.c
+    ${DSE_NCODEC_SOURCE_DIR}/stream/buffer.c
+    ${DSE_NCODEC_SOURCE_DIR}/codec/ab/codec.c
+    ${DSE_NCODEC_SOURCE_DIR}/codec/ab/frame_fbs.c
+    ${DSE_NCODEC_SOURCE_DIR}/codec/ab/pdu_fbs.c
+    ${DSE_NCODEC_SOURCE_DIR}/codec.c
+    ${FLATCC_SOURCE_DIR}/builder.c
+    ${FLATCC_SOURCE_DIR}/emitter.c
+    ${FLATCC_SOURCE_DIR}/refmap.c
+)
+target_include_directories(ncodec
+    PRIVATE
+        ${DSE_NCODEC_INCLUDE_DIR}
+        ${DSE_CLIB_INCLUDE_DIR}
+        ${FLATCC_INCLUDE_DIR}
+)
+install(TARGETS ncodec
+    LIBRARY DESTINATION
+        ${EXAMPLE_PATH}/lib
+    RUNTIME DESTINATION
+        ${EXAMPLE_PATH}/lib
+)
+install(
+    FILES
+        simulation.py
+    DESTINATION
+        ${EXAMPLE_PATH}
+)

dse/ncodec/examples/fmpy/README.md

@@ -0,0 +1,39 @@
+<!--
+Copyright 2025 Robert Bosch GmbH
+
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# DSE NCodec Example: FMPy Integration
+
+
+```bash
+# Clone and build the examples.
+$ git clone https://github.com/boschglobal/dse.ncodec.git
+$ cd dse.ncodec
+$ make
+
+# Change to the FMPy example directory.
+$ cd dse/ncodec/build/_out/examples/fmpy
+
+# Download the FMU.
+$ export FMU_URL=https://github.com/boschglobal/dse.fmi/releases/download/v1.1.28/example-network-linux-amd64-fmi2.fmu
+$ curl -fSL $FMU_URL --create-dirs --output fmu/example-network-fmi2.fmu
+$ ls -1shR
+.:
+total 8.0K
+   0 fmu/
+   0 lib/
+8.0K simulation.py*
+
+./fmu:
+total 688K
+688K example-network-fmi2.fmu*
+
+./lib:
+total 500K
+500K libncodec.so*
+
+# Run the simulation.
+$ python simulation.py
+```

dse/ncodec/examples/fmpy/ncodec.c

@@ -0,0 +1,77 @@
+// Copyright 2025 Robert Bosch GmbH
+//
+// SPDX-License-Identifier: Apache-2.0
+
+#include <stdlib.h>
+#include <stdint.h>
+#include <string.h>
+#include <dse/platform.h>
+#include <dse/ncodec/codec.h>
+#include <dse/ncodec/stream/stream.h>
+
+
+DLL_PUBLIC void* ncodec_open_with_stream(const char* mime_type)
+{
+    NCODEC* nc = ncodec_create(mime_type);
+    if (nc) {
+        NCodecInstance*     _nc = (NCodecInstance*)nc;
+        NCodecStreamVTable* stream = ncodec_buffer_stream_create(0);
+        _nc->stream = stream;
+    }
+    return (void*)nc;
+}
+
+DLL_PUBLIC int64_t ncodec_write_pdu_msg(
+    void* nc, uint32_t id, uint8_t* payload, size_t payload_len)
+{
+    return ncodec_write(nc, &(struct NCodecPdu){
+        .id = id,
+        .payload = payload,
+        .payload_len = payload_len
+    });
+}
+
+DLL_PUBLIC int64_t ncodec_read_pdu_msg(void* nc,
+    uint32_t* id, uint8_t** payload, size_t* payload_len)
+{
+    NCodecPdu msg = {};
+    int64_t          rc = ncodec_read(nc, &msg);
+    if (rc > 0) {
+        *id = msg.id;
+        *payload = malloc(msg.payload_len);
+        memcpy(*payload, msg.payload, msg.payload_len);
+        *payload_len = msg.payload_len;
+    }
+    return rc;
+}
+
+DLL_PUBLIC int64_t ncodec_write_stream(void* nc, uint8_t* buffer, size_t len)
+{
+    NCodecInstance* _nc = (NCodecInstance*)nc;
+    if (_nc && _nc->stream && _nc->stream->write) {
+        ncodec_truncate(nc);
+        _nc->stream->write((NCODEC*)nc, buffer, len);
+        ncodec_seek(nc, 0, NCODEC_SEEK_SET);  // Position for reading.
+        return len;
+    } else {
+        return -ENOSTR;
+    }
+}
+
+DLL_PUBLIC size_t ncodec_read_stream(void* nc, uint8_t** buffer)
+{
+    NCodecInstance* _nc = (NCodecInstance*)nc;
+    if (_nc && _nc->stream && _nc->stream->read) {
+        size_t len = ncodec_flush(nc);
+        ncodec_seek(nc, 0, NCODEC_SEEK_SET);
+        *buffer = malloc(len);
+        return _nc->stream->read(nc, buffer, &len, NCODEC_POS_UPDATE);
+    } else {
+        return -ENOSTR;
+    }
+}
+
+DLL_PUBLIC void ncodec_free(uint8_t* buffer)
+{
+    free(buffer);
+}

dse/ncodec/examples/fmpy/simulation.py

@@ -0,0 +1,114 @@
+# Copyright 2025 Robert Bosch GmbH
+# SPDX-License-Identifier: Apache-2.0
+
+import base64
+import ctypes
+from ctypes import *
+from typing import List
+from fmpy import simulate_fmu, read_model_description, extract
+from fmpy.fmi2 import FMU2Slave
+from ncodec.codec_interface import CodecFactory, ICodec
+from ncodec.pdu import PduMessage
+
+
+# Global objects:
+fmu = None
+nc = None
+network_rx_ref = None
+network_tx_ref = None
+id : c_uint32 = 0
+
+
+def do_network():
+    global nc
+    global id
+
+    # Fetch the stream from FMU and write to NCodec:
+    stream = bytearray()
+    tx_message_hex = fmu.getString([network_tx_ref])[0]
+    if tx_message_hex is not None:
+        stream = base64.a85decode(tx_message_hex)
+
+    # Read messages from the NCodec:
+    nc.Stream = stream
+    read_pdus : List[PduMessage] = nc.Read()
+    for pdus in read_pdus:
+        print(f'  Received PDU from model: {pdus.id}')
+
+    # Write messages to the NCodec:
+    message = PduMessage(
+        id,
+        bytes([0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x07]),
+        0
+    )
+    id += 1
+
+    # Send the NCodec stream to FMU:
+    nc.Write([message])
+    nc.Flush()
+    rx_message_hex = base64.a85encode(nc.Stream, adobe=False)
+    rx_message_hex = rx_message_hex.decode('utf-8')
+    fmu.setString([network_rx_ref], [rx_message_hex])
+    nc.Truncate()
+
+
+def simulation_setup():
+    # Setup FMU:
+    global fmu
+    fmu_filename = r"fmu/example-network-fmi2.fmu"
+    unzipdir = extract(r"fmu/example-network-fmi2.fmu")
+    model_desc = read_model_description(fmu_filename)
+    fmu = FMU2Slave(
+        guid=model_desc.guid,
+        modelIdentifier=model_desc.coSimulation.modelIdentifier,
+        unzipDirectory=unzipdir,
+        instanceName="network"
+    )
+
+    # Locate network variables
+    global network_rx_ref
+    global network_tx_ref
+    network_rx_var = next((var for var in model_desc.modelVariables if var.name == "pdu_rx"), None)
+    network_tx_var = next((var for var in model_desc.modelVariables if var.name == "pdu_tx"), None)
+    if network_rx_var:
+        network_rx_ref = network_rx_var.valueReference
+    else:
+        raise ValueError("pdu_rx not found in the FMU.")
+    if network_tx_var:
+        network_tx_ref = network_tx_var.valueReference
+    else:
+        raise ValueError("pdu_tx not found in the FMU.")
+
+
+def simulation_instantiate():
+    global fmu
+    fmu.instantiate(loggingOn=True)
+    fmu.setupExperiment(startTime=0.0, stopTime=4.0, tolerance=None)
+    fmu.enterInitializationMode()
+    fmu.exitInitializationMode()
+
+    global nc
+    nc = CodecFactory.create_codec("application/x-automotive-bus; interface=stream; type=pdu; schema=fbs; swc_id=1; ecu_id=7", bytearray(), "Test", 0.0)
+
+
+def simulation_terminate():
+    global fmu
+    fmu.terminate()
+    fmu.freeInstance()
+
+    # global nc
+    # ncodecDLL.ncodec_close(nc)
+
+
+# Run the simulation
+simulation_setup()
+simulation_instantiate()
+
+step_size = 0.0005
+current_time = 0.0
+while current_time < 4.0:
+    do_network()
+    fmu.doStep(currentCommunicationPoint=current_time, communicationStepSize=step_size)
+    current_time += step_size
+
+simulation_terminate()

extra/python/.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+.dist-info/
+build/
+*.pyc
+.venv/

extra/python/README.md

@@ -0,0 +1,41 @@
+# extra.python
+
+This directory contains the Python package for the `dse.ncodec` project.
+
+Layout
+- `src/` — source packages (importable packages live under `src/`)
+- `tests/` — unit tests (pytest)
+- `pyproject.toml` — build metadata and packaging configuration
+- `README.md` — this file
+
+Install
+
+Create a virtual environment and install the package in editable mode for development:
+
+```bash
+cd dse.ncodec/extra/python
+python -m venv .venv
+source .venv/bin/activate
+pip install -e .[test]
+```
+
+Quick usage
+
+```py
+
+```
+
+Run tests
+
+```bash
+pytest
+```
+
+Development notes
+
+- The package uses a src-layout so packaging tools find packages under `src/`.
+- Follow the repository license in the top-level `LICENSE` file.
+
+Contact
+
+Please update `pyproject.toml` authors and metadata with the appropriate project contact information.

extra/python/pyproject.toml

@@ -0,0 +1,25 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "ncodec"
+version = "0.0.1"
+description = "Python implementation of the dse.ncodec project."
+authors = [
+    { name = "Your Name", email = "your.email@example.com" }
+]
+readme = "README.md"
+requires-python = ">=3.8"
+license = { file = "../LICENSE" }
+classifiers = [
+    "Programming Language :: Python :: 3",
+    "Operating System :: OS Independent",
+]
+keywords = ["ncodec", "codec", "bosch"]
+
+[project.optional-dependencies]
+test = ["pytest>=6.0","flatbuffers==1.12"]
+
+[tool.setuptools.packages.find]
+where = ["src"]

extra/python/pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+minversion = 6.0
+addopts = -q
+testpaths = tests

extra/python/src/AutomotiveBus/Register/Can/BufferDirection.py

@@ -0,0 +1,8 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: Can
+
+class BufferDirection(object):
+    Tx = 0
+    Rx = 1
+

extra/python/src/AutomotiveBus/Register/Can/BufferStatus.py

@@ -0,0 +1,8 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: Can
+
+class BufferStatus(object):
+    None_ = 0
+    RxError = 1
+

extra/python/src/AutomotiveBus/Register/Can/BusState.py

@@ -0,0 +1,9 @@
+# automatically generated by the FlatBuffers compiler, do not modify
+
+# namespace: Can
+
+class BusState(object):
+    BusOff = 0
+    Idle = 1
+    Sync = 2
+