Skip to content

data

neo4j_db()

Connects to Neo4j database.

Source code in api/data.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
@st.experimental_singleton
def neo4j_db():
    """Connects to Neo4j database."""
    neo4j_uri = os.environ.get("NEO4J_URI", "neo4j://localhost:7687")
    neo4j_password = os.environ.get("NEO4J_PASSWORD")
    neo4j_database = os.environ.get("NEO4J_DATABASE", "neo4j")
    _con = Neo4jClient(
        uri=neo4j_uri,
        neo4j_password=neo4j_password,
        database=neo4j_database,
    )
    _con.driver.verify_connectivity()
    return _con

routes_to_graph(routes, rgm, pos_color='#FF4B4B', neg_color='#002941')

Generates a graph object used by streamlit for visualizing the routes.

Duplicated reaction nodes are dropped but edges are preserved. Edge widths are proportional to the highest-scoring route they belong to.

Source code in api/data.py 
 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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def routes_to_graph(routes, rgm: ReactionGeneMap, pos_color="#FF4B4B", neg_color="#002941"):
    """Generates a graph object used by streamlit for visualizing the routes.

    Duplicated reaction nodes are dropped but edges are preserved. Edge widths are proportional to
    the highest-scoring route they belong to.
    """
    nodes, edges = {}, {}
    max_score, min_score = -float("inf"), float("inf")

    for route_id, route in enumerate(routes):
        # Save max/min scores for later scaling
        max_score = max(max_score, abs(route["score"]))
        min_score = min(min_score, abs(route["score"]))

        # Iterate over all nodes
        r = route["route"]
        route_color = pos_color if route["score"] > 0 else neg_color
        for i, step in enumerate(r):
            # compound->reaction->compound|compounds->reaction->...
            if step["nodeType"] == "Compound":
                # Default state for new nodes
                if step["id"] not in nodes:
                    n = Node(
                        id=step["id"],
                        label=step["name"],
                        size=6,
                        shape="dot",
                        color="#EA9C70",
                    )
                    nodes[step["id"]] = n
                # TODO: color nodes based on structural similarity?

            elif step["nodeType"] in {"Reaction", "ReverseReaction"}:
                n_src = r[i - 1]["id"]
                _ = edges.setdefault(n_src, {})

                k = i + 1
                while k < len(r) and r[k]["nodeType"] == "Compound":
                    n_tgt = r[k]["id"]

                    if n_tgt not in edges[n_src]:
                        rxn_exp = rgm.rxn_exp.get(step["id"])
                        if rxn_exp is None:
                            rxn_exp = route["score"]
                        edges[n_src][n_tgt] = Edge(
                            source=n_src,
                            target=n_tgt,
                            title=f"route {route_id}: {step['name']}",
                            value=abs(rxn_exp),
                            color=route_color,
                        )

                    k += 1

            else:
                raise ValueError(f"Unknown node type {step['nodeType']}")

    # Scale edge widths
    final_edges = [x for n in edges.values() for x in n.values()]
    scaling_factor = max_score - min_score
    for e in final_edges:
        e.value = (e.value - min_score) / scaling_factor
        e.value = min(e.value, 0.75)

    return list(nodes.values()), final_edges