Разработка UI

Client.run

real_script module-attribute

real_script = 'app_client.py'

Client.app_client

selected module-attribute

selected = option_menu('Разделы', ['EDA', 'Обучение модели', 'Инференс'])

Client.eda_page

show_bar

show_bar(classes, counts)

Функция для построения столбчатых диаграмм.

Source code in Client/eda_page.py

def show_bar(classes: list[str], counts: list[int]) -> None:
    """Функция для построения столбчатых диаграмм."""
    plt.figure(figsize=(35, 20))
    plt.bar(classes, counts, color="#008080")
    y_pos = np.arange(len(classes))
    plt.xticks(y_pos, classes, rotation=90, ha="right", fontsize=30)
    plt.subplots_adjust(bottom=0.3)
    plt.ylabel("Количество изображений", fontsize=30)
    st.pyplot(plt)
    plt.close()

show_images

show_images(url_server)

Функция для отображения примеров изображений с каждого класса.

Source code in Client/eda_page.py

def show_images(url_server: str) -> None:
    """Функция для отображения примеров изображений с каждого класса."""
    st.subheader("Примеры изображений по классам")
    try:
        with st.spinner("Ожидаем загрузки изображений..."):
            response = requests.get(url_server + "dataset/samples", timeout=90)
            img = Image.open(BytesIO(response.content))
            st.image(img, caption="Загруженные изображения", use_container_width=True)
            logger.info("Изображения успешно загружены и отображены для клиента")

    except requests.exceptions.HTTPError as http_err:
        logger.error(f"HTTP ошибка при получении изображений с датасета: {http_err}")
        st.error("Ошибка получения изображений с датасета: Проверьте сервер.")

    except requests.exceptions.Timeout:
        st.error("Превышено время ожидания ответа от сервера.")
        logger.error("Превышено время ожидания ответа от сервера")

    except requests.exceptions.RequestException as req_err:
        logger.error(f"Ошибка сети при получении изображений с датасета: {req_err}")
        st.error("Ошибка получения изображений с датасета: Проверьте соединение.")

    except OSError as io_err:
        logger.error(f"Ошибка обработки изображения: {io_err}")
        st.error("Ошибка обработки изображения: Не удалось открыть изображение.")

show_bar_std_mean_rgb

show_bar_std_mean_rgb(rgb_df, cls)

Функция для отображения графика отклонений по каналам RGB для конкретного класса.

Source code in Client/eda_page.py

def show_bar_std_mean_rgb(rgb_df: DataFrame, cls: str) -> None:
    """Функция для отображения графика отклонений по каналам RGB для конкретного класса."""
    rows = rgb_df[rgb_df["class"] == cls].values
    mean_r = np.mean(rows[:, 2])
    mean_g = np.mean(rows[:, 3])
    mean_b = np.mean(rows[:, 4])
    std_r = np.std(rows[:, 5])
    std_g = np.std(rows[:, 6])
    std_b = np.std(rows[:, 7])

    means = [mean_r, mean_g, mean_b]
    stds = [std_r, std_g, std_b]
    channels = ["R", "G", "B"]

    fig = go.Figure()
    fig.add_trace(
        go.Bar(
            x=channels,
            y=means,
            error_y={"type": "data", "array": stds, "visible": True},
            marker={"color": ["red", "green", "blue"]},
            opacity=0.7,
        )
    )
    fig.update_layout(xaxis_title="Каналы", yaxis_title="Значение пикселей", width=800, height=400)

    st.plotly_chart(fig)
    logger.info(f"Отображение графика отклонений по каналам RGB для класса {cls}")

show_eda

show_eda(url_server)

Функция для отображения основных статистик датасета.

Source code in Client/eda_page.py

def show_eda(url_server: str) -> None:
    """Функция для отображения основных статистик датасета."""
    try:
        response = requests.get(url_server + "dataset/info", timeout=90)
        response_data = json.loads(response.text)
        dataset_info = DatasetInfo(**response_data)
        st.subheader("Основные статистики:")
        size_df = pd.DataFrame(dataset_info.sizes.rows, columns=dataset_info.sizes.columns)
        st.write(
            f"**Средний размер изображений**: "
            f"ширина: {round(size_df['width'].mean(), 0)}, "
            f"высота: {round(size_df['height'].mean(), 0)}"
        )
        logger.info("Вывод основных статистик для датасета")

        st.subheader("График распределения изображений по классам:")
        show_bar(list(dataset_info.classes.keys()), list(dataset_info.classes.values()))
        logger.info("Отображение графика распределения изображений по классам")

        if dataset_info.duplicates is not None:
            st.subheader("График распределения дубликатов по классам:")
            show_bar(list(dataset_info.duplicates.keys()), list(dataset_info.duplicates.values()))
            logger.info("Отображение графика распределения дубликатов по классам")
        else:
            st.write("**Дубликатов нет**")

        st.subheader("Среднее значение и стандартное отклонение по каналам (R, G, B)")
        rgb_df = pd.DataFrame(dataset_info.colors.rows, columns=dataset_info.colors.columns)
        classes = rgb_df["class"].unique()
        cls = st.selectbox("Выберите класс", classes)
        show_bar_std_mean_rgb(rgb_df, cls)

    except requests.exceptions.HTTPError as http_err:
        logger.error(f"HTTP ошибка при получении EDA данных: {http_err}")
        st.error("Ошибка получения EDA данных: Проверьте сервер.")

    except requests.exceptions.Timeout:
        st.error("Превышено время ожидания ответа от сервера.")
        logger.error("Превышено время ожидания ответа от сервера")

    except requests.exceptions.RequestException as req_err:
        logger.error(f"Ошибка сети при получении EDA данных: {req_err}")
        st.error("Ошибка получения EDA данных: Проверьте соединение.")

    except json.JSONDecodeError as json_err:
        logger.error(f"Ошибка декодирования JSON данных: {json_err}")
        st.error("Ошибка получения EDA данных: Неверный формат данных.")

eda_page

eda_page(url_server)

Функция для заполнения страницы с EDA.

Source code in Client/eda_page.py

def eda_page(url_server: str) -> None:
    """Функция для заполнения страницы с EDA."""
    st.header("EDA для датасета изображений")
    st.subheader("Загрузка данных")
    st.info("Пожалуйста, убедитесь, что ваш датасет соответствует следующим требованиям:")
    st.markdown(
        """
        - **Формат изображений**: JPEG.
        - **Аннотации**: метки классов должны быть представлены в отдельной структуре папок.
        """
    )
    uploaded_file = st.file_uploader("Выберите файл с датасетом", type=["zip"])
    if uploaded_file is not None:
        files = {"file": (uploaded_file.name, uploaded_file.getvalue(), uploaded_file.type)}
        with st.spinner("Ожидаем загрузки датасета на сервер..."):
            try:
                response = requests.post(url_server + "dataset/load", files=files, timeout=90)
                if response.status_code == 201:
                    st.session_state.uploaded_file = uploaded_file
                    st.success(f"Новый датасет {uploaded_file.name} успешно загружен на сервер")
                    logger.info("Датасет успешно загружен на сервер")
                    show_eda(url_server)
                    show_images(url_server)
                else:
                    logger.error(f"Произошла ошибка: {response.text}")
                    st.error(f"Произошла ошибка: {response.text}")

            except requests.exceptions.Timeout:
                st.error("Превышено время ожидания ответа от сервера.")
                logger.error("Превышено время ожидания ответа от сервера")

    elif "uploaded_file" in st.session_state and st.session_state.uploaded_file is not None:
        logger.info(f"На сервере уже есть датасет {st.session_state.uploaded_file.name}")
        st.subheader(f"**Датасет:** {st.session_state.uploaded_file.name}")
        show_eda(url_server)
        show_images(url_server)

Client.model_inference

make_prediction

make_prediction(url_server, files, use_probability)

Функция для получения предсказания на обученной модели.

Source code in Client/model_inference.py

def make_prediction(url_server: str, files: dict[str, tuple[str, bytes, str]], use_probability: bool) -> dict | None:
    """Функция для получения предсказания на обученной модели."""
    try:
        endpoint = "models/predict_proba" if use_probability else "models/predict"
        response = requests.post(f"{url_server}{endpoint}", files=files, timeout=90)
        response.raise_for_status()
        return response.json()

    except requests.exceptions.HTTPError as http_err:
        st.error(f"HTTP ошибка во время предсказания: {http_err}")
        logger.error(f"HTTP ошибка во время предсказания: {http_err}")
        return None

    except requests.exceptions.Timeout:
        st.error("Превышено время ожидания ответа от сервера.")
        logger.error("Превышено время ожидания ответа от сервера")
        return None

    except requests.exceptions.RequestException as req_err:
        st.error(f"Сетевая ошибка во время предсказания: {req_err}")
        logger.error(f"Сетевая ошибка во время предсказания: {req_err}")
        return None

    except json.JSONDecodeError as json_err:
        st.error(f"Ошибка декодирования JSON: {json_err}")
        logger.error(f"Ошибка декодирования JSON: {json_err}")
        return None

download_trained_model

download_trained_model(url_server, selected_model_info)

Функция загрузки обученной модели на сервер

Source code in Client/model_inference.py

def download_trained_model(url_server: str, selected_model_info: ModelInfo) -> bool:
    """Функция загрузки обученной модели на сервер"""
    with st.spinner("Загрузка модели для предсказания..."):
        try:
            logger.info(f"Началась загрузка модели {selected_model_info.name} для предсказания")

            load_model = LoadRequest(id=selected_model_info.id)
            load_json = load_model.model_dump()

            requests.post(f"{url_server}models/load", json=load_json, timeout=90)
            st.success(f"Модель {selected_model_info.name} успешно подготовлена для предсказания")
            logger.info(f"Модель {selected_model_info.name} успешно подготовлена для предсказания")
            return True

        except requests.exceptions.Timeout:
            st.error("Превышено время ожидания ответа от сервера.")
            logger.error("Превышено время ожидания ответа от сервера")
            return False

        except requests.exceptions.RequestException as e:
            st.error("Произошла ошибка при попытке загрузить модель. Проверьте соединение с сервером.")
            logger.exception(f"Ошибка получения ответа от сервера: {e}")
            return False

model_inference

model_inference(url_server)

Функция для получения предсказания на обученной модели.

Source code in Client/model_inference.py

def model_inference(url_server: str) -> None:
    """Функция для получения предсказания на обученной модели."""
    st.header("Инференс с использованием обученной модели")

    if "model_info_list" in st.session_state:
        model_info_list = st.session_state.model_info_list
        model_names = [model.name for model in model_info_list]
        selected_model_name = st.selectbox("Выберите модель", model_names)
        selected_model_info = next((model for model in model_info_list if model.name == selected_model_name), None)

        if download_trained_model(url_server, selected_model_info):
            uploaded_image = st.file_uploader("Загрузите изображение", type=["jpeg", "png", "jpg"])
            if uploaded_image is not None:
                logger.info("Изображение для предсказания успешно загружено")
                st.image(uploaded_image, caption="Загруженное изображение", use_container_width=True)

                files = {"file": (uploaded_image.name, uploaded_image.getvalue(), uploaded_image.type)}
                if selected_model_info.id == "baseline":
                    response_data = make_prediction(url_server, files, False)
                else:
                    response_data = make_prediction(
                        url_server, files, selected_model_info.hyperparameters["svc__probability"]
                    )

                if response_data:
                    if selected_model_info.id != "baseline" and selected_model_info.hyperparameters["svc__probability"]:
                        prediction_info = ProbabilityResponse(**response_data)
                        st.markdown(
                            f""":green-background[**Я думаю это {prediction_info.prediction}
                            с вероятностью {round(prediction_info.probability, 2) * 100} %**]"""
                        )
                        logger.info("Предсказание с вероятность выполнено успешно")
                    else:
                        prediction_info = PredictionResponse(**response_data)
                        st.markdown(f":green-background[**Я думаю это {prediction_info.prediction}**]")
                        logger.info("Предсказание без вероятности выполнено успешно")

Client.model_training_page

timeout_handler

timeout_handler()

Функция обработки стандартного таймаута

Source code in Client/model_training_page.py

def timeout_handler() -> None:
    """
    Функция обработки стандартного таймаута
    """
    st.error("Превышено время ожидания ответа от сервера.")
    logger.error("Превышено время ожидания ответа от сервера")

plt_learning_curve

plt_learning_curve(model_info_list)

Функция для отображения графика кривых обучения модели.

Source code in Client/model_training_page.py

def plt_learning_curve(model_info_list: list[ModelInfo]) -> None:
    """Функция для отображения графика кривых обучения модели."""
    plt.figure(figsize=(10, 6))
    colors = sns.color_palette("husl", len(model_info_list))

    for index, model in enumerate(model_info_list):

        train_scores_mean = np.mean(model.learning_curve.train_scores, axis=1)
        train_scores_std = np.std(model.learning_curve.train_scores, axis=1)

        test_scores_mean = np.mean(model.learning_curve.test_scores, axis=1)
        test_scores_std = np.std(model.learning_curve.test_scores, axis=1)

        plt.plot(
            model.learning_curve.train_sizes,
            train_scores_mean,
            marker="o",
            label=f"{model.name} - Тренировочная оценка",
            color=colors[index],
        )
        plt.fill_between(
            model.learning_curve.train_sizes,
            train_scores_mean - train_scores_std,
            train_scores_mean + train_scores_std,
            color=colors[index],
            alpha=0.2,
        )

        plt.plot(
            model.learning_curve.train_sizes,
            test_scores_mean,
            marker="o",
            label=f"{model.name} - Тестовая оценка",
            color=colors[index],
            linestyle="--",
        )

        plt.fill_between(
            model.learning_curve.train_sizes,
            test_scores_mean - test_scores_std,
            test_scores_mean + test_scores_std,
            color=colors[index],
            alpha=0.2,
        )

    plt.title("Кривые обучения")
    plt.xlabel("Размер обучающей выборки")
    plt.ylabel("f1-macro")
    plt.legend()
    plt.grid()
    st.pyplot(plt)
    plt.close()
    logger.info("Построен график кривых обучения модели")

change_models_learning_curve

change_models_learning_curve()

Функция выбора моделей, для которых будет постоен график кривых обучения.

Source code in Client/model_training_page.py

def change_models_learning_curve() -> None:
    """Функция выбора моделей, для которых будет постоен график кривых обучения."""
    if "model_info_list" in st.session_state:
        st.subheader("Построение графиков кривых обучения моделей")
        model_info_list = st.session_state.model_info_list
        model_learning_curve = []
        valid_models = []
        for model_info in model_info_list:
            if model_info.learning_curve is not None:
                valid_models.append(model_info)
                if st.checkbox(model_info.name, value=False):
                    model_learning_curve.append(model_info)

        if model_learning_curve:
            plt_learning_curve(model_learning_curve)
        elif valid_models:
            st.warning("Выберите хотя бы одну модель для отображения.")
    else:
        st.error("Список моделей не найден в состоянии сессии.")
        logger.error("Список моделей не найден в состоянии сессии.")

show_model_statistics

show_model_statistics(model_info)

Функция для отображения информации об обученной модели.

Source code in Client/model_training_page.py

def show_model_statistics(model_info: ModelInfo) -> None:
    """Функция для отображения информации об обученной модели."""

    st.subheader("Информация о модели")
    hyperparams_str = "".join([f"\n- **{key} =** {value}" for key, value in model_info.hyperparameters.items()])
    st.markdown(
        f"""
    **Название модели:** {model_info.name} <br>
    **Гиперпараметры:** {hyperparams_str}
    """,
        unsafe_allow_html=True,
    )

    logger.info("Для клиента отображена основная информация об обученной модели")

delete_model

delete_model(url_server, model_id)

Функция для удаления обученной модели.

Source code in Client/model_training_page.py

def delete_model(url_server: str, model_id: str) -> bool:
    """Функция для удаления обученной модели."""
    try:
        response = requests.delete(url_server + f"models/remove/{model_id}", timeout=90)
        if response.status_code == 200:
            return True

    except requests.exceptions.Timeout:
        timeout_handler()

    return False

delete_all_models

delete_all_models(url_server)

Функция для удаления всех обученных моделей.

Source code in Client/model_training_page.py

def delete_all_models(url_server: str) -> bool:
    """Функция для удаления всех обученных моделей."""
    try:
        response = requests.delete(url_server + "models/remove_all", timeout=90)
        if response.status_code == 200:
            return True

    except requests.exceptions.Timeout:
        timeout_handler()

    return False

get_models_list

get_models_list(url_server)

Функция для получения списка всех обученных моделей.

Source code in Client/model_training_page.py

def get_models_list(url_server: str) -> list[ModelInfo] | None:
    """Функция для получения списка всех обученных моделей."""
    try:
        with st.spinner("Загрузка списка моделей..."):
            logger.info("Загрузка списка моделей с сервера")
            model_info_list = []
            response = requests.get(url_server + "models/list_models", timeout=90)
            model_data = response.json()
            if not model_data:
                st.warning("Список моделей пуст.")
                logger.info("Список моделей пуст")
            else:
                for _, model_info in model_data.items():
                    model_info_list.append(ModelInfo(**model_info))

                st.session_state.model_info_list = model_info_list
                logger.info(f"Получен список обученных моделей {model_info_list}")
            return model_info_list

    except requests.exceptions.HTTPError as http_err:
        logger.error(f"HTTP ошибка при получении списка моделей: {http_err}")
        st.error("Ошибка получения списка моделей: Проверьте сервер.")
        return None

    except requests.exceptions.Timeout:
        timeout_handler()
        return None

    except requests.exceptions.RequestException as req_err:
        logger.error(f"Ошибка сети при получении списка моделей: {req_err}")
        st.error("Ошибка получения списка моделей: Проверьте соединение.")
        return None

    except json.JSONDecodeError as json_err:
        logger.error(f"Ошибка декодирования JSON: {json_err}")
        st.error("Ошибка при обработке данных сервера: Неверный формат ответа.")
        return None

show_models_list

show_models_list(url_server)

Функция для отображения на странице списка обученных моделей с возможностью их выбора и удаления.

Source code in Client/model_training_page.py

def show_models_list(url_server: str) -> None:
    """Функция для отображения на странице списка обученных моделей с возможностью их выбора и удаления."""
    st.subheader("Выбор существующих моделей")
    if "selected_model_name" not in st.session_state:
        st.session_state.selected_model_name = ""

    model_info_list = get_models_list(url_server)

    if model_info_list is not None:
        model_names = [model.name for model in model_info_list]
        selected_model_name = st.selectbox(
            "Выберите уже обученную модель",
            model_names,
            index=(
                model_names.index(st.session_state.selected_model_name)
                if st.session_state.selected_model_name in model_names
                else 0
            ),
        )

        selected_model_info = next((model for model in model_info_list if model.name == selected_model_name), None)
        show_model_statistics(selected_model_info)
        if selected_model_info.id != "baseline":
            if st.button(f"Удалить модель {selected_model_name}"):
                delete_model(url_server, selected_model_info.id)
                logger.info(f"Модель {selected_model_name} успешно удалена")
                st.rerun()

        if st.button("Удалить все модели"):
            delete_all_models(url_server)
            logger.info("Все обученные модели успешно удалены")
            st.rerun()

show_forms_create_model

show_forms_create_model(url_server)

Функция для отображения на странице формы подготовки модели для обучения.

Source code in Client/model_training_page.py

def show_forms_create_model(url_server: str) -> None:
    """Функция для отображения на странице формы подготовки модели для обучения."""
    st.subheader("Создание новой модели SVC и выбор гиперпараметров")

    name_model = st.text_input("Введите название модели")
    param_c = st.slider("Выберите параметр регуляризации:", 0.1, 30.0, 0.1)
    kernel = st.selectbox("Выберите ядро:", ["linear", "poly", "rbf", "sigmoid", "precomputed"])
    probability = st.toggle("Включить оценку вероятности")
    learning_curve = st.toggle("Включить learning_curve")

    fit_request_data = FitRequest(
        config={"svc__C": param_c, "svc__kernel": kernel, "svc__probability": probability},
        with_learning_curve=learning_curve,
        name=name_model,
    )

    fit_json = fit_request_data.model_dump()
    if st.button(":red[**Начать обучение модели**]"):
        with st.spinner("Обучение модели..."):
            logger.info(f"Обучение новой модели {name_model}")

            try:
                response = requests.post(url_server + "models/fit", json=fit_json, timeout=90)
                response_data = json.loads(response.text)
                model_info = ModelInfo(**response_data)
                st.session_state.selected_model_name = model_info.name
                logger.info(f"Модель {name_model} успешно обучена")
                st.rerun()

            except requests.exceptions.HTTPError as http_err:
                st.error("Ошибка сервера при обучении модели.")
                logger.error(f"HTTP ошибка: {http_err}")

            except requests.exceptions.Timeout:
                timeout_handler()

            except requests.exceptions.RequestException as req_err:
                st.error("Ошибка сети при обучении модели.")
                logger.error(f"Ошибка сети: {req_err}")

            except json.JSONDecodeError as json_err:
                st.error("Ошибка при обработке ответа сервера.")
                logger.error(f"Ошибка декодирования JSON: {json_err}")

model_training_page

model_training_page(url_server)

Функция для заполнения страницы с подготовкой модели для обучения.

Source code in Client/model_training_page.py

def model_training_page(url_server: str):
    """Функция для заполнения страницы с подготовкой модели для обучения."""
    show_models_list(url_server)
    change_models_learning_curve()
    show_forms_create_model(url_server)