<s> from twisted . internet import defer <EOL> from twisted . python import filepath , reflect <EOL> from zope import interface <EOL> from piped import util , processing <EOL> from piped . processors import base <EOL> class DiagramTrace ( base . InputOutputProcessor ) : <EOL> """<STR_LIT>""" <EOL> interface . classProvides ( processing . IProcessor ) <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , input_path = '<STR_LIT>' , output_path = '<STR_LIT>' , show_unused_processors = True , ** kw ) : <EOL> """<STR_LIT>""" <EOL> super ( DiagramTrace , self ) . __init__ ( input_path = input_path , output_path = output_path , ** kw ) <EOL> self . show_unused_processors = show_unused_processors <EOL> @ classmethod <EOL> def _get_color_for_time ( cls , t , mean_time ) : <EOL> """<STR_LIT>""" <EOL> if not ( t or mean_time ) : <EOL> return '<STR_LIT>' <EOL> if t > mean_time : <EOL> return '<STR_LIT>' % ( int ( <NUM_LIT:255> * mean_time / t ) , ) <EOL> return '<STR_LIT>' % int ( <NUM_LIT:255> * min ( <NUM_LIT:1.0> , t / mean_time ) ) <EOL> def configure ( self , runtime_environment ) : <EOL> dm = runtime_environment . dependency_manager <EOL> self . pipeline_provider_dependency = dm . add_dependency ( self , dict ( provider = '<STR_LIT>' ) ) <EOL> def process_input ( self , input , baton ) : <EOL> processors_by_pipeline , unknowns = self . _find_processors_and_unknowns ( input ) <EOL> total_time , mean_time = self . _calculate_mean_and_total_time ( input ) <EOL> graph_list = [ '<STR_LIT>' ] <EOL> for unknown in unknowns : <EOL> graph_list . append ( '<STR_LIT>' % ( id ( unknown ) , getattr ( unknown , '<STR_LIT:name>' , repr ( unknown ) ) ) ) <EOL> for trace_step in input : <EOL> if not trace_step [ '<STR_LIT>' ] : <EOL> graph_list . append ( '<STR_LIT>' % id ( trace_step [ '<STR_LIT>' ] ) ) <EOL> for pipeline , processors in processors_by_pipeline . items ( ) : <EOL> subgraph_list = self . _create_pipeline_subgraph ( pipeline , processors ) <EOL> graph_list . append ( '<STR_LIT:\n>' . join ( subgraph_list ) ) <EOL> graph_list . append ( '<STR_LIT:\n>' ) <EOL> for i , trace_step in enumerate ( input ) : <EOL> traced_edge = self . _create_traced_edge_entry ( i , trace_step , mean_time ) <EOL> graph_list . append ( traced_edge ) <EOL> graph_list . append ( '<STR_LIT:\n>' ) <EOL> graph_list . append ( '<STR_LIT>' ) <EOL> return '<STR_LIT:\n>' . join ( graph_list ) <EOL> def _calculate_mean_and_total_time ( self , trace ) : <EOL> total_time = <NUM_LIT:0> <EOL> steps_with_time = <NUM_LIT:0> <EOL> for trace_step in trace : <EOL> if '<STR_LIT>' in trace_step : <EOL> total_time += trace_step [ '<STR_LIT>' ] <EOL> steps_with_time += <NUM_LIT:1> <EOL> mean_time = total_time / max ( steps_with_time , <NUM_LIT:1> ) <EOL> return total_time , mean_time <EOL> def _find_processors_and_unknowns ( self , trace ) : <EOL> processors_by_pipeline = dict ( ) <EOL> unknowns = set ( ) <EOL> for trace_step in trace : <EOL> for maybe_processor in trace_step [ '<STR_LIT:source>' ] , trace_step [ '<STR_LIT>' ] : <EOL> if not maybe_processor : <EOL> continue <EOL> if isinstance ( maybe_processor , base . Processor ) : <EOL> processors_by_pipeline . setdefault ( maybe_processor . evaluator , set ( ) ) . add ( maybe_processor ) <EOL> else : <EOL> unknowns . add ( maybe_processor ) <EOL> return processors_by_pipeline , unknowns <EOL> def _create_pipeline_subgraph ( self , pipeline , processors ) : <EOL> subgraph_list = [ '<STR_LIT>' % ( pipeline . name , pipeline . name ) ] <EOL> for processor in processors : <EOL> subgraph_list . append ( '<STR_LIT>' % ( id ( processor ) , processor . node_name ) + '<STR_LIT:\n>' ) <EOL> if self . show_unused_processors : <EOL> used_processors = set ( processors ) <EOL> all_processors = set ( [ processor for processor in pipeline ] ) <EOL> unused_processors = all_processors . difference ( used_processors ) <EOL> for processor in used_processors : <EOL> for consumer in processor . consumers : <EOL> if consumer in unused_processors : <EOL> subgraph_list . append ( '<STR_LIT>' % ( id ( processor ) , id ( consumer ) ) ) <EOL> for error_consumer in processor . error_consumers : <EOL> if error_consumer in unused_processors : <EOL> subgraph_list . append ( '<STR_LIT>' % ( id ( processor ) , id ( error_consumer ) ) ) <EOL> for processor in unused_processors : <EOL> subgraph_list . append ( '<STR_LIT>' % ( id ( processor ) , processor . name ) + '<STR_LIT:\n>' ) <EOL> for consumer in processor . consumers : <EOL> subgraph_list . append ( '<STR_LIT>' % ( id ( processor ) , id ( consumer ) ) ) <EOL> for error_consumer in processor . error_consumers : <EOL> subgraph_list . append ( '<STR_LIT>' % ( id ( processor ) , id ( error_consumer ) ) ) <EOL> subgraph_list . append ( '<STR_LIT>' ) <EOL> return subgraph_list <EOL> def _create_traced_edge_entry ( self , i , trace_step , mean_time ) : <EOL> time_spent = trace_step . get ( '<STR_LIT>' , None ) <EOL> color = self . _get_color_for_time ( time_spent , mean_time ) if time_spent is not None else '<STR_LIT>' <EOL> fontcolor = '<STR_LIT>' <EOL> constraint = '<STR_LIT:true>' <EOL> style = '<STR_LIT>' <EOL> if trace_step . get ( '<STR_LIT>' ) is not None : <EOL> style = '<STR_LIT>' <EOL> if trace_step . get ( '<STR_LIT>' , False ) : <EOL> color = '<STR_LIT>' <EOL> fontcolor = '<STR_LIT>' <EOL> else : <EOL> color = '<STR_LIT>' <EOL> constraint = '<STR_LIT:false>' <EOL> source_id = id ( trace_step [ '<STR_LIT:source>' ] ) <EOL> destination_id = id ( trace_step [ '<STR_LIT>' ] ) <EOL> if not trace_step [ '<STR_LIT>' ] : <EOL> destination_id = id ( trace_step [ '<STR_LIT>' ] ) <EOL> label = '<STR_LIT>' % ( i + <NUM_LIT:1> ) <EOL> tooltip = '<STR_LIT>' <EOL> if time_spent : <EOL> tooltip = '<STR_LIT>' % ( time_spent * <NUM_LIT:1000> ) <EOL> return '<STR_LIT>' % ( source_id , destination_id , label , i , color , fontcolor , constraint , style , tooltip , tooltip ) <EOL> class RenderTrace ( base . InputOutputProcessor ) : <EOL> """<STR_LIT>""" <EOL> interface . classProvides ( processing . IProcessor ) <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , request_path = '<STR_LIT>' , svg_path = '<STR_LIT>' , input_path = '<STR_LIT>' , output_path = '<STR_LIT:content>' , skip_if_nonexistent = False , template = None , json_encoder = None , ** kw ) : <EOL> """<STR_LIT>""" <EOL> super ( RenderTrace , self ) . __init__ ( input_path = input_path , output_path = output_path , skip_if_nonexistent = skip_if_nonexistent , ** kw ) <EOL> self . request_path = request_path <EOL> self . svg_path = svg_path <EOL> self . template = util . expand_filepath ( template or filepath . FilePath ( __file__ ) . sibling ( '<STR_LIT:data>' ) . child ( '<STR_LIT>' ) ) <EOL> self . json_encoder_name = json_encoder or reflect . fullyQualifiedName ( util . BatonJSONEncoder ) <EOL> def configure ( self , runtime_environment ) : <EOL> dm = runtime_environment . dependency_manager <EOL> self . pipeline_provider_dependency = dm . add_dependency ( self , dict ( provider = '<STR_LIT>' ) ) <EOL> self . json_encoder = reflect . namedAny ( self . json_encoder_name ) ( ) <EOL> @ defer . inlineCallbacks <EOL> def process_input ( self , input , baton ) : <EOL> trace = input or list ( ) <EOL> pipeline_provider = yield self . pipeline_provider_dependency . wait_for_resource ( ) <EOL> dot = util . dict_get_path ( baton , self . svg_path , '<STR_LIT>' ) <EOL> steps = list ( ) <EOL> processors = dict ( ) <EOL> for trace_step in trace : <EOL> steps . append ( <EOL> dict ( <EOL> baton = trace_step [ '<STR_LIT>' ] , <EOL> failure = trace_step . get ( '<STR_LIT>' ) , <EOL> source = id ( trace_step [ '<STR_LIT:source>' ] ) , <EOL> destination = id ( trace_step [ '<STR_LIT>' ] ) , <EOL> ) <EOL> ) <EOL> for processor in trace_step [ '<STR_LIT:source>' ] , trace_step [ '<STR_LIT>' ] : <EOL> if not processor : <EOL> continue <EOL> processors [ id ( processor ) ] = dict ( name = reflect . fullyQualifiedName ( processor . __class__ ) , state = getattr ( processor , '<STR_LIT>' , dict ( ) ) ) <EOL> pipeline_names = sorted ( pipeline_provider . processor_graph_factory . pipelines_configuration . keys ( ) ) <EOL> variables = dict ( dot = dot , steps = self . json_encoder . encode ( steps ) , pipeline_names = self . json_encoder . encode ( pipeline_names ) , processors = self . json_encoder . encode ( processors ) ) <EOL> trace_template = open ( self . template ) . read ( ) <EOL> html = trace_template % variables <EOL> defer . returnValue ( html ) </s>
