<s> """<STR_LIT>""" <EOL> from monk . roles . configuration import get_config <EOL> import monk . core . api as monkapi <EOL> import logging <EOL> from monk . network . broker import KafkaBroker <EOL> from monk . network . server import taskT , Task , MonkServer <EOL> import sys <EOL> import time <EOL> from collections import deque <EOL> from tornado . web import RequestHandler <EOL> from bokeh . resources import INLINE <EOL> from bokeh . plotting import figure , file_html , decode_utf8 <EOL> import numpy as np <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> DEFAULT_MONITOR_USER = '<STR_LIT>' <EOL> class MonitorBroker ( KafkaBroker ) : <EOL> def track ( self , name , value , user = DEFAULT_MONITOR_USER ) : <EOL> self . produce ( '<STR_LIT>' , name , user = user , value = value , time = time . time ( ) ) <EOL> def aggregate ( self , name , value , user = DEFAULT_MONITOR_USER ) : <EOL> self . produce ( '<STR_LIT>' , name , user = user , value = value ) <EOL> def measure ( self , name , value , label = <NUM_LIT:1> , user = DEFAULT_MONITOR_USER ) : <EOL> self . produce ( '<STR_LIT>' , name , user = user , value = value , label = label ) <EOL> def reset_tracker ( self , name ) : <EOL> self . produce ( '<STR_LIT>' , name ) <EOL> def reset_aggregator ( self , name ) : <EOL> self . produce ( '<STR_LIT>' , name ) <EOL> def reset_measurer ( self , name ) : <EOL> self . produce ( '<STR_LIT>' , name ) <EOL> def rename_tracker ( self , name , newname ) : <EOL> self . produce ( '<STR_LIT>' , name , newname = newname ) <EOL> def rename_aggregator ( self , name , newname ) : <EOL> self . produce ( '<STR_LIT>' , name , newname = newname ) <EOL> def rename_measurer ( self , name , newname ) : <EOL> self . produce ( '<STR_LIT>' , name , newname = newname ) <EOL> def annotate_tracker ( self , name , annotator ) : <EOL> self . produce ( '<STR_LIT>' , name , annotator = annotator , time = time . time ( ) ) <EOL> class MonkMonitor ( MonkServer ) : <EOL> def init_brokers ( self , config ) : <EOL> self . trackers = { } <EOL> self . aggregators = { } <EOL> self . measurers = { } <EOL> monkapi . initialize ( config ) <EOL> self . MAINTAIN_INTERVAL = config . monitorMaintainInterval <EOL> self . POLL_INTERVAL = config . monitorPollInterval <EOL> self . EXECUTE_INTERVAL = config . monitorExecuteInterval <EOL> self . MAX_QUEUE_SIZE = config . monitorMaxQueueSize <EOL> self . monitorBroker = MonitorBroker ( config . kafkaConnectionString , <EOL> config . monitorGroup , <EOL> config . monitorTopic , <EOL> KafkaBroker . SIMPLE_CONSUMER ) <EOL> return [ self . monitorBroker ] <EOL> monitor = MonkMonitor ( ) <EOL> class Tracker ( object ) : <EOL> def __init__ ( self , retireTime = <NUM_LIT> , resolution = <NUM_LIT:1> ) : <EOL> self . valuesTimed = dict ( ) <EOL> self . numTimed = dict ( ) <EOL> self . annotatorTimed = dict ( ) <EOL> self . queue = deque ( ) <EOL> self . resolution = resolution <EOL> self . retiredTime = int ( retireTime / resolution ) <EOL> def add ( self , t , value , user ) : <EOL> t = int ( t / self . resolution ) <EOL> firstTime = self . queue [ <NUM_LIT:0> ] <EOL> if t - firstTime > self . retiredTime : <EOL> if firstTime in self . valuesTimed : <EOL> del self . valuesTimed [ firstTime ] <EOL> del self . numTimed [ firstTime ] <EOL> del self . annotatorTimed [ firstTime ] <EOL> self . queue . popleft ( ) <EOL> self . queue . append ( t ) <EOL> if t in self . valuesTimed : <EOL> if user in self . valuesTimed : <EOL> self . valuesTimed [ t ] [ user ] += value <EOL> self . numTimed [ t ] [ user ] += <NUM_LIT:1.0> <EOL> else : <EOL> self . valuesTimed [ t ] [ user ] = value <EOL> self . numTimed [ t ] [ user ] = <NUM_LIT:1.0> <EOL> else : <EOL> self . valuesTimed [ t ] = { user : value } <EOL> self . numTimed [ t ] = { user : <NUM_LIT:1.0> } <EOL> def annotate ( self , t , annotator ) : <EOL> t = int ( t / self . resolution ) <EOL> if t in self . annotatorTimed : <EOL> logger . warning ( '<STR_LIT>' . format ( self . annotatorTimed [ t ] , t * self . resolution ) ) <EOL> else : <EOL> self . annotatorTimed [ t ] = annotator <EOL> def clear ( self ) : <EOL> self . queue . clear ( ) <EOL> self . valuesTimed . clear ( ) <EOL> self . numTimed . clear ( ) <EOL> class Track ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> value = float ( self . get ( '<STR_LIT:value>' ) ) <EOL> except : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> t = float ( self . get ( '<STR_LIT:time>' ) ) <EOL> except : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> user = self . get ( '<STR_LIT:user>' ) <EOL> if key not in monitor . trackers : <EOL> monitor . trackers [ key ] = Tracker ( ) <EOL> tracker = monitor . trackers [ key ] <EOL> tracker . add ( t , value , user ) <EOL> taskT ( Track ) <EOL> class ResetTracker ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> if key in monitor . trackers : <EOL> monitor . trackers [ key ] . clear ( ) <EOL> taskT ( ResetTracker ) <EOL> class RenameTracker ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> newname = self . get ( '<STR_LIT>' ) <EOL> if key in monitor . trackers and newname : <EOL> monitor . trackers [ newname ] = monitor . trackers [ key ] <EOL> del monitor . trackers [ key ] <EOL> taskT ( RenameTracker ) <EOL> class AnnotateTracker ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> annotator = self . get ( '<STR_LIT>' ) <EOL> try : <EOL> t = float ( self . get ( '<STR_LIT:time>' ) ) <EOL> except : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> if annotator and key in monitor . trackers : <EOL> tracker = monitor . trackers [ key ] <EOL> tracker . annotate ( t , annotator ) <EOL> taskT ( AnnotateTracker ) <EOL> class Aggregator ( object ) : <EOL> def __init__ ( self , resolution = <NUM_LIT> ) : <EOL> self . resolution = resolution <EOL> self . hist = dict ( ) <EOL> self . num = dict ( ) <EOL> def clear ( self ) : <EOL> self . hist . clear ( ) <EOL> self . num . clear ( ) <EOL> def add ( self , value , user ) : <EOL> vL = int ( value / self . resolution ) <EOL> vH = vL + <NUM_LIT:1> <EOL> wL = value / self . resolution - vL <EOL> wH = <NUM_LIT:1.0> - wL <EOL> if user in self . hist : <EOL> hist = self . hist [ user ] <EOL> if vL in hist : <EOL> hist [ vL ] += wL <EOL> else : <EOL> hist [ vL ] = wL <EOL> if vH in hist : <EOL> hist [ vH ] += wH <EOL> else : <EOL> hist [ vH ] = wH <EOL> self . num [ user ] += <NUM_LIT:1.0> <EOL> else : <EOL> self . hist [ user ] = { vL : wL , vH : wH } <EOL> self . num [ user ] = <NUM_LIT:1.0> <EOL> class Aggregate ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> value = float ( self . get ( '<STR_LIT:value>' ) ) <EOL> except : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> user = self . get ( '<STR_LIT:user>' ) <EOL> if key not in monitor . aggregators : <EOL> monitor . aggregators [ key ] = Aggregator ( ) <EOL> aggregator = monitor . aggregators [ key ] <EOL> aggregator . add ( value , user ) <EOL> taskT ( Aggregate ) <EOL> class ResetAggregator ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> if key in monitor . aggregators : <EOL> monitor . aggregators [ key ] . clear ( ) <EOL> taskT ( ResetAggregator ) <EOL> class RenameAggregator ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> newname = self . get ( '<STR_LIT>' ) <EOL> if key in monitor . aggregators and newname : <EOL> monitor . aggregators [ newname ] = monitor . aggregators [ key ] <EOL> del monitor . aggregators [ key ] <EOL> taskT ( RenameAggregator ) <EOL> class Measurer ( object ) : <EOL> def __init__ ( self , resolution = <NUM_LIT> ) : <EOL> self . resolution = resolution <EOL> self . scores = dict ( ) <EOL> self . totalPos = dict ( ) <EOL> self . totalNeg = dict ( ) <EOL> self . PRCs = None <EOL> self . ROCs = None <EOL> self . invalid = True <EOL> def clear ( self ) : <EOL> self . scores . clear ( ) <EOL> self . totalPos . clear ( ) <EOL> self . totalNeg . clear ( ) <EOL> self . PRCs = None <EOL> self . ROCs = None <EOL> self . invalid = True <EOL> def add ( self , value , user , label ) : <EOL> if user not in self . scores : <EOL> self . scores [ user ] = [ ] <EOL> self . totalNeg [ user ] = <NUM_LIT:0> <EOL> self . totalPos [ user ] = <NUM_LIT:0> <EOL> self . scores [ user ] . append ( ( value , label ) ) <EOL> if label > <NUM_LIT:0> : <EOL> self . totalPos [ user ] += <NUM_LIT:1> <EOL> else : <EOL> self . totalNeg [ user ] += <NUM_LIT:1> <EOL> self . invalid = True <EOL> def intervals ( self ) : <EOL> num = int ( <NUM_LIT:1.0> / self . resolution ) + <NUM_LIT:1> <EOL> return np . linspace ( <NUM_LIT:0.0> , <NUM_LIT:1.0> , num ) <EOL> def compute_metrics ( self ) : <EOL> num = int ( <NUM_LIT:1.0> / self . resolution ) + <NUM_LIT:1> <EOL> logger . debug ( '<STR_LIT>' . format ( num ) ) <EOL> self . PRCs = None <EOL> self . ROCs = None <EOL> PRCn = np . zeros ( num , dtype = int ) <EOL> ROCn = np . zeros ( num , dtype = int ) <EOL> for user in self . scores : <EOL> if user is DEFAULT_MONITOR_USER : <EOL> continue <EOL> tp = self . totalPos [ user ] <EOL> tn = <NUM_LIT:0> <EOL> fp = self . totalNeg [ user ] <EOL> fn = <NUM_LIT:0> <EOL> totalP = tp <EOL> totalN = fp <EOL> logger . debug ( '<STR_LIT>' . format ( totalP , totalN ) ) <EOL> if totalP == <NUM_LIT:0> or totalN == <NUM_LIT:0> : <EOL> continue <EOL> score = self . scores [ user ] <EOL> logger . info ( '<STR_LIT>' . format ( score , user ) ) <EOL> score . sort ( ) <EOL> logger . info ( '<STR_LIT>' . format ( score , user ) ) <EOL> PRC = np . zeros ( num ) <EOL> ROC = np . zeros ( num ) <EOL> PRCn . fill ( <NUM_LIT:0> ) <EOL> ROCn . fill ( <NUM_LIT:0> ) <EOL> for s , label in score : <EOL> if label > <NUM_LIT:0> : <EOL> fn += <NUM_LIT:1> <EOL> tp -= <NUM_LIT:1> <EOL> else : <EOL> tn += <NUM_LIT:1> <EOL> fp -= <NUM_LIT:1> <EOL> recall = float ( tp ) / float ( totalP ) <EOL> if tp + fp == <NUM_LIT:0> : <EOL> precision = <NUM_LIT:1> <EOL> else : <EOL> precision = float ( tp ) / float ( tp + fp ) <EOL> v = int ( precision / self . resolution ) <EOL> PRC [ v ] += recall <EOL> PRCn [ v ] += <NUM_LIT:1> <EOL> fpr = float ( fp ) / float ( totalN ) <EOL> v = int ( fpr / self . resolution ) <EOL> ROC [ v ] += recall <EOL> ROCn [ v ] += <NUM_LIT:1> <EOL> logger . info ( '<STR_LIT>' . format ( PRC ) ) <EOL> logger . info ( '<STR_LIT>' . format ( PRCn ) ) <EOL> logger . info ( '<STR_LIT>' . format ( ROC ) ) <EOL> logger . info ( '<STR_LIT>' . format ( ROCn ) ) <EOL> for v in reversed ( range ( num - <NUM_LIT:1> ) ) : <EOL> if PRCn [ v ] == <NUM_LIT:0> : <EOL> PRC [ v ] = PRC [ v + <NUM_LIT:1> ] <EOL> else : <EOL> PRC [ v ] /= PRCn [ v ] <EOL> for v in range ( <NUM_LIT:2> , num ) : <EOL> if ROCn [ v ] == <NUM_LIT:0> : <EOL> ROC [ v ] = ROC [ v - <NUM_LIT:1> ] <EOL> else : <EOL> ROC [ v ] /= ROCn [ v ] <EOL> if self . ROCs is None : <EOL> self . ROCs = ROC <EOL> else : <EOL> self . ROCs = np . vstack ( ( self . ROCs , ROC ) ) <EOL> if self . PRCs is None : <EOL> self . PRCs = PRC <EOL> else : <EOL> self . PRCs = np . vstack ( ( self . PRCs , PRC ) ) <EOL> self . ROCs . sort ( axis = <NUM_LIT:0> ) <EOL> self . PRCs . sort ( axis = <NUM_LIT:0> ) <EOL> logger . info ( '<STR_LIT>' . format ( self . ROCs ) ) <EOL> logger . info ( '<STR_LIT>' . format ( self . PRCs ) ) <EOL> self . invalid = False <EOL> def set_resolution ( self , resolution ) : <EOL> try : <EOL> self . resolution = float ( resolution ) <EOL> self . invalid = True <EOL> except : <EOL> logger . warning ( '<STR_LIT>' . format ( resolution ) ) <EOL> def get_ROCs ( self ) : <EOL> if self . invalid : <EOL> self . compute_metrics ( ) <EOL> return self . ROCs <EOL> def get_PRCs ( self ) : <EOL> if self . invalid : <EOL> self . compute_metrics ( ) <EOL> return self . PRCs <EOL> class Measure ( Task ) : <EOL> def act ( self ) : <EOL> logger . info ( '<STR_LIT>' . format ( self . decodedMessage ) ) <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> value = float ( self . get ( '<STR_LIT:value>' ) ) <EOL> except : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> pos = int ( self . get ( '<STR_LIT:label>' ) ) <EOL> except : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> user = self . get ( '<STR_LIT:user>' ) <EOL> if key not in monitor . measurers : <EOL> monitor . measurers [ key ] = Measurer ( ) <EOL> measurer = monitor . measurers [ key ] <EOL> measurer . add ( value , user , pos ) <EOL> taskT ( Measure ) <EOL> class ResetMeasurer ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> if key in monitor . measurers : <EOL> monitor . measurers [ key ] . clear ( ) <EOL> taskT ( ResetMeasurer ) <EOL> class RenameMeasurer ( Task ) : <EOL> def act ( self ) : <EOL> key = self . name <EOL> if not key : <EOL> self . warning ( logger , '<STR_LIT>' ) <EOL> return <EOL> newname = self . get ( '<STR_LIT>' ) <EOL> if key in monitor . measurers and newname : <EOL> monitor . measurers [ newname ] = monitor . measurers [ key ] <EOL> del monitor . measurers [ key ] <EOL> taskT ( RenameMeasurer ) <EOL> class AccuracyHandler ( RequestHandler ) : <EOL> def draw ( self , p , accuracies , intervals , resolution , fillColor ) : <EOL> if accuracies is not None : <EOL> m = accuracies . shape [ <NUM_LIT:0> ] <EOL> logger . debug ( '<STR_LIT>' . format ( m ) ) <EOL> maxs = accuracies [ - <NUM_LIT:1> ] <EOL> mins = accuracies [ <NUM_LIT:0> ] <EOL> uppers = accuracies [ - ( <NUM_LIT:1> + m / <NUM_LIT:4> ) ] <EOL> lowers = accuracies [ m / <NUM_LIT:4> ] <EOL> logger . debug ( '<STR_LIT>' . format ( maxs ) ) <EOL> logger . debug ( '<STR_LIT>' . format ( mins ) ) <EOL> logger . debug ( '<STR_LIT>' . format ( uppers ) ) <EOL> logger . debug ( '<STR_LIT>' . format ( lowers ) ) <EOL> p . segment ( intervals , maxs , intervals , mins , color = '<STR_LIT>' , toolbar_location = '<STR_LIT:left>' ) <EOL> p . rect ( intervals , ( uppers + lowers ) / <NUM_LIT:2> , resolution / <NUM_LIT:2> , abs ( uppers - lowers ) , fill_color = fillColor , line_color = '<STR_LIT>' ) <EOL> def get ( self ) : <EOL> name = self . get_argument ( '<STR_LIT:name>' , None ) <EOL> fillColor = self . get_argument ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> resolution = self . get_argument ( '<STR_LIT>' , None ) <EOL> accType = self . get_argument ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> logger . info ( '<STR_LIT>' . format ( name , fillColor , resolution ) ) <EOL> if name and name in monitor . measurers : <EOL> measurer = monitor . measurers [ name ] <EOL> logger . debug ( '<STR_LIT>' . format ( resolution ) ) <EOL> measurer . set_resolution ( resolution ) <EOL> if accType == '<STR_LIT>' : <EOL> accuracies = measurer . get_ROCs ( ) <EOL> elif accType == '<STR_LIT>' : <EOL> accuracies = measurer . get_PRCs ( ) <EOL> else : <EOL> accuracies = None <EOL> intervals = measurer . intervals ( ) <EOL> resolution = measurer . resolution <EOL> else : <EOL> accuracies = None <EOL> intervals = None <EOL> resolution = <NUM_LIT> <EOL> logger . debug ( '<STR_LIT>' . format ( accuracies ) ) <EOL> logger . debug ( '<STR_LIT>' . format ( intervals ) ) <EOL> logger . debug ( '<STR_LIT>' . format ( resolution ) ) <EOL> TOOLS = '<STR_LIT>' <EOL> p = figure ( tools = TOOLS , plot_width = <NUM_LIT:1000> ) <EOL> self . draw ( p , accuracies , intervals , resolution , fillColor ) <EOL> p . title = '<STR_LIT>' . format ( accType , name ) <EOL> p . grid . grid_line_alpha = <NUM_LIT> <EOL> response = file_html ( p , INLINE , p . title ) <EOL> self . write ( decode_utf8 ( response ) ) <EOL> self . set_header ( "<STR_LIT:Content-Type>" , "<STR_LIT>" ) <EOL> def main ( ) : <EOL> global monitor <EOL> myname = '<STR_LIT>' <EOL> config = get_config ( sys . argv [ <NUM_LIT:1> : ] , myname , '<STR_LIT>' ) <EOL> monitor = MonkMonitor ( myname , config ) <EOL> monitor . add_application ( r'<STR_LIT>' , AccuracyHandler ) <EOL> monitor . run ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
