<s> """<STR_LIT>""" <EOL> from functools import reduce <EOL> import param <EOL> try : <EOL> from itertools import izip as zip <EOL> except : <EOL> pass <EOL> from . dimension import ViewableElement <EOL> from . element import Element , HoloMap , GridSpace , Collator <EOL> from . layout import Layout <EOL> from . overlay import NdOverlay , Overlay <EOL> from . spaces import DynamicMap <EOL> from . traversal import unique_dimkeys <EOL> from . import util <EOL> class Operation ( param . ParameterizedFunction ) : <EOL> """<STR_LIT>""" <EOL> group = param . String ( default = '<STR_LIT>' , doc = """<STR_LIT>""" ) <EOL> @ classmethod <EOL> def search ( cls , element , pattern ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( element , Layout ) : <EOL> return [ el for cell in element for el in cls . search ( cell , pattern ) ] <EOL> if isinstance ( element , ( NdOverlay , Overlay ) ) : <EOL> return [ el for el in element if el . matches ( pattern ) ] <EOL> elif isinstance ( element , Element ) : <EOL> return [ element ] if element . matches ( pattern ) else [ ] <EOL> @ classmethod <EOL> def get_overlay_label ( cls , overlay , default_label = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if all ( el . label == overlay . get ( <NUM_LIT:0> ) . label for el in overlay ) : <EOL> return overlay . get ( <NUM_LIT:0> ) . label <EOL> else : <EOL> return default_label <EOL> @ classmethod <EOL> def get_overlay_bounds ( cls , overlay ) : <EOL> """<STR_LIT>""" <EOL> if all ( el . bounds == overlay . get ( <NUM_LIT:0> ) . bounds for el in overlay ) : <EOL> return overlay . get ( <NUM_LIT:0> ) . bounds <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> class DynamicOperation ( Operation ) : <EOL> """<STR_LIT>""" <EOL> def __call__ ( self , map_obj , ** params ) : <EOL> self . p = param . ParamOverrides ( self , params ) <EOL> callback = self . _dynamic_operation ( map_obj ) <EOL> if isinstance ( map_obj , DynamicMap ) : <EOL> return map_obj . clone ( callback = callback , shared_data = False ) <EOL> else : <EOL> return self . _make_dynamic ( map_obj , callback ) <EOL> def _process ( self , element ) : <EOL> return element <EOL> def _dynamic_operation ( self , map_obj ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( map_obj , DynamicMap ) : <EOL> def dynamic_operation ( * key ) : <EOL> return self . _process ( map_obj [ key ] ) <EOL> return dynamic_operation <EOL> def dynamic_operation ( * key ) : <EOL> key = key [ <NUM_LIT:0> ] if map_obj . mode == '<STR_LIT>' else key <EOL> _ , el = util . get_dynamic_item ( map_obj , map_obj . kdims , key ) <EOL> return self . _process ( el ) <EOL> return dynamic_operation <EOL> def _make_dynamic ( self , hmap , dynamic_fn ) : <EOL> """<STR_LIT>""" <EOL> dim_values = zip ( * hmap . data . keys ( ) ) <EOL> params = util . get_param_values ( hmap ) <EOL> kdims = [ d ( values = list ( values ) ) for d , values in zip ( hmap . kdims , dim_values ) ] <EOL> return DynamicMap ( dynamic_fn , ** dict ( params , kdims = kdims ) ) <EOL> class DynamicFunction ( DynamicOperation ) : <EOL> """<STR_LIT>""" <EOL> function = param . Callable ( default = lambda x : x , doc = """<STR_LIT>""" ) <EOL> kwargs = param . Dict ( default = { } , doc = """<STR_LIT>""" ) <EOL> def _process ( self , element ) : <EOL> return self . p . function ( element , ** self . p . kwargs ) <EOL> class ElementOperation ( Operation ) : <EOL> """<STR_LIT>""" <EOL> dynamic = param . ObjectSelector ( default = '<STR_LIT:default>' , <EOL> objects = [ '<STR_LIT:default>' , True , False ] , doc = """<STR_LIT>""" ) <EOL> input_ranges = param . ClassSelector ( default = { } , <EOL> class_ = ( dict , tuple ) , doc = """<STR_LIT>""" ) <EOL> def _process ( self , view , key = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def process_element ( self , element , key , ** params ) : <EOL> """<STR_LIT>""" <EOL> self . p = param . ParamOverrides ( self , params ) <EOL> return self . _process ( element , key ) <EOL> def __call__ ( self , element , ** params ) : <EOL> self . p = param . ParamOverrides ( self , params ) <EOL> dynamic = ( ( self . p . dynamic == '<STR_LIT:default>' and <EOL> isinstance ( element , DynamicMap ) ) <EOL> or self . p . dynamic is True ) <EOL> if isinstance ( element , ViewableElement ) : <EOL> processed = self . _process ( element ) <EOL> elif isinstance ( element , GridSpace ) : <EOL> grid_data = ( ( pos , self ( cell , ** params ) ) <EOL> for pos , cell in element . items ( ) ) <EOL> processed = GridSpace ( grid_data , label = element . label , <EOL> kdims = element . kdims ) <EOL> elif dynamic : <EOL> processed = DynamicFunction ( element , function = self , kwargs = params ) <EOL> elif isinstance ( element , DynamicMap ) : <EOL> if any ( ( not d . values ) for d in element . kdims ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> samples = tuple ( d . values for d in element . kdims ) <EOL> processed = self ( element [ samples ] , ** params ) <EOL> elif isinstance ( element , HoloMap ) : <EOL> mapped_items = [ ( k , self . _process ( el , key = k ) ) <EOL> for k , el in element . items ( ) ] <EOL> refval = mapped_items [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> processed = element . clone ( mapped_items , <EOL> group = refval . group , <EOL> label = refval . label ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" % type ( element ) . __name__ ) <EOL> return processed <EOL> class MapOperation ( param . ParameterizedFunction ) : <EOL> """<STR_LIT>""" <EOL> group = param . String ( default = '<STR_LIT>' , doc = """<STR_LIT>""" ) <EOL> def __call__ ( self , vmap , ** params ) : <EOL> self . p = param . ParamOverrides ( self , params ) <EOL> if not isinstance ( vmap , HoloMap ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> return self . _process ( vmap ) <EOL> def _process ( self , view ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> class TreeOperation ( Operation ) : <EOL> """<STR_LIT>""" <EOL> def process_element ( self , element , key , ** params ) : <EOL> """<STR_LIT>""" <EOL> self . p = param . ParamOverrides ( self , params ) <EOL> maps = self . _process ( element , key ) <EOL> return reduce ( lambda x , y : x + y , maps ) <EOL> def __call__ ( self , src , ** params ) : <EOL> self . p = param . ParamOverrides ( self , params ) <EOL> dims , keys = unique_dimkeys ( src ) <EOL> if isinstance ( src , Layout ) and not src . uniform : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if not dims : <EOL> return self . process_element ( src , None ) <EOL> else : <EOL> dim_names = [ d . name for d in dims ] <EOL> values = { } <EOL> for key in keys : <EOL> selection = src . select ( ** dict ( zip ( dim_names , key ) ) ) <EOL> if not isinstance ( selection , Layout ) : <EOL> selection = Layout . from_values ( [ selection ] ) <EOL> processed = self . _process ( selection , key ) <EOL> if isinstance ( processed , list ) : <EOL> processed = Layout . from_values ( processed ) <EOL> values [ key ] = processed <EOL> return Collator ( values , kdims = dims ) ( ) <EOL> def _process ( self , tree , key = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError </s>
