generator

This module contains a list of examples that can be used when learning to use Merlin, or for setting up new workflows.

Examples are packaged in directories, with the directory name denoting the example name. This must match the name of the Merlin specification inside.

gather_all_examples()

Get all the example YAML files.

Returns:

Type	Description
List[str]	A list of file paths to all YAML files in the example directories.

Source code in merlin/examples/generator.py

def gather_all_examples() -> List[str]:
    """
    Get all the example YAML files.

    Returns:
        A list of file paths to all YAML files in the example directories.
    """
    path = os.path.join(os.path.join(EXAMPLES_DIR, ""), os.path.join("*", "*.yaml"))
    return glob.glob(path)

gather_example_dirs()

Get all the example directories.

Returns:

Type	Description
Dict[str, str]	A dictionary where the keys and values are the names of example directories.

Source code in merlin/examples/generator.py

def gather_example_dirs() -> Dict[str, str]:
    """
    Get all the example directories.

    Returns:
        A dictionary where the keys and values are the names of example directories.
    """
    result = {}
    for directory in sorted(os.listdir(EXAMPLES_DIR)):
        result[directory] = directory
    return result

list_examples()

List all available examples with their descriptions.

Returns:

Type	Description
str	A formatted string table of example names and their descriptions.

Source code in merlin/examples/generator.py

def list_examples() -> str:
    """
    List all available examples with their descriptions.

    Returns:
        A formatted string table of example names and their descriptions.
    """
    headers = ["name", "description"]
    rows = []
    for example_dir in gather_example_dirs():
        directory = os.path.join(os.path.join(EXAMPLES_DIR, example_dir), "")
        specs = glob.glob(directory + "*.yaml")
        for spec in sorted(specs):
            if "template" in spec:
                continue
            with open(spec) as f:  # pylint: disable=C0103
                try:
                    spec_metadata = yaml.safe_load(f)["description"]
                except KeyError:
                    LOG.warning(f"{spec} lacks required section 'description'")
                    continue
                except TypeError:
                    continue
            name = spec_metadata["name"]
            if name is None:
                continue
            # if there is a variable reference in the workflow name, instead list the filename (minus the yaml extension).
            if "$" in name:
                name = os.path.basename(os.path.normpath(spec)).replace(".yaml", "")
            rows.append([name, spec_metadata["description"]])
    return "\n" + tabulate.tabulate(rows, headers) + "\n"

setup_example(name, outdir)

Set up the given example by copying it to the specified output directory.

Parameters:

Name	Type	Description	Default
name	str	The name of the example to set up.	required
outdir	str	The output directory where the example will be copied.	required

Returns:

Type	Description
Union[str, None]	The name of the example if successful, or None if the example was not found or an error occurred.

Source code in merlin/examples/generator.py

def setup_example(name: str, outdir: str) -> Union[str, None]:
    """
    Set up the given example by copying it to the specified output directory.

    Args:
        name: The name of the example to set up.
        outdir: The output directory where the example will be copied.

    Returns:
        The name of the example if successful, or None if the example was not found or an error occurred.
    """
    example = None
    spec_paths = gather_all_examples()
    spec_path = None
    for spec_path in spec_paths:
        spec = os.path.basename(os.path.normpath(spec_path)).replace(".yaml", "")
        if name == spec:
            example = os.path.basename(os.path.dirname(spec_path))
            break
    if example is None:
        LOG.error(f"Example '{name}' not found.")
        return None

    # if there is only 1 file in the example, don't bother making a directory for it
    if len(os.listdir(os.path.dirname(spec_path))) == 1:
        src_path = os.path.join(EXAMPLES_DIR, os.path.join(example, example + ".yaml"))

    else:
        src_path = os.path.join(EXAMPLES_DIR, example)

        if outdir:
            outdir = os.path.join(os.getcwd(), outdir)
        else:
            outdir = os.path.join(os.getcwd(), example)

        if os.path.exists(outdir):
            LOG.error(f"File '{outdir}' already exists!")
            return None

    if outdir is None:
        outdir = os.getcwd()

    LOG.info(f"Copying example '{name}' to {outdir}")
    write_example(src_path, outdir)
    return example

write_example(src_path, dst_path)

Write out the example workflow to a file or directory.

Parameters:

Name	Type	Description	Default
src_path	str	The path to copy the example from.	required
dst_path	str	The destination path to copy the example to.	required

Source code in merlin/examples/generator.py

def write_example(src_path: str, dst_path: str):
    """
    Write out the example workflow to a file or directory.

    Args:
        src_path: The path to copy the example from.
        dst_path: The destination path to copy the example to.
    """
    if os.path.isdir(src_path):
        shutil.copytree(src_path, dst_path)
    else:
        shutil.copy(src_path, dst_path)