Merge pull request #244 from caracal-pipeline/userutil

Userutil

scabha/basetypes.py

@@ -47,7 +47,6 @@ class SkippedOutput(Unresolved):
     def __str__(self):
         return f"Skipped({self.value})"
 
-
 class URI(str):
     def __init__(self, value):
         self.protocol, self.path, self.remote = URI.parse(value)
@@ -76,6 +75,10 @@ class File(URI):
     @property
     def NAME(self):
         return File(os.path.basename(self))
+
+    @property
+    def PATH(self):
+        return File(os.path.abspath(self))
 
     @property
     def DIR(self):
@@ -92,6 +95,10 @@ def BASENAME(self):
     @property
     def EXT(self):
         return os.path.splitext(self)[1]
+
+    @property
+    def EXISTS(self):
+        return os.path.exists(self)
 
 class Directory(File):
     pass
@@ -107,6 +114,3 @@ def is_file_type(dtype):
 def is_file_list_type(dtype):
     return any(dtype == List[t] for t in FILE_TYPES)
 
-
-
-

scabha/schema_utils.py

@@ -1,17 +1,20 @@
+import os
 import re
 import click
 from scabha.exceptions import SchemaError
 from .cargo import Parameter, UNSET, _UNSET_DEFAULT
-from .basetypes import EmptyDictDefault, is_file_type
-from typing import *
+from typing import List, Union, Optional, Callable, Dict, DefaultDict, Any
+from .basetypes import EmptyDictDefault, File, is_file_type
 from dataclasses import dataclass, make_dataclass, field
 from omegaconf import OmegaConf, MISSING
 from collections.abc import MutableSet, MutableSequence, MutableMapping
+from scabha import configuratt
 
 def schema_to_dataclass(io: Dict[str, Parameter], class_name: str, bases=(), post_init: Optional[Callable] =None):
-    """Converts a scabha schema to a dataclass.
-       Each parameter in the schema will correspond to a field. Metadata of fields will contain:
-        'help' for info, 'choices' for the choices field, 'parameter' for parameter name
+    """
+    Converts a scabha schema to a dataclass.
+    Each parameter in the schema will correspond to a field. Metadata of fields will contain:
+    'help' for info, 'choices' for the choices field, 'parameter' for parameter name
 
     Args:
         io (Dict[str, Parameter]): dict of parameters
@@ -143,7 +146,7 @@ def clickify_parameters(schemas: Union[str, Dict[str, Any]]):
 
     if type(schemas) is str:
         schemas = OmegaConf.merge(OmegaConf.structured(Schema),
-                               OmegaConf.load(schemas))
+                                OmegaConf.load(schemas))
 
     decorator_chain = None
     for io in schemas.inputs, schemas.outputs:
@@ -210,3 +213,39 @@ def clickify_parameters(schemas: Union[str, Dict[str, Any]]):
                 decorator_chain = lambda x,deco=deco,chain=decorator_chain: chain(deco(x))
 
     return decorator_chain
+
+@dataclass
+class SchemaSpec:
+    inputs: Dict[str, Parameter]
+    outputs: Dict[str, Parameter]
+    libs: Dict[str, Any]
+
+def paramfile_loader(paramfiles: File|List[File], sources: File|List[File] = [], schema_spec=None, use_cache=False) -> Dict:
+    """Load a scabha-style parameter defintion using.
+
+    Args:
+        paramfiles (List[File]): Name of parameter definition files
+        sources (List[Dict], optional): Parameter definition dependencies 
+        (a.k.a files specified via_include)
+
+    Returns:
+        Dict: Schema object
+    """
+    args_defn = OmegaConf.structured(schema_spec or SchemaSpec)
+    if isinstance(paramfiles, File):
+        paramfiles = [paramfiles]
+
+    if isinstance(sources, File):
+        sources = [sources]
+
+    srcs = []
+    for src in sources:
+        if not src.EXISTS:
+            raise FileNotFoundError(f"Source file for either of {paramfiles} could not be found at {src.PATH}")
+        srcs.append(configuratt.load(src, use_cache=use_cache)[0])
+
+    struct_args, _ = configuratt.load_nested(paramfiles, structured=args_defn,
+                                            use_sources=srcs, use_cache=use_cache)
+
+    return OmegaConf.create(struct_args)
+