文章摘要
这篇文章介绍了一个使用Python语言的ROS（Robot Operating System）脚本，用于合并多个ROS bag文件。该脚本通过命令行参数接受输入，支持选择性合并文件，并可以指定压缩格式（包括不压缩、BZ2和LZ4）。脚本还使用了rosbag模块中的Bag类来读取和写入文件，并在合并过程中显示进度条，实时反馈合并进度。文章重点展示了脚本的结构、功能以及如何处理文件输入和压缩参数。
#! /usr/bin/env python3
import os
import time
import argparse
from rosbag import Bag, Compression
def parse_compression(compression):
if compression==”none” or compression==”NONE”:
compression=Compression.NONE
elif compression==”bz2″:
compression=Compression.BZ2
elif compression==”lz4″:
compression=Compression.LZ4
return compression
def get_files_list(arg_input, arg_select):
files=None
if ” ” in arg_input:
files=arg_input.split(” “)
elif “,” in arg_input:
files=arg_input.split(“,”)
elif os.path.exists(args.input) or “*” in args.input:
rfind_index=arg_input.rfind(“/”)
dir_path=arg_input[:rfind_index]
files=os.listdir(dir_path)
files.sort() # 名称排序
files=[dir_path + “/” + file for file in files]
if arg_select:
files=select_bags(files)
print(“bags list:”)
for i in range(len(files)):
print(str(i) + “.” + files[i])
return files
def select_bags(files):
for i in range(len(files)):
print(str(i) + “.” + files[i])
print(“Please input bag numbers to merge. split by , or space”, end=”:”)
s_b=input()
s_b_list=[]
if “,” in s_b:
s_b_list=s_b.split(“,”)
elif ” ” in s_b:
s_b_list=s_b.split(” “)
files=[files[int(x)] for x in s_b_list]
return files
def show_process_bar(total, i, start):
a=”*” * i
b=”.” * (total – i)
c=(i / total) * 100
dur=time.perf_counter() – start
print(“\r{:^3.0f}%[{}->{}]{:.2f}s”.format(c,a,b,dur), end=””)
def merge_bags(args):
print(“start merge bags.”)
compression=parse_compression(args.compression)
print(f”bag’s compression mode is {compression}.”)
files=get_files_list(args.input, args.select)
with Bag(args.output, “w”, compression=compression) as o:
start=time.perf_counter()
for i in range(len(files)):
show_process_bar(len(files), i+1, start)
with Bag(files[i], “r”) as ib:
for topic, msg, t in ib:
o.write(topic, msg, t)
show_process_bar(len(files), i+1, start)
if __name__==”__main__”:
parser=argparse.ArgumentParser(description=”Merge one or more bag files to one file.”)
parser.add_argument(“-o”, “–output”, help=”Output bag’s file path.”,
default=”output.bag”, required=True)
parser.add_argument(“-i”, “–input”, help=”Input files bags name or path, split by , .”,
required=True)
parser.add_argument(“-c”, “–compression”, help=”Compress the bag by bz2 or lz4″,
default=”lz4″, choices=[“none”, “lz4”, “bz2”])
parser.add_argument(“-s”, “–select”, help=”Select bags to merge.”,
default=False)
parser.add_argument(“-v”, “–verbose”, help=”Show the verbose msg.”)
args=parser.parse_args()
merge_bags(args)