Function run_python_copy

scripts/bench_rosbag2_cpp_vs_py.py:36–76

(args: Args)

def run_python_copy(args: Args):
    reader = SequentialReader()
    reader.open(StorageOptions(uri=args.input_uri, storage_id=args.storage_id),
                ConverterOptions(input_serialization_format=args.serialization, output_serialization_format=args.serialization))
    writer = SequentialWriter()
    writer.open(StorageOptions(uri=args.output_uri_py, storage_id=args.storage_id),
                ConverterOptions(input_serialization_format=args.serialization, output_serialization_format=args.serialization))

    try:
        topics = reader.get_all_topics_and_types()
    except AttributeError:
        topics = reader.get_topics_and_types()

    for t in topics:
        writer.create_topic(TopicMetadata(
            id=0,
            name=t.name,
            type=t.type,
            serialization_format=args.serialization
        ))

    has_next = reader.has_next
    read_next = reader.read_next
    write = writer.write

    gc_was_enabled = gc.isenabled()
    if gc_was_enabled:
        gc.disable()
    try:
        t0 = time.time()
        count = 0
        while has_next():
            topic, data, ts = read_next()
            write(topic, data, ts)
            count += 1
        t1 = time.time()
    finally:
        if gc_was_enabled:
            gc.enable()
            gc.collect()
    return t1 - t0, count


def run_cxx_copy(args: Args):