def add()

in src/rosbridge/timeseq.py [0:0]

• 17 lines of code
• 7 McCabe index (conditional complexity)

    def add(self, ts, channel_name, sample):
        try:
            if self.is_open:
                if self.begin_ts is None: self.begin_ts = ts
                if ts > self.end_ts: self.end_ts = ts

                chunk = chunk_of(ts)
                if self.chunks_by_channel.get(channel_name, -1) != chunk:
                    self.close_chunk(channel_name)
                    self.open_chunk(channel_name, chunk)
                f = self.files_by_channel[channel_name]
                if len(self.times_by_channel[channel_name]) > 0:
                    f.write(',')
                ujson.dump(sample, f)
                self.times_by_channel[channel_name].append(ts)
        except OverflowError:
            logger.info("Failed to serialize sample. Might it have numpy arrays with float32s? Please convert numpy arrays to float64")
            raise