File size: 4,592 Bytes
cf54850
 
47766e3
 
b854243
47766e3
b854243
cf54850
b854243
19aa24b
b854243
0c93f3d
 
eff3d67
19aa24b
 
 
 
 
 
 
4f52ddf
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
19aa24b
 
 
 
 
47766e3
 
6272dd1
c3457e0
6272dd1
 
 
b854243
 
 
 
 
47766e3
b854243
47766e3
b854243
 
 
 
 
 
 
 
47766e3
21f188b
b854243
 
 
19aa24b
b854243
 
19aa24b
b854243
 
19aa24b
b854243
 
19aa24b
b854243
 
19aa24b
b854243
 
19aa24b
b854243
 
19aa24b
b854243
 
19aa24b
b854243
 
19aa24b
b854243
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
19aa24b
 
b854243
cb54c81
cf54850
 
21f188b
b854243
 
 
 
 
 
0c93f3d
 
 
 
 
 
 
 
 
 
 
 
b854243
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
from __future__ import annotations

import os
import sys
import time
import traceback
from typing import Any

import gradio as gr
from fastapi import File, UploadFile
from fastapi.responses import RedirectResponse
from starlette.middleware import Middleware
from fastapi.middleware.cors import CORSMiddleware

from src.api.main import detect_image as api_detect_image
from src.api.main import detect_video as api_detect_video
from src.api.main import health as api_health
from src.api.main import health_models as api_health_models
from src.types import DetectionResponse


_default_unraisablehook = sys.unraisablehook


def _suppress_asyncio_fd_shutdown_noise(unraisable) -> None:
    exc = unraisable.exc_value
    if (
        isinstance(exc, ValueError)
        and str(exc) == "Invalid file descriptor: -1"
        and getattr(unraisable.object, "__name__", "") == "__del__"
    ):
        return
    _default_unraisablehook(unraisable)


sys.unraisablehook = _suppress_asyncio_fd_shutdown_noise


def _install_excepthook() -> None:
    def handle_exception(exc_type, exc_value, exc_tb):
        traceback.print_exception(exc_type, exc_value, exc_tb)

    sys.excepthook = handle_exception


def _normalize_gradio_env() -> None:
    os.environ["GRADIO_SSR_MODE"] = "False"
    os.environ.pop("GRADIO_NODE_SERVER_PORT", None)


def _build_demo() -> gr.Blocks:
    with gr.Blocks(title="GenAI-DeepDetect") as demo:
        gr.Markdown(
            """
            # GenAI-DeepDetect

            Gradio frontend with API routes attached to the same app for your external frontend.

            Available API endpoints:
            - `GET /api/health`
            - `GET /api/health/models`
            - `POST /api/detect/image`
            - `POST /api/detect/video`
            """
        )
    return demo


def _attach_api_routes(app: Any) -> None:
    async def health() -> dict:
        return await api_health()

    async def api_health_route() -> dict:
        return await health()

    async def health_models() -> dict[str, object]:
        return await api_health_models()

    async def api_health_models_route() -> dict[str, object]:
        return await health_models()

    async def detect_image(file: UploadFile = File(...)) -> DetectionResponse:
        return await api_detect_image(file)

    async def api_detect_image_route(file: UploadFile = File(...)) -> DetectionResponse:
        return await detect_image(file)

    async def detect_video(file: UploadFile = File(...)) -> DetectionResponse:
        return await api_detect_video(file)

    async def api_detect_video_route(file: UploadFile = File(...)) -> DetectionResponse:
        return await detect_video(file)

    async def gradio_compat_redirect() -> RedirectResponse:
        return RedirectResponse(url="/", status_code=307)

    app.add_api_route("/health", health, methods=["GET"])
    app.add_api_route("/api/health", api_health_route, methods=["GET"])
    app.add_api_route("/health/models", health_models, methods=["GET"])
    app.add_api_route("/api/health/models", api_health_models_route, methods=["GET"])
    app.add_api_route(
        "/detect/image",
        detect_image,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route(
        "/api/detect/image",
        api_detect_image_route,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route(
        "/detect/video",
        detect_video,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route(
        "/api/detect/video",
        api_detect_video_route,
        methods=["POST"],
        response_model=DetectionResponse,
    )
    app.add_api_route("/gradio", gradio_compat_redirect, methods=["GET"])


demo = _build_demo().queue()


if __name__ == "__main__":
    _install_excepthook()
    _normalize_gradio_env()
    try:
        app, _, _ = demo.launch(
            prevent_thread_lock=True,
            show_error=True,
            ssr_mode=False,
            app_kwargs={
                "docs_url": "/docs",
                "redoc_url": "/redoc",
                "middleware": [
                    Middleware(
                        CORSMiddleware,
                        allow_origins=["*"],
                        allow_methods=["*"],
                        allow_headers=["*"],
                    )
                ],
            },
        )
        _attach_api_routes(app)
        while True:
            time.sleep(60)
    except Exception:
        traceback.print_exc()
        sys.exit(1)