ai_special_exam/app5.py

import re
import threading
import time
from flask import Flask, jsonify,send_from_directory
from basket_cleaning_detect import start_basket_cleaning_detection,init_basket_cleaning_detection
from globals import inference_thread, stop_event,redis_client


app = Flask(__name__)


# Define the /wearing_detection endpoint

@app.route('/basket_cleaning_detection', methods=['GET'])
def basket_cleaning_detection():#开启平台搭设检测
    global inference_thread#当全局变量需要重新赋值时，需要用global关键字声明

    if inference_thread is None or not inference_thread.is_alive():#防止重复开启检测服务
        #redis_client.set("log_in_flag",'False')

        stop_event.clear()

        start_events = []#给每个线程一个事件，让我知道某个线程是否开始检测
        inference_thread = threading.Thread(target=start_basket_cleaning_detection,args=(start_events,))
        inference_thread.start()
        
            
        app.logger.info('start_platform_setup_detection')
        init_basket_cleaning_detection()


        # 等待所有YOLO线程开始检测
        for event in start_events:
            event.wait()

        return jsonify({"status": "SUCCESS"}), 200
    
    else:
        app.logger.info("reset_detection already running")   
        return jsonify({"status": "ALREADY_RUNNING"}), 200    

@app.route('/basket_cleaning_status', methods=['GET']) 
def basket_cleaning_status():#获取平台搭设状态状态
    if not redis_client.exists('basket_cleaning_order'):#平台搭设步骤还没有一个完成
        app.logger.info('basket_cleaning_order is none')

        return jsonify({"status": "NONE"}), 200
    
    else:

        basket_cleaning_order = redis_client.lrange("basket_cleaning_order", 0, -1)

        json_array = []
        for value in basket_cleaning_order:
            match = re.search(r'basket_step_(\d+)', value)
            step_number = match.group(1)
            json_object = {"step": step_number, "image": redis_client.get(f"basket_step_{step_number}")}
            json_array.append(json_object) 

        return jsonify({"status": "SUCCESS","data":json_array}), 200


@app.route('/basket_cleaning_finish', methods=['GET']) 
def basket_cleaning_finish():#开始登录时，检测是否需要复位，若需要，则发送复位信息，否则开始焊接检测

    stop_inference_internal()
    app.logger.info('basket_cleaning_order')
    return jsonify({"status": "SUCCESS"}), 200


    
def stop_inference_internal():
    global inference_thread
    if inference_thread is not None and inference_thread.is_alive():
        stop_event.set()  # 设置停止事件标志，通知推理线程停止运行
        inference_thread.join()  # 等待推理线程结束
        inference_thread = None  # 释放线程资源
        app.logger.info('detection stopped')
        return True
    else:
        app.logger.info('No inference stopped')
        return False

@app.route('/stop_detection', methods=['GET'])
def stop_inference():
    #global inference_thread
    if stop_inference_internal():
        app.logger.info('detection stopped')
        return jsonify({"status": "DETECTION_STOPPED"}), 200
    else:
        app.logger.info('No_detection_running')
        return jsonify({"status": "No_detection_running"}), 200


@app.route('/images/<filename>')
def get_image(filename):
    app.logger.info('get_image'+filename)
    return send_from_directory('static/images', filename)


if __name__ == '__main__':

    # Start the Flask server
    app.run(debug=False, host='172.16.20.163', port=5005)