PIbd-42_SSPR/main.py

import pandas as pd
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pyDOE3 import pbdesign, lhs

from db.csv_to_db import csv_to_db
from network.routes import (ch_experimentdb_experiment_data_router, experiment_data_router,
                            experiment_parameters_router, experiment_category_router)
from network.routes import load_parameters_router, recycling_parameters_router
from network.schemas import *
from new_experiment_planner import run_experiment
from new_experiment_planner_pyDOE3 import scale_design, scale_design_lhs, round_by_index

app = FastAPI()

app.include_router(ch_experimentdb_experiment_data_router.router,
                   prefix="/ch_experimentdb_experiment_data",
                   tags=["ch_experimentdb_experiment_data"])
app.include_router(experiment_category_router.router,
                   prefix="/experiment_category",
                   tags=["experiment_category"])
app.include_router(experiment_data_router.router,
                   prefix="/experiment_data",
                   tags=["experiment_data"])
app.include_router(experiment_parameters_router.router,
                   prefix="/experiment_parameters",
                   tags=["experiment_parameters"])
app.include_router(load_parameters_router.router,
                   prefix="/load_parameters",
                   tags=["load_parameters"])
app.include_router(recycling_parameters_router.router,
                   prefix="/recycling_parameters",
                   tags=["recycling_parameters"])


# Эндпоинт для запуска эксперимента
@app.post("/run_experiment/")
def run_experiment_api(params: ExperimentParameters):
    try:
        # Вызываем функцию run_experiment с параметрами
        run_experiment(
            params.outer_blades_count,
            params.outer_blades_length,
            params.outer_blades_angle,
            params.middle_blades_count,
            params.load,
            params.recycling
        )
        return {"status": "success", "message": "Experiment started successfully."}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")


# эндпоинт инициализации бд из csv файлов
@app.get('/init_db_data')
async def init_db_data(background_tasks: BackgroundTasks):
    try:
        background_tasks.add_task(csv_to_db)
        return {"status": "success", "message": "Инициализация БД запущена в фоне"}
    except Exception as e:
        print(str(e.with_traceback()))
        raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")

# Пример запроса
# {
#   "param_ranges": {
#     "outer_blades_count": [12, 48],
#     "outer_blades_length": [44, 107.5],
#     "outer_blades_angle": [30, 75],
#     "middle_blades_count": [9, 36],
#     "load": [315, 465],
#     "recycling_level": [0, 20],
#     "oxidizer_temp": [471, 493]
#   },
#   "count_exp": 1440,
#   "round_rules": [0, 1, 1, 0, 1, 1, 1]
# }

@app.post("/pyDOE3_screening_design")
def generate_screening_design(request: ExperimentParametersPyDOE3) -> List[Dict[str, float]]:
    param_ranges = request.param_ranges

    # Создаем screening design и масштабируем его
    num_factors = len(param_ranges)
    screening_design = pbdesign(num_factors)
    scaled_screening_design = scale_design(screening_design, param_ranges)

    # Преобразуем в DataFrame и возвращаем результат
    df_screening = pd.DataFrame(scaled_screening_design, columns=param_ranges.keys())
    return df_screening.to_dict(orient="records")


@app.post("/pyDOE3_lhs_design")
def generate_lhs_design(request: ExperimentParametersPyDOE3) -> List[Dict[str, float]]:
    param_ranges = request.param_ranges
    count_exp = request.count_exp
    round_rules = request.round_rules

    # Создаем lhs design и масштабируем его
    num_factors = len(param_ranges)
    lhs_samples = lhs(num_factors, samples=count_exp)
    scaled_lhs_samples = scale_design_lhs(lhs_samples, param_ranges)

    # Округляем значения
    round_scaled_lhs_samples = round_by_index(scaled_lhs_samples, round_rules)

    # Преобразуем в DataFrame и возвращаем результат
    df_lhs = pd.DataFrame(round_scaled_lhs_samples, columns=param_ranges.keys())
    return df_lhs.to_dict(orient="records")