🚸 Include resource soure information in resource-not-found errors

Author: shnizzedy

CHANGELOG.md

@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Moved `pygraphviz` from requirements to `graphviz` optional dependencies group.
 - Automatically tag untagged `subject_id` and `unique_id` as `!!str` when loading data config files.
 - Made orientation configurable (was hard-coded as "RPI").
+- Resource-not-found errors now include information about where to source those resources.
 
 ### Fixed
 

CPAC/pipeline/engine.py

@@ -420,10 +420,12 @@ def get(
             if report_fetched:
                 return (None, None)
             return None
+        from CPAC.pipeline.resource_inventory import where_to_find
+
         msg = (
             "\n\n[!] C-PAC says: None of the listed resources are in "
-            f"the resource pool:\n\n  {resource}\n\nOptions:\n- You "
-            "can enable a node block earlier in the pipeline which "
+            f"the resource pool:\n\n  {where_to_find(resource)}\n\nOptions:\n"
+            "- You can enable a node block earlier in the pipeline which "
             "produces these resources. Check the 'outputs:' field in "
             "a node block's documentation.\n- You can directly "
             "provide this required data by pulling it from another "
@@ -458,7 +460,9 @@ def copy_resource(self, resource, new_name):
         try:
             self.rpool[new_name] = self.rpool[resource]
         except KeyError:
-            msg = f"[!] {resource} not in the resource pool."
+            from CPAC.pipeline.resource_inventory import where_to_find
+
+            msg = f"[!] Not in the resource pool:\n{where_to_find(resource)}"
             raise Exception(msg)
 
     def update_resource(self, resource, new_name):
@@ -630,11 +634,13 @@ def get_strats(self, resources, debug=False):
             total_pool.append(sub_pool)
 
         if not total_pool:
+            from CPAC.pipeline.resource_inventory import where_to_find
+
             raise LookupError(
                 "\n\n[!] C-PAC says: None of the listed "
                 "resources in the node block being connected "
                 "exist in the resource pool.\n\nResources:\n"
-                "%s\n\n" % resource_list
+                "%s\n\n" % where_to_find(resource_list)
             )
 
         # TODO: right now total_pool is:

CPAC/pipeline/resource_inventory.py

@@ -214,7 +214,7 @@ def cli_parser() -> Namespace:
     return parser.parse_args()
 
 
-def _flatten_io(io: list[Iterable]) -> list[str]:
+def _flatten_io(io: Iterable[Iterable]) -> list[str]:
     """Given a list of strings or iterables thereof, flatten the list to all strings."""
     if all(isinstance(resource, str) for resource in io):
         return cast(list[str], io)
@@ -243,7 +243,7 @@ def find_directly_set_resources(package_name: str) -> dict[str, list[str]]:
     dict
         A dictionary containing the name of the resource and the name of the functions that set it.
     """
-    resources: dict[str[list[str]]] = {}
+    resources: dict[str, list[str]] = {}
     for dirpath, _, filenames in os.walk(str(files(package_name))):
         for filename in filenames:
             if filename.endswith(".py"):
@@ -280,7 +280,7 @@ def resource_inventory(package: str = "CPAC") -> dict[str, ResourceIO]:
             "CPAC.unet._torch",
         ],
     ):
-        nbf_name = f"{nbf.__module__}.{nbf.__qualname__}"
+        nbf_name = f"{nbf.name} ({nbf.__module__}.{nbf.__qualname__})"
         if hasattr(nbf, "inputs"):
             for nbf_input in _flatten_io(cast(list[Iterable], nbf.inputs)):
                 if nbf_input:
@@ -331,6 +331,24 @@ def dump_inventory_to_yaml(inventory: dict[str, ResourceIO]) -> str:
     )
 
 
+def where_to_find(resources: list[str] | str) -> str:
+    """Return a multiline string describing where each listed resource is output from."""
+    if isinstance(resources, str):
+        resources = [resources]
+    resources = _flatten_io(resources)
+    inventory = resource_inventory("CPAC")
+    output = ""
+    for resource in resources:
+        output += f"'{resource}' is output from:\n"
+        if resource in inventory:
+            for source in inventory[resource].output_from:
+                output += f"  {source}\n"
+        else:
+            output += "  !! Nowhere !!\n"
+        output += "\n"
+    return output.rstrip()
+
+
 def main() -> None:
     """Save the NodeBlock inventory to a file."""
     args = cli_parser()