Source code for obci.utils.streaming_debug

#!/usr/bin/env python3
#
# Author:
#      Mateusz Kruszynski <mateusz.kruszynski@gmail.com>

"""
Use to debug real-life streaming modules like:
- amplifier
- streamer
- filter
See signal_streamer_no_filter.py for sample use.
"""
import time


[docs]class Debug(object): def __init__(self, p_sampling, logger, per=1): """By now init externally given logger (python standar logger object) and sampling.""" self.num_of_samples = 0 self.sampling = p_sampling self.logger = logger self.per = per
[docs] def next_sample(self): """Called after every new sample received. Aftet sel.sampling sample print stats info.""" if self.num_of_samples == 0: self.start_time = time.time() self.last_pack_first_sample_ts = time.time() pre_rest = self.num_of_samples % self.sampling self.num_of_samples += self.per rest = self.num_of_samples % self.sampling if pre_rest > rest: self.logger.info(''.join( ["Time of last ", str(self.sampling), " samples / all avg: ", str(time.time() - self.last_pack_first_sample_ts), ' / ', str(self.sampling * (time.time() - self.start_time) / float(self.num_of_samples))])) self.last_pack_first_sample_ts = time.time()
[docs] def next_sample_timestamp(self, sample_timestamp): if self.num_of_samples == 0: self.first_sample_timestamp = sample_timestamp self.num_of_samples += 1 rest = self.num_of_samples % self.sampling if rest == 0: self.logger.info(''.join( ["Time of last ", str(self.sampling), " samples / all avg: ", str(sample_timestamp - self.first_st_in_pack), ' / ', str(self.sampling * (sample_timestamp - self.first_sample_timestamp) / float(self.num_of_samples))])) elif rest == 1: self.first_st_in_pack = sample_timestamp