Print task info

Python script to call after a ROMI task to print a task summary.

angles_and_internodes_info(task, task_id, db_path)

Print info about AnglesAndInternodes task output.

Source code in romitask/cli/print_task_info.py

def angles_and_internodes_info(task, task_id, db_path):
    """Print info about AnglesAndInternodes task output."""
    json_f, md_json = json_metadata(task, task_id, db_path)
    for md_info in ["angles", "internodes"]:
        try:
            data = md_json[md_info]
            print("Found the following {} between successive organs:".format(md_info))
            print([round(d, 1) for d in data])
        except KeyError as e:
            print("Could not find {} entry in {}".format(e, json_f))

clustered_mesh_info(task, task_id, db_path)

Print info about ClusteredMesh task output.

Source code in romitask/cli/print_task_info.py

def clustered_mesh_info(task, task_id, db_path):
    """Print info about ClusteredMesh task output."""
    from plantdb.commons.io import read_triangle_mesh
    out_dir = os.path.join(db_path, task_id)
    files = os.listdir(out_dir)
    ply_files = [f for f in files if f.endswith('.ply')]
    organs = {}
    for ply in ply_files:
        ply_f = os.path.join(db_path, task_id, ply)
        ply_mesh = read_triangle_mesh(ply_f)
        if ply_mesh.is_empty():
            continue  # don't count empty PLY as valid organs!
        organ = ply.split('_')[0]
        if organ in organs.keys():
            organs[organ] += 1
        else:
            organs[organ] = 1

    print(f"{task_id} organ reconstruction:\n - ", end="")
    print("\n - ".join([f"{v}x {k}" for k, v in organs.items()]))

json_metadata(task, task_id, db_path)

Get task metadata JSON.

Parameters:

Name	Type	Description	Default
task	str	Name of the task.	required
task_id	str	Id of the tasks.	required
db_path	str	Path to scan dataset.	required

Returns:

Type	Description
str	JSON location.
dict	Loaded JSON dictionary

Source code in romitask/cli/print_task_info.py

def json_metadata(task, task_id, db_path):
    """ Get task metadata JSON.

    Parameters
    ----------
    task : str
        Name of the task.
    task_id : str
        Id of the tasks.
    db_path : str
        Path to scan dataset.

    Returns
    -------
    str
        JSON location.
    dict
        Loaded JSON dictionary

    """
    json_f = os.path.join(db_path, task_id, f"{task}.json")
    return json_f, json.load(open(json_f, 'r'))

list_configured_tasks(toml_conf)

List the configured tasks in a toml file.

Parameters:

Name	Type	Description	Default
toml_conf	dict	The TOML tasks dictionary.	required

Returns:

Type	Description
list(str)	List of tasks names.

Source code in romitask/cli/print_task_info.py

def list_configured_tasks(toml_conf):
    """List the configured tasks in a toml file.

    Parameters
    ----------
    toml_conf : dict
        The TOML tasks dictionary.

    Returns
    -------
    list(str)
        List of tasks names.

    """
    return list(toml_conf.keys())

metadata_info(task, task_id, db_path)

Dump a JSON metadata file for a given task & dataset.

Source code in romitask/cli/print_task_info.py

def metadata_info(task, task_id, db_path):
    """Dump a JSON metadata file for a given task & dataset. """
    json_f, md_json = json_metadata(task, task_id, db_path)
    print(json.dumps(md_json, sort_keys=True, indent=2))

pointcloud_info(task, task_id, db_path)

Print info about PointCloud task output.

Source code in romitask/cli/print_task_info.py

def pointcloud_info(task, task_id, db_path):
    """Print info about PointCloud task output."""
    from plantdb.commons.io import read_point_cloud
    ply_f = os.path.join(db_path, task_id, f"{task}.ply")
    ply = read_point_cloud(ply_f)
    if ply.is_empty():
        print(f"PLY file for task '{task_id}' is empty!")
    else:
        print(f"Found PLY file for task '{task_id}':")
    print(f" - {len(ply.points)} points")
    print(f" - pointcloud dimensions (x, y, z): {ply.get_max_bound() - ply.get_min_bound()}")

segmentation2d_info(task, task_id, db_path)

Print info about ClusteredMesh task output.

Source code in romitask/cli/print_task_info.py

def segmentation2d_info(task, task_id, db_path):
    """Print info about ClusteredMesh task output."""
    out_dir = os.path.join(db_path, task_id)
    files = os.listdir(out_dir)
    png_files = [f for f in files if f.endswith('.png')]
    organs = {}
    for png in png_files:
        organ = os.path.splitext(png.split('_')[1])[0]
        if organ in organs.keys():
            organs[organ] += 1
        else:
            organs[organ] = 1

    n_imgs = get_dataset_size(db_path)
    print(f"{task_id} detected the following organs in RGB images: {', '.join(list(organs.keys()))}")
    print(f"{task_id} organ detected:\n - ", end="")
    print("\n - ".join([f"{v}/{n_imgs} for {k}" for k, v in organs.items()]))

segmented_pointcloud_info(task, task_id, db_path)

Print info about SegmentedPointCloud task output.

Source code in romitask/cli/print_task_info.py

def segmented_pointcloud_info(task, task_id, db_path):
    """Print info about SegmentedPointCloud task output."""
    from plantdb.commons.io import read_point_cloud
    ply_f = os.path.join(db_path, task_id, f"{task}.ply")
    ply = read_point_cloud(ply_f)
    if ply.is_empty():
        print(f"PLY file for task '{task_id}' is empty!")
    else:
        print(f"Found PLY file for task '{task_id}':")
    print(f" - {len(ply.points)} points")
    print(f" - {len(np.unique(np.asarray(ply.colors), axis=0))} unique colors")
    print(f" - pointcloud dimensions (x, y, z): {ply.get_max_bound() - ply.get_min_bound()}")

triangle_mesh_info(task, task_id, db_path)

Print info about TriangleMesh task output.

Source code in romitask/cli/print_task_info.py

def triangle_mesh_info(task, task_id, db_path):
    """Print info about TriangleMesh task output."""
    from plantdb.commons.io import read_triangle_mesh
    ply_f = os.path.join(db_path, task_id, f"{task}.ply")
    ply = read_triangle_mesh(ply_f)
    if ply.is_empty():
        print(f"PLY file for task '{task_id}' is empty!")
    else:
        print(f"Found PLY file for task '{task_id}':")
    print(f" - {len(ply.vertices)} vertices")
    print(f" - {len(ply.triangles)} triangles")
    print(f" - mesh dimensions (x, y, z): {ply.get_max_bound() - ply.get_min_bound()}")