python-service/app/main.py - main - BELEGO/GIS_APPLICATION

python-service/app/main.py
182 строки7 KB
HACKER_LAPTOP
1. Updated the unity7 method, removed the black background from the photo. 2.The create_full_mask method is now synchronous
10 фев 2026, 12:19
10 фев 2026, 12:192c43aee
from fastapi import FastAPI, Query, BackgroundTasks, Response, HTTPException
from fastapi.responses import FileResponse
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse

from ViewMasks import inspect_tif_folder
from unity762 import process_geotiff_folder
from unity7 import combine_and_visualize_tiff_bands_by_path
from create_3masks import process_3masks
from create_voting_masks import make_voting_masks
import os
from Models.GeoImage import GeoImage
from database import SessionLocal

app = FastAPI()

origins = [
    "http://localhost:5173",  # фронт
    "http://127.0.0.1:5173",  # фронт
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,       # какие источники разрешены
    allow_credentials=True,
    allow_methods=["*"],         # GET, POST, …
    allow_headers=["*"],         # любые заголовки
)

@app.get("/process-folder/")
async def process_folder(path: str = Query(..., description="Путь к папке с TIFF изображениями")):
    """Обрабатывает папку с TIFF изображениями и возвращает статус"""
    try:
        # Запуск обработки
        output_file = combine_and_visualize_tiff_bands_by_path(path)
    except Exception as e:
        # Возвращаем JSON с ошибкой
        return JSONResponse(
            status_code=500,
            content={
                "status": "error",
                "message": str(e),
                "detail": "Ошибка при обработке изображения"
            }
        )

    if not os.path.exists(output_file):
        # Возвращаем JSON с ошибкой
        return JSONResponse(
            status_code=500,
            content={
                "status": "error",
                "message": "Файл результата не найден",
                "output_file": output_file
            }
        )

    # ВОЗВРАЩАЕМ JSON
    return JSONResponse(
        status_code=200,
        content={
            "status": "success",
            "message": "Обработка завершена успешно",
            "output_file": output_file,
            "product_id": os.path.basename(output_file).split('_')[0] if '_' in os.path.basename(output_file) else None
        }
    )
@app.get("/combine-bands/")
async def combine_bands(
    folder_path: str = Query(..., description="Путь к папке с TIFF файлами"),
    target_bands: str = Query("B1,B2,B3,B4,B5,B6,B7", description="Каналы для объединения через запятую"),
    visualize_rgb_bands: str = Query("B7,B6,B2", description="Каналы для визуализации RGB через запятую")
):
    try:
        target_bands_list = tuple(target_bands.split(","))
        visualize_rgb_list = tuple(visualize_rgb_bands.split(","))
        output_file = combine_and_visualize_tiff_bands_by_path(
            relative_folder_path=folder_path,
            target_bands=target_bands_list,
            visualize_rgb_bands=visualize_rgb_list
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    if not os.path.exists(output_file):
        raise HTTPException(status_code=500, detail="Файл результата не найден")

    return FileResponse(output_file, media_type="image/tiff", filename=os.path.basename(output_file))

@app.get("/create-3masks/")
async def detect_fires(folder_path: str = Query(..., description="Путь к папке с изображением")):
    try:
        result = process_3masks(folder_path)
        return {"status": "success", "message": "Маски сформированы успешно"}
    except RuntimeError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Ошибка при детекции пожаров: {str(e)}")

@app.get("/create-voting-masks/")
async def create_voting_masks(folder_path: str = Query(..., description="Путь к папке с изображением")):
    try:
        result = make_voting_masks(folder_path)

        if result["status"] == "exists":
            raise HTTPException(status_code=400, detail="Маска (Voting) уже существуют")

        if result["status"] == "no_fire":
            return {"status": "ok", "message": "Обработка завершена, но пожары не обнаружены", "fire": False}

        return {"status": "success", "message": "Voting маски успешно созданы", "fire": result["fire"]}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/create-full-mask/")
async def create_full_mask(folder_path: str):
    """
    Синхронное создание маски
    Ожидание: ~2 минуты
    """
    try:
        # 1. Создаем 3 маски
        process_3masks(folder_path)

        # 2. Создаем voting-маску
        result = make_voting_masks(folder_path)

        if result["status"] == "exists":
            print("Маска (Voting) уже существует")
            # Убедимся, что fire - булево значение, а не None
            result["fire"] = False if result.get("fire") is None else result["fire"]
        elif result["status"] == "no_fire":
            print("Обработка завершена, но пожары не обнаружены")
        else:
            print("3 маски и Voting маска успешно созданы, fire:", result["fire"])

        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/get-mask/{product_id}")
def get_mask(product_id: str):
    """
    Возвращает PNG маску из столбца Mask для изображения с product_id
    """
    with SessionLocal() as session:  # открываем сессию
        geo_image = session.query(GeoImage).filter(
            GeoImage.ProductId == product_id  # точное соответствие
        ).first()

        if not geo_image or not geo_image.Mask:
            raise HTTPException(status_code=404, detail="Маска не найдена")

        return Response(content=geo_image.Mask, media_type="image/png")

@app.get("/get-mask/{entity_id}")
def get_mask(entity_id: str):
    """
    Возвращает PNG маску для entity_id
    """
    with SessionLocal() as session:
        # Находим GeoImage по entity_id
        geo_image = session.query(GeoImage).filter(
            GeoImage.EntityId == entity_id  # изменено на EntityId
        ).first()

        if not geo_image or not geo_image.Mask:
            raise HTTPException(status_code=404, detail="Маска не найдена")

        return Response(content=geo_image.Mask, media_type="image/png")

@app.get("/inspect")
def inspect(folder_path: str = Query(..., description="Путь к папке с .tif файлами")):
    """
    Вызывает функцию визуального осмотра tif-файлов в указанной папке.
    """
    try:
        inspect_tif_folder(folder_path)
        return {"status": "OK", "message": f"Обработка папки {folder_path} завершена."}
    except Exception as e:
        return {"status": "error", "detail": str(e)}