Разработка API

Информация о API сервиса

Backend.app.api.v1.api_route

models module-attribute

models = {}

active_model module-attribute

active_model = {'model': None, 'info': None}

dataset_info module-attribute

dataset_info = DatasetInfo(is_empty=True, classes={}, duplicates={}, sizes=TableModel(columns=[], rows=[]), colors=TableModel(columns=[], rows=[]))

router_models module-attribute

router_models = APIRouter(prefix='/api/v1/models')

router_dataset module-attribute

router_dataset = APIRouter(prefix='/api/v1/dataset')

load_dataset async

load_dataset(file)

Загрузка датасета. На вход должен подаваться архив, содержащий папки с изображениями классов.

Source code in Backend/app/api/v1/api_route.py

@router_dataset.post(
    "/load",
    response_model=Annotated[DatasetInfo, "Информация о датасете"],
    status_code=HTTPStatus.CREATED,
    description="Загрузка датасета",
)
async def load_dataset(
    file: Annotated[UploadFile, File(..., description="Архив с классами изображений")],
) -> DatasetInfo:
    """
    Загрузка датасета.
    На вход должен подаваться архив, содержащий папки с изображениями классов.
    """
    if file.filename.lower().endswith(".zip") is False:
        logger.exception("Неверный формат файла. Должен загружаться zip-архив!")
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST, detail="Неверный формат файла. Должен загружаться zip-архив!"
        )
    try:
        archive = await file.read()
        # разархивировал картинки
        preprocess_archive(archive)
        # удалил прошлое превью, если было
        remove_preview()
        dataset_info.classes = classes_info()
        dataset_info.duplicates = duplicates_info()
        dataset_info.sizes = TableModel(rows=sizes_info(), columns=["class", "name", "width", "height"])
        dataset_info.colors = TableModel(
            rows=colors_info(), columns=["class", "name", "mean_R", "mean_G", "mean_B", "std_R", "std_G", "std_B"]
        )
        dataset_info.is_empty = False
        return dataset_info
    except Exception as e:
        logger.error(str(e))
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e)) from e

get_dataset_info async

get_dataset_info()

Получение информации о датасете. Возвращается количество изображений в каждом классе, дубли, таблица размеров и цветов.

Source code in Backend/app/api/v1/api_route.py

@router_dataset.get(
    "/info",
    response_model=Annotated[DatasetInfo, "Информация о датасете"],
    status_code=HTTPStatus.OK,
    description="Получение информации о датасете",
)
async def get_dataset_info() -> DatasetInfo:
    """
    Получение информации о датасете.
    Возвращается количество изображений в каждом классе, дубли, таблица размеров и цветов.
    """
    dataset_uploaded = check_dataset_uploaded()
    if not dataset_uploaded:
        message = "Нет загруженного набора данных!"
        logger.info(message)
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=message)
    if dataset_info.is_empty:
        dataset_info.classes = classes_info()
        dataset_info.duplicates = duplicates_info()
        dataset_info.sizes = TableModel(rows=sizes_info(), columns=["class", "name", "width", "height"])
        dataset_info.colors = TableModel(
            rows=colors_info(), columns=["class", "name", "mean_R", "mean_G", "mean_B", "std_R", "std_G", "std_B"]
        )
        dataset_info.is_empty = False
    return dataset_info

dataset_samples async

dataset_samples()

Возвращает картинку с примерами изображений по каждому классу

Source code in Backend/app/api/v1/api_route.py

@router_dataset.get(
    "/samples",
    response_class=Annotated[StreamingResponse, "Пример с изображениями"],
    status_code=HTTPStatus.OK,
    description="Изображения из классов",
)
async def dataset_samples() -> StreamingResponse:
    """
    Возвращает картинку с примерами изображений по каждому классу
    """
    dataset_uploaded = check_dataset_uploaded()
    if not dataset_uploaded:
        logger.exception("Нет загруженного набора данных!")
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Нет загруженного набора данных!")
    try:
        buffer = preview_dataset(3)
        return StreamingResponse(buffer, media_type="image/png")
    except Exception as e:
        logger.error(str(e))
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e)) from e

fit async

fit(request)

Обучение модели. По истечении 10 секунд обучение прерывается. Есть возможность дополнительно получить кривую обучения, указав with_learning_curve=True Также для обучения модели передаются гиперпараметры вида pca__ и svc__

Source code in Backend/app/api/v1/api_route.py

@router_models.post(
    "/fit",
    response_model=Annotated[ModelInfo, "Информация об обученной модели"],
    status_code=HTTPStatus.CREATED,
    description="Обучение модели",
)
async def fit(request: Annotated[FitRequest, "Параметры для обучения модели"]) -> ModelInfo:
    """
    Обучение модели. По истечении 10 секунд обучение прерывается.
    Есть возможность дополнительно получить кривую обучения, указав `with_learning_curve=True`
    Также для обучения модели передаются гиперпараметры вида `pca__` и `svc__`
    """
    dataset_uploaded = check_dataset_uploaded()
    if not dataset_uploaded:
        logger.exception("Нет загруженного набора данных!")
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Нет загруженного набора данных!")
    manager = Manager()
    model_manager = manager.dict()
    try:
        preprocess_dataset((64, 64))
        new_model = create_model(request.config)
        images, labels = load_colored_images_and_labels()

        stratified_cv = StratifiedKFold(n_splits=3)
        curve = None
        if request.with_learning_curve:
            train_sizes, train_scores, test_scores = learning_curve(
                new_model,
                images,
                labels,
                cv=stratified_cv,
                scoring="f1_macro",
                shuffle=True,
                random_state=42,
                train_sizes=[0.3, 0.6, 0.9],
                error_score=0,
            )
            curve = LearningCurveInfo(test_scores=test_scores, train_scores=train_scores, train_sizes=train_sizes)

        model_id = str(uuid4())
        process = Process(target=train_model, args=(new_model, images, labels, model_id, model_manager))
        process.start()
        # Через 10 сек. обучение прервется, т.к. считается долгим
        process.join(timeout=10)
        if process.is_alive():
            process.terminate()
            raise HTTPException(status_code=HTTPStatus.REQUEST_TIMEOUT, detail="Время обучения модели истекло")

        new_model = model_manager.get(model_id, None)
        if new_model is None:
            raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Ошибка обучения модели!")

        models[model_id] = {
            "id": model_id,
            "model": new_model,
            "type": ModelType.custom,
            "name": request.name,
            "hyperparameters": request.config,
            "learning_curve": curve,
        }
        return ModelInfo(
            name=request.name, id=model_id, type=ModelType.custom, hyperparameters=request.config, learning_curve=curve
        )
    except Exception as e:
        logger.error(str(e))
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e)) from e

predict async

predict(file)

Предсказание изображенного фрукта или овоща

Source code in Backend/app/api/v1/api_route.py

@router_models.post(
    "/predict",
    response_model=Annotated[PredictionResponse, "Предсказание"],
    status_code=HTTPStatus.OK,
    description="Предсказание класса",
)
async def predict(
    file: Annotated[UploadFile, File(..., description="Файл изображения для предсказания")],
) -> PredictionResponse:
    """
    Предсказание изображенного фрукта или овоща
    """
    if active_model.get("model", None) is None:
        logger.exception("Не выбрана модель")
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Не выбрана модель")
    try:
        contents = await file.read()
        image = preprocess_image(contents)
        model: Pipeline = active_model["model"]
        return PredictionResponse(prediction=model.predict([image])[0])
    except Exception as e:
        logger.error(str(e))
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e)) from e

predict_proba async

predict_proba(file)

Предсказание с вероятностью изображенного фрукта или овоща

Source code in Backend/app/api/v1/api_route.py

@router_models.post(
    "/predict_proba",
    response_model=Annotated[ProbabilityResponse, "Предсказание с вероятностью"],
    status_code=HTTPStatus.OK,
    description="Предсказание класса с вероятностью",
)
async def predict_proba(
    file: Annotated[UploadFile, File(..., description="Файл изображения для предсказания")],
) -> ProbabilityResponse:
    """
    Предсказание с вероятностью изображенного фрукта или овоща
    """
    if active_model.get("model", None) is None:
        logger.exception("Не выбрана модель")
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Не выбрана модель")
    try:
        contents = await file.read()
        image = preprocess_image(contents)
        model: Pipeline = active_model["model"]
        probability = max(model.predict_proba([image])[0])
        prediction = model.predict([image])[0]
        return ProbabilityResponse(prediction=prediction, probability=probability)
    except Exception as e:
        logger.error(str(e))
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e)) from e

load async

load(request)

Загрузка пользовательской модели для использования. Модель загружается по id

Source code in Backend/app/api/v1/api_route.py

@router_models.post(
    "/load",
    response_model=Annotated[ModelInfo, "Информация о выбранной модели"],
    status_code=HTTPStatus.OK,
    description="Загрузка одной из моделей",
)
async def load(request: LoadRequest) -> ModelInfo:
    """
    Загрузка пользовательской модели для использования. Модель загружается по id
    """
    if request.id not in models:
        logger.exception(f"Модель '{request.id}' не была найдена!")
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail=f"Модель '{request.id}' не была найдена!",
        )
    model = models[request.id]
    info = ModelInfo(
        id=request.id,
        hyperparameters=model["hyperparameters"],
        type=model["type"],
        learning_curve=model["learning_curve"],
        name=model["name"],
    )
    active_model["model"] = model["model"]
    active_model["info"] = info
    return info

unload async

unload()

Выгрузка модели. Если модель была выгружена, то предсказания не будут работать пока не загрузят новую модель

Source code in Backend/app/api/v1/api_route.py

@router_models.post(
    "/unload",
    response_model=Annotated[ApiResponse, "Сообщение о выгрузке модели"],
    status_code=HTTPStatus.OK,
    description="Выгрузка модели из памяти",
)
async def unload() -> ApiResponse:
    """
    Выгрузка модели.
    Если модель была выгружена, то предсказания не будут работать пока не загрузят новую модель
    """
    active_model["model"] = None
    active_model["info"] = None
    return ApiResponse(message="Модель выгружена из памяти")

list_models async

list_models()

Возврат списка всех доступных моделей

Source code in Backend/app/api/v1/api_route.py

@router_models.get(
    "/list_models",
    response_model=Annotated[dict[str, ModelInfo], "Информация о моделях на сервере"],
    status_code=HTTPStatus.OK,
    description="Получение списка моделей",
)
async def list_models() -> dict[str, ModelInfo]:
    """
    Возврат списка всех доступных моделей
    """
    return {
        model_id: ModelInfo(
            id=model_id,
            type=model["type"],
            hyperparameters=model["hyperparameters"],
            learning_curve=model["learning_curve"],
            name=model["name"],
        )
        for model_id, model in models.items()
    }

model_info async

model_info(model_id)

Возвращает информацию по модели с указанным id. В информацию входит: - id - тип модели (baseline/custom) - гиперпараметры (какие были использованы при обучении) - кривая обучения (если получалась при обучении) - пользовательское название модели

Source code in Backend/app/api/v1/api_route.py

@router_models.get(
    "/info/{model_id}",
    response_model=Annotated[ModelInfo, "Информация о модели"],
    status_code=HTTPStatus.OK,
    description="Получение информации о модели",
)
async def model_info(model_id: Annotated[str, "Id модели"]) -> ModelInfo:
    """
    Возвращает информацию по модели с указанным id.
    В информацию входит:
    - id
    - тип модели (baseline/custom)
    - гиперпараметры (какие были использованы при обучении)
    - кривая обучения (если получалась при обучении)
    - пользовательское название модели
    """
    if model_id not in models:
        logger.exception(f"Модель '{model_id}' не была найдена!")
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST,
            detail=f"Модель '{model_id}' не была найдена!",
        )
    model = models[model_id]
    return ModelInfo(
        id=model["id"],
        type=model["type"],
        hyperparameters=model["hyperparameters"],
        learning_curve=model["learning_curve"],
        name=model["name"],
    )

remove async

remove(model_id)

Удалит модель из памяти, ее больше нельзя будет загрузить для работы.

Source code in Backend/app/api/v1/api_route.py

@router_models.delete(
    "/remove/{model_id}",
    response_model=Annotated[dict[str, ModelInfo], "Оставшиеся модели"],
    status_code=HTTPStatus.OK,
    description="Удаление модели",
)
async def remove(model_id: Annotated[str, "Id модели, которую нужно удалить"]) -> dict[str, ModelInfo]:
    """
    Удалит модель из памяти, ее больше нельзя будет загрузить для работы.
    """
    if model_id not in models:
        logger.exception(f"Нет модели с id '{model_id}'")
        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail=f"Нет модели с id '{model_id}'")
    if model_id == "baseline":
        logger.exception("Нельзя удалить baseline модель!")
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail="Нельзя удалить baseline модель!")
    del models[model_id]
    return {
        model_id: ModelInfo(
            id=model_id,
            type=model["type"],
            hyperparameters=model["hyperparameters"],
            learning_curve=model["learning_curve"],
            name=model["name"],
        )
        for model_id, model in models.items()
    }

remove_all async

remove_all()

Удалит все пользовательские модели, бейзлайн модель не удаляется

Source code in Backend/app/api/v1/api_route.py

@router_models.delete(
    "/remove_all",
    response_model=Annotated[dict[str, ModelInfo], "Baseline модели"],
    status_code=HTTPStatus.OK,
    description="Удаление пользовательских моделей",
)
async def remove_all() -> dict[str, ModelInfo]:
    """
    Удалит все пользовательские модели, бейзлайн модель не удаляется
    """
    for model_id, model_item in models.items():
        if model_item["type"] != ModelType.baseline:
            del models[model_id]
    return {
        model_id: ModelInfo(
            id=model_id,
            type=model["type"],
            hyperparameters=model["hyperparameters"],
            learning_curve=model["learning_curve"],
            name=model["name"],
        )
        for model_id, model in models.items()
    }

Pydantic модели для API

Backend.app.api.models

ApiResponse

Bases: BaseModel

Ответ сервера на запрос с клиента

message instance-attribute

message

data class-attribute instance-attribute

data = None

PredictionResponse

Bases: BaseModel

Предсказание класса изображения

prediction instance-attribute

prediction

ProbabilityResponse

Bases: PredictionResponse

Предсказание класса изображения с вероятностью

probability instance-attribute

probability

prediction instance-attribute

prediction

ModelType

Bases: Enum

Типы моделей

baseline class-attribute instance-attribute

baseline = 'baseline'

custom class-attribute instance-attribute

custom = 'custom'

LearningCurveInfo

Bases: BaseModel

Информация о кривой обучения модели

train_sizes instance-attribute

train_sizes

train_scores instance-attribute

train_scores

test_scores instance-attribute

test_scores

ModelInfo

Bases: BaseModel

Информация об обученной модели

id instance-attribute

id

hyperparameters instance-attribute

hyperparameters

type instance-attribute

type

learning_curve instance-attribute

learning_curve

name instance-attribute

name

LoadRequest

Bases: BaseModel

Запрос на загрузку обученной модели

id instance-attribute

id

FitRequest

Bases: BaseModel

Запрос на обучение модели с указанием названия модели, гиперпараметров и сохранения кривой обучения

config instance-attribute

config

with_learning_curve instance-attribute

with_learning_curve

name instance-attribute

name

TableModel

Bases: BaseModel

Модель таблицы, содержащая ее столбцы и строки

columns instance-attribute

columns

rows instance-attribute

rows

DatasetInfo

Bases: BaseModel

Информация о датасете: - количестве изображений в каждом классе - дубликаты в каждом классе - таблица размеров - таблица цветов - загружен датасет или нет

classes instance-attribute

classes

duplicates instance-attribute

duplicates

sizes instance-attribute

sizes

colors instance-attribute

colors

is_empty instance-attribute

is_empty