""" purpose: in order to analyze the data loss of bluetooth, a small code is needed. Guess that there are three type of data loss. In order to prove this idea, several tests are needed types of data loss: 1. packet loss: bluetooth notification does not need any negotiation. if there is any packet loss during the connection, it will not be lost forever 2. program delay causes loss in client(headstage): too many effort on data packaging, it would cause data delay 3. program delay causes loss in host(controller): ...todo and need to define Assume that we have lost several bytes data. We want to analyze the type of data lost. method: compare ramp data with time stamp, there are several conditions if data_delta * 1 / sampling_rate == time_delta: this should be packet loss else this should be program delay method to coped with the corresponding problem: 1. try 'Indication' to check data loss is reduced or not. 2. modified the procedure of data packaging 3. after the upper two problems are excluded, this problem should be left over """ import os.path import sys from collections import Counter from typing import List, Optional, Dict, Set, Any class Options: def __init__(self): self.skip = False self.round = 2 self.overflow_pow: Optional[int] = None # print options self.quiet = False # graph options self.graph: Optional[str] = None self.graph_log_scale = False self.graph_save_path: Optional[str] = None # fft graph options self.fft_x_log = False self.fft_y_norm = False self.fft_channel = 0 @staticmethod def print_help(): prg = sys.argv[0] print(prg, '[OPTIONS]', 'FILE', '...') print() print('OPTIONS:') print(' -h, --help : print help') print(' --skip : skip if FILE not found') print(' --round VALUE : floating number round') print(' --overflow VALUE : overflow number. 2^VALUE') print() print('OPTIONS (print):') print(' -q : do not print table') print() print('OPTIONS (graph):') print(' --graph TYPE : use matplotlib to generate graph. could be:') print(' ramp : ramp data, calculate value miss') print(' fft : sin wave data, calculate fft') print(' --graph-log-scale :') print(' --graph-save PATH : save graph file path') print() print('OPTIONS (graph=fft):') print(' --fft-channel CH : filter channel') print(' --fft-log : log x-axis') print(' --fft-norm : log x-axis normalize') print(' --fft [opt,...] : above') print() print('ARGUMENTS:') print(' FILE : txt data file') print() class Result: def __init__(self, data_file: str, options: Options): # data file information self.data_file = data_file self.date_time: Optional[str] = None self.device_name: Optional[str] = None self.device_address: Optional[str] = None self.parameter: Dict[str, str] = {} self.channel: Set[int] = set() self.total_duration: float = 0.0 # analysis data self.data_count = 0 self.time_delta_counter = Counter() self.value_delta_counter = Counter() # analysis options self.round = options.round self.overflow: Optional[int] = None if options.overflow_pow is not None: self.overflow = 2 ** options.overflow_pow # temp data self._prev_time: Optional[float] = None self._prev_value: Optional[int] = None self._close = False @property def sample_rate(self) -> int: try: sample_rate = self.parameter['SAMPLE_RATE'] except KeyError as e: raise RuntimeError('data file do not contain SAMPLE_RATE') from e try: return int(sample_rate) except ValueError as e: raise RuntimeError('data file SAMPLE_RATE value not a value : ' + str(sample_rate)) from e def update(self, time: float, channel: int, value: int): if self._close: raise RuntimeError('closed') self.data_count += 1 if self._prev_time is not None: delta = round(time - self._prev_time, self.round) self.time_delta_counter[delta] += 1 if self._prev_value is not None: delta = value - self._prev_value self.value_delta_counter[delta] += 1 self._prev_time = time self.channel.add(channel) self._prev_value = value self.total_duration = time def close(self): self._close = True if self.overflow is not None: self.value_delta_counter[1] += self.value_delta_counter[1 - self.overflow] del self.value_delta_counter[1 - self.overflow] i = 2 while i - self.overflow in self.value_delta_counter: self.value_delta_counter[i] += self.value_delta_counter[i - self.overflow] del self.value_delta_counter[i - self.overflow] i += 1 def calculate(txt_file: str, options: Options) -> Result: if not os.path.exists(txt_file): raise FileNotFoundError(txt_file) result = Result(txt_file, options) parsing_state = 0 with open(txt_file) as _file: for line, content in enumerate(_file): # type: int, str content = content.strip() if len(content) == 0: continue if parsing_state == 0: if content.startswith('#'): if result.date_time is None: result.date_time = content[1:].strip() elif content.startswith('# device_name'): result.device_name = content[len('# device_name'):].strip() elif content.startswith('# mac_address'): result.device_address = content[len('# mac_address'):].strip() elif content.startswith('# parameter'): parsing_state = 1 elif parsing_state == 1: if content.startswith('#'): if content.startswith('# time_stamp'): parsing_state = 2 else: part = content[2:].split(' ', maxsplit=2) result.parameter[part[0]] = part[1] elif parsing_state == 2: if content.startswith('#'): continue part = content.split(' ', maxsplit=3) try: time = part[0] channel = part[1] value = part[2] except IndexError: print('@%d' % (line + 1), 'incomplete data', ':', content) continue check_pass = True try: time = float(time) except ValueError: print('@%d' % (line + 1), 'incomplete time', ':', time) check_pass = False try: channel = int(channel) except ValueError: print('@%d' % (line + 1), 'incomplete channel', ':', channel) check_pass = False try: value = int(value) except ValueError: print('@%d' % (line + 1), 'incomplete value', ':', value) check_pass = False if check_pass: result.update(time, channel, value) result.close() return result def print_result(result: Result, options: Options): def print_table(counter: Dict[Any, Any], key_format='%10d', value_format='%d'): f = ' %s : %s' % (key_format, value_format) for k in sorted(counter.keys()): print(f % (k, counter[k])) if not options.quiet: print(result.data_file) print('data_count', result.data_count) print('channel', ' '.join(list(map(str, sorted(result.channel))))) print() inv_map = {v: k for k, v in result.time_delta_counter.items()} print('data loss rate', 1 - max(inv_map.keys()) * inv_map[max(inv_map.keys())] / result.total_duration) try: sample_rate = result.sample_rate print('sample_rate [1/s]', sample_rate) print('1/sample_rate [ms]', 1000 / sample_rate) except RuntimeError: print('sample_rate', '(E)') print('time_delta') print_table(result.time_delta_counter, '%10.2f') print() print('value_delta') print_table(result.value_delta_counter) print() # packet miss rate # if options.graph == 'ramp': delta_value = list(filter(lambda it: it > 0, result.value_delta_counter)) total = sum(map(lambda it: it * result.value_delta_counter[it], delta_value)) miss = sum(map(lambda it: (it - 1) * result.value_delta_counter[it], delta_value)) print('value_miss (ramp)', '%d/%d' % (miss, total), '%.2f%%' % (100 * miss / total)) valid_time = list(filter(lambda it: it < 50, result.time_delta_counter)) total_valid_t = sum(map(lambda it: it * result.time_delta_counter[it], valid_time)) print('time_waste (ramp)', '%d/%d' % (total_valid_t, result.total_duration), '%.2f%%' % ( 100 * total_valid_t / result.total_duration)) timesofwaste = list(filter(lambda it: it > 50, result.time_delta_counter)) timesintotal = sum(map(lambda it: result.time_delta_counter[it], timesofwaste)) print('waste time occur times (ramp)', '%d' % timesintotal) print() paint_ramp_result(result, options) elif options.graph == 'fft': paint_fft_result(result, options) def calculate_(result: Result): print(max(result)) def paint_fft_result(result: Result, options: Options): import matplotlib.pyplot as plt import numpy as np from scipy import signal from scipy.interpolate import interp1d from scipy.fftpack import fft time_seq = [] value_seq = [] time_offset = None total_duration = 0.0 with open(result.data_file) as _file: for line in _file: line = line.strip() if len(line) == 0 or line.startswith('#'): continue else: part = line.split(' ', maxsplit=3) channel = int(part[1]) if channel == options.fft_channel: time = float(part[0]) if time_offset is None: time_seq.append(0.0) time_offset = time else: total_duration = time - time_offset time_seq.append(time - time_offset) value_seq.append(int(part[2])) if len(time_seq) == 0: raise RuntimeError('empty channel data : ' + str(options.fft_channel)) plt.figure(1) f1, pxx_den = signal.periodogram(value_seq, 1e3) plt.semilogy(f1, pxx_den) plt.ylim([1e-7, 1e7]) plt.xlabel('frequency [hz]') plt.ylabel('PSD[v**2/Hz') plt.figure(2) sample_rate = result.sample_rate time_delta = 1 / sample_rate total_time_step = int(total_duration / (time_delta * 1000)) f = interp1d(time_seq, value_seq, copy=False) t = np.linspace(0, total_duration, total_time_step) fft_result = fft(f(t)) fft_result = fft_result[:len(fft_result) // 2] fft_x = np.linspace(0, 1.0/(2.0*time_delta), len(fft_result)//2) if options.fft_x_log: fft_x = np.log10(fft_x) if options.fft_y_norm: fft_result /= sum(fft_result) plt.plot(fft_x, 2.0 / len(fft_result) * np.abs(fft_result[0:len(fft_result)//2])) if options.graph_save_path is not None: plt.savefig(options.graph_save_path, dpi=600) else: plt.show() def paint_ramp_result(result: Result, options: Options): import matplotlib.pyplot as plt fig, (at, av) = plt.subplots(2, 1) time_delta_data = list(result.time_delta_counter) expect_time_delta = 1000 / result.sample_rate time_delta_bins = int((max(time_delta_data) - min(time_delta_data)) / expect_time_delta) time_delta_weight = [result.time_delta_counter[k] for k in time_delta_data] at.hist(time_delta_data, bins=time_delta_bins, weights=time_delta_weight, log=options.graph_log_scale, align='left') value_delta_data = list(result.value_delta_counter) value_delta_bins = max(value_delta_data) - min(value_delta_data) value_delta_weight = [result.value_delta_counter[k] for k in value_delta_data] av.hist(value_delta_data, bins=value_delta_bins, weights=value_delta_weight, log=options.graph_log_scale, align='left') fig.tight_layout() if options.graph_save_path is not None: plt.savefig(options.graph_save_path, dpi=600) else: plt.show() def main(argv: List[str]): """ :param argv: :return: """ # df = pd.read_csv("C:/Users/yichin/Downloads/last-2019-07-18-16-30-15-7.csv", sep=",", skiprows=12, header=None) # sel_df = pd.DataFrame(df) # column, row = sel_df.shape # sel_df = sel_df.fillna(value=0) # delta_df = sel_df.diff(periods=1) # delta_df = delta_df.fillna(value=0) # delta_df = delta_df.drop(column - 1) # delta_df.set_axis(['time_delta', 'data_delta'], axis='columns', inplace=True) o = Options() i = 0 while i < len(argv): arg = argv[i] if arg in ('-h', '--help'): o.print_help() return elif arg == '--skip': o.skip = True i += 1 elif arg == '--round': o.round = int(argv[i + 1]) i += 2 elif arg.startswith('--round='): o.round = int(arg[len('--round='):]) i += 1 elif arg == '--overflow': o.overflow_pow = int(argv[i + 1]) i += 2 elif arg.startswith('--overflow='): o.overflow_pow = int(arg[len('--overflow='):]) i += 1 elif arg == '-q': o.quiet = True i += 1 elif arg == '--graph': o.graph = argv[i + 1] i += 2 elif arg.startswith('--graph='): o.graph = arg[len('--graph='):] i += 1 elif arg == '--graph-log-scale': o.graph_log_scale = True i += 1 elif arg == '--graph-save': o.graph_save_path = argv[i + 1] i += 2 elif arg.startswith('--graph-save='): o.graph = arg[len('--graph-save='):] i += 1 elif arg == '--fft-channel': o.fft_channel = int(argv[i + 1]) i += 2 elif arg.startswith('--fft-channel='): o.fft_channel = int(arg[len('--fft-channel='):]) i += 1 elif arg == '--fft-log': o.fft_x_log = True i += 1 elif arg == '--fft-norm': o.fft_y_norm = True i += 1 elif arg.startswith('--fft'): if arg == '--fft': c = argv[i + 1] i += 2 elif arg.startswith('--fft='): c = arg[len('--fft='):] i += 1 else: raise ValueError('unknown options : ' + arg) for a in c.split(','): if a == 'log': o.fft_x_log = True elif a == 'norm': o.fft_y_norm = True elif a.startswith('channel='): o.fft_channel = int(a[len('channel='):]) else: raise ValueError('unknown options : --fft-' + a) elif arg.startswith('-'): raise ValueError('unknown options : ' + arg) else: i += 1 try: result = calculate(arg, o) except FileNotFoundError: if o.skip: continue else: raise else: print_result(result, o) if __name__ == '__main__': main(sys.argv[1:])