Source code for tsad.base.task

import abc
import pandas as pd

from enum import Enum


[docs]class TaskStatus(Enum): """ # TaskStatus Класс TaskStatus является перечислением, содержащим возможные статусы задачи. """ UNKNOWN = "UNKNOWN" RUNNING = "RUNNING" SUCCEEDED = "SUCCEEDED" FAILED = "FAILED"
[docs]class TaskResult(abc.ABC): """ # Документация для класса TaskResult Класс TaskResult является абстрактным базовым классом, предназначенным для сохранения и отображения результатов задач. ### Методы: - save() -> str: абстрактный метод, возвращающий строку, содержащую сохраненные результаты задачи. - show() -> None: абстрактный метод, отображающий результаты задачи. """
[docs] def save(self) -> str: pass
[docs] @abc.abstractmethod def show(self) -> None: pass
[docs]class Task(abc.ABC): """ # Документация для класса Task Класс Task является абстрактным базовым классом для задач, которые могут быть выполнены на наборе данных. ### Атрибуты: - name: str: имя задачи. - status: TaskStatus: текущий статус задачи. ### Методы: - __init__(name: str | None = None) -> None: конструктор класса, инициализирующий атрибуты name и status. - fit_predict(df: pd.DataFrame) -> tuple[pd.DataFrame, TaskResult]: абстрактный метод, выполняющий обучение задачи на наборе данных и возвращающий результаты обучения вместе с обновленным набором данных. - predict(df: pd.DataFrame) -> tuple[pd.DataFrame, TaskResult]: абстрактный метод, выполняющий предсказание задачи на наборе данных и возвращающий результаты предсказания вместе с исходным набором данных. ### Пример использования: ```python class CustomTask(Task): def fit_predict(self, df: pd.DataFrame) -> tuple[pd.DataFrame, TaskResult]: # реализация обучения задачи result = TaskResult() # ... return df, result def predict(self, df: pd.DataFrame) -> tuple[pd.DataFrame, TaskResult]: # реализация предсказания задачи result = TaskResult() # ... return df, result task = CustomTask("Моя задача") df = pd.DataFrame(...) output_df, result = task.fit_predict(df) print(output_df) result.show() """ name: str status: TaskStatus def __init__(self, name: str | None = None): self.name = name
[docs] @abc.abstractmethod def fit_predict(self, df: pd.DataFrame) -> tuple[pd.DataFrame, TaskResult]: pass
[docs] @abc.abstractmethod def predict(self, df: pd.DataFrame) -> tuple[pd.DataFrame, TaskResult]: pass