def __flush_bag()

in a2d2/src/rosbag_producer.py [0:0]

• 34 lines of code
• 11 McCabe index (conditional complexity)

    def __flush_bag(self):
        try:
            if self.bag_lock:
                self.bag_lock.acquire()

            _ntopics = len(self.topic_list)
              
            msg = None
            ts = None
            topic = None

            _ntopics_flushed = 0
            # rotate through topics and flush them to bag
            self.logger.info("Flushing  ROS topics to bag")
            for _ in range(_ntopics):
                self.topic_index = (self.topic_index + 1) % _ntopics
                _topic = self.topic_list[ self.topic_index ]
                
                if self.topic_dict[_topic]:
                    front = self.topic_dict[_topic].pop(0)
                    msg = front.msg
                    ts = front.ts
                    topic = _topic
                else:
                    _ntopics_flushed += 1

                if topic and msg and ts:
                    self.__write_ros_msg_to_bag(topic=topic, msg=msg, ts=ts, s3_client=None, flush=True)
                
                if _ntopics == _ntopics_flushed:
                    break
            self.logger.info("Flushed ROS topics to bag")

        except Exception as _:
            exc_type, exc_value, exc_traceback = sys.exc_info()
            traceback.print_tb(exc_traceback, limit=20, file=sys.stdout)
            self.logger.error(str(exc_type))
            self.logger.error(str(exc_value))
            raise
        finally:    
            if self.bag_lock:
                self.bag_lock.release()