Dump sensor_msgs/CompressedImage from bagfile into a folder as png or jpeg files

extract_images_from_bagfile.py

#!/usr/bin/env python3
"""Extract sensor_msgs/CompressedImage from bagfile and save into a folder as png or jpeg files

Example usage:
./extract_images_from_bagfile.py bagfile.bag "/camera_topic" output_folder/images

Then files will be saved inside:
  output_folder/images/image_1594120767.5438294.jpeg
  output_folder/images/image_1594120767.8838294.jpeg
  output_folder/images/image_1594120767.9538294.jpeg
  ...
"""
import cv2
import numpy as np
import rosbag
import argparse
import logging
import os

__license__ = 'MIT'


def main():
    logging.getLogger().setLevel(logging.INFO)
    supported_image_formats = ['jpeg', 'png']

    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)

    parser.add_argument(
        "bagfile", help="Path to bagfile from which to extract images")
    parser.add_argument("topic", help="Topic name")

    format_help_string = "File format (e.g. {})".format(",".join(supported_image_formats))
    parser.add_argument("--format",
                        help=format_help_string,
                        default="jpeg")

    parser.add_argument("output", help="Name of the output folder")

    args = parser.parse_args()

    if args.format not in supported_image_formats:
        logging.error("Supported image formats {}".format(", ".join(supported_image_formats)))
        return None

    bag = rosbag.Bag(args.bagfile)
    for topic, msg, t in bag.read_messages(topics=args.topic):

        arr = np.fromstring(msg.data, np.uint8)
        image = cv2.imdecode(arr, cv2.IMREAD_COLOR)

        stamp = msg.header.stamp
        timestamp = "{}.{}".format(stamp.secs, stamp.nsecs)
        filename = os.path.join(args.output, "image_{}.{}".format(timestamp, args.format))

        if not os.path.exists(args.output):
            os.makedirs(args.output)

        logging.info("Saving {}".format(filename))

        cv2.imwrite(filename, image)

    bag.close()


if __name__ == "__main__":
    main()