Source code for storx.hierarchy.construct_hierarchy_from_file_m

"""Contains functionality to construct a hierarchy class from a file."""
from pathlib import Path
from .hierarchy_m import Hierarchy
from .nested_dict_tools_m import convert_list_to_nested_dict, \
                                 get_nested_dict_value, set_nested_dict_value
from .. import library_dir

code_name = None

def get_hierarchy_path():
    """Return the path of the currently active hierarchy file."""
    return library_dir / "hierarchy" / f"hierarchy_data_{code_name}.yml"



def set_code_name(name: str):
    """Set the code name used in choosing which hierarchy file to use."""
    global code_name
    code_name = name


def check_code_name():
    """Check the code name for validity."""
    if code_name == None:
        raise RuntimeError(r"The code name was not initialized. Please call " \
            "set_code_name to initialize the correct code name.")
    if not get_hierarchy_path().exists():
        raise RuntimeError(f"Invalid code name {code_name}. Corresponding " \
            "hierarchy not found.")



def get_active_hierarchy():
    """Return the hierarchy that is active, i.e. specified by the code name."""
    check_code_name()
    path = get_hierarchy_path()
    return Hierarchy.initialize_from_yaml(path)




def get_modifier_hierarchy():
    """Return the hierarchy modifiers."""
    path = library_dir / "hierarchy" / "hierarchy_modifiers.yml"
    return Hierarchy.initialize_from_yaml(path)


def get_description_for_storage_entry(string: str, h_active: Hierarchy, \
                                      h_modif: Hierarchy):
    """Create a description from the string of an entry in the storage."""
    # Split-off last part of the string and get its description
    keys = string.split("/")
    path = h_active.search_for_key(keys[-1])
    if not path:
        raise RuntimeError(f"Found key {keys[-1]} in storage which is not " \
                           "registered in current active hierarchy. Please " \
                           "check if a different storx version was used and " \
                           "update the hierarchy files present.")
    desc = get_nested_dict_value(h_active, path[0])

    # Iterate through the remaining parts and append their description
    # NOTE: These are modifiers, thus name and value are given in the string
    #       separated by a dash.
    for entry in keys[:-1]:
        path = h_modif.search_for_key(entry.split("-")[0])
        if len(path) > 0:
            desc += ", " + get_nested_dict_value(h_modif, path[0])
            desc += "=" + entry.split("-")[1]
    return desc



def create_hierarchy_from_storage(path):
    """Create a hierarchy object from a storage located at the given path."""
    ha = get_active_hierarchy()
    hm = get_modifier_hierarchy()

    path = Path(path)
    list_of_found_paths = list(path.glob("**/*.zarr"))

    # Return empty object if no path was found
    if(list_of_found_paths == []):
        return Hierarchy({})

    # Get the index of the top level of the storage, as the found paths contain
    # relative paths, the top level might be located within multiple nested
    # directory levels
    i_top_level = [i for i, s in enumerate(list_of_found_paths[0].parts) \
                   if "level" in s][0]
    # Separate the leading directory path (header) and remove it from all paths
    # found
    header = "/".join(list_of_found_paths[0].parts[:i_top_level]) + "/"
    header = header.replace("//", "/")
    list_of_entries = [str(e).replace(".zarr", "").replace(header, "") \
                       for e in list_of_found_paths]

    # Create a dict of entries that is then converted to the Hierarchy
    entry_dict = convert_list_to_nested_dict(list_of_entries)
    for entry in list_of_entries:
        desc = get_description_for_storage_entry(entry, ha, hm)
        set_nested_dict_value(entry_dict, entry, desc)

    return Hierarchy(entry_dict)