source: trunk/bin/train @ 4

#!/usr/bin/env python3
#-*- mode: Python;-*-

import sys
import os
import time
import random
import statistics
import functools
import argparse
import threading
import queue
import pprint
import json


VERSION = "{DEVELOPMENT}"
if VERSION == "{DEVELOPMENT}":
    script_dir = '.'
    try:
        script_dir = os.path.dirname(os.path.realpath(__file__))
    except:
        try:
            script_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
        except:
            pass
    sys.path.append("%s/../lib" % script_dir)

from nanownlib import *
import nanownlib.storage
from nanownlib.stats import boxTest,multiBoxTest,subsample,bootstrap,bootstrap2,trimean,midhinge,midhingeTest,samples2Distributions,samples2MeanDiffs

parser = argparse.ArgumentParser(
    description="")
#parser.add_argument('-c', dest='cases', type=str, default='{"short":10000,"long":1010000}',
#                    help='JSON representation of echo timing cases. Default: {"short":10000,"long":1010000}')
parser.add_argument('session_data', default=None,
                    help='Database file storing session information')
options = parser.parse_args()



class WorkerThreads(object):
    workq = None
    resultq = None
    target = None
    
    def __init__(self, num_workers, target):
        self.workq = queue.Queue()
        self.resultq = queue.Queue()
        self.target = target
        
        self.workers = []
        for i in range(num_workers):
            t = threading.Thread(target=self._worker)
            t.daemon = True
            t.start()
            self.workers.append(t)

    def _worker(self):
        while True:
            item = self.workq.get()
            if item == None:
                self.workq.task_done()
                break

            job_id,args = item
            self.resultq.put((job_id, self.target(*args)))
            self.workq.task_done()

    def addJob(self, job_id, args):
        self.workq.put((job_id, args))
            
    def wait(self):
        self.workq.join()

    def stop(self):
        for i in range(0,len(workers)):
            self.workq.put(None)
        for w in self.workers:
            w.join()

            
def trainBoxTest(db, test_cases, longest, subsample_size):

    def trainAux(low,high,num_trials):
        estimator = functools.partial(boxTest, {'low':low, 'high':high})
        estimates = bootstrap(estimator, db, 'train', test_cases, subsample_size, num_trials)
        null_estimates = bootstrap(estimator, db, 'train_null', test_cases, subsample_size, num_trials)

        #XXX: need to have a configurable policy on what we're looking for.
        #     which is longest or which is shortest?
        bad_estimates = len([e for e in estimates if e != longest])
        bad_null_estimates = len([e for e in null_estimates if e != None])
        
        false_negatives = 100.0*bad_estimates/num_trials
        false_positives = 100.0*bad_null_estimates/num_trials
        return false_positives,false_negatives

    start = time.time()
    wt = WorkerThreads(2, trainAux)
    
    width = 2.0
    performance = []
    percentiles = list(range(0,50,2))
    for low in percentiles:
        wt.addJob(low, (low,low+width,200))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    print(time.time()-start)
    
    lows = [p[1] for p in performance[0:5]]
    widths = [w/10.0 for w in range(0,105,5)]
    performance = []
    for width in widths:
        false_positives = []
        false_negatives = []
        for low in lows:
            wt.addJob(low,(low,low+width,150))
        wt.wait()
        while not wt.resultq.empty():
            job_id,errors = wt.resultq.get()
            fp,fn = errors
            false_negatives.append(fn)
            false_positives.append(fp)

        #print(width, false_negatives)
        #print(width, false_positives)
        performance.append(((statistics.mean(false_positives)+statistics.mean(false_negatives))/2.0,
                            width, statistics.mean(false_negatives), statistics.mean(false_positives)))
    performance.sort()
    pprint.pprint(performance)
    good_width = performance[0][1]
    print("good_width:",good_width)


    lc = {}
    for low in lows:
        if low-1 > 0:
            lc[low-1] = None
        lc[low] = None
        lc[low+1] = None
    lows = lc.keys()
    
    performance = []
    for low in lows:
        wt.addJob(low, (low,low+good_width,300))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    best_low = performance[0][1]
    print("best_low:", best_low)

    
    widths = [good_width-0.4,good_width-0.3,good_width-0.2,good_width-0.1,
              good_width,good_width+0.1,good_width+0.2,good_width+0.3,good_width+0.4]
    performance = []
    for width in widths:
        wt.addJob(width, (best_low,best_low+width,200))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    best_width=performance[0][1]
    print("best_width:",best_width)
    print("final_performance:", performance[0][0])

    return {"low":best_low,"high":best_low+good_width}


def trainBoxTest2(db, unusual_case, greater, subsample_size):

    def trainAux(low,high,num_trials):
        estimator = functools.partial(multiBoxTest, {'low':low, 'high':high}, unusual_case, greater)
        estimates = bootstrap2(estimator, db, 'train', subsample_size, num_trials)
        null_estimates = bootstrap2(estimator, db, 'train_null', subsample_size, num_trials)

        bad_estimates = len([e for e in estimates if e != 1])
        bad_null_estimates = len([e for e in null_estimates if e != 0])
        
        false_negatives = 100.0*bad_estimates/num_trials
        false_positives = 100.0*bad_null_estimates/num_trials
        return false_positives,false_negatives

    start = time.time()
    wt = WorkerThreads(2, trainAux)
    
    num_trials = 200
    width = 2.0
    performance = []
    percentiles = list(range(0,50,2))
    for low in percentiles:
        wt.addJob(low, (low,low+width,num_trials))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    print(time.time()-start)
    
    num_trials = 150
    lows = [p[1] for p in performance[0:5]]
    widths = [w/10.0 for w in range(0,105,5)]
    performance = []
    for width in widths:
        false_positives = []
        false_negatives = []
        for low in lows:
            wt.addJob(low,(low,low+width,num_trials))
        wt.wait()
        while not wt.resultq.empty():
            job_id,errors = wt.resultq.get()
            fp,fn = errors
            false_negatives.append(fn)
            false_positives.append(fp)

        #print(width, false_negatives)
        #print(width, false_positives)
        performance.append(((statistics.mean(false_positives)+statistics.mean(false_negatives))/2.0,
                            width, statistics.mean(false_negatives), statistics.mean(false_positives)))
    performance.sort()
    pprint.pprint(performance)
    good_width = performance[0][1]
    print("good_width:",good_width)


    lc = {}
    for low in lows:
        if low-1 >= 0:
            lc[low-1] = None
        lc[low] = None
        lc[low+1] = None
    lows = lc.keys()
    print("candidate lows:")
    pprint.pprint(lows)
    
    num_trials = 300
    performance = []
    for low in lows:
        wt.addJob(low, (low,low+good_width,num_trials))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    best_low = performance[0][1]
    print("best_low:", best_low)

    num_trials = 200
    widths = [good_width-0.4,good_width-0.3,good_width-0.2,good_width-0.1,
              good_width,good_width+0.1,good_width+0.2,good_width+0.3,good_width+0.4]
    performance = []
    for width in widths:
        wt.addJob(width, (best_low,best_low+width,num_trials))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    best_width=performance[0][1]
    print("best_width:",best_width)
    print("final_performance:", performance[0][0])
    
    params = json.dumps({"low":best_low,"high":best_low+good_width})
    return {'algorithm':"boxtest",
            'params':params,
            'sample_size':subsample_size,
            'num_trials':num_trials,
            'trial_type':"train",
            'false_positives':performance[0][3],
            'false_negatives':performance[0][2]}


def trainMidhinge(db, unusual_case, greater, subsample_size):

    def trainAux(distance, threshold, num_trials):
        estimator = functools.partial(midhingeTest, {'distance':distance,'threshold':threshold}, unusual_case, greater)
        estimates = bootstrap2(estimator, db, 'train', subsample_size, num_trials)
        null_estimates = bootstrap2(estimator, db, 'train_null', subsample_size, num_trials)

        bad_estimates = len([e for e in estimates if e != 1])
        bad_null_estimates = len([e for e in null_estimates if e != 0])
        
        false_negatives = 100.0*bad_estimates/num_trials
        false_positives = 100.0*bad_null_estimates/num_trials
        return false_positives,false_negatives

    #determine expected delta based on differences
    start = time.time()
    mean_diffs = list(samples2MeanDiffs(subsample(db, 'train'), 'packet_rtt', unusual_case))
    threshold = trimean(mean_diffs)/2.0
    print("initial threshold:", threshold)
    print("median threshold:", statistics.median(mean_diffs)/2.0)
    print("midhinge threshold:", midhinge(mean_diffs)/2.0)
    print("trimean threshold:", trimean(mean_diffs)/2.0)
    
    mean_diffs = list(samples2MeanDiffs(subsample(db, 'train_null'), 'packet_rtt', unusual_case))
    print(len(mean_diffs))
    print("null mean:", statistics.mean(mean_diffs))
    print("null median:", statistics.median(mean_diffs))
    print("null midhinge:", midhinge(mean_diffs))
    print("null trimean:", trimean(mean_diffs))
    print(time.time()-start)

    
    start = time.time()
    wt = WorkerThreads(1, trainAux)
    
    num_trials = 200
    performance = []
    #for distance in range(1,46,4):
    for distance in range(25,46,4):
        wt.addJob(distance, (distance,threshold,num_trials))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    print(time.time()-start)
    good_distance = performance[0][1]
    print("good_distance:",good_distance)

    
    num_trials = 200
    start = time.time()
    performance = []
    for t in range(80,125,5):
        wt.addJob(threshold*(t/100.0), (good_distance,threshold*(t/100.0),num_trials))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    print(time.time()-start)
    good_threshold = performance[0][1]
    print("good_threshold:", good_threshold)

    
    num_trials = 200
    start = time.time()
    performance = []
    for d in range(-4,5):
        wt.addJob(good_distance+d, (good_distance+d,good_threshold,num_trials))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    print(time.time()-start)
    best_distance = performance[0][1]
    print("best_distance:",best_distance)

    num_trials = 200
    start = time.time()
    performance = []
    for t in range(95,106):
        wt.addJob(good_threshold*(t/100.0), (best_distance,good_threshold*(t/100.0),num_trials))
    wt.wait()
    while not wt.resultq.empty():
        job_id,errors = wt.resultq.get()
        fp,fn = errors
        performance.append(((fp+fn)/2.0, job_id, fn, fp))
    performance.sort()
    pprint.pprint(performance)
    print(time.time()-start)
    best_threshold = performance[0][1]
    print("best_threshold:", best_threshold)

    params = json.dumps({'distance':best_distance,'threshold':best_threshold})
    return {'algorithm':"midhinge",
            'params':params,
            'sample_size':subsample_size,
            'num_trials':num_trials,
            'trial_type':"train",
            'false_positives':performance[0][3],
            'false_negatives':performance[0][2]}


#classifiers = {'boxtest':{'train':trainBoxTest2, 'test':multiBoxTest},
#               'midhinge':{'train':trainMidhinge, 'test':midhinge}}


db = nanownlib.storage.db(options.session_data)
#cursor = db.cursor()
#cursor.execute("SELECT min(sample) min, max(sample) max FROM probes")
#train_start,test_end = cursor.fetchone()
#train_end = int(test_end-train_start)
#test_start = train_end+1
#subsample_size = min(10000,(train_end-train_start+1)/4)

start = time.time()
unusual_case,unusual_diff = findUnusualTestCase(db)
greater = (unusual_diff > 0)
print("unusual_case:", unusual_case)
print("unusual_diff:", unusual_diff)
end = time.time()
print(":", end-start)

start = time.time()
results = trainMidhinge(db, unusual_case, greater, 6000)
db.addClassifierResults(results)
print("midhinge result:", results)
end = time.time()
print(":", end-start)

start = time.time()
results = trainBoxTest2(db, unusual_case, greater, 6000)
db.addClassifierResults(results)
print("multi box test result:", results)
end = time.time()
print(":", end-start)

#start = time.time()
#print("box test params:", trainBoxTest(db, test_cases, 'long', 100))
#end = time.time()
#print(":", end-start)