Changeset 11 for trunk/bin

Timestamp:

07/16/15 20:40:01 (9 years ago)

Author:

tim

Message:

.

Location:

trunk/bin

Files:

• graph (modified) (4 diffs)
• train (modified) (7 diffs)

trunk/bin/graph

@@ -40 +40 @@
 
 
-def differences(db, unusual_case, column='packet_rtt'):
-    cursor = db.conn.cursor()
-    query="""
-      SELECT %(column)s-(SELECT avg(%(column)s) FROM probes,analysis
-                         WHERE analysis.probe_id=probes.id AND probes.test_case!=:unusual_case AND probes.type in ('train','test') AND sample=u.sample)
-      FROM (SELECT probes.sample,%(column)s FROM probes,analysis 
-                         WHERE analysis.probe_id=probes.id AND probes.test_case =:unusual_case AND probes.type in ('train','test')) u
-      """ % {"column":column}
-    params = {"unusual_case":unusual_case}
-    cursor.execute(query, params)
-    for row in cursor:
-        yield row[0]
+def differences(db, unusual_case, rtt_type='packet'):
+    ret_val = [s['unusual_'+rtt_type]-s['other_'+rtt_type] for s in db.subseries('train', unusual_case)]
+    ret_val += [s['unusual_'+rtt_type]-s['other_'+rtt_type] for s in db.subseries('test', unusual_case)]
+    return ret_val
+
+def null_differences(db, unusual_case, rtt_type='packet'):
+    ret_val = [s['unusual_'+rtt_type]-s['other_'+rtt_type] for s in db.subseries('train_null', unusual_case)]
+    return ret_val
 
 
@@ -91 +87 @@
 print('packet_rtt diff ubersummary: %f' % ubersummary(diffs))
 print('packet_rtt diff MAD: %f' % mad(diffs))
-print('reported diff trimean: %f' % trimean(reported_diffs))
-print('reported diff quadsummary: %f' % quadsummary(reported_diffs))
-print('reported diff ubersummary: %f' % ubersummary(reported_diffs))
-print('reported diff MAD: %f' % mad(reported_diffs))
-
-import cProfile
-kresults = kfilter({},diffs)
-#print('packet_rtt diff kfilter: ', numpy.mean(kresults['est']), kresults['var'])
-print('packet_rtt diff kfilter: ', kresults['est'][-1], kresults['var'][-1])
-kresults = kfilter({},reported_diffs)
-#print('reported diff kfilter: ', numpy.mean(kresults['est']), kresults['var'][-1])
-print('reported diff kfilter: ', kresults['est'][-1], kresults['var'][-1])
+try:
+    print('reported diff trimean: %f' % trimean(reported_diffs))
+    print('reported diff quadsummary: %f' % quadsummary(reported_diffs))
+    print('reported diff ubersummary: %f' % ubersummary(reported_diffs))
+    print('reported diff MAD: %f' % mad(reported_diffs))
+
+    import cProfile
+    start = time.time()
+    kresults = kfilter({},diffs)
+    #print('packet_rtt diff kfilter: ', numpy.mean(kresults['est']), kresults['var'])
+    print('packet_rtt diff kfilter: ', kresults['est'][-1], kresults['var'][-1])
+    kresults = kfilter({},reported_diffs)
+    #print('reported diff kfilter: ', numpy.mean(kresults['est']), kresults['var'][-1])
+    print('reported diff kfilter: ', kresults['est'][-1], kresults['var'][-1])
+    print("kfilter time: %f" % (time.time()-start))
+except:
+    pass
+
+print('tsval diff mean: %f' % numpy.mean(differences(db, 'long', 'tsval')))
+print('tsval null diff mean: %f' % numpy.mean(null_differences(db, 'long', 'tsval')))
+print('tsval diff weighted mean: %f' % tsvalwmean(db.subseries('train','long')+db.subseries('test','long')))
+print('tsval null diff weighted mean: %f' % tsvalwmean(db.subseries('train_null','long')))
 
 
@@ -110 +116 @@
 #cut_off_high = all_data[int(len(all_data)*0.997)]
 
+
+def plotSingleProbe(probe_id=None):
+    if probe_id == None:
+        cursor = db.conn.cursor()
+        query="""SELECT probe_id FROM analysis WHERE suspect='' ORDER BY probe_id DESC limit 1 OFFSET 10"""
+        cursor.execute(query)
+        probe_id = cursor.fetchone()[0]
+    
+    cursor = db.conn.cursor()
+    query="""SELECT observed,payload_len FROM packets WHERE probe_id=? AND sent=1"""
+    cursor.execute(query, (probe_id,))
+    pkts = cursor.fetchall()
+    sent_payload = [row[0] for row in pkts if row[1] != 0]
+    sent_other = [row[0] for row in pkts if row[1] == 0]
+    
+    query="""SELECT observed,payload_len FROM packets WHERE probe_id=? AND sent=0"""
+    cursor.execute(query, (probe_id,))
+    pkts = cursor.fetchall()
+    rcvd_payload = [row[0] for row in pkts if row[1] != 0]
+    rcvd_other = [row[0] for row in pkts if row[1] == 0]
+    
+    #query="""SELECT reported,time_of_day FROM probes WHERE id=?"""
+    #cursor.execute(query, (probe_id,))
+    #reported,tod = cursor.fetchone()
+    #userspace_times = [sent_times[0]-reported/3.0, sent_times[0]+reported]
+
+    print("single probe counts:",len(sent_payload),len(sent_other),len(rcvd_payload),len(rcvd_other))
+    plt.clf()
+    plt.title("Single HTTP Request - Packet Times")
+    sp = plt.eventplot(sent_payload, colors=('red',), lineoffsets=8, linewidths=2, alpha=0.6,label='sent')
+    so = plt.eventplot(sent_other, colors=('red',), lineoffsets=6, linewidths=2, alpha=0.6,label='sent')
+    rp = plt.eventplot(rcvd_payload, colors=('blue',), lineoffsets=4, linewidths=2, alpha=0.6,label='received')
+    ro = plt.eventplot(rcvd_other, colors=('blue',), lineoffsets=2, linewidths=2, alpha=0.6,label='received')
+    #plt.legend((s,r), ('sent','received'))
+    #plt.savefig('../img/http-packet-times.svg')
+    plt.show()
+
+#plotSingleProbe()
+
+
+def graphTestResults():
+    plt.clf()
+    plt.title("Test Results")
+    plt.xlabel('sample size')
+    plt.ylabel('error rate')
+    legend = []
+    colors = ['red','blue','green','purple','orange','black','brown']
+    color_id = 0
+
+    cursor = db.conn.cursor()
+    query = """
+      SELECT classifier FROM classifier_results GROUP BY classifier ORDER BY classifier;
+    """
+    cursor.execute(query)
+    classifiers = []
+    for c in cursor:
+        classifiers.append(c[0])
+
+    for classifier in classifiers:
+        query="""
+        SELECT params FROM classifier_results 
+        WHERE trial_type='test'
+         AND classifier=:classifier
+         AND (false_positives+false_negatives)/2.0 < 5.0 
+        ORDER BY num_observations,(false_positives+false_negatives) 
+        LIMIT 1
+        """
+        cursor.execute(query, {'classifier':classifier})
+        row = cursor.fetchone()
+        if row == None:
+            query="""
+            SELECT params FROM classifier_results 
+            WHERE trial_type='test' and classifier=:classifier
+            ORDER BY (false_positives+false_negatives),num_observations
+            LIMIT 1
+            """
+            cursor.execute(query, {'classifier':classifier})
+            row = cursor.fetchone()
+            if row == None:
+                sys.stderr.write("WARN: couldn't find test results for classifier '%s'.\n" % classifier)
+                continue
+
+        best_params = row[0]
+        query="""
+        SELECT num_observations,(false_positives+false_negatives)/2.0 FROM classifier_results 
+        WHERE trial_type='test'
+         AND classifier=:classifier
+         AND params=:params 
+        ORDER BY num_observations
+        """
+        cursor.execute(query, {'classifier':classifier,'params':best_params})
+
+        num_obs = []
+        performance = []
+        for row in cursor:
+            num_obs.append(row[0])
+            performance.append(row[1])
+        #print(num_obs,performance)
+        path = plt.scatter(num_obs, performance, color=colors[color_id], s=4, alpha=0.8, linewidths=3.0)
+        plt.plot(num_obs, performance, color=colors[color_id], alpha=0.8)
+        legend.append((classifier,path))
+        color_id = (color_id+1) % len(colors)
+
+    plt.legend([l[1] for l in legend], [l[0] for l in legend], scatterpoints=1, fontsize='xx-small')
+    plt.show()
+        
+graphTestResults()
+
+sys.exit(0)
 
 plt.clf()
@@ -122 +237 @@
 plt.show()
 
+
+
+plt.clf()
+plt.title("Simple HTTP Request")
+plt.xlabel('Time of Day')
+plt.ylabel('')
+s = plt.scatter(sent_times, [2]*len(sent_times), s=3, color='red', alpha=0.9)
+r = plt.scatter(rcvd_times, [1]*len(rcvd_times), s=3, color='blue', alpha=0.9)
+plt.legend((s,r), ('sent','received'), scatterpoints=1)
+plt.show()
+
+sys.exit(0)
 short_overtime,long_overtime,diff_overtime = None,None,None
 

trunk/bin/train

@@ -25 +25 @@
     sys.path.append("%s/../lib" % script_dir)
 
+
 from nanownlib import *
 from nanownlib.stats import *
+from nanownlib.train import *
 from nanownlib.parallel import WorkerThreads
 import nanownlib.storage
@@ -36 +38 @@
 #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('--retrain', action='append', default=[], help='Force a classifier to be retrained.  May be specified multiple times.')
+parser.add_argument('--unusual-case', action='store', default=None, help='Specify the unusual case and whether it is greater than the other cases.  Format: {case name},{1 or 0}')
+parser.add_argument('--retrain', action='append', default=[], help='Force a classifier to be retrained (and retested).  May be specified multiple times.')
 parser.add_argument('--retest', action='append', default=[], help='Force a classifier to be retested.  May be specified multiple times.')
 parser.add_argument('session_data', default=None,
                     help='Database file storing session information')
 options = parser.parse_args()
+db = nanownlib.storage.db(options.session_data)
 
 
-def trainBoxTest(db, unusual_case, greater, num_observations):
-    db.resetOffsets()
-    
-    def trainAux(low,high,num_trials):
-        estimator = functools.partial(multiBoxTest, {'low':low, 'high':high}, greater)
-        estimates = bootstrap3(estimator, db, 'train', unusual_case, num_observations, num_trials)
-        null_estimates = bootstrap3(estimator, db, 'train_null', unusual_case, num_observations, 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 = 1.0
-    performance = []
-    for low in range(0,50):
-        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 = 200
-    lows = [p[1] for p in performance[0:5]]
-    widths = [w/10.0 for w in range(5,65,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.append((abs(statistics.mean(false_positives)-statistics.mean(false_negatives)),
-                            width, statistics.mean(false_negatives), statistics.mean(false_positives)))
-    performance.sort()
-    #pprint.pprint(performance)
-    good_width = performance[0][1]
-    #print("good_width:",good_width)
-
-
-    num_trials = 500
-    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 = 500
-    widths = [good_width+(x/100.0) for x in range(-70,75,5) if good_width+(x/100.0) > 0.0]
-    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.append((abs(fp-fn), 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])
-
-    wt.stop()
-    params = json.dumps({"low":best_low,"high":best_low+best_width})
-    return {'trial_type':"train",
-            'num_observations':num_observations,
-            'num_trials':num_trials,
-            'params':params,
-            'false_positives':performance[0][3],
-            'false_negatives':performance[0][2]}
-
-
-def trainSummary(summaryFunc, db, unusual_case, greater, num_observations):
-    db.resetOffsets()
-    stest = functools.partial(summaryTest, summaryFunc)
-    
-    def trainAux(distance, threshold, num_trials):
-        estimator = functools.partial(stest, {'distance':distance,'threshold':threshold}, greater)
-        estimates = bootstrap3(estimator, db, 'train', unusual_case, num_observations, num_trials)
-        null_estimates = bootstrap3(estimator, db, 'train_null', unusual_case, num_observations, 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
-    mean_diffs = [s['unusual_case']-s['other_cases'] for s in db.subseries('train', unusual_case)]
-    threshold = summaryFunc(mean_diffs)/2.0
-    #print("init_threshold:", threshold)
-    
-    wt = WorkerThreads(2, trainAux)
-    
-    num_trials = 500
-    performance = []
-    for distance in range(1,50):
-        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)
-    good_distance = performance[0][1]
-    #print("good_distance:",good_distance)
-
-    
-    num_trials = 500
-    performance = []
-    for t in range(80,122,2):
-        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.append((abs(fp-fn), job_id, fn, fp))
-    performance.sort()
-    #pprint.pprint(performance)
-    good_threshold = performance[0][1]
-    #print("good_threshold:", good_threshold)
-
-    
-    num_trials = 500
-    performance = []
-    for d in [good_distance+s for s in range(-4,5) if good_distance+s > -1]:
-        wt.addJob(d, (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)
-    best_distance = performance[0][1]
-    #print("best_distance:",best_distance)
-
-    
-    num_trials = 500
-    performance = []
-    for t in range(90,111):
-        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.append((abs(fp-fn), job_id, fn, fp))
-    performance.sort()
-    #pprint.pprint(performance)
-    best_threshold = performance[0][1]
-    #print("best_threshold:", best_threshold)
-
-    wt.stop()
-    params = json.dumps({'distance':best_distance,'threshold':best_threshold})
-    return {'trial_type':"train",
-            'num_observations':num_observations,
-            'num_trials':num_trials,
-            'params':params,
-            'false_positives':performance[0][3],
-            'false_negatives':performance[0][2]}
-
-
-def trainKalman(db, unusual_case, greater, num_observations):
-    db.resetOffsets()
-
-    def trainAux(params, num_trials):
-        estimator = functools.partial(kalmanTest, params, greater)
-        estimates = bootstrap3(estimator, db, 'train', unusual_case, num_observations, num_trials)
-        null_estimates = bootstrap3(estimator, db, 'train_null', unusual_case, num_observations, 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
-    
-    mean_diffs = [s['unusual_case']-s['other_cases'] for s in db.subseries('train', unusual_case)]
-    good_threshold = kfilter({},mean_diffs)['est'][-1]/2.0
-
-    wt = WorkerThreads(2, trainAux)
-    num_trials = 200
-    performance = []
-    for t in range(90,111):
-        params = {'threshold':good_threshold*(t/100.0)}
-        wt.addJob(good_threshold*(t/100.0), (params,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.append((abs(fp-fn), job_id, fn, fp))
-    performance.sort()
-    #pprint.pprint(performance)
-    best_threshold = performance[0][1]
-    #print("best_threshold:", best_threshold)
-    params = {'threshold':best_threshold}
-
-    wt.stop()
-    
-    return {'trial_type':"train",
-            'num_observations':num_observations,
-            'num_trials':num_trials,
-            'params':json.dumps(params),
-            'false_positives':performance[0][3],
-            'false_negatives':performance[0][2]}
-
-    
-    #determine expected delta based on differences
-classifiers = {'boxtest':{'train':trainBoxTest, 'test':multiBoxTest, 'train_results':[]},
-               'midsummary':{'train':functools.partial(trainSummary, midsummary), 'test':midsummaryTest, 'train_results':[]},
-               #'ubersummary':{'train':functools.partial(trainSummary, ubersummary), 'test':ubersummaryTest, 'train_results':[]},
-               'quadsummary':{'train':functools.partial(trainSummary, quadsummary), 'test':quadsummaryTest, 'train_results':[]},
-               'kalman':{'train':trainKalman, 'test':kalmanTest, 'train_results':[]},
-               #'_trimean':{'train':None, 'test':trimeanTest, 'train_results':[]},
-              }
-
-
-db = nanownlib.storage.db(options.session_data)
-
-import cProfile
 
 def trainClassifier(db, unusual_case, greater, classifier, retrain=False):
@@ -324 +72 @@
         print("number of observations: %d | error: %f | false_positives: %f | false_negatives: %f | train time: %s | params: %s"
               % (num_obs, error, result['false_positives'],result['false_negatives'], train_time, result['params']))
-        db.addClassifierResults(result)
+        db.addClassifierResult(result)
         classifiers[classifier]['train_results'].append(result)
 
@@ -355 +103 @@
 
 
+    def getResult(classifier, params, num_obs, num_trials):
+        jparams = json.dumps(params, sort_keys=True)
+        result = db.fetchClassifierResult(classifier, 'test', num_obs, jparams)
+        if result:
+            fp = result['false_positives']
+            fn = result['false_negatives']
+        else:
+            fp,fn = testAux(params, num_trials, num_obs)
+            result = {'classifier':classifier,
+                      'trial_type':"test",
+                      'num_observations':num_obs,
+                      'num_trials':num_trials,
+                      'params':jparams,
+                      'false_positives':fp,
+                      'false_negatives':fn}
+            db.addClassifierResult(result)
+        return ((fp+fn)/2.0,result)
+    
     if retest:
         print("Dropping stored test results...")
@@ -368 +134 @@
     
         print("initial test")
-        fp,fn = testAux(params, num_trials, num_obs)
-        error = (fp+fn)/2.0
+        error,result = getResult(classifier,params,num_obs,num_trials)
         print("walking up")
         while (error > target_error) and (num_obs < max_obs):
@@ -375 +140 @@
             #print("increase_factor:", increase_factor)
             num_obs = min(int(increase_factor*num_obs), max_obs)
-            fp,fn = testAux(params, num_trials, num_obs)
-            error = (fp+fn)/2.0
+            error,result = getResult(classifier,params,num_obs,num_trials)
 
         print("walking down")
         while (num_obs > 0):
-            current_best = (num_obs,error,params,fp,fn)
+            current_best = (error,result)
             num_obs = int(0.95*num_obs)
-            fp,fn = testAux(params, num_trials, num_obs)
-            error = (fp+fn)/2.0
+            error,result = getResult(classifier,params,num_obs,num_trials)
             if error > target_error:
                 break
         
-        test_results.append(current_best)
-
-    test_results.sort()
-    best_obs,error,best_params,fp,fn = test_results[0]
-    
-    return {'classifier':classifier,
-            'trial_type':"test",
-            'num_observations':best_obs,
-            'num_trials':num_trials,
-            'params':best_params,
-            'false_positives':fp,
-            'false_negatives':fn}
+    return current_best
 
 
-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)
+if options.unusual_case != None:
+    unusual_case,greater = options.unusual_case.split(',')
+    greater = bool(int(greater))
+else:
+    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)
+
 
 for c in sorted(classifiers.keys()):
@@ -424 +181 @@
     start = time.time()
     print("Testing %s..." % c)
-    result = testClassifier(db, unusual_case, greater, c, c in options.retest)
+    error,result = testClassifier(db, unusual_case, greater, c, c in (options.retest+options.retrain))
     print("%s result:" % c)
     pprint.pprint(result)
-    classifiers[c]['test_error'] = (result['false_positives']+result['false_negatives'])/2.0
+    classifiers[c]['test_error'] = error
     print("completed in:", time.time()-start)