diff --git a/docs/http-api.md b/docs/http-api.md index d881081..fa3221d 100644 --- a/docs/http-api.md +++ b/docs/http-api.md @@ -113,7 +113,7 @@ GET /query_connectivity?upstream_type=DA1 lPN&downstream_type=Kenyon cell |---|---| | `upstream_type`, `downstream_type` | Neuron type labels, synonyms or FBbt ids. **At least one is required**; giving one asks "everything downstream of / upstream of this". | | `weight` | Minimum synapse count for a connection to be reported. Default 5. | -| `group_by_class` | `true` aggregates to one row per class pair, with `pairwise_connections`, `average_weight` and `percent_connected`. Default is one row per neuron pair. | +| `group_by_class` | `true` aggregates by class, rolled up over the subclass hierarchy: a connection counts toward **every** (upstream level, downstream level) pair up to the queried type(s), so a row appears for the queried type itself *and* for each subclass that has data — with `pairwise_connections`, `total_weight`, `average_weight` and `percent_connected`. A single connection therefore contributes to several rows, so the per-row figures do not sum to the raw connection count. This matches `vfb_connect`'s `get_connected_neurons_by_type`. Default is one row per neuron pair. | | `exclude_dbs` | Comma-separated datasets to leave out. Defaults to `hb,fafb` — see below. Pass `exclude_dbs=` (empty) for every dataset. A symbol (`mc`), a short_form (`male_cns_v0_9`), a label (`male-cns`) or the whole label all name the same dataset; an unrecognised one is a **400** with suggestions. | | `include_graph` | Attach a graph structure alongside the table. | | `force_refresh` | Bypass the cache. | diff --git a/src/test/test_graph_builder.py b/src/test/test_graph_builder.py index ddc9131..f5934e9 100644 --- a/src/test/test_graph_builder.py +++ b/src/test/test_graph_builder.py @@ -178,6 +178,10 @@ def fake_batch(ids): } import vfbquery.graph_builder as gb monkeypatch.setattr(gb, "batch_lookup_ids", fake_batch) + # Containment edges need a live graph DB; default to none so unit tests + # don't reach the network. Tests that exercise the compound graph override + # this with their own stub. + monkeypatch.setattr(gb, "subclass_containment_edges", lambda ids: []) class TestGraphFromQueryConnectivity: @@ -205,8 +209,43 @@ def test_class_level(self, monkeypatch): assert len(g["nodes"]) == 2 assert len(g["edges"]) == 1 assert g["edges"][0]["weight"] == 5000 + assert g["edges"][0]["relation"] == "synapsed_to" assert g["directed"] is True + def test_class_level_compound_graph(self, monkeypatch): + """Rolled-up results add containment edges, tagged distinctly from the + synapsed_to connectivity edges so a renderer can style them apart.""" + _mock_batch_lookup(monkeypatch) + import vfbquery.graph_builder as gb + # Stub the graph-DB lookup: child class is a subclass of the parent. + monkeypatch.setattr(gb, "subclass_containment_edges", lambda ids: [ + {"source": "FBbt_child", "target": "FBbt_parent", + "relation": "SUBCLASSOF", "label": "subclass of", "weight": 0, + **gb.SUBCLASS_EDGE_STYLE}, + ] if "FBbt_parent" in ids and "FBbt_child" in ids else []) + connections = [ + {"upstream_class": "EPG", "upstream_class_id": "FBbt_parent", + "downstream_class": "ExR1", "downstream_class_id": "FBbt_d1", + "total_upstream_count": 10, "connected_upstream_count": 9, + "percent_connected": 90, "pairwise_connections": 20, + "total_weight": 500, "average_weight": 25}, + {"upstream_class": "EPG_PB1", "upstream_class_id": "FBbt_child", + "downstream_class": "ExR1", "downstream_class_id": "FBbt_d1", + "total_upstream_count": 4, "connected_upstream_count": 4, + "percent_connected": 100, "pairwise_connections": 5, + "total_weight": 80, "average_weight": 16}, + ] + g = graph_from_query_connectivity(connections, group_by_class=True, + upstream_type="EPG", + downstream_type="ExR1") + rels = [e.get("relation") for e in g["edges"]] + assert rels.count("synapsed_to") == 2 + assert rels.count("SUBCLASSOF") == 1 + sub = next(e for e in g["edges"] if e["relation"] == "SUBCLASSOF") + assert sub["source"] == "FBbt_child" and sub["target"] == "FBbt_parent" + assert sub["style"] == "dashed" # pattern-coded + assert sub["color"] != g["edges"][0].get("color") # colour-coded apart + def test_per_neuron(self, monkeypatch): _mock_batch_lookup(monkeypatch) connections = [ @@ -321,6 +360,32 @@ def test_filters_to_input_term_block(self, monkeypatch): assert g["edges"][0]["target"] == "FBbt_d1" assert g["edges"][0]["weight"] == 5000 # input-term block, not the subclass's 500 + def test_compound_graph_tags_edges(self, monkeypatch): + """Partner classes are rolled up over the hierarchy, so containment + edges are added and tagged apart from the synapsed_to edges.""" + _mock_batch_lookup(monkeypatch) + import vfbquery.graph_builder as gb + monkeypatch.setattr(gb, "subclass_containment_edges", lambda ids: [ + {"source": "FBbt_d_child", "target": "FBbt_d_parent", + "relation": "SUBCLASSOF", "label": "subclass of", "weight": 0, + **gb.SUBCLASS_EDGE_STYLE}, + ] if "FBbt_d_parent" in ids and "FBbt_d_child" in ids else []) + rows = [ + {"id": "FBbt_d_parent", "query_id": "FBbt_primary", + "downstream_class": "[ExR1](FBbt_d_parent)", + "pairwise_connections": 200, "total_weight": 5000}, + {"id": "FBbt_d_child", "query_id": "FBbt_primary", + "downstream_class": "[ExR1 DM4](FBbt_d_child)", + "pairwise_connections": 50, "total_weight": 1000}, + ] + g = graph_from_downstream_class(rows, "FBbt_primary", "EPG") + rels = [e.get("relation") for e in g["edges"]] + assert rels.count("synapsed_to") == 2 + assert rels.count("SUBCLASSOF") == 1 + sub = next(e for e in g["edges"] if e["relation"] == "SUBCLASSOF") + assert sub["source"] == "FBbt_d_child" and sub["target"] == "FBbt_d_parent" + assert sub["style"] == "dashed" + class TestGraphFromUpstreamClass: def test_basic(self, monkeypatch): diff --git a/src/test/test_vfb_connectivity.py b/src/test/test_vfb_connectivity.py index 1af8df4..b285cf3 100644 --- a/src/test/test_vfb_connectivity.py +++ b/src/test/test_vfb_connectivity.py @@ -138,6 +138,72 @@ def test_group_by_class(self): assert "downstream_class" in conn +class TestQueryConnectivityRollup: + """Grouped queries roll up over the subclass hierarchy, so a row appears for + every level with data up to each named query term — the two-ended analogue + of the single-ended DownstreamClassConnectivity / UpstreamClassConnectivity + rollup. The distinguishing case is EPG <-> ExR1: across datasets EPG is + typed both to its parent class and per-glomerulus, and ExR1 both to its + parent and to DM3/DM4 lineage, so only a rollup produces the parent-to-parent + row. exclude_dbs=[] keeps every dataset in scope.""" + + #: EPG parent and ExR1 parent. EPG has per-glomerulus subclasses; ExR1 has + #: DM3/DM4 lineage subclasses. Both parents have directly-typed instances in + #: some datasets, which is what makes the parent-to-parent row real. + ROLLUP_UP = "FBbt_00047030" # EPG + ROLLUP_DOWN = "FBbt_00003655" # ExR1 + + @pytest.fixture(scope="class") + def rollup_result(self): + return query_connectivity( + upstream_type=self.ROLLUP_UP, + downstream_type=self.ROLLUP_DOWN, + group_by_class=True, + exclude_dbs=[], + ) + + @pytest.mark.integration + def test_top_level_query_term_row_present(self, rollup_result): + # The row the non-rolled-up grouping never emitted: both sides at the + # named query term, aggregating every level beneath them. + conns = rollup_result["connections"] + assert conns + assert any( + c["upstream_class_id"] == self.ROLLUP_UP + and c["downstream_class_id"] == self.ROLLUP_DOWN + for c in conns + ), "expected a rolled-up EPG->ExR1 parent-to-parent row" + + @pytest.mark.integration + def test_finer_levels_also_present(self, rollup_result): + # Rollup adds levels, it does not replace them: subclass-level rows still + # appear alongside the parent-to-parent row. + conns = rollup_result["connections"] + assert any( + c["upstream_class_id"] != self.ROLLUP_UP + or c["downstream_class_id"] != self.ROLLUP_DOWN + for c in conns + ) + + @pytest.mark.integration + def test_parent_row_dominates_its_children(self, rollup_result): + # The parent-to-parent row is a set-union over every child pair, so its + # weight is at least that of any single child pair sharing an endpoint. + conns = rollup_result["connections"] + top = next( + c for c in conns + if c["upstream_class_id"] == self.ROLLUP_UP + and c["downstream_class_id"] == self.ROLLUP_DOWN + ) + children = [ + c for c in conns + if c["upstream_class_id"] == self.ROLLUP_UP + and c["downstream_class_id"] != self.ROLLUP_DOWN + ] + for child in children: + assert top["total_weight"] >= child["total_weight"] + + class TestQueryConnectivityWeightFiltering: @pytest.mark.integration def test_higher_weight_fewer_results(self): diff --git a/src/vfbquery/graph_builder.py b/src/vfbquery/graph_builder.py index cc06788..10da7bb 100644 --- a/src/vfbquery/graph_builder.py +++ b/src/vfbquery/graph_builder.py @@ -17,6 +17,15 @@ MAX_NODES = 80 MAX_EDGES = 200 + +#: Above this many class nodes, skip the subclass-containment lookup. The +#: transitive-reduction query is fine for a focused graph (a bounded subtree, +#: e.g. EPG→ExR1 at ~14 classes) but explodes when a one-sided rolled-up query +#: pulls in high-level classes up to the neuron root (hundreds of classes, +#: broad ``SUBCLASSOF`` walks) — enough to time a query out. A big flat graph +#: gains little from containment nesting anyway, so it degrades to the plain +#: (non-compound) graph. See :func:`subclass_containment_edges`. +MAX_CONTAINMENT_CLASSES = 50 GRAPH_VERSION = 1 # Neurotransmitter group colours (matching VFBchat conventions) @@ -97,6 +106,69 @@ def batch_lookup_ids(ids): return {} +#: Visual encoding for containment (subclass) edges, so a renderer can tell +#: structural ``SUBCLASSOF`` edges apart from ``synapsed_to`` connectivity edges. +#: Dashed + muted so the connectivity edges stay the focus. +SUBCLASS_EDGE_STYLE = {"style": "dashed", "color": "#9aa4b2"} + + +def subclass_containment_edges(class_ids): + """Containment edges (``child`` → ``parent``) among *class_ids*, as the + transitive reduction restricted to the present set — each class links only + to its *nearest present* ancestor, so the result is a clean nesting rather + than an edge to every ancestor. + + Used to turn a rolled-up class-connectivity graph into a compound graph: + the same connection appears at several hierarchy levels, and these edges + make that nesting explicit instead of leaving parent and child as unrelated + siblings. Each edge is tagged ``relation="SUBCLASSOF"`` and carries dashed/ + muted style hints (see :data:`SUBCLASS_EDGE_STYLE`) so a renderer can style + it differently from the ``synapsed_to`` connectivity edges. Returns ``[]`` + for a graph with more than :data:`MAX_CONTAINMENT_CLASSES` class nodes (the + query is too costly there and the nesting adds little), and on any error, so + an unavailable graph DB degrades to the plain (non-compound) graph. + + :param class_ids: ids of the class nodes already in the graph + :return: list of edge dicts ``{source, target, relation, ...}`` + """ + ids = [i for i in dict.fromkeys(class_ids) if i] + if len(ids) < 2 or len(ids) > MAX_CONTAINMENT_CLASSES: + return [] + id_list = str(ids) + try: + nc = Neo4jConnect() + cypher = ( + "MATCH (child:Class)-[:SUBCLASSOF*1..]->(parent:Class) " + f"WHERE child.short_form IN {id_list} " + f"AND parent.short_form IN {id_list} AND child <> parent " + "AND NOT EXISTS { " + "MATCH (child)-[:SUBCLASSOF*1..]->(mid:Class)-[:SUBCLASSOF*1..]->(parent) " + f"WHERE mid.short_form IN {id_list} AND mid <> child AND mid <> parent " + "} " + "RETURN DISTINCT child.short_form AS child, parent.short_form AS parent" + ) + results = nc.commit_list([cypher]) + if not results: + return [] + rows = dict_cursor(results) + except Exception: + return [] + edges = [] + for r in rows: + child, parent = r.get("child"), r.get("parent") + if not child or not parent: + continue + edges.append({ + "source": child, + "target": parent, + "relation": "SUBCLASSOF", + "label": "subclass of", + "weight": 0, + **SUBCLASS_EDGE_STYLE, + }) + return edges + + # --------------------------------------------------------------------------- # Group assignment # --------------------------------------------------------------------------- @@ -167,9 +239,16 @@ def build_graph(nodes, edges, title=None, directed=True, layout="force"): orig_node_count = len(deduped_nodes) orig_edge_count = len(edges) - # Truncate edges — keep highest weight first + # Truncate edges — keep highest-weight connectivity first, but never drop + # structural (containment) edges: they carry no weight, so a plain weight + # sort would push them to the bottom and cut them, losing the nesting. if len(edges) > MAX_EDGES: - edges = sorted(edges, key=lambda e: e.get("weight") or 0, reverse=True)[:MAX_EDGES] + structural = [e for e in edges if e.get("relation") == "SUBCLASSOF"] + weighted = [e for e in edges if e.get("relation") != "SUBCLASSOF"] + budget = max(0, MAX_EDGES - len(structural)) + weighted = sorted(weighted, key=lambda e: e.get("weight") or 0, + reverse=True)[:budget] + edges = weighted + structural # Truncate nodes — keep those with highest degree if len(deduped_nodes) > MAX_NODES: @@ -289,7 +368,14 @@ def graph_from_query_connectivity(connections, group_by_class, "source": up_id, "target": dn_id, "weight": c.get("total_weight", 0), + "relation": "synapsed_to", }) + # Compound graph: a rolled-up result carries the same connection at + # several class levels, so add containment edges linking each class to + # its nearest present ancestor. Tagged distinctly (dashed/muted) from + # the synapsed_to edges above so subclass structure reads as nesting + # rather than as more connectivity. + edges.extend(subclass_containment_edges(list(nodes.keys()))) else: # Per-neuron results for c in connections: @@ -565,8 +651,14 @@ def graph_from_downstream_class(rows, primary_id, primary_label=None): "source": primary_id, "target": rid, "weight": weight, + "relation": "synapsed_to", }) + # Partner classes are rolled up over the subclass hierarchy (the single-ended + # rollup walks partners to the neuron root), so add containment edges — tagged + # apart from the synapsed_to edges — to show that nesting rather than leaving + # a partner parent and its subclasses as unrelated siblings. + edges.extend(subclass_containment_edges([n["id"] for n in nodes])) disp = _node_display_label(primary_info) or primary_label or primary_id return build_graph(nodes, edges, title=f"Downstream of {disp}", directed=True) @@ -638,7 +730,13 @@ def graph_from_upstream_class(rows, primary_id, primary_label=None): "source": rid, "target": primary_id, "weight": weight, + "relation": "synapsed_to", }) + # Partner classes are rolled up over the subclass hierarchy (the single-ended + # rollup walks partners to the neuron root), so add containment edges — tagged + # apart from the synapsed_to edges — to show that nesting rather than leaving + # a partner parent and its subclasses as unrelated siblings. + edges.extend(subclass_containment_edges([n["id"] for n in nodes])) disp = _node_display_label(primary_info) or primary_label or primary_id return build_graph(nodes, edges, title=f"Upstream of {disp}", directed=True) diff --git a/src/vfbquery/owlery_client.py b/src/vfbquery/owlery_client.py index af12656..0272afb 100644 --- a/src/vfbquery/owlery_client.py +++ b/src/vfbquery/owlery_client.py @@ -65,8 +65,9 @@ def __init__(self, owlery_endpoint: str = "http://owl.virtualflybrain.org/kbs/vf """ self.owlery_endpoint = owlery_endpoint.rstrip('/') - def get_subclasses(self, query: str, query_by_label: bool = True, - verbose: bool = False, direct: bool = False) -> List[str]: + def get_subclasses(self, query: str, query_by_label: bool = True, + verbose: bool = False, direct: bool = False, + timeout: int = 2400) -> List[str]: """ Query Owlery for subclasses matching an OWL class expression. @@ -129,7 +130,7 @@ def convert_short_form_to_iri(match): response = session.get( f"{self.owlery_endpoint}/subclasses", params=params, - timeout=2400 + timeout=timeout ) if verbose: diff --git a/src/vfbquery/vfb_connectivity.py b/src/vfbquery/vfb_connectivity.py index 82f003a..67271d2 100644 --- a/src/vfbquery/vfb_connectivity.py +++ b/src/vfbquery/vfb_connectivity.py @@ -3,8 +3,8 @@ Uses VFBquery's Neo4jConnect client to run Cypher queries directly against the VFB Neo4j database, without depending on vfb_connect. -Two behaviours here are worth knowing before reading the code, because both -are the difference between an empty answer and a correct one. +Three behaviours here are worth knowing before reading the code, because each +is the difference between an empty, wrong, or correct answer. **Queries are expanded over the subclass hierarchy.** Asking for "Kenyon cell" means "Kenyon cell *or any of its subclasses*". This is not a nicety: in FBbt @@ -14,10 +14,21 @@ named class alone returns nothing at all for the single most obvious query in the mushroom body. See :func:`_subclass_closure`. +**Class-aggregated queries (`group_by_class=True`) roll up over the +hierarchy.** A connection is attributed to every (upstream level, downstream +level) pair up to the queried type(s), so a row appears for the queried type +itself *and* for each subclass with data — not just the directly asserted +class. One connection therefore appears in several rows, and per-row +``pairwise_connections`` / ``total_weight`` do not sum to the raw connection +count. This mirrors the single-ended class-connectivity queries in +``vfb_queries`` and ``vfb_connect``'s ``get_connected_neurons_by_type``. See +:func:`_rollup_grouped_connections`. + **Some connectome datasets are excluded by default.** See :data:`DEFAULT_EXCLUDE_DBS` for which, and why. """ from .neo4j_client import Neo4jConnect, dict_cursor +from .owlery_client import SimpleVFBConnect #: Connectome datasets excluded unless the caller says otherwise. @@ -55,6 +66,16 @@ #: would not be usable regardless. MAX_SUBCLASS_IDS = 10000 +#: How long to wait for Owlery's subclass reasoning before giving up and falling +#: back to a Neo4j ``SUBCLASSOF`` closure. The Owlery client's own default is 40 +#: minutes (tuned for heavy OWL reasoning); that is far too long here, where an +#: Owlery outage would otherwise hang every connectivity query — and the CI +#: graph test — for the full 40 minutes. Owlery answers a neuron-class subclass +#: query in well under this, so a miss means Owlery is unhealthy, not slow, and +#: the Neo4j fallback (asserted SUBCLASSOF, a close approximation) keeps results +#: correct rather than empty. See :func:`_subclass_closure`. +OWLERY_TIMEOUT_SECONDS = 30 + _NC = None @@ -75,6 +96,26 @@ def _get_nc(): return _NC +_VC = None + + +def _get_vc(): + """Return a process-wide :class:`SimpleVFBConnect`, used only for its Owlery + client (``vc.vfb.oc``). + + The subclass closure is taken from the Owlery reasoner rather than a Neo4j + ``SUBCLASSOF`` traversal so that this module's expansion matches the + single-ended class-connectivity queries in ``vfb_queries`` exactly — Owlery + is described there as the canonical subclass set used throughout VFBquery. + Instances are still counted in Neo4j (Owlery's ``get_instances`` has been + observed to hang), so only ``get_subclasses`` is used here. + """ + global _VC + if _VC is None: + _VC = SimpleVFBConnect() + return _VC + + def _cypher_str(value): """Quote a Python string as a Cypher single-quoted literal. @@ -196,14 +237,44 @@ def _resolve_neuron_type_label(nc, label, notes=None): ) +def _neo4j_subclass_ids(nc, class_id): + """Neo4j ``SUBCLASSOF`` closure of a neuron class (the class plus every + asserted subclass beneath it), used as the fallback when Owlery is + unavailable. Returns ``[]`` on error so the caller can still proceed with + the class itself. + """ + quoted = _cypher_str(class_id) + try: + results = nc.commit_list([ + f"MATCH (c:Class:Neuron {{short_form: {quoted}}})\n" + "OPTIONAL MATCH (c)<-[:SUBCLASSOF*0..]-(sub:Class)\n" + "RETURN collect(DISTINCT sub.short_form) AS ids" + ]) + dc = dict_cursor(results) + if dc: + return [i for i in (dc[0].get("ids") or []) if i] + except Exception as e: + print(f"Neo4j subclass fallback failed for {class_id}: {e}") + return [] + + def _subclass_closure(nc, class_id): """Return ``(label, subclass_ids, instance_count)`` for a neuron class. - ``subclass_ids`` is the class itself plus every class beneath it under - ``SUBCLASSOF``, and ``instance_count`` is how many non-deprecated - connectivity individuals are typed to any of them. Both come from one - round-trip because the caller needs both: the ids to query with, and the - count to decide which side of a two-sided query to drive from. + ``subclass_ids`` is the class itself plus every class beneath it, taken from + the **Owlery reasoner** (``get_subclasses``) so the closure matches the + single-ended class-connectivity queries in ``vfb_queries`` exactly. Owlery + excludes the queried class itself, so it is added back. ``instance_count`` + is how many non-deprecated connectivity individuals are typed to any class + in the closure, counted in Neo4j. + + Owlery is called with a short timeout (:data:`OWLERY_TIMEOUT_SECONDS`), not + its 40-minute default: an Owlery outage must not hang every connectivity + query. On timeout or error we fall back to the Neo4j ``SUBCLASSOF`` closure + (:func:`_neo4j_subclass_ids`) — asserted rather than reasoned, but a close + approximation that keeps results correct instead of collapsing to the + queried class alone. ``subs is None`` distinguishes an Owlery failure (fall + back) from Owlery legitimately returning no subclasses (a leaf class). Why the count matters is worth stating plainly, since getting it wrong is a two-minute query instead of a half-second one: expanding *both* sides of a @@ -211,22 +282,32 @@ def _subclass_closure(nc, class_id): complete in reasonable time. Driving from the smaller side and filtering the partner by an id list does. See :func:`_build_connectivity_cypher`. """ + subs = None + try: + subs = _get_vc().vfb.oc.get_subclasses( + query=f"<{class_id}>", query_by_label=False, verbose=False, + timeout=OWLERY_TIMEOUT_SECONDS, + ) + except Exception as e: + print(f"Owlery subclass query failed for {class_id}: {e}; " + "falling back to Neo4j SUBCLASSOF closure") + if subs is None: + subs = _neo4j_subclass_ids(nc, class_id) + ids = sorted(set(subs or []) | {class_id}) + quoted = _cypher_str(class_id) results = nc.commit_list([ f"MATCH (c:Class:Neuron) WHERE c.short_form = {quoted}\n" "WITH c LIMIT 1\n" - "OPTIONAL MATCH (c)<-[:SUBCLASSOF*0..]-(sub:Class)\n" - "WITH c, collect(DISTINCT sub.short_form) AS ids\n" "OPTIONAL MATCH (s:Class)<-[:INSTANCEOF]-" "(n:Individual:Neuron:has_neuron_connectivity)\n" - "WHERE s.short_form IN ids AND NOT n:Deprecated\n" - "RETURN c.label AS label, ids AS ids, count(DISTINCT n) AS instances" + f"WHERE s.short_form IN {_id_list(ids)} AND NOT n:Deprecated\n" + "RETURN c.label AS label, count(DISTINCT n) AS instances" ]) dc = dict_cursor(results) if not dc: - return class_id, [class_id], 0 + return class_id, ids, 0 row = dc[0] - ids = sorted(set(row.get("ids") or []) | {class_id}) return row.get("label") or class_id, ids, row.get("instances") or 0 @@ -318,7 +399,12 @@ def query_connectivity(upstream_type=None, downstream_type=None, weight=5, :param upstream_type: Presynaptic neuron type label (optional) :param downstream_type: Postsynaptic neuron type label (optional) :param weight: Minimum synapse count threshold (default 5) - :param group_by_class: Aggregate by neuron class (default False) + :param group_by_class: Aggregate by class, rolled up over the subclass + hierarchy (default False). A connection counts toward every (upstream + level, downstream level) pair up to the queried type(s), so a row + appears for each level with data — matching vfb_connect's + ``get_connected_neurons_by_type``. See + :func:`_rollup_grouped_connections`. :param exclude_dbs: Dataset symbols to exclude; defaults to :data:`DEFAULT_EXCLUDE_DBS`, which documents why. Pass ``[]`` for every dataset. @@ -429,11 +515,22 @@ def _query_connectivity_uncached(upstream_type=None, downstream_type=None, weigh # For a one-sided query the supplied side is the only candidate. anchor = min(sides, key=lambda s: sides[s][1]) + # Class-aggregated queries roll up over the subclass hierarchy so a row + # appears for every level with data up to each named query term — the + # two-ended analogue of the single-ended DownstreamClassConnectivity / + # UpstreamClassConnectivity rollup. See :func:`_rollup_grouped_connections`. + if group_by_class: + connections = _rollup_grouped_connections( + nc, resolved, sides, weight, exclude_dbs, anchor + ) + return {"connections": connections, "warnings": warnings, + "count": len(connections), "resolved": resolved} + cypher = _build_connectivity_cypher( upstream_ids=sides["upstream"][0] if "upstream" in sides else None, downstream_ids=sides["downstream"][0] if "downstream" in sides else None, weight=weight, - group_by_class=group_by_class, + group_by_class=False, exclude_dbs=exclude_dbs, anchor=anchor, ) @@ -448,6 +545,265 @@ def _query_connectivity_uncached(upstream_type=None, downstream_type=None, weigh "resolved": resolved} +def _class_instance_membership(nc, class_ids, exclude_dbs): + """For each class in ``class_ids`` that has connectivity individuals, return + the set of individuals in its ``SUBCLASSOF`` closure plus the class label. + + Individuals carry the same deprecation and dataset filters the connectivity + match applies, so this doubles as the ``total_upstream_count`` denominator + for the block (presynaptic) side: counting a neuron here that the dataset + filter excludes from the numerator would depress the whole class's percent. + + :return: ``(class_to_instances, labels)`` — classes with no surviving + individuals simply do not appear. + """ + if not class_ids: + return {}, {} + dbf = _db_filter_predicate("n", exclude_dbs) if exclude_dbs else None + q = ( + "MATCH (c:Class)<-[:SUBCLASSOF*0..]-(:Class)<-[:INSTANCEOF]-" + "(n:Individual:has_neuron_connectivity)\n" + f"WHERE c.short_form IN {_id_list(class_ids)}\n" + "AND NOT n:Deprecated" + + (f"\nAND {dbf}" if dbf else "") + + "\nRETURN c.short_form AS cid, c.label AS label, " + "collect(DISTINCT n.short_form) AS iids" + ) + class_to_instances = {} + labels = {} + for r in dict_cursor(nc.commit_list([q])): + cid = r.get("cid") + iids = set(r.get("iids") or []) + if not cid or not iids: + continue + class_to_instances[cid] = iids + labels[cid] = r.get("label") or cid + return class_to_instances, labels + + +def _partner_class_membership(nc, partner_instances, partner_ids): + """Map each partner individual to the set of neuron classes it belongs to, + with labels, for the set-union rollup. + + Bounded by ``partner_ids`` (the partner's own subclass closure) when the + partner side was named, so the rollup stops at that query term; for a + one-sided query ``partner_ids`` is ``None`` and every ``:Neuron`` ancestor + is kept, walking up to the neuron root as the single-ended partner rollup + does. Set-union over these memberships is what keeps FBbt multi-inheritance + from double-counting. + + :return: ``(instance_to_classes, labels)`` + """ + if not partner_instances: + return {}, {} + bound = "" + if partner_ids: + bound = f"\nAND c.short_form IN {_id_list(partner_ids)}" + q = ( + "MATCH (n:Individual)-[:INSTANCEOF]->(:Class:Neuron)" + "-[:SUBCLASSOF*0..]->(c:Class:Neuron)\n" + f"WHERE n.short_form IN {_id_list(sorted(partner_instances))}" + bound + + "\nRETURN n.short_form AS iid, " + "collect(DISTINCT c.short_form) AS cids, " + "collect(DISTINCT [c.short_form, c.label]) AS labs" + ) + inst_to_classes = {} + labels = {} + for r in dict_cursor(nc.commit_list([q])): + iid = r.get("iid") + if not iid: + continue + inst_to_classes[iid] = set(r.get("cids") or []) + for pair in r.get("labs") or []: + if isinstance(pair, (list, tuple)) and len(pair) == 2 and pair[0]: + labels[pair[0]] = pair[1] or pair[0] + return inst_to_classes, labels + + +def _partner_class_totals(nc, class_ids, exclude_dbs): + """Count the connectivity individuals in each partner class's ``SUBCLASSOF`` + closure, deprecation- and dataset-filtered. + + This is the ``total_upstream_count`` denominator only when the partner side + is presynaptic — i.e. a downstream-only query, where the upstream partner is + the source of each connection. When the block side is presynaptic the + denominator comes from :func:`_class_instance_membership` instead. + """ + if not class_ids: + return {} + dbf = _db_filter_predicate("n", exclude_dbs) if exclude_dbs else None + q = ( + "MATCH (c:Class)<-[:SUBCLASSOF*0..]-(:Class)<-[:INSTANCEOF]-" + "(n:Individual:has_neuron_connectivity)\n" + f"WHERE c.short_form IN {_id_list(class_ids)}\n" + "AND NOT n:Deprecated" + + (f"\nAND {dbf}" if dbf else "") + + "\nRETURN c.short_form AS cid, count(DISTINCT n) AS total" + ) + totals = {} + for r in dict_cursor(nc.commit_list([q])): + cid = r.get("cid") + if cid: + totals[cid] = r.get("total") or 0 + return totals + + +def _rollup_grouped_connections(nc, resolved, sides, weight, exclude_dbs, anchor): + """Class-aggregated connections rolled up over the subclass hierarchy, so a + row appears for every level with data up to each named query term. + + This is the two-ended analogue of the single-ended rollup in + ``vfb_queries._aggregate_class_connectivity`` (added in ca94f17), and mirrors + it: one side is enumerated as **blocks** — its query term first (aggregated + over its whole instance population), then each subclass that has data — while + the partner side is **rolled up** over its ancestor classes by set-union on + instance memberships. Together the two produce a row for every (upstream + level, downstream level) pair that has a connection, including the top-level + (query term, query term) pair the non-rolled-up grouping never emitted. + + Differences from single-ended, all deliberate: edges come from the live + Neo4j ``synapsed_to`` match (not the Solr per-instance cache), so ``weight`` + and ``exclude_dbs`` apply and every dataset is visible; the partner rollup is + bounded by the partner's own query closure when that side was named, rather + than always walking to the neuron root. + + When both sides are named the **upstream** side supplies the blocks and the + downstream side is the rolled-up partner; for a one-sided query the named + side supplies the blocks and the partner rolls up to the neuron root. + ``percent_connected`` normalizes on the presynaptic side, matching the + non-rolled-up path and VFB_connect: the returned ``total_upstream_count`` / + ``connected_upstream_count`` always describe the presynaptic (upstream) + class of each row. + """ + from collections import defaultdict + + up_ids = sides["upstream"][0] if "upstream" in sides else None + down_ids = sides["downstream"][0] if "downstream" in sides else None + up_term = resolved.get("upstream", {}).get("id") + down_term = resolved.get("downstream", {}).get("id") + + # 1. Filtered per-pair edges from the live query (same filters as the + # non-grouped path). Deduplicate by (n1, n2) so a neuron with several + # data-source cross-references is not counted once per source. + cypher = _build_connectivity_cypher( + upstream_ids=up_ids, downstream_ids=down_ids, weight=weight, + group_by_class=False, exclude_dbs=exclude_dbs, anchor=anchor, + ) + edge_weight = {} + for r in dict_cursor(nc.commit_list([cypher])): + u = r.get("upstream_neuron_id") + d = r.get("downstream_neuron_id") + w = r.get("weight") + if not u or not d or w is None: + continue + try: + edge_weight[(u, d)] = float(w) + except (TypeError, ValueError): + continue + if not edge_weight: + return [] + + # 2. Block side = named query side (upstream when both are named); the + # partner side is the other. found[block_instance] = [(partner_inst, w)]. + block_side = "upstream" if up_ids else "downstream" + block_term = up_term if block_side == "upstream" else down_term + block_ids = up_ids if block_side == "upstream" else down_ids + partner_ids = down_ids if block_side == "upstream" else up_ids + queried_is_presynaptic = (block_side == "upstream") + + found = defaultdict(list) + partner_instances = set() + for (u, d), w in edge_weight.items(): + if block_side == "upstream": + found[u].append((d, w)) + partner_instances.add(d) + else: + found[d].append((u, w)) + partner_instances.add(u) + + # 3. Memberships. Block side: (sub)class -> filtered instance set + labels + # (also the presynaptic denominator). Partner side: instance -> {ancestor + # classes} + labels, for the set-union rollup. + block_class_instances, block_labels = _class_instance_membership( + nc, block_ids, exclude_dbs) + if not block_class_instances: + return [] + partner_inst_classes, partner_labels = _partner_class_membership( + nc, partner_instances, partner_ids) + partner_totals = {} + if not queried_is_presynaptic: + all_partner_classes = set() + for cs in partner_inst_classes.values(): + all_partner_classes |= cs + partner_totals = _partner_class_totals( + nc, all_partner_classes, exclude_dbs) + + def block_for(query_id): + instances = block_class_instances.get(query_id) or set() + if not instances: + return [] + total_queried = len(instances) + buckets = defaultdict(lambda: { + "edges": set(), "weight_sum": 0.0, + "connected_queried": set(), "connected_partner": set(), + }) + for n1 in instances: + for n2, w in found.get(n1, ()): + if w <= 0: + continue + for c in partner_inst_classes.get(n2, ()): + b = buckets[c] + b["edges"].add((n1, n2)) + b["weight_sum"] += w + b["connected_queried"].add(n1) + b["connected_partner"].add(n2) + block = [] + for cid, b in buckets.items(): + pw = len(b["edges"]) + tw = b["weight_sum"] + if queried_is_presynaptic: + total = total_queried + connected = len(b["connected_queried"]) + else: + total = partner_totals.get(cid, 0) + connected = len(b["connected_partner"]) + pct = round((connected / total) * 100) if total else 0 + # Integer weight and integer division, matching VFB_connect's + # get_connected_neurons_by_type and the previous Cypher grouping + # (both produced integers); synapse counts are integers. + avg = (tw // pw) if pw else 0 + if block_side == "upstream": + up_id, up_label = query_id, block_labels.get(query_id, query_id) + dn_id, dn_label = cid, partner_labels.get(cid, cid) + else: + up_id, up_label = cid, partner_labels.get(cid, cid) + dn_id, dn_label = query_id, block_labels.get(query_id, query_id) + block.append({ + "upstream_class": up_label, + "upstream_class_id": up_id, + "downstream_class": dn_label, + "downstream_class_id": dn_id, + "total_upstream_count": total, + "connected_upstream_count": connected, + "percent_connected": pct, + "pairwise_connections": pw, + "total_weight": int(tw), + "average_weight": int(avg), + }) + block.sort( + key=lambda r: (r["pairwise_connections"], r["average_weight"]), + reverse=True, + ) + return block + + # 4. Query term first, then one block per subclass (ordered by class id) — + # the same layout as the single-ended queries. + rows = block_for(block_term) + for cid in sorted(c for c in block_class_instances if c != block_term): + rows.extend(block_for(cid)) + return rows + + def _build_connectivity_cypher(upstream_ids, downstream_ids, weight, group_by_class, exclude_dbs, anchor="upstream"): """Build the Cypher query for connectivity. diff --git a/src/vfbquery/vfb_queries.py b/src/vfbquery/vfb_queries.py index 9543632..3f13adc 100644 --- a/src/vfbquery/vfb_queries.py +++ b/src/vfbquery/vfb_queries.py @@ -109,6 +109,39 @@ def get_dict_cursor(): # Replace VfbConnect with SimpleVFBConnect vc = SimpleVFBConnect() +#: Timeout (seconds) for the *simple* Owlery subclass-closure calls used by the +#: connectivity queries — a plain ```` reasoning query, which is fast +#: when the endpoint is healthy: even ``adult neuron`` (~15,400 subclasses, +#: bigger than any realistic connectivity query) returns in under 2s, and normal +#: neuron classes in well under a second. The Owlery client's own default is 40 +#: minutes — tuned for arbitrary class *expressions* that need heavy reasoning — +#: so an Owlery outage would otherwise hang every connectivity query for 40 +#: minutes. A miss here means Owlery is unhealthy, not that the class is large, +#: so we fall back to the Neo4j ``SUBCLASSOF`` closure (:func:`_neo4j_subclass_ids`), +#: which returns the same closure for neuron classes in a fraction of a second. +#: NB: the reasoning-heavy callers (``_owlery_ids_cached``, ``get_anatomy_scrnaseq``) +#: deliberately keep the 40-minute default — do not route them through this. +OWLERY_SUBCLASS_TIMEOUT = 30 + + +def _neo4j_subclass_ids(short_form): + """Neo4j ``SUBCLASSOF`` closure (the class plus every asserted subclass), + the fallback used when Owlery is unavailable or too slow. Returns ``[]`` on + error so the caller can still proceed with the class itself. + """ + quoted = short_form.replace("\\", "\\\\").replace("'", "\\'") + try: + recs = get_dict_cursor()(vc.nc.commit_list([ + f"MATCH (c:Class {{short_form: '{quoted}'}})\n" + "OPTIONAL MATCH (c)<-[:SUBCLASSOF*0..]-(sub:Class)\n" + "RETURN collect(DISTINCT sub.short_form) AS ids" + ])) + if recs: + return [i for i in (recs[0].get("ids") or []) if i] + except Exception as e: + print(f"Neo4j subclass fallback failed for {short_form}: {e}") + return [] + # --------------------------------------------------------------------------- # Canonical VFB term link # --------------------------------------------------------------------------- @@ -4335,11 +4368,13 @@ def _fetch_connectivity_entries(short_form: str, solr_field: str, subclass_ids=N owl_query = f"<{short_form}>" try: subclass_ids = vc.vfb.oc.get_subclasses( - query=owl_query, query_by_label=False, verbose=False + query=owl_query, query_by_label=False, verbose=False, + timeout=OWLERY_SUBCLASS_TIMEOUT, ) except Exception as e: - print(f"Owlery subclass query failed for {short_form}: {e}") - subclass_ids = [] + print(f"Owlery subclass query failed for {short_form}: {e}; " + "falling back to Neo4j SUBCLASSOF closure") + subclass_ids = _neo4j_subclass_ids(short_form) # Always include the queried class itself; normalise to a list. subclass_ids = list(subclass_ids) @@ -4551,11 +4586,13 @@ def _aggregate_class_connectivity(short_form, direction, try: owl_query = f"<{short_form}>" subclass_ids = vc.vfb.oc.get_subclasses( - query=owl_query, query_by_label=False, verbose=False + query=owl_query, query_by_label=False, verbose=False, + timeout=OWLERY_SUBCLASS_TIMEOUT, ) except Exception as e: - print(f"Owlery subclass query failed for {short_form}: {e}") - subclass_ids = [] + print(f"Owlery subclass query failed for {short_form}: {e}; " + "falling back to Neo4j SUBCLASSOF closure") + subclass_ids = _neo4j_subclass_ids(short_form) query_class_ids = {short_form, *(subclass_ids or [])} # 1b. queried (sub)class -> its instances (SUBCLASSOF closure), with labels.