Source code for bosdyn.client.data_chunk

# Copyright (c) 2023 Boston Dynamics, Inc.  All rights reserved.
# Downloading, reproducing, distributing or otherwise using the SDK Software
# is subject to the terms and conditions of the Boston Dynamics Software
# Development Kit License (20191101-BDSDK-SL).

from bosdyn.api import data_chunk_pb2

[docs]def split_serialized(serialized: bytes, data_chunk_byte_size: int): """Split a byte string into appropriately-sized chunks.""" start_index = 0 end_index = 0 total_size = len(serialized) while end_index < total_size: end_index = start_index + data_chunk_byte_size yield serialized[start_index:min(end_index, total_size)] start_index = end_index
[docs]def chunk_serialized(serialized: bytes, data_chunk_byte_size: int): """Yield DataChunks for the given bytes.""" total_bytes_size = len(serialized) for data in split_serialized(serialized, data_chunk_byte_size): yield data_chunk_pb2.DataChunk(total_size=total_bytes_size, data=data)
[docs]def chunk_message(message, data_chunk_byte_size: int): """Take a message, and split it into data chunks Args: data_chunk_byte_size: max size of each streamed message """ return chunk_serialized(message.SerializeToString(), data_chunk_byte_size)
[docs]def parse_from_chunks(iterable_chunks, out_msg): """Parse out a message from chunks.""" return out_msg.ParseFromString(serialized_from_chunks(iterable_chunks))
[docs]def serialized_from_messages(iterable_messages): """Assemble from messages that define a DataChunk chunk field.""" return serialized_from_strings( for msg in iterable_messages)
[docs]def serialized_from_chunks(iterable_chunks): """Assemble from data chunks directly without wrapper messages.""" return serialized_from_strings( for chunk in iterable_chunks)
[docs]def serialized_from_strings(iterable_strings): """Concatenate bytes together.""" return b''.join(iterable_strings)