hrm_omero.tree

  1"""Functions related to OMERO's tree view."""
  2
  3from loguru import logger as log
  4
  5from .decorators import connect_and_set_group
  6
  7
  8def gen_obj_dict(obj, id_pfx=""):
  9    """Create a dict from an OMERO object.
 10
 11    Parameters
 12    ----------
 13    obj : omero.gateway._*Wrapper
 14        The OMERO object to process.
 15    id_pfx : str, optional
 16        A string prefix that will be added to the `id` value, by default ''.
 17
 18    Returns
 19    -------
 20    dict
 21        A dictionary with the following structure:
 22        ```
 23            {
 24                'children': [],
 25                'id': 'Project:1154',
 26                'label': 'HRM_TESTDATA',
 27                'owner': u'demo01',
 28                'class': 'Project'
 29            }
 30        ```
 31    """
 32    obj_dict = {}
 33    obj_dict["label"] = obj.getName()
 34    obj_dict["class"] = obj.OMERO_CLASS
 35    if obj.OMERO_CLASS == "Experimenter":
 36        obj_dict["owner"] = obj.getId()
 37        obj_dict["label"] = obj.getFullName()
 38    elif obj.OMERO_CLASS == "ExperimenterGroup":
 39        # for some reason getOwner() et al. return nothing on a group, so we
 40        # simply put it to None for group objects:
 41        obj_dict["owner"] = None
 42    else:
 43        obj_dict["owner"] = obj.getOwnerOmeName()
 44    obj_dict["id"] = id_pfx + f"{obj.OMERO_CLASS}:{obj.getId()}"
 45    obj_dict["children"] = []
 46    return obj_dict
 47
 48
 49@connect_and_set_group
 50def gen_children(conn, omero_id):
 51    """Get the children for a given node.
 52
 53    Parameters
 54    ----------
 55    conn : omero.gateway.BlitzGateway
 56        The OMERO connection object.
 57    omero_id : hrm_omero.misc.OmeroId
 58        An object denoting an OMERO target.
 59
 60    Returns
 61    -------
 62    list
 63        A list with children nodes (of type `dict`), having the `load_on_demand`
 64        property set to `True` required by the jqTree JavaScript library (except for
 65        nodes of type `Dataset` as they are the last / lowest level).
 66    """
 67    if omero_id.obj_type == "BaseTree":
 68        return gen_base_tree(conn)
 69
 70    log.debug(f"generating children for [{omero_id}]")
 71
 72    # conn.SERVICE_OPTS.setOmeroGroup(gid)
 73    obj = conn.getObject(omero_id.obj_type, omero_id.obj_id)
 74    # we need different child-wrappers, depending on the object type:
 75    if omero_id.obj_type == "Experimenter":
 76        children_wrapper = []
 77        for proj in conn.listProjects(omero_id.obj_id):
 78            children_wrapper.append(proj)
 79        # OMERO.web is showing "orphaned" datasets (i.e. that do NOT belong to a
 80        # certain project) at the top level, next to the projects - so we are going to
 81        # add them to the tree at the same hierarchy level:
 82        for dataset in conn.listOrphans("Dataset", eid=omero_id.obj_id):
 83            children_wrapper.append(dataset)
 84
 85    elif omero_id.obj_type == "ExperimenterGroup":
 86        log.warning(
 87            f"{__name__} has been called with omero_id='{str(omero_id)}', but "
 88            "'ExperimenterGroup' trees should be generated via `gen_group_tree()`!",
 89        )
 90        return []
 91
 92    else:
 93        children_wrapper = obj.listChildren()
 94
 95    # now process children:
 96    children = []
 97    for child in children_wrapper:
 98        children.append(gen_obj_dict(child, "G:" + omero_id.group + ":"))
 99    children = sorted(children, key=lambda d: d["label"].lower())
100
101    # set the on-demand flag unless the children are the last level:
102    if not omero_id.obj_type == "Dataset":
103        for child in children:
104            child["load_on_demand"] = True
105
106    return children
107
108
109def gen_base_tree(conn):
110    """Generate all group trees with their members as the basic tree.
111
112    Parameters
113    ----------
114    conn : omero.gateway.BlitzGateway
115        The OMERO connection object.
116
117    Returns
118    -------
119    list
120        A list of grouptree dicts as generated by `gen_group_tree()`.
121    """
122    log.debug("Generating base tree...")
123    tree = []
124    for group in conn.getGroupsMemberOf():
125        tree.append(gen_group_tree(conn, group))
126    tree_sorted = sorted(tree, key=lambda d: d["label"].lower())
127    return tree_sorted
128
129
130def gen_group_tree(conn, group=None):
131    """Create the tree nodes for a group and its members.
132
133    Parameters
134    ----------
135    conn : omero.gateway.BlitzGateway
136        The OMERO connection object.
137    group : int or str or omero.gateway._ExperimenterGroupWrapper, optional
138        The group object (or the group ID as int or str) to generate the tree for, by
139        default `None` which will result in the group being derived from the current
140        connection's context.
141
142    Returns
143    -------
144    dict
145        A nested dict of the given group (or the default group if not specified
146        explicitly) and its members as a list of dicts in the `children` item, starting
147        with the current user as the first entry.
148    """
149    if group is None:
150        log.debug("Getting group from current context...")
151        group = conn.getGroupFromContext()
152
153    if isinstance(group, (int, str)):
154        target_gid = int(group)
155        group = None
156        for candidate in conn.getGroupsMemberOf():
157            if int(candidate.getId()) == target_gid:
158                log.debug(f"Found group object for ID {target_gid}!")
159                group = candidate
160                break
161        if group is None:
162            msg = f"Unable to identify group with ID {target_gid}!"
163            log.error(msg)
164            raise RuntimeError(msg)
165
166    gid = str(group.getId())
167    log.debug(f"Generating tree for group {gid}...")
168    conn.setGroupForSession(gid)
169
170    group_dict = gen_obj_dict(group)
171    # add the user's own tree first:
172    user = conn.getUser()
173    user_dict = gen_obj_dict(user, "G:" + gid + ":")
174    user_dict["load_on_demand"] = True
175    group_dict["children"].append(user_dict)
176    all_user_dicts = []
177    # then add the trees for other group members
178    for user in conn.listColleagues():
179        user_dict = gen_obj_dict(user, "G:" + gid + ":")
180        user_dict["load_on_demand"] = True
181        all_user_dicts.append(user_dict)
182
183    group_dict["children"] += sorted(all_user_dicts, key=lambda d: d["label"].lower())
184    return group_dict