For context, I’m using Iceoryx2 version 0.9.0.
I see that y’all expose the C++ type iox2::bb::StaticVector<>. Is cross-language communication between a C++ process (publisher) and a Python process (subscriber) possible, where the payload type is StaticVector<>? If so, how would that look on the receiving Python side? I don’t see a StaticVector equivalent in the Python bindings (unless I’m missing something), so I don’t know what type to provide when calling publish_subscribe(...) on the ServiceBuilder.
I’ll provide an example below.
C++ Side
auto main() -> int {
auto node = iox2::NodeBuilder()
.create<iox2::ServiceType::Ipc>()
.value();
auto service = node.service_builder(ServiceName::create("Example").value())
.publish_subscribe<iox2::bb::StaticVector<float, 64>>()
.open_or_create()
.value()
auto publisher = service.publisher_builder().create().value();
# Publish
std::vector<float> vec(64);
auto sample = publisher.loan_uninit().value();
auto initialized_sample =
sample.write_payload(*iox2::bb::StaticVector<float, 64>::from_range_unchecked(vec));
send(std::move(initialized_sample)).value();
}
Python Side
node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)
service = (
node.service_builder(iox2.ServiceName.new("Example"))
.publish_subscribe("What type do I pass?")
.open_or_create()
)
# How would parsing the received payload look?
Much thanks in advance. I greatly appreciate the work y’all have done.
The memory layout compatible containers (String and Vector) are only implemented for C++ and Rust at the moment.
We would love to implement them also for Python but currently we are looking for a company who would like to sponsor/contract this feature - any support here is highly appreciated.
I ran into the same issue that there is currently no way to use the iox2::bb::StaticVector<> with Python.
However, I managed to use following workaround by simply using a custom structure in C++ and class in Python
C++ Side
struct StaticVectorWorkaround{
// Type name important for Cross Language Communication (Python)
static constexpr const char *IOX2_TYPE_NAME = "StaticVectorWorkaround";
//iox2::bb::StaticVector<uint8_t, 8192> buffer;
std::array<uint8_t, 8192> buffer; // <-- use this instead of StaticVector
uint32_t size{0}; // Optional buffer size
};
Python Side
import ctypes
class StaticVectorWorkaround(ctypes.Structure):
_fields_ = [
("buffer", ctypes.c_uint8 * 8192),
("size", ctypes.c_uint32),
]
def __str__(self) -> str:
"""Returns human-readable string of the contents."""
return f"StaticVectorWorkaround(size={self.payloadSize})"
@staticmethod
def type_name() -> str:
"""Returns system-wide unique type name required for communication."""
return "StaticVectorWorkaround"
Since you also ask about how to parse this structure, on the python side the code would look something like this:
node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)
service = (
node.service_builder(iox2.ServiceName.new("Example"))
.publish_subscribe(StaticVectorWorkaround, StaticVectorWorkaround)
.open_or_create()
)
client = service.client_builder().create()
def send_and_receive(request_data: bytes) -> bytes:
iox_request = client.loan_uninit()
buffer_size = len(request_data)
if buffer_size > 8192:
raise BufferError(f"Maximum buffer size of {8192} exceeded")
# This creates the StaticVectorWorkaround and writes its fileds
payload = StaticVectorWorkaround()
payload.size = buffer_size
payload.buffer[:buffer_size] = request_data
iox_request = iox_request.write_payload(payload)
pending_response = iox_request.send()
# Receive all responses for this request
while True:
iox_response = pending_response.receive()
if iox_response:
payload: StaticVectorWorkaround= iox_response.payload().contents
return bytes(payload.buffer[:payload.size])
I hope this is helpful, for me it was a fitting replacement for StaticVector, but I also did not plan on using any of the iox2::bb::StaticVector<>class’ functionality