Inherits NetworkObject, Generic, and DatasetType.

Public Member Functions
Dict[str, Any]	parameters (self)

Path	taskPath (self)

DatasetType	dataset (self)

bool	updateStatus (self, Optional[TaskRunStatus] status=None, Optional[str] message=None, bool notifyServer=True)

None	createMetrics (self, List[Metric] metrics)

bool	submitMetrics (self, Dict[str, Tuple[float, float]] metricValues)

None	submitOutput (self, str parameterName, Any value)

None	submitOutputs (self, Dict[str, Any] outputs)

bool	downloadTask (self)

Optional[Artifact]	createArtifact (self, Union[Path, str] localFilePath, str remoteFilePath, Optional[str] mimeType=None)

Self	run (cls, int taskId, Union[int, str] nodeId, Optional[str] name, Optional[str] description=None, Optional[List[Dict[str, Any]]] parameters=None)

Self	runLocal (cls, int projectId, bool saveSnapshot, Optional[str] name, Optional[str] description=None, Optional[List[Dict[str, Any]]] parameters=None, Optional[str] entryPoint=None)

str	generateEntityName (self)

Detailed Description

    Represents TaskRun entity from Coretex.ai

    Properties
    ----------
    datasetId : int
        id of dataset
    name : str
        name of TaskRun
    description : str
        description of TaskRun
    meta : Dict[str, Any]
        meta data of TaskRun
    status : TaskRunStatus
        status of TaskRun
    projectId : int
        id of Coretex Project
    projectName : str
        name of Coretex Project
    projectType : ProjectType
        appropriate project type
    taskId : int
        id of task
    taskName : str
        name of task
    createdById : str
        id of created TaskRun
    useCachedEnv : bool
        if True chached env will be used, otherwise new environment will be created

Definition at line 45 of file task_run.py.

Member Function Documentation

◆ createArtifact()

Optional[Artifact] coretex.entities.task_run.task_run.TaskRun.createArtifact	(		self,
		Union[Path, str]	localFilePath,
		str	remoteFilePath,
		Optional[str]	mimeType = `None`
	)

    Creates Artifact for the current TaskRun on Coretex.ai

    Parameters
    ----------
    localFilePath : Union[Path, str]
        local path of Artifact file
    remoteFilePath : str
        path of Artifact file on Coretex
    mimeType : Optional[str]
        mimeType (not required) if not passed guesMimeType() function is used

    Returns
    -------
    Optional[Artifact] -> if response is True returns Artifact object, None otherwise

Definition at line 447 of file task_run.py.

     ) -> Optional[Artifact]:
  
         """
             Creates Artifact for the current TaskRun on Coretex.ai
  
             Parameters
             ----------
             localFilePath : Union[Path, str]
                 local path of Artifact file
             remoteFilePath : str
                 path of Artifact file on Coretex
             mimeType : Optional[str]
                 mimeType (not required) if not passed guesMimeType() function is used
  
             Returns
             -------
             Optional[Artifact] -> if response is True returns Artifact object, None otherwise
         """
  
         return Artifact.create(self.id, localFilePath, remoteFilePath, mimeType)
  

◆ createMetrics()

None coretex.entities.task_run.task_run.TaskRun.createMetrics	(		self,
		List[Metric]	metrics
	)

    Creates specified metrics for the TaskRun

    Parameters
    ----------
    values : List[Metric]]
        List of Metric meta objects in this format
        Metric("name", "x_label", "x_type", "y_label", "y_type", "x_range", "y_range")

    Returns
    -------
    List[Metric] -> List of Metric objects

    Raises
    ------
    NetworkRequestError -> if the request failed

    Example
    -------
    >>> from coretex import ExecutingTaskRun, MetricType
    \b
    >>> metrics = ExecutingTaskRun.current().createMetrics([
            Metric.create("loss", "epoch", MetricType.int, "value", MetricType.float, None, [0, 100]),
            Metric.create("accuracy", "epoch", MetricType.int, "value", MetricType.float, None, [0, 100])
        ])
    >>> if len(metrics) == 0:
            print("Failed to create metrics")

Definition at line 270 of file task_run.py.

     def createMetrics(self, metrics: List[Metric]) -> None:
         """
             Creates specified metrics for the TaskRun
  
             Parameters
             ----------
             values : List[Metric]]
                 List of Metric meta objects in this format
                 Metric("name", "x_label", "x_type", "y_label", "y_type", "x_range", "y_range")
  
             Returns
             -------
             List[Metric] -> List of Metric objects
  
             Raises
             ------
             NetworkRequestError -> if the request failed
  
             Example
             -------
             >>> from coretex import ExecutingTaskRun, MetricType
             \b
             >>> metrics = ExecutingTaskRun.current().createMetrics([
                     Metric.create("loss", "epoch", MetricType.int, "value", MetricType.float, None, [0, 100]),
                     Metric.create("accuracy", "epoch", MetricType.int, "value", MetricType.float, None, [0, 100])
                 ])
             >>> if len(metrics) == 0:
                     print("Failed to create metrics")
         """
  
         parameters: Dict[str, Any] = {
             "experiment_id": self.id,
             "metrics": [metric.encode() for metric in metrics]
         }
  
         response = networkManager.post(f"{self._endpoint()}/metrics-meta", parameters)
         if response.hasFailed():
             raise NetworkRequestError(response, "Failed to create metrics")
  
         self.metrics.extend(metrics)
  

◆ dataset()

DatasetType coretex.entities.task_run.task_run.TaskRun.dataset ( self )

    Value of the parameter with name "dataset" assigned to this TaskRun

    Returns
    -------
    Dataset object if there was a parameter with name "dataset" entered when the TaskRun was started

    Raises
    ------
    ValueError -> if there is not parameter with name "dataset"

Definition at line 121 of file task_run.py.

     def dataset(self) -> DatasetType:
         """
             Value of the parameter with name "dataset" assigned to this TaskRun
  
             Returns
             -------
             Dataset object if there was a parameter with name "dataset" entered when the TaskRun was started
  
             Raises
             ------
             ValueError -> if there is not parameter with name "dataset"
         """
  
         dataset = self.parameters.get("dataset")
         if dataset is None:
             raise ValueError(f">> [Coretex] TaskRun \"{self.id}\" does not have a parameter named \"dataset\"")
  
         return dataset  # type: ignore
  

◆ downloadTask()

bool coretex.entities.task_run.task_run.TaskRun.downloadTask ( self )

    Downloads task snapshot linked to the TaskRun

    Returns
    -------
    bool -> True if task downloaded successfully, False if task download has failed

Definition at line 419 of file task_run.py.

     def downloadTask(self) -> bool:
         """
             Downloads task snapshot linked to the TaskRun
  
             Returns
             -------
             bool -> True if task downloaded successfully, False if task download has failed
         """
  
         params = {
             "model_queue_id": self.id
         }
  
         zipFilePath = f"{self.taskPath}.zip"
         response = networkManager.download(f"workspace/download", zipFilePath, params)
  
         if response.hasFailed():
             logging.getLogger("coretexpylib").info(">> [Coretex] Task download has failed")
             return False
  
         with ZipFile(zipFilePath) as zipFile:
             zipFile.extractall(self.taskPath)
  
         # remove zip file after extract
         os.unlink(zipFilePath)
  
         return not response.hasFailed()
  

◆ generateEntityName()

str coretex.entities.task_run.task_run.TaskRun.generateEntityName ( self )

    Combines the id and the name of the task run into a name for enitites like
    datasets or models.

    Returns
    -------
    str -> the generated name

Definition at line 657 of file task_run.py.

     def generateEntityName(self) -> str:
         """
             Combines the id and the name of the task run into a name for enitites like
             datasets or models.
  
             Returns
             -------
             str -> the generated name
         """
  
         name = f"{self.id}-{self.name}"
         return name[:50]

◆ parameters()

Dict[str, Any] coretex.entities.task_run.task_run.TaskRun.parameters ( self )

    Returns
    -------
    Dict[str, Any] -> Parameters for TaskRun

Definition at line 101 of file task_run.py.

     def parameters(self) -> Dict[str, Any]:
         """
             Returns
             -------
             Dict[str, Any] -> Parameters for TaskRun
         """
  
         return self.__parameters
  

◆ run()

Self coretex.entities.task_run.task_run.TaskRun.run	(		cls,
		int	taskId,
		Union[int, str]	nodeId,
		Optional[str]	name,
		Optional[str]	description = `None`,
		Optional[List[Dict[str, Any]]]	parameters = `None`
	)

    Schedules an TaskRun for execution on the specified
    Node on Coretex.ai

    Parameters
    ----------
    taskId : int
        id of task that is being used for starting TaskRun
    nodeId : Union[int, str]
        id of node that is being used for starting TaskRun
    name : Optional[str]
        name of TaskRun (not required)
    description : Optional[str]
        TaskRun description (not required)
    parameters : Optional[List[Dict[str, Any]]]
        list of parameters (not required)

    Returns
    -------
    Self -> TaskRun object

    Raises
    ------
    NetworkRequestError -> if the request failed

    Example
    -------
    >>> from coretex import TaskRun
    >>> from coretex.networking import NetworkRequestError
    \b
    >>> parameters = [
            {
                "name": "dataset",
                "description": "Dataset id that is used for fetching dataset from coretex.",
                "value": null,
                "data_type": "dataset",
                "required": true
            }
        ]
    \b
    >>> try:
            taskRun = TaskRun.run(
                taskId = 1023,
                nodeId = 23,
                name = "Dummy Custom TaskRun
                description = "Dummy description",
                parameters = parameters
            )

            print(f"Created TaskRun with name: {taskRun.name}")
    >>> except NetworkRequestError:
            print("Failed to create TaskRun")

Definition at line 505 of file task_run.py.

     ) -> Self:
  
         """
             Schedules an TaskRun for execution on the specified
             Node on Coretex.ai
  
             Parameters
             ----------
             taskId : int
                 id of task that is being used for starting TaskRun
             nodeId : Union[int, str]
                 id of node that is being used for starting TaskRun
             name : Optional[str]
                 name of TaskRun (not required)
             description : Optional[str]
                 TaskRun description (not required)
             parameters : Optional[List[Dict[str, Any]]]
                 list of parameters (not required)
  
             Returns
             -------
             Self -> TaskRun object
  
             Raises
             ------
             NetworkRequestError -> if the request failed
  
             Example
             -------
             >>> from coretex import TaskRun
             >>> from coretex.networking import NetworkRequestError
             \b
             >>> parameters = [
                     {
                         "name": "dataset",
                         "description": "Dataset id that is used for fetching dataset from coretex.",
                         "value": null,
                         "data_type": "dataset",
                         "required": true
                     }
                 ]
             \b
             >>> try:
                     taskRun = TaskRun.run(
                         taskId = 1023,
                         nodeId = 23,
                         name = "Dummy Custom TaskRun
                         description = "Dummy description",
                         parameters = parameters
                     )
  
                     print(f"Created TaskRun with name: {taskRun.name}")
             >>> except NetworkRequestError:
                     print("Failed to create TaskRun")
         """
  
         if isinstance(nodeId, int):
             nodeId = str(nodeId)
  
         if parameters is None:
             parameters = []
  
         response = networkManager.post("run", {
             "sub_project_id": taskId,
             "service_id": nodeId,
             "name": name,
             "description": description,
             "execution_type": ExecutionType.remote.value,
             "parameters": parameters
         })
  
         if response.hasFailed():
             raise NetworkRequestError(response, "Failed to create TaskRun")
  
         responseJson = response.getJson(dict)
         return cls.fetchById(responseJson["experiment_ids"][0])
  

◆ runLocal()

Self coretex.entities.task_run.task_run.TaskRun.runLocal	(		cls,
		int	projectId,
		bool	saveSnapshot,
		Optional[str]	name,
		Optional[str]	description = `None`,
		Optional[List[Dict[str, Any]]]	parameters = `None`,
		Optional[str]	entryPoint = `None`
	)

    Creates TaskRun on Coretex.ai with the provided parameters,
    which will be run on the same machine which created it immidiately
    after running the entry point file of the task

    Parameters
    ----------
    projectId : int
        id of project that is being used for starting TaskRun
    saveSnapshot : bool
        true if snapshot of local files will be saved to Coretex
    name : Optional[str]
        name of TaskRun (not required)
    description : Optional[str]
        TaskRun description (not required)
    parameters : Optional[List[Dict[str, Any]]]
        list of parameters (not required)
    entryPoint : Optional[str]
        relative path to the script inside of the project

    Returns
    -------
    Self -> TaskRun object

    Raises
    ------
    NetworkRequestError -> if the request failed

Definition at line 590 of file task_run.py.

     ) -> Self:
  
         """
             Creates TaskRun on Coretex.ai with the provided parameters,
             which will be run on the same machine which created it immidiately
             after running the entry point file of the task
  
             Parameters
             ----------
             projectId : int
                 id of project that is being used for starting TaskRun
             saveSnapshot : bool
                 true if snapshot of local files will be saved to Coretex
             name : Optional[str]
                 name of TaskRun (not required)
             description : Optional[str]
                 TaskRun description (not required)
             parameters : Optional[List[Dict[str, Any]]]
                 list of parameters (not required)
             entryPoint : Optional[str]
                 relative path to the script inside of the project
  
             Returns
             -------
             Self -> TaskRun object
  
             Raises
             ------
             NetworkRequestError -> if the request failed
         """
  
         if parameters is None:
             parameters = []
  
         params = {
             "project_id": projectId,
             "name": name,
             "description": description,
             "execution_type": ExecutionType.local.value,
             "parameters": json.dumps(parameters)
         }
  
         if entryPoint is not None:
             params["entry_point"] = entryPoint
  
         # Create snapshot
         if saveSnapshot:
             files = [FileData.createFromPath("file", createSnapshot())]
         else:
             files = None
  
         response = networkManager.formData("run", params, files)
         if response.hasFailed():
             raise NetworkRequestError(response, "Failed to create TaskRun")
  
         responseJson = response.getJson(dict)
         return cls.fetchById(responseJson["experiment_ids"][0])
  
  

◆ submitMetrics()

bool coretex.entities.task_run.task_run.TaskRun.submitMetrics	(		self,
		Dict[str, Tuple[float, float]]	metricValues
	)

    Appends metric values for the provided metrics

    Parameters
    ----------
    metricValues : Dict[str, Tuple[float, float]]
        Values of metrics in this format {"name": x, y}

    Example
    -------
    >>> from coretex import ExecutingTaskRun
    \b
    >>> result = ExecutingTaskRun.current().submitMetrics({
            "loss": (epoch, logs["loss"]),
            "accuracy": (epoch, logs["accuracy"]),
        })
    >>> print(result)
    True

Definition at line 311 of file task_run.py.

     def submitMetrics(self, metricValues: Dict[str, Tuple[float, float]]) -> bool:
         """
             Appends metric values for the provided metrics
  
             Parameters
             ----------
             metricValues : Dict[str, Tuple[float, float]]
                 Values of metrics in this format {"name": x, y}
  
             Example
             -------
             >>> from coretex import ExecutingTaskRun
             \b
             >>> result = ExecutingTaskRun.current().submitMetrics({
                     "loss": (epoch, logs["loss"]),
                     "accuracy": (epoch, logs["accuracy"]),
                 })
             >>> print(result)
             True
         """
  
         metrics = [{
             "timestamp": value[0] if self._isInterval(key) else time.time(),
             "metric": key,
             "x": value[0],
             "y": value[1]
         } for key, value in metricValues.items()]
  
         parameters: Dict[str, Any] = {
             "experiment_id": self.id,
             "metrics": metrics
         }
  
         response = networkManager.post(f"{self._endpoint()}/metrics", parameters)
         return not response.hasFailed()
  

◆ submitOutput()

None coretex.entities.task_run.task_run.TaskRun.submitOutput	(		self,
		str	parameterName,
		Any	value
	)

    Submit an output of this task to act as a parameter in tasks
    downstream in the Workflow

    Parameters
    ----------
    parameterName : str
        name of the parameter
    value : Any
        value that will be sent (id of the object will be sent
        in case Coretex objects like CustomDataset, Model etc. is submited)

    Raises
    ------
    NetworkRequestError -> if the request failed

    Example
    -------
    >>> from coretex import TaskRun
    \b
    >>> ExecutingTaskRun.current().submitOutputParameter(
            parameterName = "outputDataset"
            value = outputDataset
        )

Definition at line 347 of file task_run.py.

     def submitOutput(self, parameterName: str, value: Any) -> None:
         """
             Submit an output of this task to act as a parameter in tasks
             downstream in the Workflow
  
             Parameters
             ----------
             parameterName : str
                 name of the parameter
             value : Any
                 value that will be sent (id of the object will be sent
                 in case Coretex objects like CustomDataset, Model etc. is submited)
  
             Raises
             ------
             NetworkRequestError -> if the request failed
  
             Example
             -------
             >>> from coretex import TaskRun
             \b
             >>> ExecutingTaskRun.current().submitOutputParameter(
                     parameterName = "outputDataset"
                     value = outputDataset
                 )
         """
  
         self.submitOutputs({parameterName: value})
  

◆ submitOutputs()

None coretex.entities.task_run.task_run.TaskRun.submitOutputs	(		self,
		Dict[str, Any]	outputs
	)

    Submit multiple outputs of this task to act as parameters in tasks
    downstream in the Workflow

    Parameters
    ----------
    outputs : Dict[str, Any]
        dictionary with outputs, with key being the name of the parameter
        (id of the object will be sent in case Coretex objects like
        CustomDataset, Model etc. is submited)

    Raises
    ------
    NetworkRequestError -> if the request failed

    Example
    -------
    >>> from coretex import TaskRun
    \b
    >>> result = ExecutingTaskRun.current().submitOutputParameters({
            "outputDataset": outputDataset,
            "numbers": 123
        })

Definition at line 376 of file task_run.py.

     def submitOutputs(self, outputs: Dict[str, Any]) -> None:
         """
             Submit multiple outputs of this task to act as parameters in tasks
             downstream in the Workflow
  
             Parameters
             ----------
             outputs : Dict[str, Any]
                 dictionary with outputs, with key being the name of the parameter
                 (id of the object will be sent in case Coretex objects like
                 CustomDataset, Model etc. is submited)
  
             Raises
             ------
             NetworkRequestError -> if the request failed
  
             Example
             -------
             >>> from coretex import TaskRun
             \b
             >>> result = ExecutingTaskRun.current().submitOutputParameters({
                     "outputDataset": outputDataset,
                     "numbers": 123
                 })
         """
  
         outputParameters: List[Dict[str, Any]] = []
  
         for key, value in outputs.items():
             if isinstance(value, NetworkObject):
                 value = value.id
  
             outputParameters.append({key: value})
  
         parameters: Dict[str, Any] = {
             "id": self.id,
             "parameters": outputParameters
         }
  
         response = networkManager.post(f"{self._endpoint()}/output-parameter", parameters)
         if response.hasFailed():
             raise NetworkRequestError(response, ">> [Coretex] Failed to submit outputs")
  

◆ taskPath()

Path coretex.entities.task_run.task_run.TaskRun.taskPath ( self )

    Returns
    -------
    Path -> Path for TaskRun

Definition at line 111 of file task_run.py.

     def taskPath(self) -> Path:
         """
             Returns
             -------
             Path -> Path for TaskRun
         """
  
         return folder_manager.temp / str(self.id)
  

◆ updateStatus()

bool coretex.entities.task_run.task_run.TaskRun.updateStatus	(		self,
		Optional[TaskRunStatus]	status = `None`,
		Optional[str]	message = `None`,
		bool	notifyServer = `True`
	)

    Updates TaskRun status, if message parameter is None
    default message value will be used\n
    Some TaskRun statuses do not have default message

    Parameters
    ----------
    status : Optional[TaskRunStatus]
        Status to which the TaskRun will be updated to
    message : Optional[str]
        Descriptive message for TaskRun status, it is diplayed
        when the status is hovered on the Coretex Web App
    notifyServer : bool
        if True update request will be sent to Coretex.ai

    Example
    -------
    >>> from coretex import ExecutingTaskRun, TaskRunStatus
    \b
    >>> ExecutingTaskRun.current().updateStatus(
            TaskRunStatus.completedWithSuccess
        )
    True

Definition at line 208 of file task_run.py.

     ) -> bool:
  
         """
             Updates TaskRun status, if message parameter is None
             default message value will be used\n
             Some TaskRun statuses do not have default message
  
             Parameters
             ----------
             status : Optional[TaskRunStatus]
                 Status to which the TaskRun will be updated to
             message : Optional[str]
                 Descriptive message for TaskRun status, it is diplayed
                 when the status is hovered on the Coretex Web App
             notifyServer : bool
                 if True update request will be sent to Coretex.ai
  
             Example
             -------
             >>> from coretex import ExecutingTaskRun, TaskRunStatus
             \b
             >>> ExecutingTaskRun.current().updateStatus(
                     TaskRunStatus.completedWithSuccess
                 )
             True
         """
  
         if status is not None:
             self.status = status
  
         if notifyServer:
             if status is not None and message is None:
                 message = status.defaultMessage
  
             parameters: Dict[str, Any] = {
                 "id": self.id
             }
  
             if status is not None:
                 parameters["status"] = status
  
             if message is not None:
                 parameters["message"] = message
  
             # TODO: Should API rename this too?
             endpoint = f"{self._endpoint()}/job-status-update"
             response = networkManager.post(endpoint, parameters)
  
             if response.hasFailed():
                 logging.getLogger("coretexpylib").error(">> [Coretex] Error while updating TaskRun status")
             elif status is not None:
                 logging.getLogger("coretexpylib").info(f">> [Coretex] Updated Task Run status to \"{status.name}\"")
  
             return not response.hasFailed()
  
         return True
  

The documentation for this class was generated from the following file:

coretex/entities/task_run/task_run.py

Public Member Functions