from typing import List, Optional, Dict, Union
from .plot import Plot
from .data import Data
from .file import File
from .document import Document
from ._private._field_api import FieldAPI
from ._private._processing_api import ProcessingAPI
from ._private._well_dto import PlotPropertiesDto
from ._private._user_task_dto import UserTaskItemDto
from ._private.dto_converters._plot_dto_converter import PlotDtoConverter
from ._private.dto_converters._user_task_dto_converter import UserTaskDtoConverter
from .field_data_types_catalog import FieldDataTypesCatalog
[docs]
class UserTask:
""" User task object.
Presents a KAPPA Automate user task information object.
"""
[docs]
def __init__(self, field_id: str, well_id: str, custom_workflow_id: Optional[str], user_task_id: str, name: str, user_task_instance_id: str, data_types_catalog: FieldDataTypesCatalog,
outputs: List[Data], plots: List[Plot], files: List[File], field_api: FieldAPI, processing_api: ProcessingAPI, plot_dto_converter: PlotDtoConverter, user_task_dto_converter: UserTaskDtoConverter):
self.__field_id: str = field_id
self.__well_id: str = well_id
self.__custom_workflow_id: Optional[str] = custom_workflow_id
self.__id: str = user_task_id
self.__name: str = name
self.__user_task_instance_id: str = user_task_instance_id
self.__outputs: List[Data] = outputs
self.__field_api: FieldAPI = field_api
self.__processing_api: ProcessingAPI = processing_api
self.__plots: List[Plot] = plots
self.__data_types_catalog: FieldDataTypesCatalog = data_types_catalog
self.__plot_dto_converter: PlotDtoConverter = plot_dto_converter
self.__user_task_dto_converter: UserTaskDtoConverter = user_task_dto_converter
self.__user_task_definition_id: Optional[str] = None
self.__user_task_definition_name: Optional[str] = None
self.__inputs: Optional[List[Dict[str, Union[bool, str, float]]]] = None
self.__user_task_definition: Optional[UserTaskItemDto] = None
self.__files: List[File] = files
[docs]
def create_plot(self, plot_name: str, pane_name: Optional[str] = None, square_log_cycles: bool = False, stacked_bars: bool = False,
x_label: str = "", is_x_log: bool = False, labels: Optional[List[str]] = None) -> Plot:
"""
Create a Kappa Automate Plot instance under the user task
Parameters
----------
plot_name:
Name of plot
pane_name:
Name of the pane
square_log_cycles:
Whether or not use Square log cycles
stacked_bars:
Whether or not use stacked bars
x_label:
x label
is_x_log:
Whether or not use logarithmic scale for x values
labels:
Add some labels to the plot
Returns
-------
:class:`Plot`:
The new Plot instance created under the task.
"""
if labels is None:
labels = list()
plot_properties = PlotPropertiesDto(name=plot_name, pane_name=pane_name, square_log_cycles=square_log_cycles, stacked_bars=stacked_bars,
x_label=x_label, is_x_log=is_x_log, labels=labels)
plot_instance_dto = self.__plot_dto_converter.get_plot_instance_dto(plot_properties)
plot_id, plot_name = self.__field_api.create_plot(self.__field_id, self.__well_id, plot_instance_dto, self.__id)
return Plot(self.__field_id, self.__well_id, plot_id, plot_name, self.__field_api, self.__data_types_catalog, self.__plot_dto_converter)
@property
def id(self) -> str:
""" Gets the id of the :class:`UserTask` object."""
return self.__id
@property
def custom_workflow_id(self) -> Optional[str]:
""" Get the custom workflow id of the :class:`UserTask` object."""
return self.__custom_workflow_id
@property
def name(self) -> str:
""" Gets the name of the :class:`UserTask`."""
return self.__name
@property
def outputs(self) -> List[Data]:
""" Gets the list of outputs of the :class:`UserTask`."""
return self.__outputs
@property
def plots(self) -> List[Plot]:
""" Gets the list of the plot instances of the :class:`UserTask`."""
self.__plots = [Plot(self.__field_id, self.__well_id, str(x.id), x.name, self.__field_api, self.__data_types_catalog, self.__plot_dto_converter) for x in self.__field_api.get_user_task_dto(self.__field_id, self.__well_id, self.__id).customPlots]
return self.__plots
@property
def files(self) -> List[File]:
""" Gets the list of the files under this :class:`UserTask`."""
self.__files = [self.__user_task_dto_converter.build_file_from_file_dto(None, file, None) for file in self.__field_api.get_user_task_dto(self.__field_id, self.__well_id, self.__id).files]
return self.__files
@property
def documents(self) -> List[Document]:
""" Gets the list of the documents under this :class:`UserTask`."""
document_list = list()
for file in self.files:
document = file.as_kw_document()
if document is not None:
document_list.append(document)
return document_list
@property
def user_task_instance_id(self) -> str:
""" Gets the user task id (processing service) of the :class:`UserTask` object."""
return self.__user_task_instance_id
@property
def user_task_definition_name(self) -> str:
""" Gets the user task definition name of the :class:`UserTask` object."""
if self.__user_task_definition_name is None:
self.__user_task_definition_name = self.__get_user_task_definition_dto().userTaskDefinitionName
return self.__user_task_definition_name
[docs]
def rename(self, new_name: str) -> None:
""" Rename the User Task"""
dto = {'name': new_name}
self.__name = self.__field_api.rename_user_task(self.__field_id, self.__well_id, self.__id, dto)
[docs]
def delete(self) -> None:
""" Delete the User Task"""
self.__field_api.delete_user_task(self.__field_id, self.__well_id, self.__id)
def __get_user_task_definition_dto(self) -> UserTaskItemDto:
if self.__user_task_definition is None:
self.__user_task_definition = self.__processing_api.get_user_task_dto(self.__user_task_instance_id)
return self.__user_task_definition
@property
def inputs(self) -> List[Dict[str, Union[bool, str, float]]]:
"""
Retrieves the user task inputs.
Returns:
List[Dict[str, Union[bool, str, float]]]: A list of dictionaries, each containing the name and value of a user task input.
The value can be of type bool, str, or float.
"""
if self.__inputs is None:
self.__inputs = [{"name": key, "value": value.currentValue} for key, value in self.__get_user_task_definition_dto().inputs.items()]
return self.__inputs
