past-project-past_code_cc2650/python_test_code/uni/briefly_data_analysis.py

"""
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, Union


class Options:
    def __init__(self):
        self.skip = False
        self.round = 2
        self.overflow_pow = 10
        self.graph: Union[bool, str] = False
        self.graph_log_scale = False


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: Dict[float, int] = Counter()
        self.value_delta_counter: Dict[int, int] = Counter()

        # analysis options
        self.round = options.round
        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 = self._prev_value - value
            self.value_delta_counter[delta] += 1

        self._prev_time = time
        self.channel.add(channel)
        self._prev_value = value

    def close(self):
        self._close = True


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]))

    print(result.data_file)
    print('data_count', result.data_count)
    print('channel', ' '.join(list(map(str, sorted(result.channel)))))

    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('value_delta')
    print_table(result.value_delta_counter)

    print()


def paint_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()
    plt.show()


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('  --graph[=FILE]    : use matplotlib to generate graph')
    print('  --graph-log-scale :')
    print()
    print('ARGUMENTS:')
    print('  FILE           : txt data file')
    print()


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'):
            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 == '--graph':
            o.graph = True
            i += 1

        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.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 o.graph:
                    paint_result(result, o)


if __name__ == '__main__':
    main(sys.argv[1:])