Inputs and interaction universes

Universe normalizes an interaction source into the DataFrame schema consumed by Network. Built-in adapters cover OmniPath, SIGNOR, HuRI, and PhosphoSitePlus; a custom DataFrame can be supplied directly.

from neko.inputs import Universe, signor

omnipath_resources = Universe("omnipath")
signor_resources = signor()
custom_resources = Universe(my_interaction_dataframe)

A bare Universe() is intentionally empty. Use an explicit resource name when data should be loaded.

Custom interaction tables

A custom DataFrame needs source and target columns. Give interaction signs either as boolean is_stimulation / is_inhibition columns or as one lowercase effect column. Recognised effect values (case-insensitive) are:

Meaning	Values
Stimulation	stimulation, activation, activate(s), stimulate(s), positive, +, True, 1
Inhibition	inhibition, inhibit(s), repression, repress(es), negative, -, False, -1
Both	bimodal, both
Undefined	undefined, unknown, or an empty value
Complex formation	form complex, complex

Any other value raises ValueError and names the offending values; NeKo does not guess a sign. Boolean flag columns (is_directed, is_stimulation, is_inhibition, form_complex) accept True/False, yes/no, t/f, y/n and 1/0/-1 in any case, and raise ValueError for other text.

import pandas as pd
from neko.core.network import Network

table = pd.DataFrame({
    "source": ["EGFR", "PTEN"],
    "target": ["PIK3CA", "AKT1"],
    "effect": ["stimulation", "inhibition"],
})
net = Network(["EGFR", "AKT1"], resources=Universe(table).interactions)

Universe

Universe

Universe(resources: Literal['omnipath'] | DataFrame = None, **param)

Load and preprocess a generic network from databases or files.

Source code in neko/inputs/_universe.py

def __init__(
        self,
        resources: Literal['omnipath'] | pd.DataFrame = None,
        **param
    ):
    """
    Load and preprocess a generic network from databases or files.
    """

    self._resources = {}
    self._directed = {}
    self._resource = resources
    self.database_name = param.get('name') or (
        'OmniPath' if isinstance(resources, str) and resources == 'omnipath'
        else getattr(resources, 'attrs', {}).get('database_name')
    )
    self.interactions = None
    self.add_resources(resources, **param)
    self.build()

Attributes

network property writable

network: DataFrame

The network as it's been read from the original source.

method property

method: Callable

The method that loads the data.

param are to be passed to this method.

Methods:

merge staticmethod

merge(df1: DataFrame, df2: DataFrame) -> pd.DataFrame

This function concatenates the provided df with the existing one in the resources object, aligning columns and filling in missing data with NaN.

Parameters: df1 (pd.DataFrame): The DataFrame to be added. df2 (pd.DataFrame): The DataFrame to be added.

Raises: ValueError: If the 'df' parameter is not a pandas DataFrame.

Returns: None

Source code in neko/inputs/_universe.py

@staticmethod
def merge(df1: pd.DataFrame, df2: pd.DataFrame) -> pd.DataFrame:
    """
    This function concatenates the provided df with the existing one in the resources object,
    aligning columns and filling in missing data with NaN.

    Parameters:
        df1 (pd.DataFrame): The DataFrame to be added.
        df2 (pd.DataFrame): The DataFrame to be added.

    Raises:
        ValueError: If the 'df' parameter is not a pandas DataFrame.

    Returns:
        None
    """

    # Align columns of both dataframes, filling missing columns with NaN
    all_columns = set(df1.columns).union(set(df2.columns))
    df1 = df1.reindex(columns=all_columns, fill_value=None)
    df2 = df2.reindex(columns=all_columns, fill_value=None)
    df1 = pd.concat([df1, df2])

    return df1.copy()

load

load() -> None

Acquire the input data according to parameters.

Source code in neko/inputs/_universe.py

def load(self) -> None:
    """
    Acquire the input data according to parameters.
    """

    self._network = self.method(**self.param)

check

check() -> bool

The network is loaded and contains the mandatory variables.

Source code in neko/inputs/_universe.py

def check(self) -> bool:
    """
    The network is loaded and contains the mandatory variables.
    """

    return (
        hasattr(self, '_network') and
        not _REQUIRED_COLS - set(self._network.columns)
    )

Adapter functions

network_universe

network_universe(resource: Literal['omnipath'] | DataFrame = 'omnipath', **kwargs) -> Universe

Generic networks from databases, files and standard formats.

Args: resource: Name of the resource or a ready data frame to bypass the built-in loading method. kwargs: Passed to the source specific method. See the specific methods in this module for details.

Note: currently OmniPath PPI is the single available option and serves as a placeholder. Later we will dispatch all inputs through this API.

Source code in neko/inputs/_universe.py

def network_universe(
        resource: Literal['omnipath'] | pd.DataFrame = 'omnipath',
        **kwargs
    ) -> Universe:
    """
    Generic networks from databases, files and standard formats.

    Args:
        resource:
            Name of the resource or a ready data frame to bypass the built-in
            loading method.
        kwargs:
            Passed to the source specific method. See the specific methods in
            this module for details.

    Note: currently OmniPath PPI is the single available option and serves
    as a placeholder. Later we will dispatch all inputs through this API.
    """

    if isinstance(resource, Universe):
        return resource
    return Universe(resource, **kwargs)

omnipath

omnipath(**kwargs) -> Universe

Source code in neko/inputs/_universe.py

def omnipath(**kwargs) -> Universe:

    return network_universe('omnipath', **kwargs)

signor

signor(path: str | None = None, **kwargs) -> Universe

Source code in neko/inputs/_universe.py

def signor(path: str | None = None, **kwargs) -> Universe:
    # A missing legacy path falls back to NeKo's validated SIGNOR cache. The
    # adapter downloads and populates that cache only when it has no usable
    # cached release.
    if path and os.path.exists(path):
        return Universe(_signor.signor(path, **kwargs), name='SIGNOR')
    else:
        if path:
            logging.warning(
                'SIGNOR path does not exist: %s. Falling back to the '
                'managed NeKo cache.',
                os.path.abspath(path),
            )

        return Universe(_signor.signor(**kwargs), name='SIGNOR')

phosphosite

phosphosite(organism: Literal['human', 'mouse', 'rat'] = 'human', kinase_substrate: str | None = None, regulatory_sites: str | None = None, **kwargs) -> Universe

Source code in neko/inputs/_universe.py

def phosphosite(
        organism: Literal["human", "mouse", "rat"] = 'human',
        kinase_substrate: str | None = None,
        regulatory_sites: str | None = None,
        **kwargs
    ) -> Universe:

    df = _psp.psp(organism, kinase_substrate, regulatory_sites)

    return Universe(df, name = 'phosphosite')

huri

huri(dataset: str = 'HI-union') -> Universe

Source code in neko/inputs/_universe.py

def huri(dataset: str = 'HI-union') -> Universe:

    df = _huri.huri(dataset)

    return Universe(df, directed = False, name = 'huri')