<s> from __future__ import absolute_import , division , print_function <EOL> from contextlib import contextmanager <EOL> from warnings import warn <EOL> import networkx as nx <EOL> from datashape import discover <EOL> from . utils import expand_tuples , ignoring <EOL> ooc_types = set ( ) <EOL> class FailedConversionWarning ( UserWarning ) : <EOL> def __init__ ( self , src , dest , exc ) : <EOL> self . src = src <EOL> self . dest = dest <EOL> self . exc = exc <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . src . __name__ , self . dest . __name__ , self . exc , <EOL> ) <EOL> class NetworkDispatcher ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> self . graph = nx . DiGraph ( ) <EOL> def register ( self , a , b , cost = <NUM_LIT:1.0> ) : <EOL> sigs = expand_tuples ( [ a , b ] ) <EOL> def _ ( func ) : <EOL> for a , b in sigs : <EOL> self . graph . add_edge ( b , a , cost = cost , func = func ) <EOL> return func <EOL> return _ <EOL> def path ( self , * args , ** kwargs ) : <EOL> return path ( self . graph , * args , ** kwargs ) <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> return _transform ( self . graph , * args , ** kwargs ) <EOL> def _transform ( graph , target , source , excluded_edges = None , ooc_types = ooc_types , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> x = source <EOL> excluded_edges = excluded_edges or set ( ) <EOL> with ignoring ( NotImplementedError ) : <EOL> if '<STR_LIT>' not in kwargs : <EOL> kwargs [ '<STR_LIT>' ] = discover ( x ) <EOL> pth = path ( graph , type ( source ) , target , <EOL> excluded_edges = excluded_edges , <EOL> ooc_types = ooc_types ) <EOL> try : <EOL> for ( A , B , f ) in pth : <EOL> x = f ( x , excluded_edges = excluded_edges , ** kwargs ) <EOL> return x <EOL> except NotImplementedError as e : <EOL> if kwargs . get ( '<STR_LIT>' ) : <EOL> raise <EOL> warn ( FailedConversionWarning ( A , B , e ) ) <EOL> new_exclusions = excluded_edges | set ( [ ( A , B ) ] ) <EOL> return _transform ( graph , target , source , excluded_edges = new_exclusions , <EOL> ** kwargs ) <EOL> def path ( graph , source , target , excluded_edges = None , ooc_types = ooc_types ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( source , type ) : <EOL> source = type ( source ) <EOL> if not isinstance ( target , type ) : <EOL> target = type ( target ) <EOL> if source not in graph : <EOL> for cls in valid_subclasses : <EOL> if issubclass ( source , cls ) : <EOL> source = cls <EOL> break <EOL> if ooc_types : <EOL> oocs = tuple ( ooc_types ) <EOL> if issubclass ( source , oocs ) and issubclass ( target , oocs ) : <EOL> graph = graph . subgraph ( [ n for n in graph . nodes ( ) <EOL> if issubclass ( n , oocs ) ] ) <EOL> with without_edges ( graph , excluded_edges ) as g : <EOL> pth = nx . shortest_path ( g , source = source , target = target , weight = '<STR_LIT>' ) <EOL> result = [ ( src , tgt , graph . edge [ src ] [ tgt ] [ '<STR_LIT>' ] ) <EOL> for src , tgt in zip ( pth , pth [ <NUM_LIT:1> : ] ) ] <EOL> return result <EOL> from collections import Iterator <EOL> import numpy as np <EOL> valid_subclasses = [ Iterator , np . ndarray ] <EOL> @ contextmanager <EOL> def without_edges ( g , edges ) : <EOL> edges = edges or [ ] <EOL> held = dict ( ) <EOL> for a , b in edges : <EOL> held [ ( a , b ) ] = g . edge [ a ] [ b ] <EOL> g . remove_edge ( a , b ) <EOL> try : <EOL> yield g <EOL> finally : <EOL> for ( a , b ) , kwargs in held . items ( ) : <EOL> g . add_edge ( a , b , ** kwargs ) </s>
