<s> '''<STR_LIT>''' <EOL> import itertools <EOL> import collections <EOL> import functools <EOL> from python_toolbox import cute_iter_tools <EOL> from python_toolbox import misc_tools <EOL> from python_toolbox import address_tools <EOL> class Emitter : <EOL> '''<STR_LIT>''' <EOL> _is_atomically_pickleable = False <EOL> def __init__ ( self , inputs = ( ) , outputs = ( ) , name = None ) : <EOL> '''<STR_LIT>''' <EOL> from python_toolbox import sequence_tools <EOL> inputs = sequence_tools . to_tuple ( inputs , <EOL> item_type = Emitter ) <EOL> outputs = sequence_tools . to_tuple ( outputs , <EOL> item_type = ( collections . Callable , <EOL> Emitter ) ) <EOL> self . _inputs = set ( ) <EOL> '''<STR_LIT>''' <EOL> self . _outputs = set ( ) <EOL> '''<STR_LIT>''' <EOL> for output in outputs : <EOL> self . add_output ( output ) <EOL> self . __total_callable_outputs_cache = None <EOL> '''<STR_LIT>''' <EOL> self . _recalculate_total_callable_outputs ( ) <EOL> for input in inputs : <EOL> self . add_input ( input ) <EOL> self . name = name <EOL> '''<STR_LIT>''' <EOL> def get_inputs ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . _inputs <EOL> def get_outputs ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . _outputs <EOL> def _get_input_layers ( self ) : <EOL> '''<STR_LIT>''' <EOL> input_layers = [ self . _inputs ] <EOL> current_layer = self . _inputs <EOL> while current_layer : <EOL> next_layer = functools . reduce ( <EOL> set . union , <EOL> ( input . _inputs for input in current_layer ) , <EOL> set ( ) <EOL> ) <EOL> for ancestor_layer in input_layers : <EOL> assert isinstance ( next_layer , set ) <EOL> next_layer -= ancestor_layer <EOL> input_layers . append ( next_layer ) <EOL> current_layer = next_layer <EOL> return input_layers <EOL> def _recalculate_total_callable_outputs_recursively ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . _recalculate_total_callable_outputs ( ) <EOL> input_layers = self . _get_input_layers ( ) <EOL> for input_layer in input_layers : <EOL> for input in input_layer : <EOL> input . _recalculate_total_callable_outputs ( ) <EOL> def _recalculate_total_callable_outputs ( self ) : <EOL> '''<STR_LIT>''' <EOL> children_callable_outputs = functools . reduce ( <EOL> set . union , <EOL> ( emitter . get_total_callable_outputs ( ) for emitter <EOL> in self . _get_emitter_outputs ( ) if emitter is not self ) , <EOL> set ( ) <EOL> ) <EOL> self . __total_callable_outputs_cache = children_callable_outputs . union ( self . _get_callable_outputs ( ) ) <EOL> def add_input ( self , emitter ) : <EOL> '''<STR_LIT>''' <EOL> assert isinstance ( emitter , Emitter ) <EOL> self . _inputs . add ( emitter ) <EOL> emitter . _outputs . add ( self ) <EOL> emitter . _recalculate_total_callable_outputs_recursively ( ) <EOL> def remove_input ( self , emitter ) : <EOL> '''<STR_LIT>''' <EOL> assert isinstance ( emitter , Emitter ) <EOL> self . _inputs . remove ( emitter ) <EOL> emitter . _outputs . remove ( self ) <EOL> emitter . _recalculate_total_callable_outputs_recursively ( ) <EOL> def add_output ( self , thing ) : <EOL> '''<STR_LIT>''' <EOL> assert isinstance ( thing , ( Emitter , collections . Callable ) ) <EOL> self . _outputs . add ( thing ) <EOL> if isinstance ( thing , Emitter ) : <EOL> thing . _inputs . add ( self ) <EOL> self . _recalculate_total_callable_outputs_recursively ( ) <EOL> def remove_output ( self , thing ) : <EOL> '''<STR_LIT>''' <EOL> assert isinstance ( thing , ( Emitter , collections . Callable ) ) <EOL> self . _outputs . remove ( thing ) <EOL> if isinstance ( thing , Emitter ) : <EOL> thing . _inputs . remove ( self ) <EOL> self . _recalculate_total_callable_outputs_recursively ( ) <EOL> def disconnect_from_all ( self ) : <EOL> '''<STR_LIT>''' <EOL> for input in self . _inputs : <EOL> self . remove_input ( input ) <EOL> for output in self . _outputs : <EOL> self . remove_output ( output ) <EOL> def _get_callable_outputs ( self ) : <EOL> '''<STR_LIT>''' <EOL> return set ( filter ( callable , self . _outputs ) ) <EOL> def _get_emitter_outputs ( self ) : <EOL> '''<STR_LIT>''' <EOL> return { output for output in self . _outputs if isinstance ( output , Emitter ) } <EOL> def get_total_callable_outputs ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . __total_callable_outputs_cache <EOL> def emit ( self ) : <EOL> '''<STR_LIT>''' <EOL> for callable_output in self . __total_callable_outputs_cache : <EOL> callable_output ( ) <EOL> def __repr__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> return '<STR_LIT>' % ( <EOL> address_tools . describe ( type ( self ) , shorten = True ) , <EOL> '<STR_LIT>' . join ( ( "<STR_LIT:'>" , self . name , "<STR_LIT>" ) ) if self . name else '<STR_LIT>' , <EOL> hex ( id ( self ) ) <EOL> ) <EOL> """<STR_LIT>""" </s>
