diff --git a/src/test/test_expression_pattern_individual_queries.py b/src/test/test_expression_pattern_individual_queries.py
index e3b9630..288dea0 100644
--- a/src/test/test_expression_pattern_individual_queries.py
+++ b/src/test/test_expression_pattern_individual_queries.py
@@ -33,6 +33,49 @@ def _menu(term_info):
return out
+def _stock_anchors(term_info):
+ """Every short_form a FindStocks query in the menu is anchored on."""
+ return sorted(
+ q.get("takes", {}).get("default", {}).get("short_form")
+ for q in (term_info or {}).get("Queries", []) or []
+ if isinstance(q, dict) and q.get("query") == "FindStocks"
+ )
+
+
+class TestExpressionPatternStockQueries(unittest.TestCase):
+ """FindStocks is propagated to expression patterns via the driver feature(s)
+ reached through the graph (`expresses` / `has_hemidriver`), not by parsing
+ the VFBexp_ short_form. The query anchors on the FlyBase feature (FBtp/...),
+ since find_stocks cannot route the VFBexp_/VFB_ id itself."""
+
+ EP_CLASS = "VFBexp_FBtp0060056" # P{GMR40G10-GAL4} expression pattern
+ EP_INDIVIDUAL = "VFB_00020530" # R40G10 image, has its own `expresses` edge
+ SPLIT_CLASS = "VFBexp_FBtp0129935FBtp0129968" # intersectional, two `has_hemidriver`
+ SPLIT_INDIVIDUAL = "VFB_00070031" # split image, no own driver edge
+
+ def _anchors_or_skip(self, short_form):
+ ti = get_term_info(short_form, preview=False)
+ if not ti:
+ self.skipTest("term_info unavailable (no live VFB backend)")
+ return _stock_anchors(ti)
+
+ def test_class_stock_query_on_driver_feature(self):
+ self.assertEqual(self._anchors_or_skip(self.EP_CLASS), ["FBtp0060056"])
+
+ def test_instance_stock_query_matches_its_driver(self):
+ # Regular EP image carries its own `expresses` edge — same feature as the class.
+ self.assertEqual(self._anchors_or_skip(self.EP_INDIVIDUAL), ["FBtp0060056"])
+
+ def test_split_class_offers_a_stock_query_per_hemidriver(self):
+ self.assertEqual(self._anchors_or_skip(self.SPLIT_CLASS),
+ ["FBtp0129935", "FBtp0129968"])
+
+ def test_split_instance_inherits_hemidriver_stock_queries(self):
+ # No own driver edge: features come from the pattern class it instantiates.
+ self.assertEqual(self._anchors_or_skip(self.SPLIT_INDIVIDUAL),
+ ["FBtp0129935", "FBtp0129968"])
+
+
class TestExpressionPatternIndividualQueries(unittest.TestCase):
"""R40G10 expression-pattern image inherits its class's menu."""
@@ -58,10 +101,16 @@ def test_instance_menu_includes_everything_the_class_offers(self):
# Queries the class offers must all appear on the instance...
missing = set(class_menu) - set(ind_menu)
self.assertFalse(missing, f"instance is missing class queries: {sorted(missing)}")
- # ...and each such inherited query must run on the class, not the instance.
+ # ...anchored identically to the class. Most inherited queries run on the
+ # class itself; FindStocks is the exception — on both the class and the
+ # instance it anchors on the embedded FlyBase feature (find_stocks cannot
+ # route the VFBexp_ id), so parity means "same anchor as the class".
for qid in class_menu:
- self.assertEqual(ind_menu[qid], self.EP_CLASS,
- f"{qid} on the instance should anchor on {self.EP_CLASS}")
+ self.assertEqual(ind_menu[qid], class_menu[qid],
+ f"{qid} on the instance should anchor as it does on the class")
+ if qid != "FindStocks":
+ self.assertEqual(class_menu[qid], self.EP_CLASS,
+ f"{qid} should run on the class {self.EP_CLASS}")
def test_expected_ep_queries_present(self):
if not self.ind:
@@ -70,11 +119,21 @@ def test_expected_ep_queries_present(self):
# SubclassesOf is intentionally NOT expected: it is gated on has_subClass,
# and this expression-pattern class is a leaf (no subclasses), so the query
# would only ever return empty.
- for qid in ("AnatomyExpressedIn", "epFrag", "ListAllAvailableImages",
- "NeuronsPartHere", "PartsOf"):
+ for qid in ("AnatomyExpressedIn", "epFrag", "ListAllAvailableImages"):
self.assertIn(qid, ind_menu, f"expected {qid} inherited onto the EP instance")
self.assertEqual(ind_menu[qid], self.EP_CLASS)
+ def test_guaranteed_empty_queries_excluded(self):
+ """PartsOf / NeuronsPartHere are gated out for expression patterns: they
+ match the Anatomy facet but have no class-level parts or overlapping
+ neuron classes, so they would only ever return empty (epFrag covers an
+ expression pattern's actual parts)."""
+ if not self.ind or not self.cls:
+ self.skipTest("term_info unavailable (no live VFB backend)")
+ for qid in ("PartsOf", "NeuronsPartHere"):
+ self.assertNotIn(qid, _menu(self.cls), f"{qid} should be gated out on the EP class")
+ self.assertNotIn(qid, _menu(self.ind), f"{qid} should be gated out on the EP instance")
+
def test_no_query_is_anchored_on_the_individual(self):
"""Inherited class queries run on the class; none should target the VFB_ instance."""
for qid, anchor in _menu(self.ind).items():
diff --git a/src/vfbquery/vfb_queries.py b/src/vfbquery/vfb_queries.py
index 4a36c2f..f95ad1b 100644
--- a/src/vfbquery/vfb_queries.py
+++ b/src/vfbquery/vfb_queries.py
@@ -1118,10 +1118,15 @@ def term_info_parse_object(results, short_form):
queries.append(q)
# NeuronsPartHere query - for anatomical regions (neuropils, ganglia, etc.)
- # Gate: Class + (Synaptic_neuropil OR Anatomy), but NOT Cell.
- # Excluded for cell classes (neurons, glia, neuroblasts): "neurons with some
- # part in " is not a meaningful query. Cell subsumes Neuron.
- if contains_all_tags(termInfo["SuperTypes"], ["Class"]) and "Cell" not in termInfo["SuperTypes"] and (
+ # Gate: Class + (Synaptic_neuropil OR Anatomy), but NOT Cell or
+ # Expression_pattern.
+ # - NOT Cell (neurons, glia, neuroblasts): "neurons with some part in
+ # " is not a meaningful query. Cell subsumes Neuron.
+ # - NOT Expression_pattern: expression patterns carry the Anatomy tag but
+ # have no neuron class overlapping them at the class level (0 of ~27.5k),
+ # so the query is guaranteed-empty; excluding it also avoids the wasted
+ # Owlery preview call (cf. SubclassesOf / has_subClass below).
+ if contains_all_tags(termInfo["SuperTypes"], ["Class"]) and "Cell" not in termInfo["SuperTypes"] and "Expression_pattern" not in termInfo["SuperTypes"] and (
"Synaptic_neuropil" in termInfo["SuperTypes"] or
"Anatomy" in termInfo["SuperTypes"]
):
@@ -1153,7 +1158,7 @@ def term_info_parse_object(results, short_form):
queries.append(q)
# PartsOf query - for anatomical classes that are not individual cells
- # Gate: Class + Anatomy, but NOT Cell.
+ # Gate: Class + Anatomy, but NOT Cell or Expression_pattern.
# - Anatomy: "part of" is only meaningful for anatomical structures; it
# excludes non-anatomy classes (genes, features, GO process terms,
# deprecated/stage classes, and data-less expression patterns lacking
@@ -1161,9 +1166,13 @@ def term_info_parse_object(results, short_form):
# - NOT Cell: excludes individual cell types (neurons, glia, neuroblasts),
# whose anatomical sub-parts are not modelled usefully at the class
# level. Cell subsumes Neuron, so this also keeps neuron classes out.
- # Retains neuropils, tracts/nerves, clones, ganglia, whole regions, and
- # imaged expression patterns/splits (all Anatomy, not Cell).
- if contains_all_tags(termInfo["SuperTypes"], ["Class", "Anatomy"]) and "Cell" not in termInfo["SuperTypes"]:
+ # - NOT Expression_pattern: expression patterns carry the Anatomy tag but
+ # have no class-level parts (0 of ~27.5k) — their parts are individual
+ # fragments, which epFrag returns — so PartsOf is guaranteed-empty here;
+ # excluding it also avoids the wasted Owlery preview call (cf.
+ # SubclassesOf below).
+ # Retains neuropils, tracts/nerves, clones, ganglia, and whole regions.
+ if contains_all_tags(termInfo["SuperTypes"], ["Class", "Anatomy"]) and "Cell" not in termInfo["SuperTypes"] and "Expression_pattern" not in termInfo["SuperTypes"]:
q = PartsOf_to_schema(termInfo["Name"], {"short_form": vfbTerm.term.core.short_form})
queries.append(q)
@@ -1409,6 +1418,30 @@ def term_info_parse_object(results, short_form):
# FindComboPublications is offered as a run_query query_type above;
# the dedicated find_combo_publications MCP tool was retired.
+ # FlyBase stocks for expression patterns. An expression pattern's stocks
+ # are those of the FlyBase feature(s) it drives, reached by navigating
+ # the knowledge graph rather than parsing the VFBexp_ short_form:
+ # the `expresses` / `has_hemidriver` edges (already loaded in the term's
+ # relationships) name the driver construct, or both halves of a split.
+ # find_stocks routes strictly by FB* prefix, so anchor a FindStocks
+ # query on each feature id — the sf gate above never fires for the
+ # VFBexp_/VFB_ ids these terms carry. Anchoring on the feature (not the
+ # VFBexp_ class) is deliberate: it gives the class and its instances the
+ # same working stock query.
+ if "Expression_pattern" in (termInfo["SuperTypes"] or []):
+ ep_feature_ids = _stock_features_from_relationships(vfbTerm)
+ if not ep_feature_ids and termInfo["IsIndividual"]:
+ # Split-GAL4 image instances carry no driver edge of their own;
+ # follow the cached parent link to the pattern class's features.
+ ep_feature_ids = _stock_features_via_parent_pattern(vfbTerm)
+ multi = len(ep_feature_ids) > 1
+ for fb_id in ep_feature_ids:
+ # Disambiguate the label only when a pattern drives several
+ # features (a split), so single-feature patterns keep a clean name.
+ stock_name = f"{termInfo['Name']} ({fb_id})" if multi else termInfo["Name"]
+ q = FindStocks_to_schema(stock_name, {"short_form": fb_id})
+ queries.append(q)
+
# Bring the parent class's query menu down onto selected Individual
# instances, reproducing the legacy term-info builder
# (VFBProcessTermInfoCachedJson, gate ~line 1757): an Individual of one
@@ -1458,7 +1491,7 @@ def term_info_parse_object(results, short_form):
# own menu above so an instance shows exactly what its class shows.
inheritable_class_queries = (
(ListAllAvailableImages_to_schema, lambda p: {"Class", "Anatomy"} <= p),
- (NeuronsPartHere_to_schema, lambda p: "Class" in p and "Cell" not in p and ("Synaptic_neuropil" in p or "Anatomy" in p)),
+ (NeuronsPartHere_to_schema, lambda p: "Class" in p and "Cell" not in p and "Expression_pattern" not in p and ("Synaptic_neuropil" in p or "Anatomy" in p)),
(NeuronsSynaptic_to_schema, lambda p: "Class" in p and "Cell" not in p and "Nervous_system" in p),
(NeuronsPresynapticHere_to_schema, lambda p: "Class" in p and "Cell" not in p and "Nervous_system" in p),
(NeuronsPostsynapticHere_to_schema, lambda p: "Class" in p and "Cell" not in p and "Nervous_system" in p),
@@ -1475,7 +1508,7 @@ def term_info_parse_object(results, short_form):
(TargetNeurons_to_schema, lambda p: {"Class", "Split"} <= p),
(DownstreamClassConnectivity_to_schema, lambda p: {"Class", "Neuron"} <= p),
(UpstreamClassConnectivity_to_schema, lambda p: {"Class", "Neuron"} <= p),
- (PartsOf_to_schema, lambda p: {"Class", "Anatomy"} <= p and "Cell" not in p),
+ (PartsOf_to_schema, lambda p: {"Class", "Anatomy"} <= p and "Cell" not in p and "Expression_pattern" not in p),
(SubclassesOf_to_schema, lambda p: {"Class", "has_subClass"} <= p),
)
for schema_fn, predicate in inheritable_class_queries:
@@ -2453,6 +2486,96 @@ def FindComboPublications_to_schema(name, take_default):
)
+# term_info SOLR loaders: fetch one term's term_info doc by short_form and
+# return it either as a deserialized object (attribute access) or as the raw
+# JSON dict, whichever the caller works with.
+def _load_term_info(short_form):
+ """Deserialize a term's SOLR term_info doc, or None if unavailable."""
+ try:
+ results = vfb_solr.search(q=f'id:{short_form}', fl='term_info', rows=1)
+ if results.docs and 'term_info' in results.docs[0]:
+ return deserialize_term_info(results.docs[0]['term_info'][0])
+ except Exception as e:
+ print(f"Warning: could not load term_info for {short_form}: {e}")
+ return None
+
+
+def _load_term_info_dict(short_form):
+ """Return a term's raw term_info dict from SOLR, or None if unavailable."""
+ try:
+ results = vfb_solr.search(q=f'id:{short_form}', fl='term_info', rows=1)
+ if results.docs and 'term_info' in results.docs[0]:
+ raw = results.docs[0]['term_info']
+ return json.loads(raw[0] if isinstance(raw, list) else raw)
+ except Exception as e:
+ print(f"Warning: could not load term_info for {short_form}: {e}")
+ return None
+
+
+# Helpers for the expression-pattern stock gate in term_info_parse_object.
+# Prefixes flybase_stocks.find_stocks can route a stock query on.
+_STOCK_FEATURE_PREFIXES = ("FBgn", "FBal", "FBti", "FBtp", "FBco", "FBst")
+# Relations that connect an expression pattern to its FlyBase driver
+# feature(s): `expresses` (RO_0002292) on ordinary patterns and their image
+# instances, and `has_hemidriver` (VFBext_0000008) on split/intersectional
+# patterns, which carry no single `expresses` edge. All targets in the graph
+# are FBtp/FBti/FBal driver features (no bare genes), so every match is a
+# stock-routable feature.
+_EXPRESSION_FEATURE_RELATION_IDS = {"RO_0002292", "VFBext_0000008"}
+_EXPRESSION_FEATURE_RELATION_LABELS = {"expresses", "has hemidriver", "has_hemidriver"}
+
+
+def _stock_features_from_relationships(vfbTerm):
+ """Stock-routable FlyBase feature IDs an expression pattern expresses.
+
+ Navigates the loaded term_info relationships (graph edges) rather than
+ parsing the ``VFBexp_`` short_form: ``expresses`` gives the driver of
+ an ordinary pattern (and of its image instances); ``has_hemidriver`` gives
+ both halves of a split/intersectional pattern. Returns the FBgn/FBal/FBti/
+ FBtp/FBco/FBst targets in edge order, de-duplicated.
+ """
+ feature_ids = []
+ for rel in (getattr(vfbTerm, "relationships", None) or []):
+ relation = getattr(rel, "relation", None)
+ obj = getattr(rel, "object", None)
+ if relation is None or obj is None:
+ continue
+ rel_id = getattr(relation, "short_form", None)
+ if not rel_id and getattr(relation, "iri", None):
+ rel_id = relation.iri.split("/")[-1]
+ if (rel_id not in _EXPRESSION_FEATURE_RELATION_IDS
+ and getattr(relation, "label", None) not in _EXPRESSION_FEATURE_RELATION_LABELS):
+ continue
+ obj_sf = getattr(obj, "short_form", None)
+ if obj_sf and obj_sf.startswith(_STOCK_FEATURE_PREFIXES) and obj_sf not in feature_ids:
+ feature_ids.append(obj_sf)
+ return feature_ids
+
+
+def _stock_features_via_parent_pattern(vfbTerm):
+ """Feature IDs for an expression-pattern instance with no driver edge.
+
+ Split-GAL4 image instances carry no ``expresses``/``has_hemidriver`` edge
+ of their own, but their term_info already names the expression-pattern
+ class(es) they instantiate among ``parents`` (a cached link, no query).
+ Follow that link to each class's term_info and read its driver edges with
+ the same relationship walker — no short_form parsing, and the class page is
+ typically already cached.
+ """
+ feature_ids = []
+ for parent in (getattr(vfbTerm, "parents", None) or []):
+ if "Expression_pattern" not in (getattr(parent, "types", None) or []):
+ continue
+ parent_sf = getattr(parent, "short_form", None)
+ parent_term = _load_term_info(parent_sf) if parent_sf else None
+ if parent_term is None:
+ continue
+ for fb_id in _stock_features_from_relationships(parent_term):
+ if fb_id not in feature_ids:
+ feature_ids.append(fb_id)
+ return feature_ids
+
+
def serialize_solr_output(results):
# Create a copy of the document and remove Solr-specific fields
doc = dict(results.docs[0])
@@ -4960,23 +5083,12 @@ def _short_form_to_iri(short_form: str) -> str:
return f"http://purl.obolibrary.org/obo/{short_form}"
# For other cases, query SOLR to get the IRI from term_info
- try:
- results = vfb_solr.search(
- q=f'id:{short_form}',
- fl='term_info',
- rows=1
- )
-
- if results.docs and 'term_info' in results.docs[0]:
- term_info_str = results.docs[0]['term_info'][0]
- term_info = json.loads(term_info_str)
- iri = term_info.get('term', {}).get('core', {}).get('iri')
- if iri:
- return iri
- except Exception as e:
- # If SOLR query fails, fall back to OBO default
- print(f"Warning: Could not fetch IRI for {short_form} from SOLR: {e}")
-
+ term_info = _load_term_info_dict(short_form)
+ if term_info:
+ iri = term_info.get('term', {}).get('core', {}).get('iri')
+ if iri:
+ return iri
+
# Default to OBO for other IDs (FBbi_, etc.)
return f"http://purl.obolibrary.org/obo/{short_form}"
@@ -7191,11 +7303,9 @@ def _get_all_children(term_id):
def _term_info_parents(term_id):
"""Return [(parent_sf, parent_label), ...] from SOLR term_info."""
try:
- results = vfb_solr.search(f'id:{term_id}', fl='term_info', rows=1)
- if not results.docs or 'term_info' not in results.docs[0]:
+ ti = _load_term_info_dict(term_id)
+ if ti is None:
return []
- raw = results.docs[0]['term_info']
- ti = json.loads(raw[0] if isinstance(raw, list) else raw)
if relationship == 'subclass_of':
return [(p['short_form'], p.get('label', p['short_form'])) for p in ti.get('parents', [])]
else:
@@ -7381,11 +7491,9 @@ def _build_ancestors_subclass(term_id, depth, visited):
visited.add(term_id)
try:
- results = vfb_solr.search(f'id:{term_id}', fl='term_info', rows=1)
- if not results.docs or 'term_info' not in results.docs[0]:
+ ti = _load_term_info_dict(term_id)
+ if ti is None:
return []
- raw = results.docs[0]['term_info']
- ti = json.loads(raw[0] if isinstance(raw, list) else raw)
parents = ti.get('parents', [])
except Exception:
return []