Task module

Task Management System for ROMI Project

A comprehensive task management framework that provides base classes and utilities for defining, executing, and managing data processing tasks in the ROMI (RObot for MIcrofarm) project. This module implements a dependency-aware task system with database integration for managing plant imaging and analysis workflows.

Key Features

• Task dependency management through upstream task declarations
• File and dataset existence verification
• Database-integrated file management system
• Parameter handling with serialization support
• Task hierarchy support with base classes for:
  • Generic tasks (RomiTask)
  • File processing tasks (FileByFileTask)
  • Dataset verification tasks
  • File existence checking
• Clean-up utilities for managing task outputs
• Virtual plant object handling
• Error handling and failure management

Usage Examples

>>> # Creating a simple task
>>> class MyProcessingTask(RomiTask):
...     upstream_task = "PreviousTask"
...
...     def requires(self):
...         return {"previous": self.upstream_task()}
...
...     def run(self):
...         # Task implementation
...         pass

>>> # Using the file management system
>>> class MyFileTask(FileByFileTask):
...     query = "SELECT * FROM files"
...     type = "process"
...
...     def f(self, file):
...         # Process individual file
...         return processed_data

Clean

Bases: RomiTask

Cleanup a scan, keeping only the "images" fileset and removing all computed pipelines.

Attributes:

Name	Type	Description
upstream_task	TaskParameter	The upstream task.
scan_id	(Parameter, optional)	The scan id to use to get or create the FilesetTarget.
no_confirm	BoolParameter	Do not ask for confirmation of the cleaning in the command prompt. Default to False.
keep_metadata	listParameter	list of metadata to keep (retain) in the images fileset metadata. Default to IMAGES_MD.
keep_task	(Parameter, optional)	Task name to keep along with all its upstream dependencies. If specified, this task and all anterior tasks will be preserved. Default to empty string (keep nothing).

See Also

romitask.task.RomiTask romitask.task.IMAGES_MD

complete()

Indicate the task as complete.

Source code in romitask/task.py

def complete(self):
    """Indicate the task as complete."""
    return False  # there is no output

output()

Override inherited method to avoid FilesetTarget.

Source code in romitask/task.py

def output(self):
    """Override inherited method to avoid ``FilesetTarget``."""
    return None  # we do not want the cleaning task to add a Fileset!

requires()

No requirements here.

Source code in romitask/task.py

def requires(self):
    """No requirements here."""
    return []

run()

Execute the cleaning workflow.

Source code in romitask/task.py

def run(self):
    """Execute the cleaning workflow."""
    scan = ScanConfiguration().scan
    logger.info(f"Cleaning task started for scan '{scan.id}'...")

    # - Create the list of metadata to keep (retain) to a set and add the ones defined in `IMAGES_MD`
    metadata_whitelist = self._merge_metadata_keep_list(self.keep_metadata)

    # - Compute the set of filesets to preserve
    filesets_to_preserve = {"images"}  # Always keep images

    tasks_to_preserve = []
    if self.keep_task:
        # Parse the pipeline configuration to get task dependencies
        task_dependencies = self._parse_pipeline_toml(scan)

        if task_dependencies:
            # Compute the full hierarchy of tasks to preserve
            tasks_to_preserve = self._compute_task_hierarchy(
                self.keep_task, task_dependencies
            )

            # Fileset IDs need to be matched to task names as they have specific suffixes
            task_fs_map = locate_task_filesets(scan, tasks_to_preserve)
            # Add task-related fileset IDs to the preserve list
            filesets_to_preserve.update(list(task_fs_map.values()))

            logger.info(
                "Preserving task %r and its ancestors: %s",
                self.keep_task,
                tasks_to_preserve
            )
        else:
            logger.warning(
                "Could not parse task dependencies, only keeping %r",
                self.keep_task
            )
            filesets_to_preserve.add(self.keep_task)

    # - Handle the necessity to confirm prior to dataset & metadata cleaning.
    if not self.no_confirm:
        del_msg = (
            f"This will delete all filesets except: {filesets_to_preserve} "
            "and 'VirtualPlant' filesets."
        )
        logger.warning(del_msg)
        if not self._confirm():
            logger.info("User aborted the cleaning operation.")
            return

    # - Delete unwanted filesets.
    filesets_to_remove = self._filesets_to_remove(scan, exclude=filesets_to_preserve)
    if filesets_to_remove:
        self._delete_filesets(scan, filesets_to_remove)
    else:
        logger.info("No filesets to delete (preserved: %s).", filesets_to_preserve)

    # - Clean the metadata of the ``images`` fileset if Colmap task is not preserved.
    # Colmap is the only task that writes to the 'images' metadata.
    if 'Colmap' not in tasks_to_preserve:
        images_fs = scan.get_fileset("images")
        if images_fs is None:
            logger.warning("Could not locate the 'images' fileset in scan %s.", scan.id)
        else:
            self._clean_images_metadata(images_fs, metadata_whitelist)

    # - Clean orphan metadata files and directories.
    with open(scan.path() / "files.json", "r") as f:
        json_data = json.load(f)["filesets"]
        known_fs = [fs['id'] for fs in json_data]
    metadata_dir = Path(scan.path()) / "metadata"
    self._clean_orphan_json_files(metadata_dir, known_fs)
    self._clean_orphan_directories(metadata_dir, known_fs)

    # - Optionally remove the pipeline backup.
    if not self.keep_pipeline_cfg:
        self._remove_pipeline_backup(scan)

    logger.info("Cleaning task finished for scan %r.", scan.id)
    return

DatasetExists

Bases: RomiTask

A task that require a given dataset id (scan name) to exist.

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	Parameter	The dataset id (scan name) that should exist.

See Also

romitask.task.RomiTask

complete()

Indicate the task as complete.

Source code in romitask/task.py

def complete(self):
    """Indicate the task as complete."""
    return False  # there is no output

output()

Returns nothing.

Source code in romitask/task.py

def output(self):
    """Returns nothing."""
    return None

requires()

Require nothing.

Source code in romitask/task.py

def requires(self):
    """Require nothing."""
    return []

run()

Check the dataset exist.

Raises:

Type	Description
OSError	If the scan_id does not exist.

Source code in romitask/task.py

def run(self):
    """Check the dataset exist.

    Raises
    ------
    OSError
        If the `scan_id` does not exist.
    """
    db = ScanConfiguration().scan.db
    if db.get_scan(self.scan_id) is None:
        raise OSError(f"Scan {self.scan_id} does not exist!")
    return

DummyTask

Bases: RomiTask

A dummy task.

Does nothing and requires nothing.

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	(Parameter, optional)	The scan id to use to get or create the FilesetTarget.

See Also

romitask.task.RomiTask

requires()

Require nothing.

Source code in romitask/task.py

def requires(self):
    """Require nothing."""
    return []

run()

Do nothing.

Source code in romitask/task.py

def run(self):
    """Do nothing."""
    logger.info("Dummy task started.")
    logger.info(f"Got a scan named '{self.scan_id}'...")
    logger.info("Dummy task done.")
    return

FSDBConfiguration

Bases: Config

Configuration for a plantdb.commons.fsdb.FSBD database.

Attributes:

Name	Type	Description
scan	FSBD	The database to use for configuration.

Examples:

>>> from romitask.task import FSDBConfiguration
>>> from plantdb.commons.test_database import dummy_db
>>> # - First, let's create a dummy FSDB database to play with:
>>> db = dummy_db()
>>> db.connect()
>>> db_cfg = FSDBConfiguration(db)
>>> type(db_cfg.db)
plantdb.commons.fsdb.FSDB

FSDBParameter

Bases: Parameter

A custom Luigi Parameter for handling FSDB objects.

This parameter class extends Luigi's basic Parameter to support passing FSDB objects directly as parameters to Luigi tasks.

parse(db_path)

Parse the parameter value.

If the value is a string, attempt to reconstruct an FSDB object. Otherwise, return the value directly.

Parameters:

Name	Type	Description	Default
db_path	FSDB	The value to be parsed, either a string path or an FSDB object.	required

Returns:

Type	Description
FSDB	An FSDB object if the input was a string path to a valid FSDB, otherwise the unmodified input value.

Source code in romitask/task.py

def parse(self, db_path):
    """Parse the parameter value.

    If the value is a string, attempt to reconstruct an FSDB object.
    Otherwise, return the value directly.

    Parameters
    ----------
    db_path : plantdb.commons.fsdb.FSDB
        The value to be parsed, either a string path or an FSDB object.

    Returns
    -------
    plantdb.commons.fsdb.FSDB
        An FSDB object if the input was a string path to a valid FSDB,
        otherwise the unmodified input value.
    """
    if db_path is None:
        return None
    if isinstance(db_path, str):
        # Try to interpret the string as a path to an FSDB
        db_path = Path(db_path)

    if not db_path.exists():
        raise ValueError(f"Given path {db_path} does not exist")
    if not db_path.is_dir():
        raise NotADirectoryError(f"Given path {db_path} is not a directory")
    if not _is_fsdb(db_path):
        raise NotAnFSDBError(f"Given path {db_path} is not an FSDB")

    db = FSDB(db_path)
    db.connect()
    return db

serialize(db)

Serialize the parameter value to a string.

For FSDB objects, converts to a string representation of the base directory path. For other values, uses the default string conversion.

Parameters:

Name	Type	Description	Default
db	FSDB	The value to be serialized, preferably an FSDB object.	required

Returns:

Type	Description
str	String representation of the value, which for FSDB objects is the path.

Source code in romitask/task.py

def serialize(self, db):
    """Serialize the parameter value to a string.

    For FSDB objects, converts to a string representation of the base directory path.
    For other values, uses the default string conversion.

    Parameters
    ----------
    db : plantdb.commons.fsdb.FSDB
        The value to be serialized, preferably an FSDB object.

    Returns
    -------
    str
        String representation of the value, which for FSDB objects is the path.
    """
    if db is not None:
        return str(db.path())
    else:
        return None

FileByFileTask

Bases: RomiTask

Abstract task to sequentially apply a function to each File of a Fileset.

Attributes:

Name	Type	Description
upstream_task	TaskParameter	The upstream task.
scan_id	(Parameter, optional)	The scan id to use to get or create the FilesetTarget. If unspecified (default), the current active scan will be used.
query	(DictParameter, optional)	A filtering dictionary to apply on input `Fileset metadata. Key(s) and value(s) must be found in metadata to select the File. By default, no filtering is performed, all inputs are used.
type	None	???
reader	None	???
writer	None	???

Notes

Input Files metadata are copied to the target/output Files metadata.

f(f, outfs)

Function applied to every file in the fileset must return a file object.

Parameters:

Name	Type	Description	Default
f		Input file.	required
outfs		Output fileset.	required

Returns:

Type	Description
File	Tis file must be created in outfs.

Source code in romitask/task.py

def f(self, f, outfs):
    """Function applied to every file in the fileset must return a file object.

    Parameters
    ----------
    f: plantdb.commons.fsdb.FSDB.File
        Input file.
    outfs: plantdb.commons.fsdb.FSDB.Fileset
        Output fileset.

    Returns
    -------
    plantdb.commons.fsdb.FSDB.File
        Tis file must be created in `outfs`.
    """
    raise NotImplementedError

run()

Run the task on every Files from a Fileset that fulfill the query.

Source code in romitask/task.py

def run(self):
    """Run the task on every `File`s from a `Fileset` that fulfill the ``query``."""
    input_fileset = self.input().get()
    output_fileset = self.output().get()

    in_files = input_fileset.get_files(query=self.query)
    logger.debug(f"Got {len(in_files)} input files:")
    logger.debug(f"{', '.join([f.id for f in in_files])}")
    logger.debug(f"Got a filtering query: '{self.query}'.")

    for fi in tqdm(in_files, unit="file"):
        outfi = self.f(fi, output_fileset)
        if outfi is not None:
            m = fi.get_metadata()
            outm = outfi.get_metadata()
            outfi.set_metadata({**m, **outm})
    return

FileExists

Bases: RomiTask

A task that require a given file id to exist.

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	(Parameter, optional)	The dataset id (scan name) where to get the Fileset.
fileset_id	Parameter	Name of the fileset where to find the file.
file_id	Parameter	Name of the file that should exist.

See Also

romitask.task.RomiTask

output()

The output of the task is the fileset.

Source code in romitask/task.py

def output(self):
    """The output of the task is the fileset."""
    self.task_id = self.fileset_id  # name the task using the fileset
    return super().output()

output_file(file_id=None, create=False)

The output file should exist.

Source code in romitask/task.py

def output_file(self, file_id=None, create=False):
    """The output file should exist."""
    if file_id is None:
        file_id = self.file_id
    return super().output_file(file_id, create=False)

requires()

Require nothing.

Source code in romitask/task.py

def requires(self):
    """Require nothing."""
    return []

run()

Check the fileset and files exist.

Raises:

Type	Description
OSError	If the fileset_id does not exist.
OSError	If the file_id does not exist.

Source code in romitask/task.py

def run(self):
    """Check the fileset and files exist.

    Raises
    ------
    OSError
        If the `fileset_id` does not exist.
    OSError
        If the `file_id` does not exist.
    """
    if self.output().get() is None:
        raise OSError(f"Fileset {self.fileset_id} does not exist")
    if self.output().get().get_file(self.file_id) is None:
        raise OSError(f"File {self.fileset_id}/{self.file_id} does not exist")
    return

FilesetExists

Bases: RomiTask

A task that require a given fileset id to exist.

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	(Parameter, optional)	The dataset id (scan name) where to get the Fileset.
fileset_id	Parameter	The name of the fileset that should exist.

See Also

romitask.task.RomiTask

output()

The output of the task is the fileset.

Source code in romitask/task.py

def output(self):
    """The output of the task is the fileset."""
    self.task_id = self.fileset_id  # name the task using the fileset
    return super().output()

requires()

Require nothing.

Source code in romitask/task.py

def requires(self):
    """Require nothing."""
    return []

run()

Check the fileset exist.

Raises:

Type	Description
OSError	If the fileset_id does not exist.

Source code in romitask/task.py

def run(self):
    """Check the fileset exist.

    Raises
    ------
    OSError
        If the `fileset_id` does not exist.
    """
    if self.output().get() is None:
        raise OSError(f"Fileset {self.fileset_id} does not exist!")
    return

FilesetTarget(scan, fileset_id)

Bases: Target

Subclass luigi.Target for Fileset as defined in romitask plantdb.commons.fsdb.FSDB API.

A FilesetTarget is used by luigi.Task (or subclass) methods: * requires to assert the existence of the Fileset prior to starting the task; * output to create a Fileset after running a task.

Attributes:

Name	Type	Description
db	FSDB	An FSDB database instance.
scan	Scan	A Scan dataset instance within db.
fileset_id	str	Name of the target Fileset instance within scan.

Notes

A luigi Target subclass requires to implement the exists method.

Examples:

>>> from romitask.task import FilesetTarget
>>> from plantdb.commons.fsdb import FSDB
>>> from plantdb.commons.test_database import dummy_db
>>> # - First, let's create a dummy FSDB database to play with:
>>> db = dummy_db()
>>> db.connect()
>>> scan = db.create_scan("007")  # Add a `Scan` named `007` to the `FSDB` instance
>>> fs = scan.create_fileset("required_fs")  # Add a `Fileset` named `required_fs` to the `Scan` instance

>>> # - Now let's use `FilesetTarget` to check if a given `Fileset` exists in the FSDB (as used in Tasks `require` methods):
>>> fst = FilesetTarget(scan, "required_fs")
>>> fst.exists()  # `Fileset` exist but is empty
False

>>> # - Add a dummy test file to the `Fileset`:
>>> fs.create_file('dummy_test_file')
>>> fst.exists()  # `Fileset` exist and is not empty
True

>>> # - Now let's use `FilesetTarget` to create a new `Fileset` (as used in Tasks `output` methods):
>>> out_fst = FilesetTarget(scan, "output_fs")
>>> out_fs = out_fst.create()
>>> type(out_fs)
plantdb.commons.fsdb.Fileset
>>> print(out_fs.id)
'output_fs'

FilesetTarget constructor.

Parameters:

Name	Type	Description	Default
scan	Scan	The Scan dataset instance where to find/create the Fileset.	required
fileset_id	str	Name of the target Fileset.	required

Source code in romitask/task.py

def __init__(self, scan, fileset_id):
    """FilesetTarget constructor.

    Parameters
    ----------
    scan : plantdb.commons.fsdb.Scan
        The ``Scan`` dataset instance where to find/create the ``Fileset``.
    fileset_id : str
        Name of the target ``Fileset``.

    """
    self.db = scan.db
    self.scan = scan
    self.fileset_id = fileset_id

create()

Creates a Fileset using the plantdb FSDB API.

The name of the created Fileset is given by self.fileset_id.

Returns:

Type	Description
Fileset	The created Fileset instance.

Source code in romitask/task.py

def create(self):
    """Creates a ``Fileset`` using the ``plantdb`` FSDB API.

    The name of the created ``Fileset`` is given by ``self.fileset_id``.

    Returns
    -------
    plantdb.commons.fsdb.Fileset
        The created `Fileset` instance.
    """
    return self.scan.create_fileset(self.fileset_id)

exists()

Assert the target Fileset exists.

Returns:

Type	Description
bool	True if the target fileset exists and is not empty, else False.

Source code in romitask/task.py

def exists(self):
    """Assert the target ``Fileset`` exists.

    Returns
    -------
    bool
        ``True`` if the target fileset exists and is not empty, else ``False``.
    """
    return self.scan.fileset_exists(self.fileset_id) and len(self.scan.get_fileset(self.fileset_id).get_files()) > 0

get()

Returns the target Fileset instance.

Returns:

Type	Description
Fileset	The fetched Fileset instance.

Source code in romitask/task.py

def get(self):
    """Returns the target ``Fileset`` instance.

    Returns
    -------
    plantdb.commons.fsdb.Fileset
        The fetched ``Fileset`` instance.
    """
    return self.scan.get_fileset(self.fileset_id)

ImagesFilesetExists

Bases: FilesetExists

Task to assert the presence of a fileset containing the images.

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	(Parameter, optional)	The dataset id (scan name) where to get the Fileset.
fileset_id	Parameter	Name of the fileset containing the images. Defaults to 'images'.

ModelFilesetExists

Bases: FilesetExists

Task to assert the presence of a fileset containing the trained weight model for CNN prediction.

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	(Parameter, optional)	The dataset id (scan name) where to get the Fileset.
fileset_id	Parameter	Name of the fileset containing the trained weight model. Defaults to 'models'.

ParallelFileTask

Bases: RomiTask

Abstract task to parallelly apply a function to each File of a Fileset.

Attributes:

Name	Type	Description
upstream_task	TaskParameter	The upstream task.
scan_id	(Parameter, optional)	The scan id to use to get or create the FilesetTarget. If unspecified (default), the current active scan will be used.
query	(DictParameter, optional)	A filtering dictionary to apply on input `Fileset metadata. Key(s) and value(s) must be found in metadata to select the File. By default, no filtering is performed, all inputs are used.
n_workers	(IntParameter, optional)	Number of worker threads to use for parallel processing. Defaults to -1, which uses the default ThreadPoolExecutor behavior.
parallel	(BoolParameter, optional)	Flag to enable/disable parallel processing. Defaults to True.

Notes

Input Files metadata are copied to the target/output Files metadata. This task runs the processing in parallel using ThreadPoolExecutor.

f(f, outfs)

Function applied to every file in the fileset must return a file object.

Parameters:

Name	Type	Description	Default
f		Input file.	required
outfs		Output fileset.	required

Returns:

Type	Description
File	This file must be created in outfs.

Source code in romitask/task.py

def f(self, f, outfs):
    """Function applied to every file in the fileset must return a file object.

    Parameters
    ----------
    f: plantdb.commons.fsdb.FSDB.File
        Input file.
    outfs: plantdb.commons.fsdb.FSDB.Fileset
        Output fileset.

    Returns
    -------
    plantdb.commons.fsdb.FSDB.File
        This file must be created in `outfs`.
    """
    raise NotImplementedError

run(input_fileset, output_fileset)

Run the task on every Files from a Fileset that fulfill the query in parallel.

Source code in romitask/task.py

def run(self, input_fileset, output_fileset):
    """Run the task on every `File`s from a `Fileset` that fulfill the ``query`` in parallel."""
    in_files = input_fileset.get_files(query=self.query)
    logger.debug(f"Got {len(in_files)} input files:")
    logger.debug(f"{', '.join([f.id for f in in_files])}")
    logger.debug(f"Got a filtering query: '{self.query}'.")

    # Helper function for parallel processing
    def _process_file(fi):
        outfi = self.f(fi, output_fileset)
        if outfi is not None:
            m = fi.get_metadata()
            outm = outfi.get_metadata()
            outfi.set_metadata({**m, **outm})
        return outfi

    self.n_workers = None if self.n_workers == -1 else self.n_workers
    self.parallel = self.n_workers != 1
    if not self.parallel:
        # Sequential processing when parallel is disabled
        for fi in tqdm(in_files, unit="file"):
            _process_file(fi)
    else:
        # Parallel processing
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.n_workers) as executor:
            list(tqdm(executor.map(_process_file, in_files),
                      total=len(in_files), unit="file"))

    return

RetryTracker

Bases: RomiTask

Tracks retry counts for tasks by storing them in a JSON file.

This class extends RomiTask to provide tracking of retry attempts for other tasks. It maintains a JSON file that maps task IDs to their retry counts, allowing the system to keep track of how many times a task has been attempted.

Parameters:

Name	Type	Description	Default
task_id	str	Identifier of the task for which to track retry attempts.	required
retry_count	int	The current retry attempt number for the specified task.	required

Attributes:

Name	Type	Description
task_id	Parameter	The parameter storing the identifier of the task being tracked.
retry_count	IntParameter	The parameter storing the number of retry attempts for the task.

Notes

The retry information is stored in a JSON file named 'retry_tracker.json' in the output directory assigned to this task. If the file exists, it is read and updated; if not, a new tracking dictionary is created.

output()

Define the output target for the retry tracker file.

Source code in romitask/task.py

def output(self):
    """Define the output target for the retry tracker file."""
    self.task_id = self.__class__.__name__  # name the task using the class's name
    return super().output()

requires()

Nothing required.

Source code in romitask/task.py

def requires(self):
    """Nothing required."""
    return []

run()

Execute the retry tracking operation.

Creates or updates a JSON file with information about task retries. Maps the task_id to its retry count in a dictionary stored as JSON.

Source code in romitask/task.py

def run(self):
    """Execute the retry tracking operation.

    Creates or updates a JSON file with information about task retries.
    Maps the task_id to its retry count in a dictionary stored as JSON.
    """
    # Get the path for retry tracker file or create it if it doesn't exist
    outfile = self.output_file(f'retry_tracker', create=True)

    try:
        # Attempt to load existing retry tracking data
        retry_dict = read_json(outfile)
    except:
        # If file doesn't exist or isn't valid JSON, initialize empty dictionary
        retry_dict = {}

    # Update the retry count for the current task
    retry_dict[self.task_id] = self.retry_count
    # Save the updated tracking dictionary back to the JSON file
    write_json(outfile, retry_dict)
    logger.info(f"Tracked retry {self.retry_count} for {self.task_id}.")

RomiTask

Bases: Task

ROMI implementation of a luigi.Task, working with the plantdb.commons.db.DB API.

Attributes:

Name	Type	Description
upstream_task	TaskParameter	The upstream task.
scan_id	(Parameter, optional)	The dataset id (scan name) to use to get, or create, the FilesetTarget. If unspecified (default), the current active scan will be used.

Notes

The task parameters are exported automatically as fileset metadata.

The task name is also exported automatically as fileset metadata, under a 'task_name' entry.

get_task_name()

Helper method to get the name of the current task.

Returns:

Type	Description
str	The name of the current task.

Source code in romitask/task.py

def get_task_name(self):
    """Helper method to get the name of the current task.

    Returns
    -------
    str
        The name of the current task.
    """
    return self.get_task_family().split('.')[-1]

input_file(file_id=None, suffix=None)

Helper method to get a file from the input fileset.

Parameters:

Name	Type	Description	Default
file_id	str	Name of the input file. Defaults to None.	None
suffix	str	A suffix of the input file name. Defaults to None.	None

Returns:

Type	Description
File	The input file.

Source code in romitask/task.py

def input_file(self, file_id=None, suffix=None):
    """Helper method to get a file from the input fileset.

    Parameters
    ----------
    file_id : str, optional
        Name of the input file. Defaults to ``None``.
    suffix : str, optional
        A suffix of the input file name. Defaults to ``None``.

    Returns
    -------
    plantdb.commons.db.File
        The input file.
    """
    return self.upstream_task().output_file(file_id, suffix=suffix, create=False)

output()

Defines the returned Target, for a RomiTask it is a FileSetTarget.

The fileset ID being the task ID.

Returns:

Type	Description
FilesetTarget	A set of file(s) in the ???

Notes

The task parameters are exported automatically as fileset metadata, under a 'task_params' entry.

The task name is also exported automatically as fileset metadata, under a 'task_name' entry.

Source code in romitask/task.py

def output(self):
    """Defines the returned ``Target``, for a ``RomiTask`` it is a ``FileSetTarget``.

    The fileset ID being the task ID.

    Returns
    -------
    FilesetTarget
        A set of file(s) in the ???

    Notes
    -----
    The task parameters are exported automatically as fileset metadata, under a 'task_params' entry.

    The task name is also exported automatically as fileset metadata, under a 'task_name' entry.
    """
    # Get the `Fileset` id from the `task_id` attribute generated by `luigi.Task`
    # Can be overriding in inheriting class as for the `Visualization` task
    # This will be used as DIRECTORY NAME!
    fileset_id = self.task_id

    if str(self.scan_id) == "":
        fs_target = FilesetTarget(ScanConfiguration().scan, fileset_id)
    else:
        fs_target = FilesetTarget(db.get_scan(self.scan_id), fileset_id)

    try:
        fs = fs_target.create()  # create the fileset
    except FilesetExistsError:
        fs = fs_target.get()  # get the fileset
    # Export all the task parameters as a dictionary:
    params = dict(self.to_str_params(only_significant=False, only_public=False))

    # Try to fix empty "scan_id":
    if params["scan_id"] == "":
        params["scan_id"] = fs.get_scan().get_id()

    # Check if it needs JSON parsing:
    for k in params.keys():
        try:
            params[k] = json.loads(params[k])
        except KeyError:
            continue
        except JSONDecodeError:
            continue

    # Save the task parameter as fileset metadata under "task_params":
    fs.set_metadata("task_params", params)
    # Save the task name as fileset metadata under "task_name":
    fs.set_metadata("task_name", self.get_task_name())
    return fs_target

output_file(file_id=None, suffix=None, create=True)

Helper method to create & get a file from the output fileset.

Parameters:

Name	Type	Description	Default
file_id	str	Name of the output file. Defaults to None.	None
suffix	str	Add a suffix to the output file name. Defaults to None.	None
create	bool	Define if the output file should be created. Defaults to True	True

Returns:

Type	Description
File	The (created) output file.

Source code in romitask/task.py

def output_file(self, file_id=None, suffix=None, create=True):
    """Helper method to create & get a file from the output fileset.

    Parameters
    ----------
    file_id : str, optional
        Name of the output file. Defaults to ``None``.
    suffix : str, optional
        Add a suffix to the output file name. Defaults to ``None``.
    create : bool, optional
        Define if the output file should be created. Defaults to ``True``

    Returns
    -------
    plantdb.commons.fsdb.File
        The (created) output file.
    """
    if file_id is None:
        file_id = self.get_task_name()
    if suffix is not None:
        file_id += suffix

    try:
        f = self.output().get().create_file(file_id)
    except IOError:
        f = self.output().get().get_file(file_id)
    return f

requires()

Specify dependencies to other Task object.

This method will be overridden by the classes inheriting RomiTask.

Returns:

Type	Description
TaskParameter	The upstream task.

Source code in romitask/task.py

def requires(self):
    """Specify dependencies to other Task object.

    This method will be overridden by the classes inheriting ``RomiTask``.

    Returns
    -------
    luigi.TaskParameter
        The upstream task.
    """
    return self.upstream_task()

ScanConfiguration

Bases: Config

Configuration for a plantdb.commons.fsdb.Scan scan dataset.

Attributes:

Name	Type	Description
scan	Scan	The scan dataset to use for configuration.

Examples:

>>> from romitask.task import ScanConfiguration
>>> from plantdb.commons.test_database import dummy_db
>>> # - First, let's create a dummy FSDB database to play with:
>>> db = dummy_db()
>>> db.connect()
>>> scan = db.create_scan("007")  # Add a `Scan` named `007` to the `FSDB` instance
>>> scan_cfg = ScanConfiguration(scan)
>>> type(scan_cfg.scan)
plantdb.commons.fsdb.Scan

ScanParameter

Bases: Parameter

Register a luigi.Parameter object to access plantdb.commons.fsdb.Scan class.

Override the default implementation methods parse & serialize.

Notes

The parse method connect to the given path to a plantdb.commons.fsdb.FSDB database.

parse(scan_path)

Convert the scan path to a plantdb.commons.fsdb.Scan object.

Override the default implementation method for specialized parsing.

Parameters:

Name	Type	Description	Default
scan_path	str	The value to parse, here the path to an plantdb.commons.fsdb.Scan dataset.	required

Returns:

Type	Description
Scan	The object corresponding to given path.

Notes

Uses given path, e.g. /db/root/path/scan_id, to defines: - the database root dir with /db/root/path - the scan dataset id with scan_id

If the given scan dataset id does not exist, it is created.

Source code in romitask/task.py

def parse(self, scan_path):
    """Convert the scan path to a ``plantdb.commons.fsdb.Scan`` object.

    Override the default implementation method for specialized parsing.

    Parameters
    ----------
    scan_path : str
        The value to parse, here the path to an ``plantdb.commons.fsdb.Scan`` dataset.

    Returns
    -------
    plantdb.commons.fsdb.Scan
        The object corresponding to given path.

    Notes
    -----
    Uses given path, e.g. ``/db/root/path/scan_id``, to defines:
      - the database root dir with ``/db/root/path``
      - the scan dataset id with ``scan_id``

    If the given scan dataset id does not exist, it is created.
    """
    from plantdb.commons.fsdb.core import FSDB
    global db
    path = scan_path.rstrip('/')
    path = path.split('/')
    # Defines the database root dir
    db_path = '/'.join(path[:-1])
    # Defines the scan dataset id
    scan_id = path[-1]
    # create & connect to `db` if not defined:
    if db is None:  # TODO: cannot change DB during run...
        no_auth = os.getenv("ROMI_DB_NOAUTH")=='1' or False
        db = FSDB(db_path, no_auth=no_auth)
        db.connect()
        if not no_auth:
            db.login(username=os.getenv("ROMI_DB_USER"), password=os.getenv("ROMI_DB_PASSWORD"))
    # Get the scan dataset object or create one & return it
    if db.scan_exists(scan_id):
        scan = db.get_scan(scan_id)
    else:
        scan = db.create_scan(scan_id)
    return scan

serialize(scan)

Converts plantdb.commons.fsdb.Scan in its path.

Opposite of parse().

Parameters:

Name	Type	Description	Default
scan	Scan	The value to serialize, here a plantdb.commons.fsdb.Scan object.	required

Returns:

Type	Description
str	The path to the corresponding plantdb.commons.fsdb.Scan object.

Source code in romitask/task.py

def serialize(self, scan):
    """Converts ``plantdb.commons.fsdb.Scan`` in its path.

    Opposite of ``parse()``.

    Parameters
    ----------
    scan : plantdb.commons.fsdb.Scan
        The value to serialize, here a ``plantdb.commons.fsdb.Scan`` object.

    Returns
    -------
    str
        The path to the corresponding ``plantdb.commons.fsdb.Scan`` object.
    """
    db_path = scan.db.basedir
    scan_id = scan.id
    return str(db_path / scan_id)

Segmentation2DGroundTruthFilesetExists

Bases: FilesetExists

Task to assert the presence of a fileset containing the ground-truth images for CNN training.

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	(Parameter, optional)	The dataset id (scan name) where to get the Fileset.
fileset_id	Parameter	Name of the fileset containing the ground-truth images. Defaults to 'Segmentation2DGroundTruth'.

VirtualPlantObj

Bases: FileExists

The VirtualPlantObj task returns a 3D plant model file.

The 3D model should be stored as '.obj' and '.mtl' files.

Use as follows, using VirtualScan as an example:

.. code-block::

[VirtualScan]
obj_fileset = "VirtualPlantObj"

This example will seek for the default file with ID VirtualPlant in the fileset VirtualPlant in the active scan directory.

These default values can be overriden as follows:

.. code-block::

[VirtualPlantObj]
scan_id = "AnotherScan"
fileset_id = "AnotherFileset"
fileset_id_prefix = "AnotherFilesetPrefix"
file_id = "FileID"

[VirtualScan]
obj_fileset = "VirtualPlantObj"

Attributes:

Name	Type	Description
upstream_task	None	No upstream task is required.
scan_id	(Parameter, optional)	The scan id where to look for the fileset. If unspecified (default), the current active scan will be used.
fileset_id	(Parameter, optional)	The ID of the fileset to use. Defaults to "VirtualPlant".
fileset_id_prefix	(Parameter, optional)	If the ID of the fileset cannot be provided, a prefix can be an alternative to search for a fileset id. Defaults to "VirtualPlant".
file_id	(Parameter, optional)	The ID of the file to use. Defaults to "VirtualPlant".

output()

Return the fileset containing the model files.

Source code in romitask/task.py

def output(self):
    """Return the fileset containing the model files."""
    if self.scan_id == "":
        scan = ScanConfiguration().scan
    else:
        scan = db.get_scan(self.scan_id)

    t = FilesetTarget(scan, self.fileset_id)
    if not t.exists():  # backup solution : search for a fileset id beginning with a specific prefix
        filesets_with_prefix = [f for f in scan.get_filesets() if f.id.startswith(self.fileset_id_prefix)]
        if len(filesets_with_prefix) == 0:
            raise FileNotFoundError(f"Fileset with {self.fileset_id_prefix} prefix does not exist")
        elif len(filesets_with_prefix) > 1:
            raise ValueError(f"Two or more Filesets with {self.fileset_id_prefix} prefix found")
        else:
            self.fileset_id = filesets_with_prefix[0].id
            t = FilesetTarget(scan, self.fileset_id)

    fs = t.get()

    params = dict(self.to_str_params(only_significant=False, only_public=False))
    for k in params.keys():
        try:
            params[k] = json.loads(params[k])
        except KeyError:
            continue
        except JSONDecodeError:
            continue
    fs.set_metadata("task_params", params)

    self.task_id = self.fileset_id
    return t

mourn_failure(task, exception)

In the case of failure of a task, remove the corresponding fileset from the database.

Parameters:

Name	Type	Description	Default
task	RomiTask	The task which has failed.	required
exception	Exception	The exception raised by the failure.	required

Source code in romitask/task.py

@RomiTask.event_handler(luigi.Event.FAILURE)
def mourn_failure(task, exception):
    """In the case of failure of a task, remove the corresponding fileset from the database.

    Parameters
    ----------
    task : RomiTask
        The task which has failed.
    exception : Exception
        The exception raised by the failure.
    """
    # Log the failure:
    logger.critical(exception)