hardware_robustness

plantimager.cli.hardware_robustness

_get_diff_between_dict

_get_diff_between_dict(d1, d2)

Return the entries that are different between two dictionaries.

Source code in plantimager/cli/hardware_robustness.py

def _get_diff_between_dict(d1, d2):
    """Return the entries that are different between two dictionaries."""
    diff_keys = list(dict(set(d1.items()) ^ set(d2.items())).keys())
    diff1 = {k: d1.get(k, None) for k in diff_keys}
    diff2 = {k: d2.get(k, None) for k in diff_keys}
    return diff1, diff2

check_scan_configs

check_scan_configs(scans_list)

Check the scan configuration is the same as the reference.

Parameters:

Name	Type	Description	Default
scans_list	list of plantsb.FSDB.Scan	List of Scan instances to compare.	required

Returns:

Type	Description
bool	True if the scan configurations are the same, else False.

Examples:

>>> import os, toml
>>> from pathlib import Path
>>> from plantdb.fsdb import FSDB
>>> db = FSDB(os.environ.get('ROMI_DB', '/home/aurele/Downloads/20221010_Jo/Scans/'))
>>> db.connect()  # print the list of scan dataset names
>>> check_scan_configs(db.get_scans())
>>> db.disconnect()

Source code in plantimager/cli/hardware_robustness.py

def check_scan_configs(scans_list):
    """Check the scan configuration is the same as the reference.

    Parameters
    ----------
    scans_list : list of plantsb.FSDB.Scan
        List of ``Scan`` instances to compare.

    Returns
    -------
    bool
        ``True`` if the scan configurations are the same, else ``False``.

    Examples
    --------
    >>> import os, toml
    >>> from pathlib import Path
    >>> from plantdb.fsdb import FSDB
    >>> db = FSDB(os.environ.get('ROMI_DB', '/home/aurele/Downloads/20221010_Jo/Scans/'))
    >>> db.connect()  # print the list of scan dataset names
    >>> check_scan_configs(db.get_scans())
    >>> db.disconnect()

    """
    same_cfg = True
    ref_scan_cfg = {}

    for scan in scans_list:
        bak_cfg_file = Path(scan.path()) / "scan.toml"
        if not bak_cfg_file.is_file():
            logger.warning(f"Could not load backup `scan.toml` for scan id '{scan.id}'!")
            continue
        cfg = toml.load(str(bak_cfg_file))
        scan_cfg = cfg['ScanPath']
        if ref_scan_cfg == {}:
            ref_scan_cfg = scan_cfg.deepcopy()
        else:
            same_type = compare_scan_type(ref_scan_cfg, scan_cfg)
            same_params = compare_scan_params(ref_scan_cfg, scan_cfg)
    return

make_animation

make_animation(db_location, imgs, img_fname, delay)

Make the animation as a sequence of images taken at the same position.

Parameters:

Name	Type	Description	Default
db_location	Path or str	Path to the database directory.	required
imgs	dict	Scan.id indexed dict of selected img_fname	required
img_fname	str	Name (without extension) of the image (from the .	required
delay	int	The delay between successive frames	required

Source code in plantimager/cli/hardware_robustness.py

def make_animation(db_location, imgs, img_fname, delay):
    """Make the animation as a sequence of images taken at the same position.

    Parameters
    ----------
    db_location : pathlib.Path or str
       Path to the database directory.
    imgs : dict
        ``Scan.id`` indexed dict of selected `img_fname`
    img_fname : str
        Name (without extension) of the image (from the .
    delay : int
        The delay between successive frames

    Returns
    -------

    """
    # Initialize a temporary directory use to store scan dataset before cleaning and running previous tasks to task to analyse:
    tmp_dir = tempfile.mkdtemp()
    # - Get the path of the temporary ROMI database
    tmp_path = Path(tmp_dir)
    logger.debug(f"Create a temporary directory '{tmp_path}' to modify the images prior to animation...")

    # Copy the image to temporary directory:
    annotated_imgs = {}
    for n, (scan_id, img) in enumerate(imgs.items()):
        img_fname = str(Path(img).stem) + f'_{n}' + '.jpg'
        annotated_imgs[scan_id] = f"{tmp_path}/{img_fname}"
        subprocess.run(["cp", img, annotated_imgs[scan_id]])

    # Add the scan name to image
    for scan_id, img in annotated_imgs.items():
        cmd = ['mogrify', '-font', 'Liberation-Sans', '-fill', 'white', '-undercolor', "black", '-pointsize', '26',
               '-gravity', 'NorthEast', '-annotate', '+10+10', f"'{scan_id}/{img_fname}'", img]
        subprocess.run(cmd)

    gif_fname = str(db_location / Path(str(Path(img_fname).stem) + '.gif'))
    cmd = ["convert", "-delay", f"{delay}"] + sorted(map(str, annotated_imgs.values())) + [gif_fname]
    subprocess.run(cmd, check=True)
    rmtree(tmp_path)
    logger.info(f"Generated the GIF animation: '{gif_fname}'.")

romi_run_task

romi_run_task(scan_path, task_name, cfg_file)

Run configured pipeline for given task on a scan.

Parameters:

Name	Type	Description	Default
scan_path	Path or str	Path to the scan dataset directory.	required
task_name	str	Name of the ROMI task to run.	required
cfg_file	str	Path to the configuration file to use to run the pipeline (romi_run_task).	required

Source code in plantimager/cli/hardware_robustness.py

def romi_run_task(scan_path, task_name, cfg_file):
    """Run configured pipeline for given task on a scan.

    Parameters
    ----------
    scan_path : pathlib.Path or str
       Path to the scan dataset directory.
    task_name : str
        Name of the ROMI task to run.
    cfg_file : str
        Path to the configuration file to use to run the pipeline (``romi_run_task``).

    """
    logger.info(f"Executing task '{task_name}' on scan dataset '{str(scan_path)}'.")
    cmd = ["romi_run_task", task_name, str(scan_path), "--config", cfg_file, "--local-scheduler"]
    subprocess.run(cmd, check=True)
    return