Skip to content

utils

Utility Functions for ROMI Data HandlingLink

A small toolbox of helper functions used throughout the ROMI project.
These utilities simplify common tasks such as image loading, temporary file‑system handling, task‑fileset discovery, angle unit detection, date formatting, user prompts, and file backup.

Key FeaturesLink

  • Image I/O: read_image_from_file loads an image with Pillow and guarantees it is fully loaded into memory.
  • Task‑fileset mapping: locate_task_filesets returns a dictionary that links task names to the corresponding fileset identifiers in a scan.
  • Angle unit guessing: is_radians heuristically decides whether a list of angles is expressed in radians.
  • Filesystem helpers: to_file, fsdb_file_from_local_file, tmpdir_from_fileset, backup_filename and backup_file make it easy to move data between the in‑memory FSDB representation and the real file system.
  • String matching: partial_match performs deep, optional fuzzy matching of nested dictionaries, lists, or strings.
  • ISO‑8601 timestamps: iso_date_now returns the current local time in ISO‑8601 format.
  • Interactive prompts: yes_no_choice and yes_no_abort_choice provide robust yes/no (with optional abort) dialogs.
  • Name sanitization: sanitize_name validates and cleans identifiers for safe use.

backup_file Link

backup_file(file)

Back up a file by creating a timestamped copy.

Parameters:

Name Type Description Default

file Link

str or Path

The path to the file to back up.

required

Returns:

Type Description
Path

The path to the backup.

See Also

backup_filename

Source code in plantdb/commons/utils.py
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def backup_file(file: Path) -> Path:
    """Back up a file by creating a timestamped copy.

    Parameters
    ----------
    file : str or pathlib.Path
        The path to the file to back up.

    Returns
    -------
    pathlib.Path
        The path to the backup.

    See Also
    --------
    backup_filename
    """
    from shutil import copy
    bak_fname = backup_filename(file)
    copy(file, bak_fname)
    return bak_fname

backup_filename Link

backup_filename(file)

Create a backup a filename by adding a timestamp.

Parameters:

Name Type Description Default

file Link

str or Path

The path to the file to back up.

required

Returns:

Type Description
Path

The path to the backup.

Examples:

>>> from plantdb.commons.utils import backup_filename
>>> backup_filename("test/file.json")
PosixPath('test/file_230601_105815.json')
Source code in plantdb/commons/utils.py
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
def backup_filename(file: Path) -> Path:
    """Create a backup a filename by adding a timestamp.

    Parameters
    ----------
    file : str or pathlib.Path
        The path to the file to back up.

    Returns
    -------
    pathlib.Path
        The path to the backup.

    Examples
    --------
    >>> from plantdb.commons.utils import backup_filename
    >>> backup_filename("test/file.json")
    PosixPath('test/file_230601_105815.json')
    """
    file = Path(file)
    now = datetime.now()
    timestamp = now.strftime("%y%m%d_%H%M%S")
    fname = file.stem
    return file.with_stem(f"{fname}_{timestamp}")

fsdb_file_from_local_file Link

fsdb_file_from_local_file(path)

Creates a temporary fsdb.core.File object from a local file.

Parameters:

Name Type Description Default

path Link

Path or str

The file path to use to create the temporary local database.

required

Returns:

Type Description
File

The temporary fsdb.core.File.

Source code in plantdb/commons/utils.py
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def fsdb_file_from_local_file(path):
    """Creates a temporary ``fsdb.core.File`` object from a local file.

    Parameters
    ----------
    path : pathlib.Path or str
        The file path to use to create the temporary local database.

    Returns
    -------
    plantdb.commons.fsdb.core.File
        The temporary ``fsdb.core.File``.
    """
    from plantdb.commons.fsdb.core import FSDB
    from plantdb.commons.fsdb.core import Scan
    from plantdb.commons.fsdb.core import Fileset
    from plantdb.commons.fsdb.core import File
    path = Path(path)
    dirname, fname = path.parent, path.name
    id = Path(fname).stem
    with tempfile.TemporaryDirectory() as tmpdir:
        # Initialise a temporary `FSDB`:
        Path(f"{tmpdir}/{MARKER_FILE_NAME}").touch()  # add the db marker file
        db = FSDB(tmpdir)
        # Initialize a `Scan` instance:
        scan = Scan(db, "tmp")
        # Initialize a `Fileset` instance:
        fileset = Fileset(scan, dirname)
        # Initialize a `File` instance & return it:
        f = File(db=db, fileset=fileset, f_id=id)
        f.filename = fname
        f.metadata = None
    return f

is_radians Link

is_radians(angles)

Guess if the sequence of angles is in radians.

Parameters:

Name Type Description Default

angles Link

list of float

Sequence of angle values.

required

Returns:

Type Description
bool

True if the sequence is in radians, else `False.

Notes

This assumes that the angles can not be greater than 360 degrees or its equivalent in radians.

Source code in plantdb/commons/utils.py
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def is_radians(angles):
    """Guess if the sequence of angles is in radians.

    Parameters
    ----------
    angles : list of float
        Sequence of angle values.

    Returns
    -------
    bool
        `True` if the sequence is in radians, else `False.

    Notes
    -----
    This assumes that the angles can not be greater than 360 degrees or its equivalent in radians.
    """
    from math import radians
    if all([angle < radians(360) for angle in angles]):
        return True
    else:
        return False

iso_date_now Link

iso_date_now()

Get current date and time in ISO format.

Returns:

Type Description
str

The current local date and time as a formatted string.

Notes

The function uses Python's datetime module to get the current date and time. The formatting follows the directives specified in the fmt parameter, which can include various placeholders like '%Y', '%m', '%d', etc. For more details on the allowed format codes, refer to the documentation of strftime in Python.

See Also

datetime.strftime : Convert a datetime object to a string based on a format. datetime.now : Return the current local date and time.

Examples:

>>> from plantdb.commons.utils import iso_date_now
>>> iso_date_now()
'2025-09-16T23:39:54.132898'
Source code in plantdb/commons/utils.py
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def iso_date_now():
    """Get current date and time in ISO format.

    Returns
    -------
    str
        The current local date and time as a formatted string.

    Notes
    -----
    The function uses Python's `datetime` module to get the current date and time.
    The formatting follows the directives specified in the `fmt` parameter, which can
    include various placeholders like '%Y', '%m', '%d', etc. For more details on the
    allowed format codes, refer to the documentation of `strftime` in Python.

    See Also
    --------
    datetime.strftime : Convert a datetime object to a string based on a format.
    datetime.now : Return the current local date and time.

    Examples
    --------
    >>> from plantdb.commons.utils import iso_date_now
    >>> iso_date_now()
    '2025-09-16T23:39:54.132898'
    """
    return datetime.isoformat(datetime.now())

locate_task_filesets Link

locate_task_filesets(scan, tasks)

Map the task names to task filesets.

Parameters:

Name Type Description Default

scan Link

Scan

A Scan instance from a local plant database (FSDB).

required

tasks Link

list of str

A list of task names to look up in the scan's list of filesets.

required

Returns:

Type Description
dict

A task indexed dictionary of fileset ids, value may be "None" if no matching fileset was found.

Notes

If more than one fileset id matches a task name, only the first one (found) will be returned!

Examples:

>>> from plantdb.commons.utils import locate_task_filesets
>>> from plantdb.commons.fsdb.core import FSDB
>>> from plantdb.commons.test_database import test_database
>>> db = test_database('real_plant_analyzed')
>>> db.connect()
>>> scan = db.get_scan('real_plant_analyzed')
>>> tasks_fs = locate_task_filesets(scan, ['Masks', 'PointCloud', 'UnknownTask'])
>>> print(tasks_fs)
{'Masks': 'Masks_1__0__1__0____channel____rgb_5619aa428d', 'PointCloud': 'PointCloud_1_0_1_0_10_0_7ee836e5a9', 'UnknownTask': 'None'}
>>> db.disconnect()
Source code in plantdb/commons/utils.py
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def locate_task_filesets(scan, tasks):
    """Map the task names to task filesets.

    Parameters
    ----------
    scan : plantdb.commons.fsdb.core.Scan
        A ``Scan`` instance from a local plant database (FSDB).
    tasks : list of str
        A list of task names to look up in the scan's list of filesets.

    Returns
    -------
    dict
        A task indexed dictionary of fileset ids, value may be "None" if no matching fileset was found.

    Notes
    -----
    If more than one fileset id matches a task name, only the first one (found) will be returned!

    Examples
    --------
    >>> from plantdb.commons.utils import locate_task_filesets
    >>> from plantdb.commons.fsdb.core import FSDB
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database('real_plant_analyzed')
    >>> db.connect()
    >>> scan = db.get_scan('real_plant_analyzed')
    >>> tasks_fs = locate_task_filesets(scan, ['Masks', 'PointCloud', 'UnknownTask'])
    >>> print(tasks_fs)
    {'Masks': 'Masks_1__0__1__0____channel____rgb_5619aa428d', 'PointCloud': 'PointCloud_1_0_1_0_10_0_7ee836e5a9', 'UnknownTask': 'None'}
    >>> db.disconnect()
    """
    # List all filesets in the scan dataset:
    fs_list = scan.list_filesets()
    # Find the fileset corresponding to the given list of tasks, if any:
    fileset_names = {}
    for task in tasks:
        try:
            # TODO: could be improved by using the saved config ('pipeline.toml') and recreate the name hash from luigi...
            fileset_names[task] = [fs for fs in fs_list if fs.startswith(task)][0]
        except IndexError:
            fileset_names[task] = "None"
    return fileset_names

partial_match Link

partial_match(source, target, fuzzy=False)

Partial matching of a reference dictionary against a target, potentially using regexp.

Parameters:

Name Type Description Default

source Link

dict or list or str

The source dictionary, list or string, with partial information, to test against the target.

required

target Link

dict or list or str

The target dictionary, list or string holding the complete information to match against.

required

fuzzy Link

bool

Whether to use fuzzy matching or not, that is the use of regular expressions.

False

Returns:

Type Description
bool

Whether the partial match is found or not.

Examples:

>>> from plantdb.commons.utils import partial_match
>>> ref = {"object": {"environment":"virtual"}}
>>> target = {'object': {'environment': 'virtual', 'plant_id': 'arabidopsis000', 'species': 'Arabidopsis Thaliana'}, 'scanner': {'workspace': {'x': [-200, 200], 'y': [-200, 200], 'z': [10, 1000]}}}
>>> partial_match(ref, target)
True
>>> ref = {"object": {"species":"Arabidopsis.*"}}
>>> partial_match(ref, target, fuzzy=True)
True
>>> ref = {"species":"Arabidopsis.*"}
>>> partial_match(ref, target, fuzzy=True)
False
Source code in plantdb/commons/utils.py
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def partial_match(source, target, fuzzy=False):
    """Partial matching of a reference dictionary against a target, potentially using regexp.

    Parameters
    ----------
    source : dict or list or str
        The source dictionary, list or string, with partial information, to test against the target.
    target : dict or list or str
        The target dictionary, list or string holding the complete information to match against.
    fuzzy : bool
        Whether to use fuzzy matching or not, that is the use of regular expressions.

    Returns
    -------
    bool
        Whether the partial match is found or not.

    Examples
    --------
    >>> from plantdb.commons.utils import partial_match
    >>> ref = {"object": {"environment":"virtual"}}
    >>> target = {'object': {'environment': 'virtual', 'plant_id': 'arabidopsis000', 'species': 'Arabidopsis Thaliana'}, 'scanner': {'workspace': {'x': [-200, 200], 'y': [-200, 200], 'z': [10, 1000]}}}
    >>> partial_match(ref, target)
    True
    >>> ref = {"object": {"species":"Arabidopsis.*"}}
    >>> partial_match(ref, target, fuzzy=True)
    True
    >>> ref = {"species":"Arabidopsis.*"}
    >>> partial_match(ref, target, fuzzy=True)
    False
    """
    from re import match

    # Check if both are dictionaries
    if isinstance(source, dict) and isinstance(target, dict):
        return all(  # Use all() to ensure every key in source is matched
            key in target and partial_match(value, target[key], fuzzy)  # Recursively check each dictionary item
            for key, value in source.items()
        )
    # Check if both are lists
    elif isinstance(source, list) and isinstance(target, list):
        return len(source) <= len(target) and all(  # Ensure source list is shorter or equal length to target list
            any(partial_match(ref_item, target_item, fuzzy) for target_item in target)
            # Use any() to check at least one match per item
            for ref_item in source
        )
    # Check if both are strings for fuzzy matching
    elif fuzzy and isinstance(source, str) and isinstance(target, str):
        # Return True if the source regex pattern matches the target string
        return bool(match(source, target))
    else:
        # Direct comparison of values
        return source == target

read_image_from_file Link

read_image_from_file(filename)

Read an image from a file and return it.

Parameters:

Name Type Description Default

filename Link

str or Path

The path to the image to read.

required

Returns:

Name Type Description
image Image

The loaded image.

Examples:

>>> from plantdb.commons.utils import read_image_from_file
>>> from plantdb.commons.test_database import test_database
>>> db = test_database('real_plant')
>>> db.connect()
>>> scan = db.get_scan('real_plant')
>>> fileset = scan.get_fileset('images')
>>> img_path = fileset.get_file('00000_rgb').path()
>>> image = read_image_from_file(img_path)
>>> print(image.size)
(1440, 1080)
>>> db.disconnect()
Source code in plantdb/commons/utils.py
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def read_image_from_file(filename):
    """Read an image from a file and return it.

    Parameters
    ----------
    filename : str or pathlib.Path
        The path to the image to read.

    Returns
    -------
    image : PIL.Image.Image
        The loaded image.

    Examples
    --------
    >>> from plantdb.commons.utils import read_image_from_file
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database('real_plant')
    >>> db.connect()
    >>> scan = db.get_scan('real_plant')
    >>> fileset = scan.get_fileset('images')
    >>> img_path = fileset.get_file('00000_rgb').path()
    >>> image = read_image_from_file(img_path)
    >>> print(image.size)
    (1440, 1080)
    >>> db.disconnect()
    """
    image = Image.open(filename)
    image.load()
    return image

sanitize_name Link

sanitize_name(name)

Sanitizes and validates the provided name.

The function ensures that the input string adheres to predefined naming rules by:

  • stripping leading/trailing spaces,
  • isolating the last segment after splitting by slashes,
  • validating the name against an alphanumeric pattern with optional underscores (_), dashes (-), or periods (.).

Parameters:

Name Type Description Default

name Link

str

The name to sanitize and validate.

required

Returns:

Type Description
str

Sanitized name that conforms to the rules.

Raises:

Type Description
ValueError

If the provided name contains invalid characters or does not meet the naming rules.

Source code in plantdb/commons/utils.py
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
def sanitize_name(name) -> str:
    """Sanitizes and validates the provided name.

    The function ensures that the input string adheres to predefined naming rules by:

    - stripping leading/trailing spaces,
    - isolating the last segment after splitting by slashes,
    - validating the name against an alphanumeric pattern
      with optional underscores (`_`), dashes (`-`), or periods (`.`).

    Parameters
    ----------
    name : str
        The name to sanitize and validate.

    Returns
    -------
    str
        Sanitized name that conforms to the rules.

    Raises
    ------
    ValueError
        If the provided name contains invalid characters or does not meet
        the naming rules.
    """
    import re

    sanitized_name = name.strip()  # Remove leading/trailing spaces
    sanitized_name = sanitized_name.split("/")[-1]  # isolate the last segment after splitting by slashes
    # Validate against an alphanumeric pattern with optional underscores, dashes, or periods
    if not re.match(r"^[a-zA-Z0-9_.-]+$", sanitized_name):
        raise ValueError(
            f"Invalid name: '{name}'. Names must be alphanumeric and can include underscores, dashes, or periods."
        )
    return sanitized_name

tmpdir_from_fileset Link

tmpdir_from_fileset(fileset)

Creates a temporary directory to host the Fileset object and write files.

Parameters:

Name Type Description Default

fileset Link

Fileset

The fileset to use to create the temporary local database.

required

Returns:

Type Description
TemporaryDirectory

The temporary directory hosting the fileset and file(s), if any.

Source code in plantdb/commons/utils.py
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def tmpdir_from_fileset(fileset):
    """Creates a temporary directory to host the ``Fileset`` object and write files.

    Parameters
    ----------
    fileset : plantdb.commons.fsdb.core.Fileset
        The fileset to use to create the temporary local database.

    Returns
    -------
    tempfile.TemporaryDirectory
        The temporary directory hosting the fileset and file(s), if any.
    """
    tmpdir = tempfile.TemporaryDirectory()
    for f in fileset.get_files():
        filepath = Path(tmpdir.name) / f.filename
        to_file(f, filepath)
    return tmpdir

to_file Link

to_file(dbfile, path)

Write a dbfile to a file in the filesystem.

Parameters:

Name Type Description Default

dbfile Link

File

The File instance to save under given path.

required

path Link

Path or str

The file path to use to save the dbfile.

required
Source code in plantdb/commons/utils.py
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def to_file(dbfile, path):
    """Write a `dbfile` to a file in the filesystem.

    Parameters
    ----------
    dbfile : plantdb.commons.fsdb.core.File
        The ``File`` instance to save under given `path`.
    path : pathlib.Path or str
        The file path to use to save the `dbfile`.
    """
    b = dbfile.read_raw()
    path = Path(path)
    with path.open(mode="wb") as fh:
        fh.write(b)
    return

yes_no_abort_choice Link

yes_no_abort_choice(question, default=True, abort_allowed=True, default_abort=False)

Prompt the user with a Yes/No question that also supports aborting.

Parameters:

Name Type Description Default

question Link

str

The question to ask.

required

default Link

bool

The default answer when the user just presses Enter (default is True).

True

abort_allowed Link

bool

If True the user can abort the operation (default is True).

True

default_abort Link

bool

If True an empty input is interpreted as abort instead of the normal default; useful when abort is the most common action.

False

Returns:

Type Description
bool | None

True for a positive answer, False for a negative answer, or None if the user aborted.

Examples:

>>> from plantdb.commons.utils import yes_no_abort_choice
>>> yes_no_abort_choice("Delete all data?")
Delete all data? [Y/n/a]>
True
>>> yes_no_abort_choice("Delete all data?", default=False)
Delete all data? [y/N/a]>
False
>>> yes_no_abort_choice("Delete all data?", default=False, default_abort=True)
Delete all data? [y/N/a]>
None
Source code in plantdb/commons/utils.py
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def yes_no_abort_choice(
        question: str,
        default: bool = True,
        abort_allowed: bool = True,
        default_abort: bool = False,
) -> bool | None:
    """Prompt the user with a Yes/No question that also supports aborting.

    Parameters
    ----------
    question : str
        The question to ask.
    default : bool, optional
        The default answer when the user just presses *Enter* (default is ``True``).
    abort_allowed : bool, optional
        If ``True`` the user can abort the operation (default is ``True``).
    default_abort : bool, optional
        If ``True`` an empty input is interpreted as *abort* instead of the normal
        ``default``; useful when abort is the most common action.

    Returns
    -------
    bool | None
        ``True`` for a positive answer, ``False`` for a negative answer, or
        ``None`` if the user aborted.

    Examples
    --------
    >>> from plantdb.commons.utils import yes_no_abort_choice
    >>> yes_no_abort_choice("Delete all data?")
    Delete all data? [Y/n/a]>
    True
    >>> yes_no_abort_choice("Delete all data?", default=False)
    Delete all data? [y/N/a]>
    False
    >>> yes_no_abort_choice("Delete all data?", default=False, default_abort=True)
    Delete all data? [y/N/a]>
    None
    """
    # Map of valid inputs
    opt = {
        "": default,
        "yes": True,
        "y": True,
        "ye": True,
        "no": False,
        "n": False,
        "a": None,
        "abort": None,
    }

    # Build the prompt suffix
    choices = "[Y/n]" if default else "[y/N]"
    if abort_allowed:
        choices = choices.replace("]", "/a]")  # add abort option
        if default_abort:
            choices = choices.replace('/N', '/n')
            choices = choices.replace('/a', '/A')
    prompt = f"{question} {choices}> "

    while True:
        kbd = input(prompt).strip().lower()
        # Empty input handling
        if kbd == "":
            if default_abort and abort_allowed:
                return None
            return default
        # Direct match
        if kbd in opt:
            return opt[kbd]
        print("Please answer with 'yes', 'no', or 'a' (abort).")

yes_no_choice Link

yes_no_choice(question, default=True)

Raise a yes/no question with a default reply and wait for a valid reply from user.

Parameters:

Name Type Description Default

question Link

str

The question to ask.

required

default Link

bool

The default answer to the question.

True

Examples:

>>> from plantdb.commons.utils import yes_no_choice
>>> yes_no_choice("Is ROMI an awesome project?")
Is ROMI an awesome project? [Y/n]>
True
>>> yes_no_choice("I am your father!", default=False)
I am your father! [y/N]>
False
Source code in plantdb/commons/utils.py
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def yes_no_choice(question: str, default: bool = True) -> bool:
    """Raise a yes/no question with a default reply and wait for a valid reply from user.

    Parameters
    ----------
    question : str
        The question to ask.
    default : bool, optional
        The default answer to the question.

    Examples
    --------
    >>> from plantdb.commons.utils import yes_no_choice
    >>> yes_no_choice("Is ROMI an awesome project?")
    Is ROMI an awesome project? [Y/n]>
    True
    >>> yes_no_choice("I am your father!", default=False)
    I am your father! [y/N]>
    False
    """
    opt = {"": default, "yes": True, "y": True, "ye": True, "no": False, "n": False}
    default_choice = " [Y/n]" if default else " [y/N]"
    prompt = f"{question} {default_choice}"
    while True:
        kbd = input(prompt).strip().lower()
        # Empty input -> take the supplied default
        if kbd == "":
            return default
        if kbd in opt:
            return opt[kbd]
        print("Please answer with 'yes' or 'no'.")