How to use nfstream - 10 common examples

def test_no_unknown_protocols_without_timeouts(self):
        files = get_files_list("tests/pcap/")
        ground_truth_ndpi = build_ground_truth_dict("tests/result/")
        print("\n----------------------------------------------------------------------")
        print(".Testing on {} applications:".format(len(files)))
        ok_files = []
        ko_files = []
        for test_file in files:
            streamer_test = NFStreamer(source=test_file, idle_timeout=31556952, active_timeout=31556952)
            test_case_name = test_file.split('/')[-1]
            result = {}
            for flow in streamer_test:
                if flow.application_name != 'Unknown':
                    try:
                        result[flow.application_name]['bytes'] += flow.bidirectional_raw_bytes
                        result[flow.application_name]['flows'] += 1
                        result[flow.application_name]['pkts'] += flow.bidirectional_packets
                    except KeyError:
                        result[flow.application_name] = {"bytes": flow.bidirectional_raw_bytes,
                                                         'flows': 1, 'pkts': flow.bidirectional_packets}
            if result == ground_truth_ndpi[test_case_name]:
                ok_files.append(test_case_name)
                print("{}\t: \033[94mOK\033[0m".format(test_case_name.ljust(60, ' ')))
            else:
                ko_files.append(test_case_name)

def test_flow_metadata_extraction(self):
        print("\n----------------------------------------------------------------------")
        streamer_test = NFStreamer(source='tests/pcap/facebook.pcap', bpf_filter="src port 52066 or dst port 52066")
        flows = []
        for flow in streamer_test:
            flows.append(flow)
        del streamer_test
        self.assertEqual(flows[0].client_info, 'facebook.com')
        self.assertEqual(flows[0].server_info, '*.facebook.com,*.facebook.net,*.fb.com,*.fbcdn.net,*.fbsbx.com,\
*.m.facebook.com,*.messenger.com,*.xx.fbcdn.net,*.xy.fbcdn.net,*.xz.fbcdn.net,facebook.com,fb.com,\
messenger.com')
        self.assertEqual(flows[0].client_info, 'facebook.com')
        self.assertEqual(flows[0].j3a_client, 'bfcc1a3891601edb4f137ab7ab25b840')
        self.assertEqual(flows[0].j3a_server, '2d1eb5817ece335c24904f516ad5da12')
        print("{}\t: \033[94mOK\033[0m".format(".Testing metadata extraction".ljust(60, ' ')))

def test_custom_expiration(self):
        class custom_expire(NFPlugin):
            def on_update(self, obs, entry):
                if entry.bidirectional_packets == 10:
                    entry.expiration_id = -1
                    entry.custom_expire = True

        print("\n----------------------------------------------------------------------")
        streamer_test = NFStreamer(source='tests/pcap/facebook.pcap',
                                   plugins=[custom_expire(volatile=True)],
                                   bpf_filter="src port 52066 or dst port 52066")
        rs = []
        for flow in streamer_test:
            rs.append(flow)
        self.assertEqual(rs[0].expiration_id, -1)
        self.assertEqual(len(rs), 2)
        del streamer_test
        print("{}\t: \033[94mOK\033[0m".format(".Testing custom expiration".ljust(60, ' ')))

def test_to_pandas_anonymized(self):
        print("\n----------------------------------------------------------------------")
        df = NFStreamer(source='tests/pcap/ethereum.pcap',
                        idle_timeout=31556952,
                        active_timeout=31556952).to_pandas(ip_anonymization=True)
        self.assertEqual(df.shape[0], 74)
        self.assertEqual(df.shape[1], 37)
        print("{}\t: \033[94mOK\033[0m".format(".Testing to Pandas ip_anonymization=True".ljust(60, ' ')))

def test_expiration_management(self):
        print("\n----------------------------------------------------------------------")
        streamer_test = NFStreamer(source='tests/pcap/facebook.pcap', active_timeout=0)
        flows = []
        for flow in streamer_test:
            flows.append(flow)
        self.assertEqual(len(flows), 60)
        print("{}\t: \033[94mOK\033[0m".format(".Testing Streamer expiration management".ljust(60, ' ')))

def test_bpf_filter(self):
        print("\n----------------------------------------------------------------------")
        streamer_test = NFStreamer(source='tests/pcap/facebook.pcap',
                                   statistics=True,
                                   bpf_filter="src port 52066 or dst port 52066")
        count = 0
        for flow in streamer_test:
            print(flow)
            print(flow.to_namedtuple())
            print(flow.to_json())
            count = count + 1
            self.assertEqual(flow.src_port, 52066)
        self.assertEqual(count, 1)
        del streamer_test
        print("{}\t: \033[94mOK\033[0m".format(".Testing BPF filtering".ljust(60, ' ')))

def test_custom_expiration(self):
        class custom_expire(NFPlugin):
            def on_update(self, obs, entry):
                if entry.bidirectional_packets == 10:
                    entry.expiration_id = -1
                    entry.custom_expire = True

        print("\n----------------------------------------------------------------------")
        streamer_test = NFStreamer(source='tests/pcap/facebook.pcap',
                                   plugins=[custom_expire(volatile=True)],
                                   bpf_filter="src port 52066 or dst port 52066")
        rs = []
        for flow in streamer_test:
            rs.append(flow)
        self.assertEqual(rs[0].expiration_id, -1)
        self.assertEqual(len(rs), 2)
        del streamer_test
        print("{}\t: \033[94mOK\033[0m".format(".Testing custom expiration".ljust(60, ' ')))

def test_user_plugins(self):
        class feat_1(NFPlugin):
            def on_update(self, obs, entry):
                if entry.bidirectional_packets == 4:
                    entry.feat_1 = obs.ip_size

        print("\n----------------------------------------------------------------------")
        streamer_test = NFStreamer(source='tests/pcap/facebook.pcap',
                                   plugins=[feat_1()],
                                   bpf_filter="src port 52066 or dst port 52066")
        rs = []
        for flow in streamer_test:
            rs.append(flow)
        self.assertEqual(rs[0].feat_1, 248)
        self.assertEqual(len(rs), 1)
        del streamer_test
        print("{}\t: \033[94mOK\033[0m".format(".Testing adding user plugins".ljust(60, ' ')))

class src2dst_max_piat_ms(NFPlugin):
    """ Flow src -> dst maximum packet inter arrival time """
    def on_init(self, obs):
        return -1  # we will set it as -1 as init value

    def on_update(self, obs, entry):
        if obs.direction == 0:
            if entry.src2dst_max_piat_ms == -1 and entry.src2dst_piat[0] >= 0:
                entry.src2dst_max_piat_ms = entry.src2dst_piat[0]
            if entry.src2dst_piat[0] > entry.src2dst_max_piat_ms:
                entry.src2dst_max_piat_ms = entry.src2dst_piat[0]


class src2dst_min_piat_ms(NFPlugin):
    """ Flow src -> dst minimum packet inter arrival time """
    def on_init(self, obs):
        return -1  # we will set it as -1 as init value

    def on_update(self, obs, entry):
        if obs.direction == 0:
            if entry.src2dst_min_piat_ms == -1 and entry.src2dst_piat[0] >= 0:
                entry.src2dst_min_piat_ms = entry.src2dst_piat[0]
            if entry.src2dst_piat[0] < entry.src2dst_min_piat_ms:
                entry.src2dst_min_piat_ms = entry.src2dst_piat[0]


class src2dst_weldord_piat_ms(NFPlugin):
    """ Flow src -> dst packet inter arrival time welford algorithm structure (volatile) """
    def on_init(self, obs):
        return [0, 0, 0]

entry.dst2src_min_piat_ms = entry.dst2src_piat[0]
            if entry.dst2src_piat[0] < entry.dst2src_min_piat_ms:
                entry.dst2src_min_piat_ms = entry.dst2src_piat[0]


class bidirectional_min_raw_ps(NFPlugin):
    """ Flow bidirectional minimum raw packet size """
    def on_init(self, obs):
        return obs.raw_size

    def on_update(self, obs, entry):
        if obs.raw_size < entry.bidirectional_min_raw_ps:
            entry.bidirectional_min_raw_ps = obs.raw_size


class bidirectional_weldord_raw_ps(NFPlugin):
    """ Flow bidirectional raw packet size welford algorithm structure (volatile)"""
    def on_init(self, obs):
        return [1, obs.raw_size, 0]

    def on_update(self, obs, entry):
        k = entry.bidirectional_weldord_raw_ps[0] + 1
        entry.bidirectional_weldord_raw_ps[0] = k
        m = entry.bidirectional_weldord_raw_ps[1]
        s = entry.bidirectional_weldord_raw_ps[2]
        entry.bidirectional_weldord_raw_ps[1] = \
            m + (obs.raw_size - m) * 1. / entry.bidirectional_weldord_raw_ps[0]
        entry.bidirectional_weldord_raw_ps[2] = \
            s + (obs.raw_size - m) * (obs.raw_size - entry.bidirectional_weldord_raw_ps[1])


class bidirectional_mean_raw_ps(NFPlugin):