omnipath-secondary-adapter/weave_knowledge_graph_original.py at main · biocypher/omnipath-secondary-adapter · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
#!/usr/bin/env python3

import argparse
import logging
import pandas as pd
import yaml

from biocypher import BioCypher
import ontoweaver


# from omnipath_seconday_adapter.adapters.example_adapter import

if __name__ == "__main__":

    usage = f"Extract nodes and edges from CSV tables of Omnipath databse: networks, enzyme-PTM, complexes, annotations and intercell."
    parser = argparse.ArgumentParser(description=usage)

    parser.add_argument(
        "-net",
        "--networks",
        metavar="TSV",
        nargs="+",
        help="Extract from the Omnipath networks TSV file.",
    )

    parser.add_argument(
        "-enz",
        "--enzyme-PTM",
        metavar="TSV",
        nargs="+",
        help="Extract from the Omnipath enzyme-PTM TSV file.",
    )

    parser.add_argument(
        "-co",
        "--complexes",
        metavar="TSV",
        nargs="+",
        help="Extract from the Omnipath complexes TSV file.",
    )

    parser.add_argument(
        "-an",
        "--annotations",
        metavar="TSV",
        nargs="+",
        help="Extract from the Omnipath annotations TSV file.",
    )

    parser.add_argument(
        "-int",
        "--intercell",
        metavar="TSV",
        nargs="+",
        help="Extract from the Omnipath intercell TSV file.",
    )

    levels = {
        "DEBUG": logging.DEBUG,
        "INFO": logging.INFO,
        "WARNING": logging.WARNING,
        "ERROR": logging.ERROR,
        "CRITICAL": logging.CRITICAL,
    }

    parser.add_argument(
        "-v",
        "--verbose",
        choices=levels.keys(),
        default="WARNING",
        help="Set the verbose level (default: %(default)s).",
    )

    asked = parser.parse_args()

    # Actually extract data.
    nodes = []
    edges = []

    # Extract from databases not requiring preprocessing.
    if asked.networks:
        logging.info(f"Weave Omnipath networks data...")

        networks_df = pd.read_csv(asked.networks[0], sep="\t")
        print(networks_df.info())

        mapping_file = "./omnipath_secondary_adapter/adapters/networks.yaml"
        with open(mapping_file) as fd:
            mapping = yaml.full_load(fd)

        adapter = ontoweaver.tabular.extract_table(
            # df=networks_df, config=mapping, separator=":", affix="suffix"
            df=networks_df,
            config=mapping,
            separator=":",
            affix="none",
        )

        nodes += adapter.nodes
        edges += adapter.edges

        logging.info(f"Wove Networks: {len(nodes)} nodes, {len(edges)} edges.")

    # Fusion step: fuse duplicated nodes and edges.
    import_file = ontoweaver.reconciliate_write(
        nodes,
        edges,
        "config/biocypher_config.yaml",
        "config/schema_config.yaml",
        separator=", ",
    )


# Example of usage in CLI:
# poetry run python weave_knowledge_graph.py -net ./data_testing/subset_interactions.tsv