flint.prefect.common.utils
==========================

.. py:module:: flint.prefect.common.utils

.. autoapi-nested-parse::

   Common prefect related utilities that can be used between flows.



Attributes
----------

.. autoapisummary::

   flint.prefect.common.utils.C
   flint.prefect.common.utils.P
   flint.prefect.common.utils.R
   flint.prefect.common.utils.SUPPORTED_IMAGE_TYPES
   flint.prefect.common.utils.T
   flint.prefect.common.utils.task_create_beam_summary
   flint.prefect.common.utils.task_create_field_summary
   flint.prefect.common.utils.task_get_fits_cube_from_paths
   flint.prefect.common.utils.task_rename_linear_to_stokes
   flint.prefect.common.utils.task_update_field_summary


Functions
---------

.. autoapisummary::

   flint.prefect.common.utils.task_archive_sbid
   flint.prefect.common.utils.task_create_object
   flint.prefect.common.utils.task_flatten
   flint.prefect.common.utils.task_get_attributes
   flint.prefect.common.utils.task_getattr
   flint.prefect.common.utils.task_sorted
   flint.prefect.common.utils.task_update_with_options
   flint.prefect.common.utils.task_zip_list_of_list
   flint.prefect.common.utils.upload_image_as_artifact


Module Contents
---------------

.. py:function:: 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

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

   :param science_folder_path: Path that contains the imaged produced
   :type science_folder_path: Path
   :param archive_path: Location to create and store the tar ball at. If None no tarball is created. Defaults to None.
   :type archive_path: Optional[Path], optional
   :param copy_path: Location to copy selected files into. If None no files are copied. Defaults to None.
   :type copy_path: Optional[Path], optional
   :param max_round: 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.
   :type max_round: Optional[int], optional
   :param update_archive_options: Additional options to provide to ArchiveOptions. Defaults to None.
   :type update_archive_options: Optional[Dict[str, Any]], optional

   :returns: The science folder files were copied from
   :rtype: Path


.. py:function:: task_create_object(object: C, **kwargs) -> C

.. py:function:: task_flatten(to_flatten: list[list[T]]) -> list[T]

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


   :param to_flatten: Input list of lists to flatten
   :type to_flatten: List[List[T]]

   :returns: Flattened form of input
   :rtype: List[T]


.. py:function:: task_get_attributes(item: Any, attribute: str) -> Any

   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.

   :param item: The item that has the input class or structure
   :type item: Any
   :param attribute: The attribute to extract
   :type attribute: str

   :returns: Value of the requested attribute
   :rtype: Any


.. py:function:: task_getattr(item: object, attribute: str, /) -> Any

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

   :param item: The item that has the input class or structure
   :type item: Any
   :param attribute: The attribute to extract
   :type attribute: str

   :returns: Value of the requested attribute
   :rtype: Any


.. py:function:: task_sorted(iterable: Iterable[T], /, *, key: Any = None, reverse: bool = False) -> list[T]

.. py:function:: task_update_with_options(input_object: T, **kwargs) -> T

   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.

   :param input_object: The object that has an `.with_options` method that will be updated
   :type input_object: T

   :returns: The updated object
   :rtype: T


.. py:function:: task_zip_list_of_list(list_of_list: list[list[T]]) -> list[tuple[T, Ellipsis]]

.. py:function:: upload_image_as_artifact(image_path: pathlib.Path, description: str | None = None) -> uuid.UUID | None

   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.

   :param image_path: Path to the image to upload
   :type image_path: Path
   :param description: A description passed to the markdown artifact. Defaults to None.
   :type description: Optional[str], optional

   :returns: Generated UUID of the registered artifact. If the upload fails ``None`` is returned.
   :rtype: UUID | None


.. py:data:: C

.. py:data:: P

.. py:data:: R

.. py:data:: SUPPORTED_IMAGE_TYPES
   :value: ('png',)


.. py:data:: T

.. py:data:: task_create_beam_summary
   :value: None


.. py:data:: task_create_field_summary
   :value: None


.. py:data:: task_get_fits_cube_from_paths
   :value: None


.. py:data:: task_rename_linear_to_stokes
   :value: None


.. py:data:: task_update_field_summary
   :value: None