signature:
def tell ( self , message , sender = no_sender ) :
    """Send a message to this actor . Asynchronous fire - and - forget .
 : param message : The message to send .
 : type message : Any
 : param sender : The sender of the message . If provided it will be made
 available to the receiving actor via the : attr : ` Actor . sender ` attribute .
 : type sender : : class : ` Actor `"""

implementation:
if sender is not no_sender and not isinstance ( sender , ActorRef ) :
    raise ValueError ( "Sender must be actor reference" )
self . _cell . send_message ( message , sender )


signature:
def get_anonymization_salt ( ts ) :
    """Get the anonymization salt based on the event timestamp ' s day ."""

implementation:
salt_key = 'stats:salt:{}' . format ( ts . date ( ) . isoformat ( ) )
salt = current_cache . get ( salt_key )
if not salt :
    salt_bytes = os . urandom ( 32 )
    salt = b64encode ( salt_bytes ) . decode ( 'utf-8' )
    current_cache . set ( salt_key , salt , timeout = 60 * 60 * 24 )
return salt


signature:
def get_geoip ( ip ) :
    """Lookup country for IP address ."""

implementation:
reader = geolite2 . reader ( )
ip_data = reader . get ( ip ) or { }
return ip_data . get ( 'country' , { } ) . get ( 'iso_code' )


signature:
def get_user ( ) :
    """User information .
 . . note : :
 * * Privacy note * * A users IP address , user agent string , and user id
 ( if logged in ) is sent to a message queue , where it is stored for about
 5 minutes . The information is used to :
 - Detect robot visits from the user agent string .
 - Generate an anonymized visitor id ( using a random salt per day ) .
 - Detect the users host contry based on the IP address .
 The information is then discarded ."""

implementation: return dict ( ip_address = request . remote_addr , user_agent = request . user_agent . string , user_id = ( current_user . get_id ( ) if current_user . is_authenticated else None ) , session_id = session . get ( 'sid_s' ) )


signature:
def default_permission_factory ( query_name , params ) :
    """Default permission factory .
 It enables by default the statistics if they don ' t have a dedicated
 permission factory ."""

implementation:
from invenio_stats import current_stats
if current_stats . queries [ query_name ] . permission_factory is None :
    return AllowAllPermission
else :
    return current_stats . queries [ query_name ] . permission_factory ( query_name , params )


signature:
def load_config ( ) :
    """Load settings from default config and optionally
 overwrite with config file and commandline parameters
 ( in that order ) ."""

implementation:
# We start with the default config
config = flatten ( default_config . DEFAULT_CONFIG )
# Read commandline arguments
cli_config = flatten ( parse_args ( ) )
if "configfile" in cli_config :
    logging . info ( "Reading config file {}" . format ( cli_config [ 'configfile' ] ) )
    configfile = parse_configfile ( cli_config [ 'configfile' ] )
    config = overwrite_config ( config , configfile )
# Parameters from commandline take precedence over all others
config = overwrite_config ( config , cli_config )
# Set verbosity level
if 'verbose' in config :
    if config [ 'verbose' ] == 1 :
        logging . getLogger ( ) . setLevel ( logging . INFO )
    elif config [ 'verbose' ] > 1 :
        logging . getLogger ( ) . setLevel ( logging . DEBUG )
return ObjectView ( config )


signature:
def parse_configfile ( configfile ) :
    """Read settings from file
 : param configfile :"""

implementation:
with open ( configfile ) as f :
    try :
        return yaml . safe_load ( f )
    except Exception as e :
        logging . fatal ( "Could not load default config file: %s" , e )
        exit ( - 1 )


signature:
def register_templates ( ) :
    """Register elasticsearch templates for events ."""

implementation:
event_templates = [ current_stats . _events_config [ e ] [ 'templates' ] for e in current_stats . _events_config ]
aggregation_templates = [ current_stats . _aggregations_config [ a ] [ 'templates' ] for a in current_stats . _aggregations_config ]
return event_templates + aggregation_templates


signature:
def check ( self , query ) :
    """: param query :"""

implementation:
if query . get_type ( ) in { Keyword . LIST , Keyword . DROP } :
    series = query . series_stmt
else :
    series = query . from_stmt
if len ( series ) >= self . min_series_name_length :
    return Ok ( True )
return Err ( "Series name too short. Please be more precise." )


signature:
def process_events ( event_types ) :
    """Index statistics events ."""

implementation:
results = [ ]
for e in event_types :
    processor = current_stats . events [ e ] . processor_class ( ** current_stats . events [ e ] . processor_config )
    results . append ( ( e , processor . run ( ) ) )
return results


signature:
def aggregate_events ( aggregations , start_date = None , end_date = None , update_bookmark = True ) :
    """Aggregate indexed events ."""

implementation:
start_date = dateutil_parse ( start_date ) if start_date else None
end_date = dateutil_parse ( end_date ) if end_date else None
results = [ ]
for a in aggregations :
    aggr_cfg = current_stats . aggregations [ a ]
    aggregator = aggr_cfg . aggregator_class ( name = aggr_cfg . name , ** aggr_cfg . aggregator_config )
    results . append ( aggregator . run ( start_date , end_date , update_bookmark ) )
return results


signature:
def ask ( actor , message ) :
    """Send a message to ` actor ` and return a : class : ` Future ` holding a possible
 reply .
 To receive a result , the actor MUST send a reply to ` sender ` .
 : param actor :
 : type actor : : class : ` ActorRef ` .
 : param message :
 : type message : : type : Any
 : return : A future holding the result ."""

implementation:
sender = PromiseActorRef ( )
actor . tell ( message , sender )
return sender . promise . future


signature:
def get_queries ( parameters ) :
    """Get a list of all queries ( q = . . . parameters ) from an URL parameter string
 : param parameters : The url parameter list"""

implementation:
parsed_params = urlparse . parse_qs ( parameters )
if 'q' not in parsed_params :
    return [ ]
queries = parsed_params [ 'q' ]
# Check if only one query string is given
# in this case make it a list
if not isinstance ( queries , list ) :
    queries = [ queries ]
return queries


signature:
def _handle_request ( self , scheme , netloc , path , headers , body = None , method = "GET" ) :
    """Run the actual request"""

implementation:
backend_url = "{}://{}{}" . format ( scheme , netloc , path )
try :
    response = self . http_request . request ( backend_url , method = method , body = body , headers = dict ( headers ) )
    self . _return_response ( response )
except Exception as e :
    body = "Invalid response from backend: '{}' Server might be busy" . format ( e . message )
    logging . debug ( body )
    self . send_error ( httplib . SERVICE_UNAVAILABLE , body )


signature:
def send_error ( self , code , message = None ) :
    """Send and log plain text error reply .
 : param code :
 : param message :"""

implementation:
message = message . strip ( )
self . log_error ( "code %d, message %s" , code , message )
self . send_response ( code )
self . send_header ( "Content-Type" , "text/plain" )
self . send_header ( 'Connection' , 'close' )
self . end_headers ( )
if message :
    self . wfile . write ( message )


signature:
def _return_response ( self , response ) :
    """: type result : HTTPResponse"""

implementation:
self . filter_headers ( response . msg )
if "content-length" in response . msg :
    del response . msg [ "content-length" ]
self . send_response ( response . status , response . reason )
for header_key , header_value in response . msg . items ( ) :
    self . send_header ( header_key , header_value )
body = response . read ( )
self . send_header ( 'Content-Length' , str ( len ( body ) ) )
self . end_headers ( )
self . wfile . write ( body )


signature:
def anonymize_user ( doc ) :
    """Preprocess an event by anonymizing user information .
 The anonymization is done by removing fields that can uniquely identify a
 user , such as the user ' s ID , session ID , IP address and User Agent , and
 hashing them to produce a ` ` visitor _ id ` ` and ` ` unique _ session _ id ` ` . To
 further secure the method , a randomly generated 32 - byte salt is used , that
 expires after 24 hours and is discarded . The salt values are stored in
 Redis ( or whichever backend Invenio - Cache uses ) . The ` ` unique _ session _ id ` `
 is calculated in the same way as the ` ` visitor _ id ` ` , with the only
 difference that it also takes into account the hour of the event . All of
 these rules effectively mean that a user can have a unique ` ` visitor _ id ` `
 for each day and unique ` ` unique _ session _ id ` ` for each hour of a day .
 This session ID generation process was designed according to the ` Project
 COUNTER Code of Practice < https : / / www . projectcounter . org / code - of -
 practice - sections / general - information / > ` _ .
 In addition to that the country of the user is extracted from the IP
 address as a ISO 3166-1 alpha - 2 two - letter country code ( e . g . " CH " for
 Switzerland ) ."""

implementation:
ip = doc . pop ( 'ip_address' , None )
if ip :
    doc . update ( { 'country' : get_geoip ( ip ) } )
user_id = doc . pop ( 'user_id' , '' )
session_id = doc . pop ( 'session_id' , '' )
user_agent = doc . pop ( 'user_agent' , '' )
# A ' User Session ' is defined as activity by a user in a period of
# one hour . timeslice represents the hour of the day in which
# the event has been generated and together with user info it determines
# the ' User Session '
timestamp = arrow . get ( doc . get ( 'timestamp' ) )
timeslice = timestamp . strftime ( '%Y%m%d%H' )
salt = get_anonymization_salt ( timestamp )
visitor_id = hashlib . sha224 ( salt . encode ( 'utf-8' ) )
# TODO : include random salt here , that changes once a day .
# m . update ( random _ salt )
if user_id :
    visitor_id . update ( user_id . encode ( 'utf-8' ) )
elif session_id :
    visitor_id . update ( session_id . encode ( 'utf-8' ) )
elif ip and user_agent :
    vid = '{}|{}|{}' . format ( ip , user_agent , timeslice )
    visitor_id . update ( vid . encode ( 'utf-8' ) )
else : # TODO : add random data ?
    pass
unique_session_id = hashlib . sha224 ( salt . encode ( 'utf-8' ) )
if user_id :
    sid = '{}|{}' . format ( user_id , timeslice )
    unique_session_id . update ( sid . encode ( 'utf-8' ) )
elif session_id :
    sid = '{}|{}' . format ( session_id , timeslice )
    unique_session_id . update ( sid . encode ( 'utf-8' ) )
elif ip and user_agent :
    sid = '{}|{}|{}' . format ( ip , user_agent , timeslice )
    unique_session_id . update ( sid . encode ( 'utf-8' ) )
doc . update ( dict ( visitor_id = visitor_id . hexdigest ( ) , unique_session_id = unique_session_id . hexdigest ( ) ) )
return doc


signature:
def hash_id ( iso_timestamp , msg ) :
    """Generate event id , optimized for ES ."""

implementation: return '{0}-{1}' . format ( iso_timestamp , hashlib . sha1 ( msg . get ( 'unique_id' ) . encode ( 'utf-8' ) + str ( msg . get ( 'visitor_id' ) ) . encode ( 'utf-8' ) ) . hexdigest ( ) )


signature:
def actionsiter ( self ) :
    """Iterator ."""

implementation:
for msg in self . queue . consume ( ) :
    try :
        for preproc in self . preprocessors :
            msg = preproc ( msg )
            if msg is None :
                break
        if msg is None :
            continue
        suffix = arrow . get ( msg . get ( 'timestamp' ) ) . strftime ( self . suffix )
        ts = parser . parse ( msg . get ( 'timestamp' ) )
        # Truncate timestamp to keep only seconds . This is to improve
        # elasticsearch performances .
        ts = ts . replace ( microsecond = 0 )
        msg [ 'timestamp' ] = ts . isoformat ( )
        # apply timestamp windowing in order to group events too close
        # in time
        if self . double_click_window > 0 :
            timestamp = mktime ( utc . localize ( ts ) . utctimetuple ( ) )
            ts = ts . fromtimestamp ( timestamp // self . double_click_window * self . double_click_window )
        yield dict ( _id = hash_id ( ts . isoformat ( ) , msg ) , _op_type = 'index' , _index = '{0}-{1}' . format ( self . index , suffix ) , _type = self . doctype , _source = msg , )
    except Exception :
        current_app . logger . exception ( u'Error while processing event' )


signature:
def run ( self ) :
    """Process events queue ."""

implementation: return elasticsearch . helpers . bulk ( self . client , self . actionsiter ( ) , stats_only = True , chunk_size = 50 )


signature:
def parse ( duration_seconds , resolution_seconds = Resolution . MAX_RESOLUTION , limit = None ) :
    """num _ datapoints = min ( duration / resolution , limit )
 : param duration _ seconds : Time duration ( in seconds ) for which datapoints should be returned
 : param resolution _ seconds : Time interval ( in seconds ) between data points
 : param limit : Maximum number of datapoints to return"""

implementation:
if not duration_seconds or duration_seconds < 0 :
    return 0
if not resolution_seconds or resolution_seconds <= 0 :
    return None
num_datapoints = duration_seconds / resolution_seconds
if limit :
    num_datapoints = min ( int ( limit ) , num_datapoints )
return int ( math . ceil ( num_datapoints ) )


signature:
def create_series ( self , num_series , batch_size = 5000 ) :
    """Write one data point for each series name to initialize the series
 : param num _ series : Number of different series names to create
 : param batch _ size : Number of series to create at the same time
 : return :"""

implementation:
datapoints = [ ]
for _ in range ( num_series ) :
    name = self . dummy_seriesname ( )
    datapoints . append ( self . create_datapoint ( name , [ "value" ] , [ [ 1 ] ] ) )
for data in tqdm ( self . batch ( datapoints , batch_size ) ) :
    self . client . write_points ( data )


signature:
def write_points ( self , series_name , start_date , end_date , resolution = 10 , batch_size = 5000 ) :
    """Create sample datapoints between two dates with the given resolution ( in seconds )
 : param series _ name :
 : param start _ date :
 : param end _ date :
 : param resolution :
 : param batch _ size :"""

implementation:
start_ts = int ( start_date . strftime ( "%s" ) )
end_ts = int ( end_date . strftime ( "%s" ) )
range_seconds = end_ts - start_ts
num_datapoints = range_seconds / resolution
timestamps = [ start_ts + i * resolution for i in range ( num_datapoints ) ]
columns = [ "time" , "value" ]
for batch in tqdm ( self . batch ( timestamps , batch_size ) ) :
    points = [ ]
    for timestamp in batch :
        point = random . randint ( 1 , 100 )
        points . append ( [ timestamp , point ] )
    datapoint = self . create_datapoint ( series_name , columns , points )
    self . client . write_points ( [ datapoint ] )


signature:
def register_events ( ) :
    """Register sample events ."""

implementation: return [ dict ( event_type = 'file-download' , templates = 'invenio_stats.contrib.file_download' , processor_class = EventsIndexer , processor_config = dict ( preprocessors = [ flag_robots , anonymize_user , build_file_unique_id ] ) ) , dict ( event_type = 'record-view' , templates = 'invenio_stats.contrib.record_view' , processor_class = EventsIndexer , processor_config = dict ( preprocessors = [ flag_robots , anonymize_user , build_record_unique_id ] ) ) ]


signature:
def register_aggregations ( ) :
    """Register sample aggregations ."""

implementation: return [ dict ( aggregation_name = 'file-download-agg' , templates = 'invenio_stats.contrib.aggregations.aggr_file_download' , aggregator_class = StatAggregator , aggregator_config = dict ( client = current_search_client , event = 'file-download' , aggregation_field = 'unique_id' , aggregation_interval = 'day' , copy_fields = dict ( file_key = 'file_key' , bucket_id = 'bucket_id' , file_id = 'file_id' , ) , metric_aggregation_fields = { 'unique_count' : ( 'cardinality' , 'unique_session_id' , { 'precision_threshold' : 1000 } ) , 'volume' : ( 'sum' , 'size' , { } ) , } , ) ) , dict ( aggregation_name = 'record-view-agg' , templates = 'invenio_stats.contrib.aggregations.aggr_record_view' , aggregator_class = StatAggregator , aggregator_config = dict ( client = current_search_client , event = 'record-view' , aggregation_field = 'unique_id' , aggregation_interval = 'day' , copy_fields = dict ( record_id = 'record_id' , pid_type = 'pid_type' , pid_value = 'pid_value' , ) , metric_aggregation_fields = { 'unique_count' : ( 'cardinality' , 'unique_session_id' , { 'precision_threshold' : 1000 } ) , } , ) ) ]


signature:
def register_queries ( ) :
    """Register queries ."""

implementation:
return [ dict ( query_name = 'bucket-file-download-histogram' , query_class = ESDateHistogramQuery , query_config = dict ( index = 'stats-file-download' , doc_type = 'file-download-day-aggregation' , copy_fields = dict ( bucket_id = 'bucket_id' , file_key = 'file_key' , ) , required_filters = dict ( bucket_id = 'bucket_id' , file_key = 'file_key' , ) ) ) , dict ( query_name = 'bucket-file-download-total' , query_class = ESTermsQuery , query_config = dict ( index = 'stats-file-download' , doc_type = 'file-download-day-aggregation' , copy_fields = dict ( # bucket _ id = ' bucket _ id ' ,
) , required_filters = dict ( bucket_id = 'bucket_id' , ) , aggregated_fields = [ 'file_key' ] ) ) , ]


signature:
def check ( self , query ) :
    """: param query :"""

implementation:
if query . get_type ( ) not in { Keyword . SELECT } : # Bailing out for non select queries
    return Ok ( True )
if query . get_resolution ( ) > 0 :
    return Ok ( True )
return Err ( "Group by statements need a positive time value (e.g. time(10s))" )


signature:
def declare_queues ( ) :
    """Index statistics events ."""

implementation: return [ dict ( name = 'stats-{0}' . format ( event [ 'event_type' ] ) , exchange = current_stats . exchange ) for event in current_stats . _events_config . values ( ) ]


signature:
def parse ( self , raw_query_string ) :
    """Parse a raw query string into fields
 : param raw _ query _ string : Raw InfluxDB query string"""

implementation:
self . _reset ( )
if not isinstance ( raw_query_string , basestring ) :
    return None
query_string = self . _cleanup ( raw_query_string )
parts = self . _split ( query_string )
parts = self . _sanitize_keywords ( parts )
tokens = self . _tokenize ( parts )
if tokens : # Run subparsers to analyze parts of the query
    self . parsed_resolution = self . _parse_resolution ( tokens )
    self . parsed_time = self . _parse_time ( tokens )
    self . parsed_time_overlap = self . _parse_duration ( self . parsed_time )
    self . parsed_datapoints = self . _parse_datapoints ( self . parsed_time_overlap . timespan_seconds ( ) , self . parsed_resolution , self . parse_keyword ( Keyword . LIMIT , tokens ) )
return self . create_query_object ( tokens )


signature:
def create_query_object ( self , tokens ) :
    """Analyze query tokens and create an InfluxDBStatement from them
 Return None on error
 : param tokens : A list of InfluxDB query tokens"""

implementation:
try :
    query_type = tokens [ 'type' ]
    return getattr ( self , 'create_%s_query' % query_type ) ( tokens )
except ( KeyError , TypeError ) :
    return self . invalid_query ( tokens )


signature:
def create_select_query ( self , tokens ) :
    """Parse tokens of select query
 : param tokens : A list of InfluxDB query tokens"""

implementation:
if not tokens [ Keyword . SELECT ] :
    return None
if not tokens [ Keyword . FROM ] :
    return None
return SelectQuery ( self . parse_keyword ( Keyword . SELECT , tokens ) , self . parse_keyword ( Keyword . FROM , tokens ) , where_stmt = self . parse_keyword ( Keyword . WHERE , tokens ) , limit_stmt = self . parse_keyword ( Keyword . LIMIT , tokens ) , group_by_stmt = self . parse_group ( tokens ) , duration = self . parsed_time_overlap . timespan_seconds ( ) , resolution = self . parsed_resolution , time_ranges = self . parsed_time , time_overlap = self . parsed_time_overlap , datapoints = self . parsed_datapoints )


signature:
def create_list_query ( self , tokens ) :
    """Parse tokens of list query
 : param tokens : A list of InfluxDB query tokens"""

implementation:
if not tokens [ Keyword . SERIES ] : # A list series keyword is allowed
# without a series name or regex
    tokens [ Keyword . SERIES ] = ''
return ListQuery ( self . parse_keyword ( Keyword . SERIES , tokens ) )


signature:
def create_drop_query ( self , tokens ) :
    """Parse tokens of drop query
 : param tokens : A list of InfluxDB query tokens"""

implementation:
if not tokens [ Keyword . SERIES ] :
    return None
return DropQuery ( self . parse_keyword ( Keyword . SERIES , tokens ) )


signature:
def create_delete_query ( self , tokens ) :
    """Parse tokens of delete query
 : param tokens : A list of InfluxDB query tokens"""

implementation:
# From keyword is required
if not tokens [ Keyword . FROM ] :
    return None
where_stmt = self . parse_keyword ( Keyword . WHERE , tokens )
if where_stmt :
    if not where_stmt . startswith ( 'time' ) :
        return None
return DeleteQuery ( self . parse_keyword ( Keyword . FROM , tokens ) , self . parse_keyword ( Keyword . WHERE , tokens ) )


signature:
def _parse_time ( self , tokens ) :
    """Parse the date range for the query
 E . g . WHERE time > now ( ) - 48h AND time < now ( ) - 24h
 would result in DateRange ( datetime _ start , datetime _ end )
 where
 datetime _ start would be parsed from now ( ) - 48h
 and
 datetime _ end would be parsed from now ( ) - 24h
 : param tokens :
 : return :"""

implementation: return self . time_parser . parse ( self . parse_keyword ( Keyword . WHERE , tokens ) )


signature:
def _parse_resolution ( self , tokens ) :
    """Parse resolution from the GROUP BY statement .
 E . g . GROUP BY time ( 10s ) would mean a 10 second resolution
 : param tokens :
 : return :"""

implementation: return self . resolution_parser . parse ( self . parse_keyword ( Keyword . GROUP_BY , tokens ) )


signature:
def _parse_datapoints ( self , parsed_duration , parsed_resolution , limit ) :
    """Parse the number of datapoints of a query .
 This can be calculated from the given duration and resolution of the query .
 E . g . if the query has a duation of 2*60*60 = 7200 seconds and a resolution of 10 seconds
 then the number of datapoints would be 7200/10 = > 7200 datapoints .
 : param parsed _ duration :
 : param parsed _ resolution :
 : param limit :
 : return :"""

implementation: return self . datapoints_parser . parse ( parsed_duration , parsed_resolution , limit )


implementation:
if query . get_type ( ) not in { Keyword . SELECT } : # Only select queries need to be checked here
# All others are not affected by this rule . Bailing out .
    return Ok ( True )
earliest_date = query . get_earliest_date ( )
if earliest_date >= self . min_start_date :
    return Ok ( True )
if query . limit_stmt :
    return Ok ( True )
return Err ( ( "Querying for data before {} is prohibited. " "Your beginning date is {}, which is before that." ) . format ( self . min_start_date . strftime ( "%Y-%m-%d" ) , earliest_date ) )


signature:
def extract_date ( self , date ) :
    """Extract date from string if necessary .
 : returns : the extracted date ."""

implementation:
if isinstance ( date , six . string_types ) :
    try :
        date = dateutil . parser . parse ( date )
    except ValueError :
        raise ValueError ( 'Invalid date format for statistic {}.' ) . format ( self . query_name )
if not isinstance ( date , datetime ) :
    raise TypeError ( 'Invalid date type for statistic {}.' ) . format ( self . query_name )
return date


signature:
def validate_arguments ( self , interval , start_date , end_date , ** kwargs ) :
    """Validate query arguments ."""

implementation:
if interval not in self . allowed_intervals :
    raise InvalidRequestInputError ( 'Invalid aggregation time interval for statistic {}.' ) . format ( self . query_name )
if set ( kwargs ) < set ( self . required_filters ) :
    raise InvalidRequestInputError ( 'Missing one of the required parameters {0} in ' 'query {1}' . format ( set ( self . required_filters . keys ( ) ) , self . query_name ) )


signature:
def build_query ( self , interval , start_date , end_date , ** kwargs ) :
    """Build the elasticsearch query ."""

implementation:
agg_query = Search ( using = self . client , index = self . index , doc_type = self . doc_type ) [ 0 : 0 ]
if start_date is not None or end_date is not None :
    time_range = { }
    if start_date is not None :
        time_range [ 'gte' ] = start_date . isoformat ( )
    if end_date is not None :
        time_range [ 'lte' ] = end_date . isoformat ( )
    agg_query = agg_query . filter ( 'range' , ** { self . time_field : time_range } )
for modifier in self . query_modifiers :
    agg_query = modifier ( agg_query , ** kwargs )
base_agg = agg_query . aggs . bucket ( 'histogram' , 'date_histogram' , field = self . time_field , interval = interval )
for destination , ( metric , field , opts ) in self . metric_fields . items ( ) :
    base_agg . metric ( destination , metric , field = field , ** opts )
if self . copy_fields :
    base_agg . metric ( 'top_hit' , 'top_hits' , size = 1 , sort = { 'timestamp' : 'desc' } )
for query_param , filtered_field in self . required_filters . items ( ) :
    if query_param in kwargs :
        agg_query = agg_query . filter ( 'term' , ** { filtered_field : kwargs [ query_param ] } )
return agg_query


signature:
def validate_arguments ( self , start_date , end_date , ** kwargs ) :
    """Validate query arguments ."""

implementation:
if set ( kwargs ) < set ( self . required_filters ) :
    raise InvalidRequestInputError ( 'Missing one of the required parameters {0} in ' 'query {1}' . format ( set ( self . required_filters . keys ( ) ) , self . query_name ) )


signature:
def build_query ( self , start_date , end_date , ** kwargs ) :
    """Build the elasticsearch query ."""

implementation:
agg_query = Search ( using = self . client , index = self . index , doc_type = self . doc_type ) [ 0 : 0 ]
if start_date is not None or end_date is not None :
    time_range = { }
    if start_date is not None :
        time_range [ 'gte' ] = start_date . isoformat ( )
    if end_date is not None :
        time_range [ 'lte' ] = end_date . isoformat ( )
    agg_query = agg_query . filter ( 'range' , ** { self . time_field : time_range } )
for modifier in self . query_modifiers :
    agg_query = modifier ( agg_query , ** kwargs )
base_agg = agg_query . aggs
def _apply_metric_aggs ( agg ) :
    for dst , ( metric , field , opts ) in self . metric_fields . items ( ) :
        agg . metric ( dst , metric , field = field , ** opts )
_apply_metric_aggs ( base_agg )
if self . aggregated_fields :
    cur_agg = base_agg
    for term in self . aggregated_fields :
        cur_agg = cur_agg . bucket ( term , 'terms' , field = term , size = 0 )
        _apply_metric_aggs ( cur_agg )
if self . copy_fields :
    base_agg . metric ( 'top_hit' , 'top_hits' , size = 1 , sort = { 'timestamp' : 'desc' } )
for query_param , filtered_field in self . required_filters . items ( ) :
    if query_param in kwargs :
        agg_query = agg_query . filter ( 'term' , ** { filtered_field : kwargs [ query_param ] } )
return agg_query


signature:
def run ( self , start_date = None , end_date = None , ** kwargs ) :
    """Run the query ."""

implementation:
start_date = self . extract_date ( start_date ) if start_date else None
end_date = self . extract_date ( end_date ) if end_date else None
self . validate_arguments ( start_date , end_date , ** kwargs )
agg_query = self . build_query ( start_date , end_date , ** kwargs )
query_result = agg_query . execute ( ) . to_dict ( )
res = self . process_query_result ( query_result , start_date , end_date )
return res


signature:
def handle_error ( self , request , client_address ) :
    """Overwrite error handling to suppress socket / ssl related errors
 : param client _ address : Address of client
 : param request : Request causing an error"""

implementation:
cls , e = sys . exc_info ( ) [ : 2 ]
if cls is socket . error or cls is ssl . SSLError :
    pass
else :
    return HTTPServer . handle_error ( self , request , client_address )


signature:
def file_download_event_builder ( event , sender_app , obj = None , ** kwargs ) :
    """Build a file - download event ."""

implementation:
event . update ( dict ( # When :
timestamp = datetime . datetime . utcnow ( ) . isoformat ( ) , # What :
bucket_id = str ( obj . bucket_id ) , file_id = str ( obj . file_id ) , file_key = obj . key , size = obj . file . size , referrer = request . referrer , # Who :
** get_user ( ) ) )
return event


signature:
def record_view_event_builder ( event , sender_app , pid = None , record = None , ** kwargs ) :
    """Build a record - view event ."""

implementation:
event . update ( dict ( # When :
timestamp = datetime . datetime . utcnow ( ) . isoformat ( ) , # What :
record_id = str ( record . id ) , pid_type = pid . pid_type , pid_value = str ( pid . pid_value ) , referrer = request . referrer , # Who :
** get_user ( ) ) )
return event


signature:
def main ( ) :
    """Setup consumer"""

implementation:
config = loader . load_config ( )
if config . version :
    show_version ( )
if config . show_rules :
    show_rules ( )
if not config . configfile and not ( hasattr ( config , "status" ) or hasattr ( config , "stop" ) ) :
    show_configfile_warning ( )
# Check if we have permissions to open the log file .
check_write_permissions ( config . logfile )
start_proxy ( config )


signature:
def check_write_permissions ( file ) :
    """Check if we can write to the given file
 Otherwise since we might detach the process to run in the background
 we might never find out that writing failed and get an ugly
 exit message on startup . For example :
 ERROR : Child exited immediately with non - zero exit code 127
 So we catch this error upfront and print a nicer error message
 with a hint on how to fix it ."""

implementation:
try :
    open ( file , 'a' )
except IOError :
    print ( "Can't open file {}. " "Please grant write permissions or change the path in your config" . format ( file ) )
    sys . exit ( 1 )


signature:
def show_rules ( ) :
    """Show the list of available rules and quit
 : return :"""

implementation:
from rules . loader import import_rules
from rules . rule_list import all_rules
rules = import_rules ( all_rules )
print ( "" )
for name , rule in rules . iteritems ( ) :
    heading = "{} (`{}`)" . format ( rule . description ( ) , name )
    print ( "#### {} ####" . format ( heading ) )
    for line in rule . reason ( ) :
        print ( line )
    print ( "" )
sys . exit ( 0 )


signature:
def start_proxy ( config ) :
    """Start the http proxy
 : param config :
 : return :"""

implementation:
protector = Protector ( config . rules , config . whitelist )
protector_daemon = ProtectorDaemon ( config = config , protector = protector )
daemon = daemonocle . Daemon ( pidfile = config . pidfile , detach = ( not config . foreground ) , shutdown_callback = shutdown , worker = protector_daemon . run )
daemon . do_action ( config . command )


signature:
def batches ( iterable , n = 1 ) :
    """From http : / / stackoverflow . com / a / 8290508/270334
 : param n :
 : param iterable :"""

implementation:
l = len ( iterable )
for ndx in range ( 0 , l , n ) :
    yield iterable [ ndx : min ( ndx + n , l ) ]


signature:
def _is_root ( ) :
    """Checks if the user is rooted ."""

implementation:
import os
import ctypes
try :
    return os . geteuid ( ) == 0
except AttributeError :
    return ctypes . windll . shell32 . IsUserAnAdmin ( ) != 0
return False


signature:
def cmdclass ( path , enable = None , user = None ) :
    """Build nbextension cmdclass dict for the setuptools . setup method .
 Parameters
 path : str
 Directory relative to the setup file that the nbextension code
 lives in .
 enable : [ str = None ]
 Extension to " enable " . Enabling an extension causes it to be loaded
 automatically by the IPython notebook .
 user : [ bool = None ]
 Whether or not the nbextension should be installed in user mode .
 If this is undefined , the script will install as user mode IF the
 installer is not sudo .
 Usage
 For automatic loading :
 # Assuming ` . / extension ` is the relative path to the JS files and
 # ` . / extension / main . js ` is the file that you want automatically loaded .
 setup (
 name = ' extension ' ,
 cmdclass = cmdclass ( ' extension ' , ' extension / main ' ) ,
 For manual loading :
 # Assuming ` . / extension ` is the relative path to the JS files .
 setup (
 name = ' extension ' ,
 cmdclass = cmdclass ( ' extension ' ) ,"""

implementation:
import warnings
from setuptools . command . install import install
from setuptools . command . develop import develop
from os . path import dirname , join , exists , realpath
from traceback import extract_stack
try : # IPython / Jupyter 4.0
    from notebook . nbextensions import install_nbextension
    from notebook . services . config import ConfigManager
except ImportError : # Pre - schism
    try :
        from IPython . html . nbextensions import install_nbextension
        from IPython . html . services . config import ConfigManager
    except ImportError :
        warnings . warn ( "No jupyter notebook found in your environment. " "Hence jupyter nbextensions were not installed. " "If you would like to have them," "please issue 'pip install jupyter'." )
        return { }
# Check if the user flag was set .
if user is None :
    user = not _is_root ( )
# Get the path of the extension
calling_file = extract_stack ( ) [ - 2 ] [ 0 ]
fullpath = realpath ( calling_file )
if not exists ( fullpath ) :
    raise Exception ( 'Could not find path of setup file.' )
extension_dir = join ( dirname ( fullpath ) , path )
# Installs the nbextension
def run_nbextension_install ( develop ) :
    import sys
    sysprefix = hasattr ( sys , 'real_prefix' )
    if sysprefix :
        install_nbextension ( extension_dir , symlink = develop , sys_prefix = sysprefix )
    else :
        install_nbextension ( extension_dir , symlink = develop , user = user )
    if enable is not None :
        print ( "Enabling the extension ..." )
        cm = ConfigManager ( )
        cm . update ( 'notebook' , { "load_extensions" : { enable : True } } )
# Command used for standard installs
class InstallCommand ( install ) :
    def run ( self ) :
        print ( "Installing Python module..." )
        install . run ( self )
        print ( "Installing nbextension ..." )
        run_nbextension_install ( False )
# Command used for development installs ( symlinks the JS )
class DevelopCommand ( develop ) :
    def run ( self ) :
        print ( "Installing Python module..." )
        develop . run ( self )
        print ( "Installing nbextension ..." )
        run_nbextension_install ( True )
return { 'install' : InstallCommand , 'develop' : DevelopCommand , }


signature:
def count_subgraph_sizes ( graph : BELGraph , annotation : str = 'Subgraph' ) -> Counter [ int ] :
    """Count the number of nodes in each subgraph induced by an annotation .
 : param annotation : The annotation to group by and compare . Defaults to ' Subgraph '
 : return : A dictionary from { annotation value : number of nodes }"""

implementation: return count_dict_values ( group_nodes_by_annotation ( graph , annotation ) )


signature:
def calculate_subgraph_edge_overlap ( graph : BELGraph , annotation : str = 'Subgraph' ) -> Tuple [ Mapping [ str , EdgeSet ] , Mapping [ str , Mapping [ str , EdgeSet ] ] , Mapping [ str , Mapping [ str , EdgeSet ] ] , Mapping [ str , Mapping [ str , float ] ] , ] :
    """Build a DatafFame to show the overlap between different sub - graphs .
 Options :
 1 . Total number of edges overlap ( intersection )
 2 . Percentage overlap ( tanimoto similarity )
 : param graph : A BEL graph
 : param annotation : The annotation to group by and compare . Defaults to ' Subgraph '
 : return : { subgraph : set of edges } , { ( subgraph 1 , subgraph2 ) : set of intersecting edges } ,
 { ( subgraph 1 , subgraph2 ) : set of unioned edges } , { ( subgraph 1 , subgraph2 ) : tanimoto similarity } ,"""

implementation:
sg2edge = defaultdict ( set )
for u , v , d in graph . edges ( data = True ) :
    if not edge_has_annotation ( d , annotation ) :
        continue
    sg2edge [ d [ ANNOTATIONS ] [ annotation ] ] . add ( ( u , v ) )
subgraph_intersection = defaultdict ( dict )
subgraph_union = defaultdict ( dict )
result = defaultdict ( dict )
for sg1 , sg2 in itt . product ( sg2edge , repeat = 2 ) :
    subgraph_intersection [ sg1 ] [ sg2 ] = sg2edge [ sg1 ] & sg2edge [ sg2 ]
    subgraph_union [ sg1 ] [ sg2 ] = sg2edge [ sg1 ] | sg2edge [ sg2 ]
    result [ sg1 ] [ sg2 ] = len ( subgraph_intersection [ sg1 ] [ sg2 ] ) / len ( subgraph_union [ sg1 ] [ sg2 ] )
return sg2edge , subgraph_intersection , subgraph_union , result


signature:
def summarize_subgraph_edge_overlap ( graph : BELGraph , annotation : str = 'Subgraph' ) -> Mapping [ str , Mapping [ str , float ] ] :
    """Return a similarity matrix between all subgraphs ( or other given annotation ) .
 : param annotation : The annotation to group by and compare . Defaults to : code : ` " Subgraph " `
 : return : A similarity matrix in a dict of dicts
 : rtype : dict"""

implementation:
_ , _ , _ , subgraph_overlap = calculate_subgraph_edge_overlap ( graph , annotation )
return subgraph_overlap


signature:
def summarize_subgraph_node_overlap ( graph : BELGraph , node_predicates = None , annotation : str = 'Subgraph' ) :
    """Calculate the subgraph similarity tanimoto similarity in nodes passing the given filter .
 Provides an alternate view on subgraph similarity , from a more node - centric view"""

implementation:
r1 = group_nodes_by_annotation_filtered ( graph , node_predicates = node_predicates , annotation = annotation )
return calculate_tanimoto_set_distances ( r1 )


signature:
def rank_subgraph_by_node_filter ( graph : BELGraph , node_predicates : Union [ NodePredicate , Iterable [ NodePredicate ] ] , annotation : str = 'Subgraph' , reverse : bool = True , ) -> List [ Tuple [ str , int ] ] :
    """Rank sub - graphs by which have the most nodes matching an given filter .
 A use case for this function would be to identify which subgraphs contain the most differentially expressed
 genes .
 > > > from pybel import from _ pickle
 > > > from pybel . constants import GENE
 > > > from pybel _ tools . integration import overlay _ type _ data
 > > > from pybel _ tools . summary import rank _ subgraph _ by _ node _ filter
 > > > import pandas as pd
 > > > graph = from _ pickle ( ' ~ / dev / bms / aetionomy / alzheimers . gpickle ' )
 > > > df = pd . read _ csv ( ' ~ / dev / bananas / data / alzheimers _ dgxp . csv ' , columns = [ ' Gene ' , ' log2fc ' ] )
 > > > data = { gene : log2fc for _ , gene , log2fc in df . itertuples ( ) }
 > > > overlay _ type _ data ( graph , data , ' log2fc ' , GENE , ' HGNC ' , impute = 0.0)
 > > > results = rank _ subgraph _ by _ node _ filter ( graph , lambda g , n : 1.3 < abs ( g [ n ] [ ' log2fc ' ] ) )"""

implementation:
r1 = group_nodes_by_annotation_filtered ( graph , node_predicates = node_predicates , annotation = annotation )
r2 = count_dict_values ( r1 )
# TODO use instead : r2 . most _ common ( )
return sorted ( r2 . items ( ) , key = itemgetter ( 1 ) , reverse = reverse )


signature:
def to_jupyter ( graph : BELGraph , chart : Optional [ str ] = None ) -> Javascript :
    """Render the graph as JavaScript in a Jupyter Notebook ."""

implementation:
with open ( os . path . join ( HERE , 'render_with_javascript.js' ) , 'rt' ) as f :
    js_template = Template ( f . read ( ) )
return Javascript ( js_template . render ( ** _get_context ( graph , chart = chart ) ) )


signature:
def to_html ( graph : BELGraph , chart : Optional [ str ] = None ) -> str :
    """Render the graph as an HTML string .
 Common usage may involve writing to a file like :
 > > > from pybel . examples import sialic _ acid _ graph
 > > > with open ( ' ideogram _ output . html ' , ' w ' ) as file :
 . . . print ( to _ html ( sialic _ acid _ graph ) , file = file )"""

implementation:
with open ( os . path . join ( HERE , 'index.html' ) , 'rt' ) as f :
    html_template = Template ( f . read ( ) )
return html_template . render ( ** _get_context ( graph , chart = chart ) )


signature:
def prerender ( graph : BELGraph ) -> Mapping [ str , Mapping [ str , Any ] ] :
    """Generate the annotations JSON for Ideogram ."""

implementation:
import bio2bel_hgnc
from bio2bel_hgnc . models import HumanGene
graph : BELGraph = graph . copy ( )
enrich_protein_and_rna_origins ( graph )
collapse_all_variants ( graph )
genes : Set [ Gene ] = get_nodes_by_function ( graph , GENE )
hgnc_symbols = { gene . name for gene in genes if gene . namespace . lower ( ) == 'hgnc' }
result = { }
hgnc_manager = bio2bel_hgnc . Manager ( )
human_genes = ( hgnc_manager . session . query ( HumanGene . symbol , HumanGene . location ) . filter ( HumanGene . symbol . in_ ( hgnc_symbols ) ) . all ( ) )
for human_gene in human_genes :
    result [ human_gene . symbol ] = { 'name' : human_gene . symbol , 'chr' : ( human_gene . location . split ( 'q' ) [ 0 ] if 'q' in human_gene . location else human_gene . location . split ( 'p' ) [ 0 ] ) , }
df = get_df ( )
for _ , ( gene_id , symbol , start , stop ) in df [ df [ 'Symbol' ] . isin ( hgnc_symbols ) ] . iterrows ( ) :
    result [ symbol ] [ 'start' ] = start
    result [ symbol ] [ 'stop' ] = stop
return result


signature:
def plot_summary_axes ( graph : BELGraph , lax , rax , logx = True ) :
    """Plots your graph summary statistics on the given axes .
 After , you should run : func : ` plt . tight _ layout ` and you must run : func : ` plt . show ` to view .
 Shows :
 1 . Count of nodes , grouped by function type
 2 . Count of edges , grouped by relation type
 : param pybel . BELGraph graph : A BEL graph
 : param lax : An axis object from matplotlib
 : param rax : An axis object from matplotlib
 Example usage :
 > > > import matplotlib . pyplot as plt
 > > > from pybel import from _ pickle
 > > > from pybel _ tools . summary import plot _ summary _ axes
 > > > graph = from _ pickle ( ' ~ / dev / bms / aetionomy / parkinsons . gpickle ' )
 > > > fig , axes = plt . subplots ( 1 , 2 , figsize = ( 10 , 4 ) )
 > > > plot _ summary _ axes ( graph , axes [ 0 ] , axes [ 1 ] )
 > > > plt . tight _ layout ( )
 > > > plt . show ( )"""

implementation:
ntc = count_functions ( graph )
etc = count_relations ( graph )
df = pd . DataFrame . from_dict ( dict ( ntc ) , orient = 'index' )
df_ec = pd . DataFrame . from_dict ( dict ( etc ) , orient = 'index' )
df . sort_values ( 0 , ascending = True ) . plot ( kind = 'barh' , logx = logx , ax = lax )
lax . set_title ( 'Number of nodes: {}' . format ( graph . number_of_nodes ( ) ) )
df_ec . sort_values ( 0 , ascending = True ) . plot ( kind = 'barh' , logx = logx , ax = rax )
rax . set_title ( 'Number of edges: {}' . format ( graph . number_of_edges ( ) ) )


signature:
def plot_summary ( graph : BELGraph , plt , logx = True , ** kwargs ) :
    """Plots your graph summary statistics . This function is a thin wrapper around : func : ` plot _ summary _ axis ` . It
 automatically takes care of building figures given matplotlib ' s pyplot module as an argument . After , you need
 to run : func : ` plt . show ` .
 : code : ` plt ` is given as an argument to avoid needing matplotlib as a dependency for this function
 Shows :
 1 . Count of nodes , grouped by function type
 2 . Count of edges , grouped by relation type
 : param plt : Give : code : ` matplotlib . pyplot ` to this parameter
 : param kwargs : keyword arguments to give to : func : ` plt . subplots `
 Example usage :
 > > > import matplotlib . pyplot as plt
 > > > from pybel import from _ pickle
 > > > from pybel _ tools . summary import plot _ summary
 > > > graph = from _ pickle ( ' ~ / dev / bms / aetionomy / parkinsons . gpickle ' )
 > > > plot _ summary ( graph , plt , figsize = ( 10 , 4 ) )
 > > > plt . show ( )"""

implementation:
fig , axes = plt . subplots ( 1 , 2 , ** kwargs )
lax = axes [ 0 ]
rax = axes [ 1 ]
plot_summary_axes ( graph , lax , rax , logx = logx )
plt . tight_layout ( )
return fig , axes


signature:
def remove_nodes_by_function_namespace ( graph : BELGraph , func : str , namespace : Strings ) -> None :
    """Remove nodes with the given function and namespace .
 This might be useful to exclude information learned about distant species , such as excluding all information
 from MGI and RGD in diseases where mice and rats don ' t give much insight to the human disease mechanism ."""

implementation: remove_filtered_nodes ( graph , function_namespace_inclusion_builder ( func , namespace ) )


signature:
def preprocessing_excel ( path ) :
    """Preprocess the excel sheet
 : param filepath : filepath of the excel data
 : return : df : pandas dataframe with excel data
 : rtype : pandas . DataFrame"""

implementation:
if not os . path . exists ( path ) :
    raise ValueError ( "Error: %s file not found" % path )
# Import Models from Excel sheet , independent for AD and PD
df = pd . read_excel ( path , sheetname = 0 , header = 0 )
# Indexes and column name
# [ log . info ( str ( x ) + ' : ' + str ( ( df . columns . values [ x ] ) ) ) for x in range ( 0 , len ( df . columns . values ) ) ]
# Starting from 4 : Pathway Name
# Fill Pathway cells that are merged and are ' NaN ' after deleting rows where there is no genes
df . iloc [ : , 0 ] = pd . Series ( df . iloc [ : , 0 ] ) . fillna ( method = 'ffill' )
# Number of gaps
# log . info ( df . ix [ : , 6 ] . isnull ( ) . sum ( ) )
df = df [ df . ix [ : , 1 ] . notnull ( ) ]
df = df . reset_index ( drop = True )
# Fill NaN to ceros in PubmedID column
df . ix [ : , 2 ] . fillna ( 0 , inplace = True )
# Number of gaps in the gene column should be already zero
if ( df . ix [ : , 1 ] . isnull ( ) . sum ( ) ) != 0 :
    raise ValueError ( "Error: Empty cells in the gene column" )
# Check current state
# df . to _ csv ( ' out . csv ' )
return df


signature:
def preprocessing_br_projection_excel ( path : str ) -> pd . DataFrame :
    """Preprocess the excel file .
 Parameters
 path : Filepath of the excel sheet"""

implementation:
if not os . path . exists ( path ) :
    raise ValueError ( "Error: %s file not found" % path )
return pd . read_excel ( path , sheetname = 0 , header = 0 )


signature:
def get_nift_values ( ) -> Mapping [ str , str ] :
    """Extract the list of NIFT names from the BEL resource and builds a dictionary mapping from the lowercased version
 to the uppercase version ."""

implementation:
r = get_bel_resource ( NIFT )
return { name . lower ( ) : name for name in r [ 'Values' ] }


signature:
def write_neurommsig_bel ( file , df : pd . DataFrame , disease : str , nift_values : Mapping [ str , str ] , ) :
    """Writes the NeuroMMSigDB excel sheet to BEL
 : param file : a file or file - like that can be writen to
 : param df :
 : param disease :
 : param nift _ values : a dictionary of lowercased to normal names in NIFT"""

implementation:
write_neurommsig_biolerplate ( disease , file )
missing_features = set ( )
fixed_caps = set ( )
nift_value_originals = set ( nift_values . values ( ) )
graph = BELGraph ( name = f'NeuroMMSigDB for {disease}' , description = f'SNP and Clinical Features for Subgraphs in {disease}' , authors = 'Daniel Domingo-Fernández, Charles Tapley Hoyt, Mufassra Naz, Aybuge Altay, Anandhi Iyappan' , contact = 'daniel.domingo.fernandez@scai.fraunhofer.de' , version = time . strftime ( '%Y%m%d' ) , )
for pathway , pathway_df in df . groupby ( pathway_column ) :
    sorted_pathway_df = pathway_df . sort_values ( genes_column )
    sliced_df = sorted_pathway_df [ columns ] . itertuples ( )
    for _ , gene , pubmeds , lit_snps , gwas_snps , ld_block_snps , clinical_features , clinical_snps in sliced_df :
        gene = ensure_quotes ( gene )
        for snp in itt . chain ( lit_snps or [ ] , gwas_snps or [ ] , ld_block_snps or [ ] , clinical_snps or [ ] ) :
            if not snp . strip ( ) :
                continue
            graph . add_association ( Gene ( 'HGNC' , gene ) , Gene ( 'DBSNP' , snp ) , evidence = 'Serialized from NeuroMMSigDB' , citation = '28651363' , annotations = { 'MeSHDisease' : disease , } , )
        for clinical_feature in clinical_features or [ ] :
            if not clinical_feature . strip ( ) :
                continue
            if clinical_feature . lower ( ) not in nift_values :
                missing_features . add ( clinical_feature )
                continue
            if clinical_feature not in nift_value_originals :
                fixed_caps . add ( ( clinical_feature , nift_values [ clinical_feature . lower ( ) ] ) )
                clinical_feature = nift_values [ clinical_feature . lower ( ) ]
            # fix capitalization
            graph . add_association ( Gene ( 'HGNC' , gene ) , Abundance ( 'NIFT' , clinical_feature ) , evidence = 'Serialized from NeuroMMSigDB' , citation = '28651363' , annotations = { 'MeSHDisease' : disease , } , )
            if clinical_snps :
                for clinical_snp in clinical_snps :
                    graph . add_association ( Gene ( 'DBSNP' , clinical_snp ) , Abundance ( 'NIFT' , clinical_feature ) , evidence = 'Serialized from NeuroMMSigDB' , citation = '28651363' , annotations = { 'MeSHDisease' : disease , } , )
if missing_features :
    log . warning ( 'Missing Features in %s' , disease )
    for feature in missing_features :
        log . warning ( feature )
if fixed_caps :
    log . warning ( 'Fixed capitalization' )
    for broken , fixed in fixed_caps :
        log . warning ( '%s -> %s' , broken , fixed )


signature:
def get_contradiction_summary ( graph : BELGraph ) -> Iterable [ Tuple [ BaseEntity , BaseEntity , str ] ] :
    """Yield triplets of ( source node , target node , set of relations ) for ( source node , target node ) pairs
 that have multiple , contradictory relations ."""

implementation:
for u , v in set ( graph . edges ( ) ) :
    relations = { data [ RELATION ] for data in graph [ u ] [ v ] . values ( ) }
    if relation_set_has_contradictions ( relations ) :
        yield u , v , relations


signature:
def get_regulatory_pairs ( graph : BELGraph ) -> Set [ NodePair ] :
    """Find pairs of nodes that have mutual causal edges that are regulating each other such that ` ` A - > B ` ` and
 ` ` B - | A ` ` .
 : return : A set of pairs of nodes with mutual causal edges"""

implementation:
cg = get_causal_subgraph ( graph )
results = set ( )
for u , v , d in cg . edges ( data = True ) :
    if d [ RELATION ] not in CAUSAL_INCREASE_RELATIONS :
        continue
    if cg . has_edge ( v , u ) and any ( dd [ RELATION ] in CAUSAL_DECREASE_RELATIONS for dd in cg [ v ] [ u ] . values ( ) ) :
        results . add ( ( u , v ) )
return results


signature:
def get_chaotic_pairs ( graph : BELGraph ) -> SetOfNodePairs :
    """Find pairs of nodes that have mutual causal edges that are increasing each other such that ` ` A - > B ` ` and
 ` ` B - > A ` ` .
 : return : A set of pairs of nodes with mutual causal edges"""

implementation:
cg = get_causal_subgraph ( graph )
results = set ( )
for u , v , d in cg . edges ( data = True ) :
    if d [ RELATION ] not in CAUSAL_INCREASE_RELATIONS :
        continue
    if cg . has_edge ( v , u ) and any ( dd [ RELATION ] in CAUSAL_INCREASE_RELATIONS for dd in cg [ v ] [ u ] . values ( ) ) :
        results . add ( tuple ( sorted ( [ u , v ] , key = str ) ) )
return results


signature:
def get_correlation_graph ( graph : BELGraph ) -> Graph :
    """Extract an undirected graph of only correlative relationships ."""

implementation:
result = Graph ( )
for u , v , d in graph . edges ( data = True ) :
    if d [ RELATION ] not in CORRELATIVE_RELATIONS :
        continue
    if not result . has_edge ( u , v ) :
        result . add_edge ( u , v , ** { d [ RELATION ] : True } )
    elif d [ RELATION ] not in result [ u ] [ v ] :
        log . log ( 5 , 'broken correlation relation for %s, %s' , u , v )
        result [ u ] [ v ] [ d [ RELATION ] ] = True
        result [ v ] [ u ] [ d [ RELATION ] ] = True
return result


signature:
def get_correlation_triangles ( graph : BELGraph ) -> SetOfNodeTriples :
    """Return a set of all triangles pointed by the given node ."""

implementation: return { tuple ( sorted ( [ n , u , v ] , key = str ) ) for n in graph for u , v in itt . combinations ( graph [ n ] , 2 ) if graph . has_edge ( u , v ) }


signature:
def get_triangles ( graph : DiGraph ) -> SetOfNodeTriples :
    """Get a set of triples representing the 3 - cycles from a directional graph .
 Each 3 - cycle is returned once , with nodes in sorted order ."""

implementation: return { tuple ( sorted ( [ a , b , c ] , key = str ) ) for a , b in graph . edges ( ) for c in graph . successors ( b ) if graph . has_edge ( c , a ) }


signature:
def get_separate_unstable_correlation_triples ( graph : BELGraph ) -> Iterable [ NodeTriple ] :
    """Yield all triples of nodes A , B , C such that ` ` A pos B ` ` , ` ` A pos C ` ` , and ` ` B neg C ` ` .
 : return : An iterator over triples of unstable graphs , where the second two are negative"""

implementation:
cg = get_correlation_graph ( graph )
for a , b , c in get_correlation_triangles ( cg ) :
    if POSITIVE_CORRELATION in cg [ a ] [ b ] and POSITIVE_CORRELATION in cg [ b ] [ c ] and NEGATIVE_CORRELATION in cg [ a ] [ c ] :
        yield b , a , c
    if POSITIVE_CORRELATION in cg [ a ] [ b ] and NEGATIVE_CORRELATION in cg [ b ] [ c ] and POSITIVE_CORRELATION in cg [ a ] [ c ] :
        yield a , b , c
    if NEGATIVE_CORRELATION in cg [ a ] [ b ] and POSITIVE_CORRELATION in cg [ b ] [ c ] and POSITIVE_CORRELATION in cg [ a ] [ c ] :
        yield c , a , b


signature:
def get_mutually_unstable_correlation_triples ( graph : BELGraph ) -> Iterable [ NodeTriple ] :
    """Yield triples of nodes ( A , B , C ) such that ` ` A neg B ` ` , ` ` B neg C ` ` , and ` ` C neg A ` ` ."""

implementation:
cg = get_correlation_graph ( graph )
for a , b , c in get_correlation_triangles ( cg ) :
    if all ( NEGATIVE_CORRELATION in x for x in ( cg [ a ] [ b ] , cg [ b ] [ c ] , cg [ a ] [ c ] ) ) :
        yield a , b , c


signature:
def jens_transformation_alpha ( graph : BELGraph ) -> DiGraph :
    """Apply Jens ' transformation ( Type 1 ) to the graph .
 1 . Induce a sub - graph over causal + correlative edges
 2 . Transform edges by the following rules :
 - increases = > increases
 - decreases = > backwards increases
 - positive correlation = > two way increases
 - negative correlation = > delete
 The resulting graph can be used to search for 3 - cycles , which now symbolize unstable triplets where ` ` A - > B ` ` ,
 ` ` A - | C ` ` and ` ` B positiveCorrelation C ` ` ."""

implementation:
result = DiGraph ( )
for u , v , d in graph . edges ( data = True ) :
    relation = d [ RELATION ]
    if relation == POSITIVE_CORRELATION :
        result . add_edge ( u , v )
        result . add_edge ( v , u )
    elif relation in CAUSAL_INCREASE_RELATIONS :
        result . add_edge ( u , v )
    elif relation in CAUSAL_DECREASE_RELATIONS :
        result . add_edge ( v , u )
return result


signature:
def jens_transformation_beta ( graph : BELGraph ) -> DiGraph :
    """Apply Jens ' Transformation ( Type 2 ) to the graph .
 1 . Induce a sub - graph over causal and correlative relations
 2 . Transform edges with the following rules :
 - increases = > backwards decreases
 - decreases = > decreases
 - positive correlation = > delete
 - negative correlation = > two way decreases
 The resulting graph can be used to search for 3 - cycles , which now symbolize stable triples where ` ` A - > B ` ` ,
 ` ` A - | C ` ` and ` ` B negativeCorrelation C ` ` ."""

implementation:
result = DiGraph ( )
for u , v , d in graph . edges ( data = True ) :
    relation = d [ RELATION ]
    if relation == NEGATIVE_CORRELATION :
        result . add_edge ( u , v )
        result . add_edge ( v , u )
    elif relation in CAUSAL_INCREASE_RELATIONS :
        result . add_edge ( v , u )
    elif relation in CAUSAL_DECREASE_RELATIONS :
        result . add_edge ( u , v )
return result


signature:
def get_jens_unstable ( graph : BELGraph ) -> Iterable [ NodeTriple ] :
    """Yield triples of nodes ( A , B , C ) where ` ` A - > B ` ` , ` ` A - | C ` ` , and ` ` C positiveCorrelation A ` ` .
 Calculated efficiently using the Jens Transformation ."""

implementation:
r = jens_transformation_alpha ( graph )
return get_triangles ( r )


signature:
def summarize_stability ( graph : BELGraph ) -> Mapping [ str , int ] :
    """Summarize the stability of the graph ."""

implementation:
regulatory_pairs = get_regulatory_pairs ( graph )
chaotic_pairs = get_chaotic_pairs ( graph )
dampened_pairs = get_dampened_pairs ( graph )
contraditory_pairs = get_contradiction_summary ( graph )
separately_unstable_triples = get_separate_unstable_correlation_triples ( graph )
mutually_unstable_triples = get_mutually_unstable_correlation_triples ( graph )
jens_unstable_triples = get_jens_unstable ( graph )
increase_mismatch_triples = get_increase_mismatch_triplets ( graph )
decrease_mismatch_triples = get_decrease_mismatch_triplets ( graph )
chaotic_triples = get_chaotic_triplets ( graph )
dampened_triples = get_dampened_triplets ( graph )
return { 'Regulatory Pairs' : _count_or_len ( regulatory_pairs ) , 'Chaotic Pairs' : _count_or_len ( chaotic_pairs ) , 'Dampened Pairs' : _count_or_len ( dampened_pairs ) , 'Contradictory Pairs' : _count_or_len ( contraditory_pairs ) , 'Separately Unstable Triples' : _count_or_len ( separately_unstable_triples ) , 'Mutually Unstable Triples' : _count_or_len ( mutually_unstable_triples ) , 'Jens Unstable Triples' : _count_or_len ( jens_unstable_triples ) , 'Increase Mismatch Triples' : _count_or_len ( increase_mismatch_triples ) , 'Decrease Mismatch Triples' : _count_or_len ( decrease_mismatch_triples ) , 'Chaotic Triples' : _count_or_len ( chaotic_triples ) , 'Dampened Triples' : _count_or_len ( dampened_triples ) }


signature:
def flatten_list_abundance ( node : ListAbundance ) -> ListAbundance :
    """Flattens the complex or composite abundance ."""

implementation: return node . __class__ ( list ( chain . from_iterable ( ( flatten_list_abundance ( member ) . members if isinstance ( member , ListAbundance ) else [ member ] ) for member in node . members ) ) )


signature:
def list_abundance_expansion ( graph : BELGraph ) -> None :
    """Flatten list abundances ."""

implementation:
mapping = { node : flatten_list_abundance ( node ) for node in graph if isinstance ( node , ListAbundance ) }
relabel_nodes ( graph , mapping , copy = False )


signature:
def list_abundance_cartesian_expansion ( graph : BELGraph ) -> None :
    """Expand all list abundances to simple subject - predicate - object networks ."""

implementation:
for u , v , k , d in list ( graph . edges ( keys = True , data = True ) ) :
    if CITATION not in d :
        continue
    if isinstance ( u , ListAbundance ) and isinstance ( v , ListAbundance ) :
        for u_member , v_member in itt . product ( u . members , v . members ) :
            graph . add_qualified_edge ( u_member , v_member , relation = d [ RELATION ] , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
    elif isinstance ( u , ListAbundance ) :
        for member in u . members :
            graph . add_qualified_edge ( member , v , relation = d [ RELATION ] , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
    elif isinstance ( v , ListAbundance ) :
        for member in v . members :
            graph . add_qualified_edge ( u , member , relation = d [ RELATION ] , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
_remove_list_abundance_nodes ( graph )


signature:
def _reaction_cartesion_expansion_unqualified_helper ( graph : BELGraph , u : BaseEntity , v : BaseEntity , d : dict , ) -> None :
    """Helper to deal with cartension expansion in unqualified edges ."""

implementation:
if isinstance ( u , Reaction ) and isinstance ( v , Reaction ) :
    enzymes = _get_catalysts_in_reaction ( u ) | _get_catalysts_in_reaction ( v )
    for reactant , product in chain ( itt . product ( u . reactants , u . products ) , itt . product ( v . reactants , v . products ) ) :
        if reactant in enzymes or product in enzymes :
            continue
        graph . add_unqualified_edge ( reactant , product , INCREASES )
    for product , reactant in itt . product ( u . products , u . reactants ) :
        if reactant in enzymes or product in enzymes :
            continue
        graph . add_unqualified_edge ( product , reactant , d [ RELATION ] , )
elif isinstance ( u , Reaction ) :
    enzymes = _get_catalysts_in_reaction ( u )
    for product in u . products : # Skip create increases edges between enzymes
        if product in enzymes :
            continue
        # Only add edge between v and reaction if the node is not part of the reaction
        # In practice skips hasReactant , hasProduct edges
        if v not in u . products and v not in u . reactants :
            graph . add_unqualified_edge ( product , v , INCREASES )
        for reactant in u . reactants :
            graph . add_unqualified_edge ( reactant , product , INCREASES )
elif isinstance ( v , Reaction ) :
    enzymes = _get_catalysts_in_reaction ( v )
    for reactant in v . reactants : # Skip create increases edges between enzymes
        if reactant in enzymes :
            continue
        # Only add edge between v and reaction if the node is not part of the reaction
        # In practice skips hasReactant , hasProduct edges
        if u not in v . products and u not in v . reactants :
            graph . add_unqualified_edge ( u , reactant , INCREASES )
        for product in v . products :
            graph . add_unqualified_edge ( reactant , product , INCREASES )


signature:
def _get_catalysts_in_reaction ( reaction : Reaction ) -> Set [ BaseAbundance ] :
    """Return nodes that are both in reactants and reactions in a reaction ."""

implementation: return { reactant for reactant in reaction . reactants if reactant in reaction . products }


signature:
def reaction_cartesian_expansion ( graph : BELGraph , accept_unqualified_edges : bool = True ) -> None :
    """Expand all reactions to simple subject - predicate - object networks ."""

implementation:
for u , v , d in list ( graph . edges ( data = True ) ) : # Deal with unqualified edges
    if CITATION not in d and accept_unqualified_edges :
        _reaction_cartesion_expansion_unqualified_helper ( graph , u , v , d )
        continue
    if isinstance ( u , Reaction ) and isinstance ( v , Reaction ) :
        catalysts = _get_catalysts_in_reaction ( u ) | _get_catalysts_in_reaction ( v )
        for reactant , product in chain ( itt . product ( u . reactants , u . products ) , itt . product ( v . reactants , v . products ) ) :
            if reactant in catalysts or product in catalysts :
                continue
            graph . add_increases ( reactant , product , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
        for product , reactant in itt . product ( u . products , u . reactants ) :
            if reactant in catalysts or product in catalysts :
                continue
            graph . add_qualified_edge ( product , reactant , relation = d [ RELATION ] , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
    elif isinstance ( u , Reaction ) :
        catalysts = _get_catalysts_in_reaction ( u )
        for product in u . products : # Skip create increases edges between enzymes
            if product in catalysts :
                continue
            # Only add edge between v and reaction if the node is not part of the reaction
            # In practice skips hasReactant , hasProduct edges
            if v not in u . products and v not in u . reactants :
                graph . add_increases ( product , v , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
            for reactant in u . reactants :
                graph . add_increases ( reactant , product , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
    elif isinstance ( v , Reaction ) :
        for reactant in v . reactants :
            catalysts = _get_catalysts_in_reaction ( v )
            # Skip create increases edges between enzymes
            if reactant in catalysts :
                continue
            # Only add edge between v and reaction if the node is not part of the reaction
            # In practice skips hasReactant , hasProduct edges
            if u not in v . products and u not in v . reactants :
                graph . add_increases ( u , reactant , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
            for product in v . products :
                graph . add_increases ( reactant , product , citation = d . get ( CITATION ) , evidence = d . get ( EVIDENCE ) , annotations = d . get ( ANNOTATIONS ) , )
_remove_reaction_nodes ( graph )


signature:
def insert_graph ( self , graph : BELGraph , ** _kwargs ) -> Network :
    """Insert a graph and return the resulting ORM object ( mocked ) ."""

implementation:
result = _Namespace ( )
result . id = len ( self . networks )
self . networks [ result . id ] = graph
return result


signature:
def get_graphs_by_ids ( self , network_ids : Iterable [ int ] ) -> List [ BELGraph ] :
    """Get several graphs by their identifiers ."""

implementation: return [ self . networks [ network_id ] for network_id in network_ids ]


signature:
def _generate_citation_dict ( graph : BELGraph ) -> Mapping [ str , Mapping [ Tuple [ BaseEntity , BaseEntity ] , str ] ] :
    """Prepare a citation data dictionary from a graph .
 : return : A dictionary of dictionaries { citation type : { ( source , target ) : citation reference }"""

implementation:
results = defaultdict ( lambda : defaultdict ( set ) )
for u , v , data in graph . edges ( data = True ) :
    if CITATION not in data :
        continue
    results [ data [ CITATION ] [ CITATION_TYPE ] ] [ u , v ] . add ( data [ CITATION ] [ CITATION_REFERENCE ] . strip ( ) )
return dict ( results )


signature:
def get_pmid_by_keyword ( keyword : str , graph : Optional [ BELGraph ] = None , pubmed_identifiers : Optional [ Set [ str ] ] = None , ) -> Set [ str ] :
    """Get the set of PubMed identifiers beginning with the given keyword string .
 : param keyword : The beginning of a PubMed identifier
 : param graph : A BEL graph
 : param pubmed _ identifiers : A set of pre - cached PubMed identifiers
 : return : A set of PubMed identifiers starting with the given string"""

implementation:
if pubmed_identifiers is not None :
    return { pubmed_identifier for pubmed_identifier in pubmed_identifiers if pubmed_identifier . startswith ( keyword ) }
if graph is None :
    raise ValueError ( 'Graph not supplied' )
return { pubmed_identifier for pubmed_identifier in iterate_pubmed_identifiers ( graph ) if pubmed_identifier . startswith ( keyword ) }


signature:
def count_citations ( graph : BELGraph , ** annotations ) -> Counter :
    """Counts the citations in a graph based on a given filter
 : param graph : A BEL graph
 : param dict annotations : The annotation filters to use
 : return : A counter from { ( citation type , citation reference ) : frequency }"""

implementation:
citations = defaultdict ( set )
annotation_dict_filter = build_edge_data_filter ( annotations )
for u , v , _ , d in filter_edges ( graph , annotation_dict_filter ) :
    if CITATION not in d :
        continue
    citations [ u , v ] . add ( ( d [ CITATION ] [ CITATION_TYPE ] , d [ CITATION ] [ CITATION_REFERENCE ] . strip ( ) ) )
return Counter ( itt . chain . from_iterable ( citations . values ( ) ) )


signature:
def count_citations_by_annotation ( graph : BELGraph , annotation : str ) -> Mapping [ str , typing . Counter [ str ] ] :
    """Group the citation counters by subgraphs induced by the annotation .
 : param graph : A BEL graph
 : param annotation : The annotation to use to group the graph
 : return : A dictionary of Counters { subgraph name : Counter from { citation : frequency } }"""

implementation:
citations = defaultdict ( lambda : defaultdict ( set ) )
for u , v , data in graph . edges ( data = True ) :
    if not edge_has_annotation ( data , annotation ) or CITATION not in data :
        continue
    k = data [ ANNOTATIONS ] [ annotation ]
    citations [ k ] [ u , v ] . add ( ( data [ CITATION ] [ CITATION_TYPE ] , data [ CITATION ] [ CITATION_REFERENCE ] . strip ( ) ) )
return { k : Counter ( itt . chain . from_iterable ( v . values ( ) ) ) for k , v in citations . items ( ) }


signature:
def count_author_publications ( graph : BELGraph ) -> typing . Counter [ str ] :
    """Count the number of publications of each author to the given graph ."""

implementation:
authors = group_as_dict ( _iter_author_publiations ( graph ) )
return Counter ( count_dict_values ( count_defaultdict ( authors ) ) )


signature:
def get_authors_by_keyword ( keyword : str , graph = None , authors = None ) -> Set [ str ] :
    """Get authors for whom the search term is a substring .
 : param pybel . BELGraph graph : A BEL graph
 : param keyword : The keyword to search the author strings for
 : param set [ str ] authors : An optional set of pre - cached authors calculated from the graph
 : return : A set of authors with the keyword as a substring"""

implementation:
keyword_lower = keyword . lower ( )
if authors is not None :
    return { author for author in authors if keyword_lower in author . lower ( ) }
if graph is None :
    raise ValueError ( 'Graph not supplied' )
return { author for author in get_authors ( graph ) if keyword_lower in author . lower ( ) }


signature:
def count_authors_by_annotation ( graph : BELGraph , annotation : str = 'Subgraph' ) -> Mapping [ str , typing . Counter [ str ] ] :
    """Group the author counters by sub - graphs induced by the annotation .
 : param graph : A BEL graph
 : param annotation : The annotation to use to group the graph
 : return : A dictionary of Counters { subgraph name : Counter from { author : frequency } }"""

implementation:
authors = group_as_dict ( _iter_authors_by_annotation ( graph , annotation = annotation ) )
return count_defaultdict ( authors )


signature:
def get_evidences_by_pmid ( graph : BELGraph , pmids : Union [ str , Iterable [ str ] ] ) :
    """Get a dictionary from the given PubMed identifiers to the sets of all evidence strings associated with each
 in the graph .
 : param graph : A BEL graph
 : param pmids : An iterable of PubMed identifiers , as strings . Is consumed and converted to a set .
 : return : A dictionary of { pmid : set of all evidence strings }
 : rtype : dict"""

implementation:
result = defaultdict ( set )
for _ , _ , _ , data in filter_edges ( graph , build_pmid_inclusion_filter ( pmids ) ) :
    result [ data [ CITATION ] [ CITATION_REFERENCE ] ] . add ( data [ EVIDENCE ] )
return dict ( result )


signature:
def count_citation_years ( graph : BELGraph ) -> typing . Counter [ int ] :
    """Count the number of citations from each year ."""

implementation:
result = defaultdict ( set )
for _ , _ , data in graph . edges ( data = True ) :
    if CITATION not in data or CITATION_DATE not in data [ CITATION ] :
        continue
    try :
        dt = _ensure_datetime ( data [ CITATION ] [ CITATION_DATE ] )
        result [ dt . year ] . add ( ( data [ CITATION ] [ CITATION_TYPE ] , data [ CITATION ] [ CITATION_REFERENCE ] ) )
    except Exception :
        continue
return count_dict_values ( result )


signature:
def get_citation_years ( graph : BELGraph ) -> List [ Tuple [ int , int ] ] :
    """Create a citation timeline counter from the graph ."""

implementation: return create_timeline ( count_citation_years ( graph ) )


signature:
def create_timeline ( year_counter : typing . Counter [ int ] ) -> List [ Tuple [ int , int ] ] :
    """Complete the Counter timeline .
 : param Counter year _ counter : counter dict for each year
 : return : complete timeline"""

implementation:
if not year_counter :
    return [ ]
from_year = min ( year_counter ) - 1
until_year = datetime . now ( ) . year + 1
return [ ( year , year_counter . get ( year , 0 ) ) for year in range ( from_year , until_year ) ]


signature:
def count_confidences ( graph : BELGraph ) -> typing . Counter [ str ] :
    """Count the confidences in the graph ."""

implementation:
return Counter ( ( 'None' if ANNOTATIONS not in data or 'Confidence' not in data [ ANNOTATIONS ] else list ( data [ ANNOTATIONS ] [ 'Confidence' ] ) [ 0 ] ) for _ , _ , data in graph . edges ( data = True ) if CITATION in data # don ' t bother with unqualified statements
)


signature:
def enrich_pubmed_citations ( graph : BELGraph , manager : Manager ) -> Set [ str ] :
    """Overwrite all PubMed citations with values from NCBI ' s eUtils lookup service .
 : return : A set of PMIDs for which the eUtils service crashed"""

implementation:
pmids = get_pubmed_identifiers ( graph )
pmid_data , errors = get_citations_by_pmids ( manager = manager , pmids = pmids )
for u , v , k in filter_edges ( graph , has_pubmed ) :
    pmid = graph [ u ] [ v ] [ k ] [ CITATION ] [ CITATION_REFERENCE ] . strip ( )
    if pmid not in pmid_data :
        log . warning ( 'Missing data for PubMed identifier: %s' , pmid )
        errors . add ( pmid )
        continue
    graph [ u ] [ v ] [ k ] [ CITATION ] . update ( pmid_data [ pmid ] )
return errors


signature:
def update_context ( universe : BELGraph , graph : BELGraph ) :
    """Update the context of a subgraph from the universe of all knowledge ."""

implementation:
for namespace in get_namespaces ( graph ) :
    if namespace in universe . namespace_url :
        graph . namespace_url [ namespace ] = universe . namespace_url [ namespace ]
    elif namespace in universe . namespace_pattern :
        graph . namespace_pattern [ namespace ] = universe . namespace_pattern [ namespace ]
    else :
        log . warning ( 'namespace: %s missing from universe' , namespace )
for annotation in get_annotations ( graph ) :
    if annotation in universe . annotation_url :
        graph . annotation_url [ annotation ] = universe . annotation_url [ annotation ]
    elif annotation in universe . annotation_pattern :
        graph . annotation_pattern [ annotation ] = universe . annotation_pattern [ annotation ]
    elif annotation in universe . annotation_list :
        graph . annotation_list [ annotation ] = universe . annotation_list [ annotation ]
    else :
        log . warning ( 'annotation: %s missing from universe' , annotation )


signature:
def highlight_nodes ( graph : BELGraph , nodes : Optional [ Iterable [ BaseEntity ] ] = None , color : Optional [ str ] = None ) :
    """Adds a highlight tag to the given nodes .
 : param graph : A BEL graph
 : param nodes : The nodes to add a highlight tag on
 : param color : The color to highlight ( use something that works with CSS )"""

implementation:
color = color or NODE_HIGHLIGHT_DEFAULT_COLOR
for node in nodes if nodes is not None else graph :
    graph . node [ node ] [ NODE_HIGHLIGHT ] = color


signature:
def is_node_highlighted ( graph : BELGraph , node : BaseEntity ) -> bool :
    """Returns if the given node is highlighted .
 : param graph : A BEL graph
 : param node : A BEL node
 : type node : tuple
 : return : Does the node contain highlight information ?
 : rtype : bool"""

implementation: return NODE_HIGHLIGHT in graph . node [ node ]


signature:
def remove_highlight_nodes ( graph : BELGraph , nodes : Optional [ Iterable [ BaseEntity ] ] = None ) -> None :
    """Removes the highlight from the given nodes , or all nodes if none given .
 : param graph : A BEL graph
 : param nodes : The list of nodes to un - highlight"""

implementation:
for node in graph if nodes is None else nodes :
    if is_node_highlighted ( graph , node ) :
        del graph . node [ node ] [ NODE_HIGHLIGHT ]


signature:
def highlight_edges ( graph : BELGraph , edges = None , color : Optional [ str ] = None ) -> None :
    """Adds a highlight tag to the given edges .
 : param graph : A BEL graph
 : param edges : The edges ( 4 - tuples of u , v , k , d ) to add a highlight tag on
 : type edges : iter [ tuple ]
 : param str color : The color to highlight ( use something that works with CSS )"""

implementation:
color = color or EDGE_HIGHLIGHT_DEFAULT_COLOR
for u , v , k , d in edges if edges is not None else graph . edges ( keys = True , data = True ) :
    graph [ u ] [ v ] [ k ] [ EDGE_HIGHLIGHT ] = color


signature:
def is_edge_highlighted ( graph : BELGraph , u , v , k ) -> bool :
    """Returns if the given edge is highlighted .
 : param graph : A BEL graph
 : return : Does the edge contain highlight information ?
 : rtype : bool"""

implementation: return EDGE_HIGHLIGHT in graph [ u ] [ v ] [ k ]


signature:
def remove_highlight_edges ( graph : BELGraph , edges = None ) :
    """Remove the highlight from the given edges , or all edges if none given .
 : param graph : A BEL graph
 : param edges : The edges ( 4 - tuple of u , v , k , d ) to remove the highlight from )
 : type edges : iter [ tuple ]"""

implementation:
for u , v , k , _ in graph . edges ( keys = True , data = True ) if edges is None else edges :
    if is_edge_highlighted ( graph , u , v , k ) :
        del graph [ u ] [ v ] [ k ] [ EDGE_HIGHLIGHT ]


signature:
def highlight_subgraph ( universe : BELGraph , graph : BELGraph ) :
    """Highlight all nodes / edges in the universe that in the given graph .
 : param universe : The universe of knowledge
 : param graph : The BEL graph to mutate"""

implementation:
highlight_nodes ( universe , graph )
highlight_edges ( universe , graph . edges ( ) )


signature:
def remove_highlight_subgraph ( graph : BELGraph , subgraph : BELGraph ) :
    """Remove the highlight from all nodes / edges in the graph that are in the subgraph .
 : param graph : The BEL graph to mutate
 : param subgraph : The subgraph from which to remove the highlighting"""

implementation:
remove_highlight_nodes ( graph , subgraph . nodes ( ) )
remove_highlight_edges ( graph , subgraph . edges ( ) )


signature:
def get_causal_out_edges ( graph : BELGraph , nbunch : Union [ BaseEntity , Iterable [ BaseEntity ] ] , ) -> Set [ Tuple [ BaseEntity , BaseEntity ] ] :
    """Get the out - edges to the given node that are causal .
 : return : A set of ( source , target ) pairs where the source is the given node"""

implementation: return { ( u , v ) for u , v , k , d in graph . out_edges ( nbunch , keys = True , data = True ) if is_causal_relation ( graph , u , v , k , d ) }


signature:
def get_causal_source_nodes ( graph : BELGraph , func : str ) -> Set [ BaseEntity ] :
    """Return a set of all nodes that have an in - degree of 0.
 This likely means that it is an external perturbagen and is not known to have any causal origin from within the
 biological system . These nodes are useful to identify because they generally don ' t provide any mechanistic insight ."""

implementation: return { node for node in graph if node . function == func and is_causal_source ( graph , node ) }


signature:
def get_causal_central_nodes ( graph : BELGraph , func : str ) -> Set [ BaseEntity ] :
    """Return a set of all nodes that have both an in - degree > 0 and out - degree > 0.
 This means that they are an integral part of a pathway , since they are both produced and consumed ."""

implementation: return { node for node in graph if node . function == func and is_causal_central ( graph , node ) }


signature:
def get_causal_sink_nodes ( graph : BELGraph , func ) -> Set [ BaseEntity ] :
    """Returns a set of all ABUNDANCE nodes that have an causal out - degree of 0.
 This likely means that the knowledge assembly is incomplete , or there is a curation error ."""

implementation: return { node for node in graph if node . function == func and is_causal_sink ( graph , node ) }


signature:
def count_top_centrality ( graph : BELGraph , number : Optional [ int ] = 30 ) -> Mapping [ BaseEntity , int ] :
    """Get top centrality dictionary ."""

implementation:
dd = nx . betweenness_centrality ( graph )
dc = Counter ( dd )
return dict ( dc . most_common ( number ) )


signature:
def get_modifications_count ( graph : BELGraph ) -> Mapping [ str , int ] :
    """Get a modifications count dictionary ."""

implementation: return remove_falsy_values ( { 'Translocations' : len ( get_translocated ( graph ) ) , 'Degradations' : len ( get_degradations ( graph ) ) , 'Molecular Activities' : len ( get_activities ( graph ) ) , } )


signature:
def remove_falsy_values ( counter : Mapping [ Any , int ] ) -> Mapping [ Any , int ] :
    """Remove all values that are zero ."""

implementation: return { label : count for label , count in counter . items ( ) if count }


signature:
def _collapse_variants_by_function ( graph : BELGraph , func : str ) -> None :
    """Collapse all of the given functions ' variants ' edges to their parents , in - place ."""

implementation:
for parent_node , variant_node , data in graph . edges ( data = True ) :
    if data [ RELATION ] == HAS_VARIANT and parent_node . function == func :
        collapse_pair ( graph , from_node = variant_node , to_node = parent_node )


signature:
def rewire_variants_to_genes ( graph : BELGraph ) -> None :
    """Find all protein variants that are pointing to a gene and not a protein and fixes them by changing their
 function to be : data : ` pybel . constants . GENE ` , in place
 A use case is after running : func : ` collapse _ to _ genes ` ."""

implementation:
mapping = { }
for node in graph :
    if not isinstance ( node , Protein ) or not node . variants :
        continue
    mapping [ node ] = Gene ( name = node . name , namespace = node . namespace , identifier = node . identifier , variants = node . variants , )
nx . relabel_nodes ( graph , mapping , copy = False )


signature:
def _collapse_edge_passing_predicates ( graph : BELGraph , edge_predicates : EdgePredicates = None ) -> None :
    """Collapse all edges passing the given edge predicates ."""

implementation:
for u , v , _ in filter_edges ( graph , edge_predicates = edge_predicates ) :
    collapse_pair ( graph , survivor = u , victim = v )


signature:
def _collapse_edge_by_namespace ( graph : BELGraph , victim_namespaces : Strings , survivor_namespaces : str , relations : Strings ) -> None :
    """Collapse pairs of nodes with the given namespaces that have the given relationship .
 : param graph : A BEL Graph
 : param victim _ namespaces : The namespace ( s ) of the node to collapse
 : param survivor _ namespaces : The namespace of the node to keep
 : param relations : The relation ( s ) to search"""

implementation:
relation_filter = build_relation_predicate ( relations )
source_namespace_filter = build_source_namespace_filter ( victim_namespaces )
target_namespace_filter = build_target_namespace_filter ( survivor_namespaces )
edge_predicates = [ relation_filter , source_namespace_filter , target_namespace_filter ]
_collapse_edge_passing_predicates ( graph , edge_predicates = edge_predicates )


signature:
def collapse_equivalencies_by_namespace ( graph : BELGraph , victim_namespace : Strings , survivor_namespace : str ) -> None :
    """Collapse pairs of nodes with the given namespaces that have equivalence relationships .
 : param graph : A BEL graph
 : param victim _ namespace : The namespace ( s ) of the node to collapse
 : param survivor _ namespace : The namespace of the node to keep
 To convert all ChEBI names to InChI keys , assuming there are appropriate equivalence relations between nodes with
 those namespaces :
 > > > collapse _ equivalencies _ by _ namespace ( graph , ' CHEBI ' , ' CHEBIID ' )
 > > > collapse _ equivalencies _ by _ namespace ( graph , ' CHEBIID ' , ' INCHI ' )"""

implementation: _collapse_edge_by_namespace ( graph , victim_namespace , survivor_namespace , EQUIVALENT_TO )


signature:
def collapse_orthologies_by_namespace ( graph : BELGraph , victim_namespace : Strings , survivor_namespace : str ) -> None :
    """Collapse pairs of nodes with the given namespaces that have orthology relationships .
 : param graph : A BEL Graph
 : param victim _ namespace : The namespace ( s ) of the node to collapse
 : param survivor _ namespace : The namespace of the node to keep
 To collapse all MGI nodes to their HGNC orthologs , use :
 > > > collapse _ orthologies _ by _ namespace ( ' MGI ' , ' HGNC ' )
 To collapse collapse both MGI and RGD nodes to their HGNC orthologs , use :
 > > > collapse _ orthologies _ by _ namespace ( [ ' MGI ' , ' RGD ' ] , ' HGNC ' )"""

implementation: _collapse_edge_by_namespace ( graph , victim_namespace , survivor_namespace , ORTHOLOGOUS )


signature:
def collapse_entrez_equivalencies ( graph : BELGraph ) :
    """Collapse all equivalence edges away from Entrez . Assumes well formed , 2 - way equivalencies ."""

implementation:
relation_filter = build_relation_predicate ( EQUIVALENT_TO )
source_namespace_filter = build_source_namespace_filter ( [ 'EGID' , 'EG' , 'ENTREZ' ] )
edge_predicates = [ relation_filter , source_namespace_filter , ]
_collapse_edge_passing_predicates ( graph , edge_predicates = edge_predicates )


signature:
def collapse_consistent_edges ( graph : BELGraph ) :
    """Collapse consistent edges together .
 . . warning : : This operation doesn ' t preserve evidences or other annotations"""

implementation:
for u , v in graph . edges ( ) :
    relation = pair_is_consistent ( graph , u , v )
    if not relation :
        continue
    edges = [ ( u , v , k ) for k in graph [ u ] [ v ] ]
    graph . remove_edges_from ( edges )
    graph . add_edge ( u , v , attr_dict = { RELATION : relation } )


signature:
def collapse_nodes_with_same_names ( graph : BELGraph ) -> None :
    """Collapse all nodes with the same name , merging namespaces by picking first alphabetical one ."""

implementation:
survivor_mapping = defaultdict ( set )
# Collapse mapping dict
victims = set ( )
# Things already mapped while iterating
it = tqdm ( itt . combinations ( graph , r = 2 ) , total = graph . number_of_nodes ( ) * ( graph . number_of_nodes ( ) - 1 ) / 2 )
for a , b in it :
    if b in victims :
        continue
    a_name , b_name = a . get ( NAME ) , b . get ( NAME )
    if not a_name or not b_name or a_name . lower ( ) != b_name . lower ( ) :
        continue
    if a . keys ( ) != b . keys ( ) : # not same version ( might have variants )
        continue
    # Ensure that the values in the keys are also the same
    for k in set ( a . keys ( ) ) - { NAME , NAMESPACE } :
        if a [ k ] != b [ k ] : # something different
            continue
    survivor_mapping [ a ] . add ( b )
    # Keep track of things that has been already mapped
    victims . add ( b )
collapse_nodes ( graph , survivor_mapping )


signature:
def main ( output ) :
    """Output the HBP knowledge graph to the desktop"""

implementation:
from hbp_knowledge import get_graph
graph = get_graph ( )
text = to_html ( graph )
print ( text , file = output )


signature:
def node_is_upstream_leaf ( graph : BELGraph , node : BaseEntity ) -> bool :
    """Return if the node is an upstream leaf .
 An upstream leaf is defined as a node that has no in - edges , and exactly 1 out - edge ."""

implementation: return 0 == len ( graph . predecessors ( node ) ) and 1 == len ( graph . successors ( node ) )


signature:
def get_unweighted_upstream_leaves ( graph : BELGraph , key : Optional [ str ] = None ) -> Iterable [ BaseEntity ] :
    """Get nodes with no incoming edges , one outgoing edge , and without the given key in its data dictionary .
 . . seealso : : : func : ` data _ does _ not _ contain _ key _ builder `
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 : return : An iterable over leaves ( nodes with an in - degree of 0 ) that don ' t have the given annotation"""

implementation:
if key is None :
    key = WEIGHT
return filter_nodes ( graph , [ node_is_upstream_leaf , data_missing_key_builder ( key ) ] )


signature:
def remove_unweighted_leaves ( graph : BELGraph , key : Optional [ str ] = None ) -> None :
    """Remove nodes that are leaves and that don ' t have a weight ( or other key ) attribute set .
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` ."""

implementation:
unweighted_leaves = list ( get_unweighted_upstream_leaves ( graph , key = key ) )
graph . remove_nodes_from ( unweighted_leaves )


signature:
def is_unweighted_source ( graph : BELGraph , node : BaseEntity , key : str ) -> bool :
    """Check if the node is both a source and also has an annotation .
 : param graph : A BEL graph
 : param node : A BEL node
 : param key : The key in the node data dictionary representing the experimental data"""

implementation: return graph . in_degree ( node ) == 0 and key not in graph . nodes [ node ]


signature:
def get_unweighted_sources ( graph : BELGraph , key : Optional [ str ] = None ) -> Iterable [ BaseEntity ] :
    """Get nodes on the periphery of the sub - graph that do not have a annotation for the given key .
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data
 : return : An iterator over BEL nodes that are unannotated and on the periphery of this subgraph"""

implementation:
if key is None :
    key = WEIGHT
for node in graph :
    if is_unweighted_source ( graph , node , key ) :
        yield node


signature:
def remove_unweighted_sources ( graph : BELGraph , key : Optional [ str ] = None ) -> None :
    """Prune unannotated nodes on the periphery of the sub - graph .
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` ."""

implementation:
nodes = list ( get_unweighted_sources ( graph , key = key ) )
graph . remove_nodes_from ( nodes )


signature:
def prune_mechanism_by_data ( graph , key : Optional [ str ] = None ) -> None :
    """Remove all leaves and source nodes that don ' t have weights .
 Is a thin wrapper around : func : ` remove _ unweighted _ leaves ` and : func : ` remove _ unweighted _ sources `
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data . Defaults to
 : data : ` pybel _ tools . constants . WEIGHT ` .
 Equivalent to :
 > > > remove _ unweighted _ leaves ( graph )
 > > > remove _ unweighted _ sources ( graph )"""

implementation:
remove_unweighted_leaves ( graph , key = key )
remove_unweighted_sources ( graph , key = key )


signature:
def generate_mechanism ( graph : BELGraph , node : BaseEntity , key : Optional [ str ] = None ) -> BELGraph :
    """Generate a mechanistic sub - graph upstream of the given node .
 : param graph : A BEL graph
 : param node : A BEL node
 : param key : The key in the node data dictionary representing the experimental data .
 : return : A sub - graph grown around the target BEL node"""

implementation:
subgraph = get_upstream_causal_subgraph ( graph , node )
expand_upstream_causal ( graph , subgraph )
remove_inconsistent_edges ( subgraph )
collapse_consistent_edges ( subgraph )
if key is not None : # FIXME when is it not pruned ?
    prune_mechanism_by_data ( subgraph , key )
return subgraph


signature:
def generate_bioprocess_mechanisms ( graph , key : Optional [ str ] = None ) -> Mapping [ BiologicalProcess , BELGraph ] :
    """Generate a mechanistic sub - graph for each biological process in the graph using : func : ` generate _ mechanism ` .
 : param graph : A BEL graph
 : param key : The key in the node data dictionary representing the experimental data ."""

implementation: return { biological_process : generate_mechanism ( graph , biological_process , key = key ) for biological_process in get_nodes_by_function ( graph , BIOPROCESS ) }


signature:
def get_neurommsig_scores ( graph : BELGraph , genes : List [ Gene ] , annotation : str = 'Subgraph' , ora_weight : Optional [ float ] = None , hub_weight : Optional [ float ] = None , top_percent : Optional [ float ] = None , topology_weight : Optional [ float ] = None , preprocess : bool = False ) -> Optional [ Mapping [ str , float ] ] :
    """Preprocess the graph , stratify by the given annotation , then run the NeuroMMSig algorithm on each .
 : param graph : A BEL graph
 : param genes : A list of gene nodes
 : param annotation : The annotation to use to stratify the graph to subgraphs
 : param ora _ weight : The relative weight of the over - enrichment analysis score from
 : py : func : ` neurommsig _ gene _ ora ` . Defaults to 1.0.
 : param hub _ weight : The relative weight of the hub analysis score from : py : func : ` neurommsig _ hubs ` .
 Defaults to 1.0.
 : param top _ percent : The percentage of top genes to use as hubs . Defaults to 5 % ( 0.05 ) .
 : param topology _ weight : The relative weight of the topolgical analysis core from
 : py : func : ` neurommsig _ topology ` . Defaults to 1.0.
 : param preprocess : If true , preprocess the graph .
 : return : A dictionary from { annotation value : NeuroMMSig composite score }
 Pre - processing steps :
 1 . Infer the central dogma with : func : ` `
 2 . Collapse all proteins , RNAs and miRNAs to genes with : func : ` `
 3 . Collapse variants to genes with : func : ` `"""

implementation:
if preprocess :
    graph = neurommsig_graph_preprocessor . run ( graph )
if not any ( gene in graph for gene in genes ) :
    logger . debug ( 'no genes mapping to graph' )
    return
subgraphs = get_subgraphs_by_annotation ( graph , annotation = annotation )
return get_neurommsig_scores_prestratified ( subgraphs = subgraphs , genes = genes , ora_weight = ora_weight , hub_weight = hub_weight , top_percent = top_percent , topology_weight = topology_weight , )


signature:
def get_neurommsig_scores_prestratified ( subgraphs : Mapping [ str , BELGraph ] , genes : List [ Gene ] , ora_weight : Optional [ float ] = None , hub_weight : Optional [ float ] = None , top_percent : Optional [ float ] = None , topology_weight : Optional [ float ] = None , ) -> Optional [ Mapping [ str , float ] ] :
    """Takes a graph stratification and runs neurommsig on each
 : param subgraphs : A pre - stratified set of graphs
 : param genes : A list of gene nodes
 : param ora _ weight : The relative weight of the over - enrichment analysis score from
 : py : func : ` neurommsig _ gene _ ora ` . Defaults to 1.0.
 : param hub _ weight : The relative weight of the hub analysis score from : py : func : ` neurommsig _ hubs ` .
 Defaults to 1.0.
 : param top _ percent : The percentage of top genes to use as hubs . Defaults to 5 % ( 0.05 ) .
 : param topology _ weight : The relative weight of the topolgical analysis core from
 : py : func : ` neurommsig _ topology ` . Defaults to 1.0.
 : return : A dictionary from { annotation value : NeuroMMSig composite score }
 Pre - processing steps :
 1 . Infer the central dogma with : func : ` `
 2 . Collapse all proteins , RNAs and miRNAs to genes with : func : ` `
 3 . Collapse variants to genes with : func : ` `"""

implementation: return { name : get_neurommsig_score ( graph = subgraph , genes = genes , ora_weight = ora_weight , hub_weight = hub_weight , top_percent = top_percent , topology_weight = topology_weight , ) for name , subgraph in subgraphs . items ( ) }


signature:
def get_neurommsig_score ( graph : BELGraph , genes : List [ Gene ] , ora_weight : Optional [ float ] = None , hub_weight : Optional [ float ] = None , top_percent : Optional [ float ] = None , topology_weight : Optional [ float ] = None ) -> float :
    """Calculate the composite NeuroMMSig Score for a given list of genes .
 : param graph : A BEL graph
 : param genes : A list of gene nodes
 : param ora _ weight : The relative weight of the over - enrichment analysis score from
 : py : func : ` neurommsig _ gene _ ora ` . Defaults to 1.0.
 : param hub _ weight : The relative weight of the hub analysis score from : py : func : ` neurommsig _ hubs ` .
 Defaults to 1.0.
 : param top _ percent : The percentage of top genes to use as hubs . Defaults to 5 % ( 0.05 ) .
 : param topology _ weight : The relative weight of the topolgical analysis core from
 : py : func : ` neurommsig _ topology ` . Defaults to 1.0.
 : return : The NeuroMMSig composite score"""

implementation:
ora_weight = ora_weight or 1.0
hub_weight = hub_weight or 1.0
topology_weight = topology_weight or 1.0
total_weight = ora_weight + hub_weight + topology_weight
genes = list ( genes )
ora_score = neurommsig_gene_ora ( graph , genes )
hub_score = neurommsig_hubs ( graph , genes , top_percent = top_percent )
topology_score = neurommsig_topology ( graph , genes )
weighted_sum = ( ora_weight * ora_score + hub_weight * hub_score + topology_weight * topology_score )
return weighted_sum / total_weight


signature:
def neurommsig_gene_ora ( graph : BELGraph , genes : List [ Gene ] ) -> float :
    """Calculate the percentage of target genes mappable to the graph .
 Assume : graph central dogma inferred , collapsed to genes , collapsed variants"""

implementation:
graph_genes = set ( get_nodes_by_function ( graph , GENE ) )
return len ( graph_genes . intersection ( genes ) ) / len ( graph_genes )


signature:
def neurommsig_hubs ( graph : BELGraph , genes : List [ Gene ] , top_percent : Optional [ float ] = None ) -> float :
    """Calculate the percentage of target genes mappable to the graph .
 Assume : graph central dogma inferred , collapsed to genes , collapsed variants , graph has more than 20 nodes
 : param graph : A BEL graph
 : param genes : A list of nodes
 : param top _ percent : The percentage of top genes to use as hubs . Defaults to 5 % ( 0.05 ) ."""

implementation:
top_percent = top_percent or 0.05
if graph . number_of_nodes ( ) < 20 :
    logger . debug ( 'Graph has less than 20 nodes' )
    return 0.0
graph_genes = set ( get_nodes_by_function ( graph , GENE ) )
bc = Counter ( { node : betweenness_centrality for node , betweenness_centrality in calculate_betweenness_centality ( graph ) . items ( ) if node in graph_genes } )
# TODO consider continuous analog with weighting by percentile
number_central_nodes = int ( len ( graph_genes ) * top_percent )
if number_central_nodes < 1 :
    number_central_nodes = 1
number_mappable_central_nodes = sum ( node in genes for node in bc . most_common ( number_central_nodes ) )
return number_mappable_central_nodes / number_central_nodes