hrm_omero.tree

Functions related to OMERO's tree view.

View Source
"""Functions related to OMERO's tree view."""

from loguru import logger as log

from .decorators import connect_and_set_group
from .misc import OmeroId


def gen_obj_dict(obj, id_pfx="", conn=None):
    """Create a dict from an OMERO object.

    Parameters
    ----------
    obj : omero.gateway._*Wrapper
        The OMERO object to process.
    id_pfx : str, optional
        A string prefix that will be added to the `id` value, by default ''.

    Returns
    -------
    dict
        A dictionary with the following structure:
        ```
            {
                'children': [],
                'id': 'Project:1154',
                'label': 'HRM_TESTDATA',
                'owner': u'demo01',
                'class': 'Project'
            }
        ```
    """
    obj_dict = {}
    obj_dict["label"] = obj.getName()
    obj_dict["class"] = obj.OMERO_CLASS
    if obj.OMERO_CLASS == "Experimenter":
        obj_dict["owner"] = obj.getId()
        obj_dict["label"] = obj.getFullName()
    elif obj.OMERO_CLASS == "ExperimenterGroup":
        # for some reason getOwner() et al. return nothing on a group, so we
        # simply put it to None for group objects:
        obj_dict["owner"] = None
    else:
        obj_dict["owner"] = obj.getOwnerOmeName()
    identifier = id_pfx + f"{obj.OMERO_CLASS}:{obj.getId()}"
    obj_dict["id"] = identifier
    obj_dict["children"] = []
    if conn and not obj.OMERO_CLASS == "Image":
        obj_dict["children"] = gen_children(conn, OmeroId(identifier))
    return obj_dict


@connect_and_set_group
def gen_children(conn, omero_id):
    """Get the children for a given node.

    Parameters
    ----------
    conn : omero.gateway.BlitzGateway
        The OMERO connection object.
    omero_id : hrm_omero.misc.OmeroId
        An object denoting an OMERO target.

    Returns
    -------
    list
        A list with children nodes (of type `dict`), having the `load_on_demand`
        property set to `True` required by the jqTree JavaScript library (except for
        nodes of type `Dataset` as they are the last / lowest level).
    """
    if omero_id.obj_type == "BaseTree":
        return gen_base_tree(conn)

    log.debug(f"generating children for [{omero_id}]")

    # conn.SERVICE_OPTS.setOmeroGroup(gid)
    obj = conn.getObject(omero_id.obj_type, omero_id.obj_id)
    # we need different child-wrappers, depending on the object type:
    if omero_id.obj_type == "Experimenter":
        children_wrapper = []
        for proj in conn.listProjects(omero_id.obj_id):
            children_wrapper.append(proj)
        # OMERO.web is showing "orphaned" datasets (i.e. that do NOT belong to a
        # certain project) at the top level, next to the projects - so we are going to
        # add them to the tree at the same hierarchy level:
        for dataset in conn.listOrphans("Dataset", eid=omero_id.obj_id):
            children_wrapper.append(dataset)

    elif omero_id.obj_type == "ExperimenterGroup":
        log.warning(
            f"{__name__} has been called with omero_id='{str(omero_id)}', but "
            "'ExperimenterGroup' trees should be generated via `gen_group_tree()`!",
        )
        return []

    else:
        children_wrapper = obj.listChildren()

    # now process children:
    children = []
    for child in children_wrapper:
        children.append(gen_obj_dict(child, "G:" + omero_id.group + ":", conn))
    children = sorted(children, key=lambda d: d["label"].lower())

    return children


def gen_base_tree(conn):
    """Generate all group trees with their members as the basic tree.

    Parameters
    ----------
    conn : omero.gateway.BlitzGateway
        The OMERO connection object.

    Returns
    -------
    list
        A list of grouptree dicts as generated by `gen_group_tree()`.
    """
    log.debug("Generating base tree...")
    tree = []
    for group in conn.getGroupsMemberOf():
        tree.append(gen_group_tree(conn, group))
    tree_sorted = sorted(tree, key=lambda d: d["label"].lower())
    return tree_sorted


def gen_group_tree(conn, group=None):
    """Create the tree nodes for a group and its members.

    Parameters
    ----------
    conn : omero.gateway.BlitzGateway
        The OMERO connection object.
    group : int or str or omero.gateway._ExperimenterGroupWrapper, optional
        The group object (or the group ID as int or str) to generate the tree for, by
        default `None` which will result in the group being derived from the current
        connection's context.

    Returns
    -------
    dict
        A nested dict of the given group (or the default group if not specified
        explicitly) and its members as a list of dicts in the `children` item, starting
        with the current user as the first entry.
    """
    if group is None:
        log.debug("Getting group from current context...")
        group = conn.getGroupFromContext()

    if isinstance(group, (int, str)):
        target_gid = int(group)
        group = None
        for candidate in conn.getGroupsMemberOf():
            if int(candidate.getId()) == target_gid:
                log.debug(f"Found group object for ID {target_gid}!")
                group = candidate
                break
        if group is None:
            msg = f"Unable to identify group with ID {target_gid}!"
            log.error(msg)
            raise RuntimeError(msg)

    gid = str(group.getId())
    log.debug(f"Generating tree for group {gid}...")
    conn.setGroupForSession(gid)

    group_dict = gen_obj_dict(group)
    # add the user's own tree first:
    user = conn.getUser()
    user_dict = gen_obj_dict(user, "G:" + gid + ":", conn)
    group_dict["children"].append(user_dict)
    all_user_dicts = []
    # then add the trees for other group members
    for user in conn.listColleagues():
        user_dict = gen_obj_dict(user, "G:" + gid + ":", conn)
        all_user_dicts.append(user_dict)

    group_dict["children"] += sorted(all_user_dicts, key=lambda d: d["label"].lower())
    return group_dict
#   def gen_obj_dict(obj, id_pfx='', conn=None):
View Source
def gen_obj_dict(obj, id_pfx="", conn=None):
    """Create a dict from an OMERO object.

    Parameters
    ----------
    obj : omero.gateway._*Wrapper
        The OMERO object to process.
    id_pfx : str, optional
        A string prefix that will be added to the `id` value, by default ''.

    Returns
    -------
    dict
        A dictionary with the following structure:
        ```
            {
                'children': [],
                'id': 'Project:1154',
                'label': 'HRM_TESTDATA',
                'owner': u'demo01',
                'class': 'Project'
            }
        ```
    """
    obj_dict = {}
    obj_dict["label"] = obj.getName()
    obj_dict["class"] = obj.OMERO_CLASS
    if obj.OMERO_CLASS == "Experimenter":
        obj_dict["owner"] = obj.getId()
        obj_dict["label"] = obj.getFullName()
    elif obj.OMERO_CLASS == "ExperimenterGroup":
        # for some reason getOwner() et al. return nothing on a group, so we
        # simply put it to None for group objects:
        obj_dict["owner"] = None
    else:
        obj_dict["owner"] = obj.getOwnerOmeName()
    identifier = id_pfx + f"{obj.OMERO_CLASS}:{obj.getId()}"
    obj_dict["id"] = identifier
    obj_dict["children"] = []
    if conn and not obj.OMERO_CLASS == "Image":
        obj_dict["children"] = gen_children(conn, OmeroId(identifier))
    return obj_dict

Create a dict from an OMERO object.

Parameters
  • obj (omero.gateway._*Wrapper): The OMERO object to process.
  • id_pfx (str, optional): A string prefix that will be added to the id value, by default ''.
Returns
  • dict: A dictionary with the following structure:
    {
        'children': [],
        'id': 'Project:1154',
        'label': 'HRM_TESTDATA',
        'owner': u'demo01',
        'class': 'Project'
    }
#  
@connect_and_set_group
def gen_children(conn, omero_id):
View Source
@connect_and_set_group
def gen_children(conn, omero_id):
    """Get the children for a given node.

    Parameters
    ----------
    conn : omero.gateway.BlitzGateway
        The OMERO connection object.
    omero_id : hrm_omero.misc.OmeroId
        An object denoting an OMERO target.

    Returns
    -------
    list
        A list with children nodes (of type `dict`), having the `load_on_demand`
        property set to `True` required by the jqTree JavaScript library (except for
        nodes of type `Dataset` as they are the last / lowest level).
    """
    if omero_id.obj_type == "BaseTree":
        return gen_base_tree(conn)

    log.debug(f"generating children for [{omero_id}]")

    # conn.SERVICE_OPTS.setOmeroGroup(gid)
    obj = conn.getObject(omero_id.obj_type, omero_id.obj_id)
    # we need different child-wrappers, depending on the object type:
    if omero_id.obj_type == "Experimenter":
        children_wrapper = []
        for proj in conn.listProjects(omero_id.obj_id):
            children_wrapper.append(proj)
        # OMERO.web is showing "orphaned" datasets (i.e. that do NOT belong to a
        # certain project) at the top level, next to the projects - so we are going to
        # add them to the tree at the same hierarchy level:
        for dataset in conn.listOrphans("Dataset", eid=omero_id.obj_id):
            children_wrapper.append(dataset)

    elif omero_id.obj_type == "ExperimenterGroup":
        log.warning(
            f"{__name__} has been called with omero_id='{str(omero_id)}', but "
            "'ExperimenterGroup' trees should be generated via `gen_group_tree()`!",
        )
        return []

    else:
        children_wrapper = obj.listChildren()

    # now process children:
    children = []
    for child in children_wrapper:
        children.append(gen_obj_dict(child, "G:" + omero_id.group + ":", conn))
    children = sorted(children, key=lambda d: d["label"].lower())

    return children

Get the children for a given node.

Parameters
  • conn (omero.gateway.BlitzGateway): The OMERO connection object.
  • omero_id (hrm_omero.misc.OmeroId): An object denoting an OMERO target.
Returns
  • list: A list with children nodes (of type dict), having the load_on_demand property set to True required by the jqTree JavaScript library (except for nodes of type Dataset as they are the last / lowest level).
#   def gen_base_tree(conn):
View Source
def gen_base_tree(conn):
    """Generate all group trees with their members as the basic tree.

    Parameters
    ----------
    conn : omero.gateway.BlitzGateway
        The OMERO connection object.

    Returns
    -------
    list
        A list of grouptree dicts as generated by `gen_group_tree()`.
    """
    log.debug("Generating base tree...")
    tree = []
    for group in conn.getGroupsMemberOf():
        tree.append(gen_group_tree(conn, group))
    tree_sorted = sorted(tree, key=lambda d: d["label"].lower())
    return tree_sorted

Generate all group trees with their members as the basic tree.

Parameters
  • conn (omero.gateway.BlitzGateway): The OMERO connection object.
Returns
#   def gen_group_tree(conn, group=None):
View Source
def gen_group_tree(conn, group=None):
    """Create the tree nodes for a group and its members.

    Parameters
    ----------
    conn : omero.gateway.BlitzGateway
        The OMERO connection object.
    group : int or str or omero.gateway._ExperimenterGroupWrapper, optional
        The group object (or the group ID as int or str) to generate the tree for, by
        default `None` which will result in the group being derived from the current
        connection's context.

    Returns
    -------
    dict
        A nested dict of the given group (or the default group if not specified
        explicitly) and its members as a list of dicts in the `children` item, starting
        with the current user as the first entry.
    """
    if group is None:
        log.debug("Getting group from current context...")
        group = conn.getGroupFromContext()

    if isinstance(group, (int, str)):
        target_gid = int(group)
        group = None
        for candidate in conn.getGroupsMemberOf():
            if int(candidate.getId()) == target_gid:
                log.debug(f"Found group object for ID {target_gid}!")
                group = candidate
                break
        if group is None:
            msg = f"Unable to identify group with ID {target_gid}!"
            log.error(msg)
            raise RuntimeError(msg)

    gid = str(group.getId())
    log.debug(f"Generating tree for group {gid}...")
    conn.setGroupForSession(gid)

    group_dict = gen_obj_dict(group)
    # add the user's own tree first:
    user = conn.getUser()
    user_dict = gen_obj_dict(user, "G:" + gid + ":", conn)
    group_dict["children"].append(user_dict)
    all_user_dicts = []
    # then add the trees for other group members
    for user in conn.listColleagues():
        user_dict = gen_obj_dict(user, "G:" + gid + ":", conn)
        all_user_dicts.append(user_dict)

    group_dict["children"] += sorted(all_user_dicts, key=lambda d: d["label"].lower())
    return group_dict

Create the tree nodes for a group and its members.

Parameters
  • conn (omero.gateway.BlitzGateway): The OMERO connection object.
  • group (int or str or omero.gateway._ExperimenterGroupWrapper, optional): The group object (or the group ID as int or str) to generate the tree for, by default None which will result in the group being derived from the current connection's context.
Returns
  • dict: A nested dict of the given group (or the default group if not specified explicitly) and its members as a list of dicts in the children item, starting with the current user as the first entry.