propagate stocks to EP pages

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():

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 <a cell>" 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
+        #   <a cell>" 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,17 +1158,21 @@ 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
         #   the Anatomy tag).
         # - 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_<FBid> 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_<FBid>`` 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 []