Connectivity: get paths from NPO instead of path models and allowing …

…tracing of NPO paths.

docs/Manifest.rst

@@ -66,14 +66,14 @@ Neuron connectivity is specified as a list consisting of a mix of:
 
 *   URLs referencing ApiNATOMY models, optained from the SCKAN endpoint.
 *   JSON dictionaries specifing individual neuron paths in an ApiNATOMY model from SCKAN.
-*   the keyword ``"NPO"`` to include all connectivity models from the NPO endpoint.
+*   the keyword ``"NPO"`` to include all connectivity paths from the NPO endpoint.
 
 When a JSON dictionary is used to define neuron connectivity:
 
-*   it MUST specify the ``"uri"`` of an ApiNATOMY model.
+*   it MUST specify either the ``"uri"`` of an ApiNATOMY model or the keyword ``"NPO"``.
 *   it MAY specify a ``"filter"``, as a JSON dictionary containing three optional lists:
 
-    -   ``"exclude"`` giving neuron path URIs from the model to exclude from the map.
+    -   ``"exclude"`` giving neuron path URIs, from the ApiNATOMY model or NPO, to exclude from the map.
     -   ``"include"`` giving neuron path URIs to render on the map.
     -   ``"trace"`` giving neuron path URIs to render and also log information about
         their routing when making the map.

mapmaker/knowledgebase/__init__.py

@@ -64,6 +64,9 @@ def get_label(entity: str) -> str:
 def get_knowledge(entity: str) -> dict[str, Any]:
     return settings['KNOWLEDGE_STORE'].entity_knowledge(entity)
 
+def npo_connectivity_paths() -> dict[str, dict[str, str]]:
+    return settings['KNOWLEDGE_STORE'].npo.connectivity_paths()
+
 def sckan_build() -> Optional[dict]:
     if (scicrunch := settings['KNOWLEDGE_STORE'].scicrunch) is not None:
         return scicrunch.build()

mapmaker/knowledgebase/sckan.py

@@ -236,19 +236,20 @@ def __init__(self, flatmap):
         self.__paths_by_id = {}
         if settings.get('ignoreSckan', False):
             return
-        connectivity_models = kb.connectivity_models('APINATOMY')
-        connectivity_models.update(kb.connectivity_models('NPO'))
-        for model in connectivity_models:
+        connectivity_paths = set()
+        for model in kb.connectivity_models('APINATOMY'):
             model_knowledege = kb.get_knowledge(model)
-            for path in model_knowledege.get('paths', []):
-                path_knowledge = kb.get_knowledge(path['id'])
-                self.__paths_by_id[path['id']] = path_knowledge
-                G = connectivity_graph_from_knowledge(path_knowledge)
-                if G:
-                    for node in G.nodes:
-                        G.nodes[node]['node-features'] = flatmap.features_for_anatomical_node(node, warn=False)
-                    self.__trim_non_existent_features(G)
-                    self.__sckan_path_nodes_by_type[G.graph['path-type']][path['id']] = SckanNodeSet(G)
+            connectivity_paths.update([path['id'] for path in model_knowledege.get('paths', [])])
+        connectivity_paths.update(kb.npo_connectivity_paths().keys())
+        for path_id in connectivity_paths:
+            path_knowledge = kb.get_knowledge(path_id)
+            self.__paths_by_id[path_id] = path_knowledge
+            G = connectivity_graph_from_knowledge(path_knowledge)
+            if G:
+                for node in G.nodes:
+                    G.nodes[node]['node-features'] = flatmap.features_for_anatomical_node(node, warn=False)
+                self.__trim_non_existent_features(G)
+                self.__sckan_path_nodes_by_type[G.graph['path-type']][path_id] = SckanNodeSet(G)
 
     def __trim_non_existent_features(self, G):
     #=========================================

mapmaker/properties/__init__.py

@@ -67,20 +67,15 @@ def __init__(self, flatmap: "FlatMap", manifest: "Manifest"):
             connectivity = FilePath(connectivity_source).get_json()
             self.__pathways.add_connectivity(connectivity)
 
-        # NPO connectivity models
-        if 'NPO' in manifest.neuron_connectivity:
-            settings['NPO'] = True
-            for connectivity_model in knowledgebase.connectivity_models('NPO').keys():
-                self.__pathways.add_connectivity_model(connectivity_model, self)
-
         # ApiNATOMY connectivity models from SciCrunch
         apinatomy_models = knowledgebase.connectivity_models('APINATOMY')
-        for connectivity_model in manifest.neuron_connectivity:
+        seen_npo = False
+        for connectivity_source in manifest.neuron_connectivity:
             path_filter = None
             traced_paths = None
-            if isinstance(connectivity_model, dict):
-                model_uri = connectivity_model['uri']
-                if (filter_lists := connectivity_model.get('filter')) is not None:
+            if isinstance(connectivity_source, dict):
+                model_source = connectivity_source['uri']
+                if (filter_lists := connectivity_source.get('filter')) is not None:
                     include_ids = filter_lists['include'] if 'include' in filter_lists else None
                     exclude_ids = filter_lists['exclude'] if 'exclude' in filter_lists else None
                     if 'trace' in filter_lists:
@@ -89,12 +84,20 @@ def __init__(self, flatmap: "FlatMap", manifest: "Manifest"):
                     path_filter = lambda path_id: ((include_ids is None or include_ids is not None and path_id in include_ids)
                                                and (exclude_ids is None or exclude_ids is not None and path_id not in exclude_ids))
             else:
-                model_uri = connectivity_model
-            if model_uri in apinatomy_models:
-                self.__pathways.add_connectivity_model(model_uri, self, path_filter=path_filter, traced_paths=traced_paths)
-            elif model_uri != 'NPO':
-                log.warning(f'Connectivity for {model_uri} not available in SCKAN')
-
+                model_source = connectivity_source
+            if model_source in apinatomy_models:
+                self.__pathways.add_connectivity_model(model_source, self,
+                    path_filter=path_filter, traced_paths=traced_paths)
+            elif model_source == 'NPO':
+                # NPO connectivity paths
+                if seen_npo:
+                    log.warning(f'`NPO` can only be specified once as a connectivity source')
+                else:
+                    seen_npo = True
+                    settings['NPO'] = True
+                    for connectivity_path in knowledgebase.npo_connectivity_paths().keys():
+                        self.__pathways.add_connectivity_path(connectivity_path,
+                            self, path_filter=path_filter, traced_paths=traced_paths)
         # Load network centreline definitions
         self.__networks = { network.get('id'): Network(flatmap, network, self)
                                 for network in properties_dict.get('networks', []) }