flint.prefect.common.utils

Common prefect related utilities that can be used between flows.

Attributes

C

P

R

SUPPORTED_IMAGE_TYPES

T

task_create_beam_summary

task_create_field_summary

task_get_fits_cube_from_paths

task_rename_linear_to_stokes

task_update_field_summary

Functions

task_archive_sbid(→ pathlib.Path)

Create a tarball of files, or copy files, from a processing folder.

task_create_object(→ C)

task_flatten(→ list[T])

Will flatten a list of lists into a single list. This

task_get_attributes(→ Any)

Retrieve an attribute from an input instance of a class or structure.

task_getattr(→ Any)

Retrieve an attribute from an input instance of a class or structure.

task_sorted(→ list[T])

task_update_with_options(→ T)

Updated any object via its .with_options() interface.

task_zip_list_of_list(→ list[tuple[T, Ellipsis]])

upload_image_as_artifact(→ uuid.UUID | None)

Create and submit a markdown artifact tracked by prefect for an

Module Contents

flint.prefect.common.utils.task_archive_sbid(science_folder_path: pathlib.Path, archive_path: pathlib.Path | None = None, copy_path: pathlib.Path | None = None, max_round: int | None = None, update_archive_options: dict[str, Any] | None = None) → pathlib.Path[source]

Create a tarball of files, or copy files, from a processing folder.

Parameters:

  • science_folder_path (Path) – Path that contains the imaged produced

  • archive_path (Optional[Path], optional) – Location to create and store the tar ball at. If None no tarball is created. Defaults to None.

  • copy_path (Optional[Path], optional) – Location to copy selected files into. If None no files are copied. Defaults to None.

  • max_round (Optional[int], optional) – The last self-calibration round performed. If provided some files form this round are copied (assuming wsclean imaging). If None, the default file patterns in ArchiveOptions are used. Defaults to None.

  • update_archive_options (Optional[Dict[str, Any]], optional) – Additional options to provide to ArchiveOptions. Defaults to None.

Returns:

The science folder files were copied from

Return type:

Path

flint.prefect.common.utils.task_create_object(object: C, **kwargs) → C[source]
flint.prefect.common.utils.task_flatten(to_flatten: list[list[T]]) → list[T][source]

Will flatten a list of lists into a single list. This is useful for when a task-descorated function returns a list.

Parameters:

to_flatten (List[List[T]]) – Input list of lists to flatten

Returns:

Flattened form of input

Return type:

List[T]

flint.prefect.common.utils.task_get_attributes(item: Any, attribute: str) → Any[source]

Retrieve an attribute from an input instance of a class or structure.

This is intended to be used when dealing with a prefect future object that has yet to be evaluated or is otherwise not immediately accessible.

Parameters:

  • item (Any) – The item that has the input class or structure

  • attribute (str) – The attribute to extract

Returns:

Value of the requested attribute

Return type:

Any

flint.prefect.common.utils.task_getattr(item: object, attribute: str, /) → Any[source]

Retrieve an attribute from an input instance of a class or structure.

Parameters:

  • item (Any) – The item that has the input class or structure

  • attribute (str) – The attribute to extract

Returns:

Value of the requested attribute

Return type:

Any

flint.prefect.common.utils.task_sorted(iterable: Iterable[T], /, *, key: Any = None, reverse: bool = False) → list[T][source]
flint.prefect.common.utils.task_update_with_options(input_object: T, **kwargs) → T[source]

Updated any object via its .with_options() interface.

All key-word arguments other than input_object are passed through to that input_object`s `.with_options() method.

Parameters:

input_object (T) – The object that has an .with_options method that will be updated

Returns:

The updated object

Return type:

T

flint.prefect.common.utils.task_zip_list_of_list(list_of_list: list[list[T]]) → list[tuple[T, Ellipsis]][source]
flint.prefect.common.utils.upload_image_as_artifact(image_path: pathlib.Path, description: str | None = None) → uuid.UUID | None[source]

Create and submit a markdown artifact tracked by prefect for an input image. Currently supporting png formatted images.

The input image is converted to a base64 encoding, and embedded directly within the markdown string. Therefore, be mindful of the image size as this is tracked in the postgres database.

Parameters:

  • image_path (Path) – Path to the image to upload

  • description (Optional[str], optional) – A description passed to the markdown artifact. Defaults to None.

Returns:

Generated UUID of the registered artifact. If the upload fails None is returned.

Return type:

UUID | None

flint.prefect.common.utils.C[source]
flint.prefect.common.utils.P[source]
flint.prefect.common.utils.R[source]
flint.prefect.common.utils.SUPPORTED_IMAGE_TYPES = ('png',)[source]
flint.prefect.common.utils.T[source]
flint.prefect.common.utils.task_create_beam_summary = None[source]
flint.prefect.common.utils.task_create_field_summary = None[source]
flint.prefect.common.utils.task_get_fits_cube_from_paths = None[source]
flint.prefect.common.utils.task_rename_linear_to_stokes = None[source]
flint.prefect.common.utils.task_update_field_summary = None[source]