service_gradio.py

import gradio as gr
import os
from core.analysis_pipeline.analysis_pipeline import AnalysisPipeline
from utils.image_utils import read_image_file, check_image_file
from utils.log_init import logger
from utils.web_utils import is_port_usable
from conf.service_args import project_root, service_config

# &#26816;&#26597;&#31471;&#21475;
gradio_port = 19192
assert is_port_usable(gradio_port), "ImageAnalysisService &#20998;&#26512; gradio &#31471;&#21475;&#34987;&#21344;&#29992;"

# &#26500;&#24314;&#20998;&#26512;&#27169;&#22359;
logger.info("&#21021;&#22987;&#20998;&#26512;&#27169;&#22359;")
image_analysis_pipeline = AnalysisPipeline()
logger.info("&#20998;&#26512;&#27169;&#22359;&#21021;&#22987;&#21270;&#25104;&#21151;")

# &#26500;&#24314;&#20219;&#21153;&#21517;&#21015;&#34920;
analysis_tasks = list(service_config["ModuleSwitch"].keys())


def call_analysis(image_data, task_name_):
    return image_analysis_pipeline.analysis_image(image_data, [task_name_])


def call_image_dir_analysis(image_dir_, task_name_):
    analysis_res = ""
    images = [os.path.join(image_dir_, one_image) for one_image in os.listdir(image_dir_)]
    for image_path in images:
        if not check_image_file(image_path):
            continue
        image_data = read_image_file(image_path)
        res_info = image_analysis_pipeline.analysis_image(image_data, [task_name_])
        task_res = res_info["data"][task_name_]
        analysis_res += f"&#22270;&#20687;:{os.path.basename(image_path)}, &#20998;&#26512;&#32467;&#26524;:{task_res} \n \n"
        yield analysis_res


if __name__ == "__main__":
    with gr.Blocks() as gradio_demo:
        with gr.Tab("&#22270;&#20687;&#20998;&#26512;"):
            with gr.Row():
                task_name = gr.Dropdown(analysis_tasks,
                                        value="ocr",
                                        label="&#20219;&#21153;&#31867;&#22411;",
                                        info="&#35831;&#36873;&#25321;&#20998;&#26512;&#20219;&#21153;&#31867;&#22411;")
                image = gr.Image(label="&#22270;&#20687;&#25991;&#20214;", image_mode="RGB")
                output = gr.JSON(label="&#35782;&#21035;&#32467;&#26524;")
            btn = gr.Button("&#20998;&#26512;")
            btn.click(fn=call_analysis, inputs=[image, task_name], outputs=output)
            gr.Examples(
                examples=os.path.join(project_root, "tests/core/seal/image"),
                inputs=[image],
                outputs=output,
                fn=call_analysis
            )
        with gr.Tab("&#26412;&#22320;&#25209;&#37327;&#20998;&#26512;"):
            with gr.Row():
                task_name = gr.Dropdown(analysis_tasks,
                                        value="ocr",
                                        label="&#20219;&#21153;&#31867;&#22411;",
                                        info="&#35831;&#36873;&#25321;&#20998;&#26512;&#20219;&#21153;&#31867;&#22411;")
                image_dir = gr.Text(label="&#26412;&#22320;&#22270;&#20687;&#25991;&#20214;&#22841;", info="&#35831;&#27880;&#24847; &#35813;&#20998;&#26512;&#20165;&#25903;&#25345;&#26412;&#22320;&#25209;&#37327;&#20998;&#26512;&#65292;&#36828;&#31243;&#35831;&#20174;&#25509;&#21475;&#35843;&#29992;")
                output = gr.Textbox(label="&#35782;&#21035;&#32467;&#26524;")
            btn = gr.Button("&#20998;&#26512;")
            btn.click(fn=call_image_dir_analysis, inputs=[image_dir, task_name], outputs=output)

    gradio_demo.launch(show_error=True, server_name="0.0.0.0", server_port=gradio_port)