# 这是一个示例 Python 脚本。 import json import os # 按 Shift+F10 执行或将其替换为您的代码。 # 按 双击 Shift 在所有地方搜索类、文件、工具窗口、操作和设置。 import SimpleView_SaveImage as simple import config import image import time # 在文件顶部导入 time 模块 from flask import Flask, jsonify, request from logConfig import get_logger app = Flask(__name__) import concurrent.futures from functools import partial # 使用示例 logger = get_logger() # 示例接口：GET 请求 @app.route("/config/update") def home(): logger.info("配置更新请求") config.load_configs() logger.info("配置更新完成") return {"message": "Hello, this service config update!", "status": "OK"} @app.route("/api/addTemplate", methods=["POST"]) def save_template(): logger.info("接收到添加模板请求") data = request.get_json() # 验证必要字段 if not data: logger.error("添加模板失败：未提供数据") return jsonify({"status": "ERROR", "message": "No data provided"}), 400 template_type = data.get("type") if not template_type: logger.error("添加模板失败：缺少'type'字段") return jsonify({"status": "ERROR", "message": "Missing 'type' field"}), 400 CONFIG_DIR = "./config/template" # 构造文件路径 filename = f"{template_type}.json" file_path = os.path.join(CONFIG_DIR, filename) # 写入文件 try: with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=4) config.load_configs() logger.info(f"模板 '{template_type}' 保存成功") return jsonify({"status": "OK", "message": f"Template '{template_type}' saved successfully."}), 200 except Exception as e: logger.error(f"保存模板失败: {str(e)}") return jsonify({"status": "ERROR", "message": f"Failed to save template: {str(e)}"}), 500 @app.route("/api/initDevice", methods=['POST']) def initDevice(): logger.info("设备初始化请求") simple.initialize_devices() logger.info("设备初始化完成") # # @app.route("/api/picComputeAll", methods=['POST']) # def picComputeAll(): # logger.info("接收到全设备图片计算请求") # start_time_total = time.time() # # # 输入参数 # # sn:设备编号 # # type:类型 # # # 获取 JSON 数据 # data = request.get_json() # type = data['type'] # template = config.TEMPLATE_CONFIG_MAP.get(type) # if not template: # logger.error("缺少模板配置: 'type'") # return jsonify({ # "message": "", # "status": "ERROR", # "error": "Missing required parameter: 'type'" # }), 400 # all_sn_list = list(config.CAMERA_CONFIG_MAP.keys()) # # # 遍历每个 sn 并判断是否都有 hole # all_have_holes = False # color_list = [] # 新增：用于保存所有 sn 的 color # # count_assemble = 0 # for sn in all_sn_list: # start_time = time.time() # 开始计时 # logger.info(f"开始处理设备 {sn} 的图片计算") # pic_info = simple.enhanced_pic(sn, type) # points = pic_info.get("point") # if points is None: # logger.warning(f"设备 {sn} 未返回点云数据") # all_have_holes = True # break # # # 提取 color 并加入列表 # color = pic_info.get("color") # if color is not None: # color_list.append(color) # # try: # real_holes, count = image.detect_large_holes(points, sn, type) # # count_assemble += count # # if len(real_holes) != 0: # all_have_holes = True # # 计算耗时 # elapsed_time = round(time.time() - start_time, 3) # # 输出日志 # logger.info(f"[INFO] picComputeAll executed in {elapsed_time} seconds {sn}") # break # 判断到空洞就直接退出 # except Exception as e: # all_have_holes = True # logger.error(f"Failed to process request: {str(e)} sn：{sn}") # # # 计算总耗时 # total_elapsed_time = round(time.time() - start_time_total, 3) # logger.info(f"全设备图片计算完成，总耗时 {total_elapsed_time} 秒，结果: {all_have_holes}") # # # 返回结果中包含 color 列表 # return jsonify({ # "message": all_have_holes, # "lack": all_have_holes, # "type": type, # "count": count_assemble, # "colors": color_list, # 添加 color 列表 # "status": "OK" # }) # @app.route("/api/picComputeAll", methods=['POST']) def picComputeAll(): logger.info("接收到全设备图片计算请求") start_time_total = time.time() # 获取 JSON 数据 data = request.get_json() type = data['type'] template = config.TEMPLATE_CONFIG_MAP.get(type) if not template: logger.error("缺少模板配置: 'type'") return jsonify({ "message": "", "status": "ERROR", "error": "Missing required parameter: 'type'" }), 400 all_sn_list = list(config.CAMERA_CONFIG_MAP.keys()) # 同步处理每个设备 all_have_holes = False color_list = [] count_assemble = 0 # 逐个同步处理设备 for sn in all_sn_list: try: start_time = time.time() logger.info(f"开始处理设备 {sn} 的图片计算") # 同步采集图片信息 pic_info = simple.enhanced_pic(sn, type) points = pic_info.get("point") if points is None: logger.warning(f"设备 {sn} 未返回点云数据") all_have_holes = True break # 提取 color 并加入列表 if pic_info.get("color") is not None: color_list.append(pic_info["color"]) # 检测空洞 real_holes, count = image.detect_large_holes(points, sn, type) elapsed_time = round(time.time() - start_time, 3) logger.info(f"设备 {sn} 处理完成，耗时 {elapsed_time} 秒，发现空洞: {len(real_holes) != 0}") # 检查是否发现空洞 if len(real_holes) != 0: all_have_holes = True count_assemble += count logger.info(f"设备 {sn} 发现空洞，停止其他任务") break # 发现空洞就直接退出 except Exception as e: logger.error(f"处理设备 {sn} 时发生异常: {str(e)}") all_have_holes = True break # 计算总耗时 total_elapsed_time = round(time.time() - start_time_total, 3) logger.info(f"全设备图片计算完成，总耗时 {total_elapsed_time} 秒，结果: {all_have_holes}") # 返回结果中包含 color 列表 return jsonify({ "message": all_have_holes, "lack": all_have_holes, "type": type, "count": count_assemble, "colors": color_list, "status": "OK" }) @app.route("/api/picBitCloudy", methods=['POST']) def picBitCloudy(): logger.info("接收到picBitCloudy请求") # 输入参数 # sn:设备编号 # type:类型 # 获取 JSON 数据 data = request.get_json() direction = data.get('direction') type = data.get('type') all_sn_list = list(config.CAMERA_CONFIG_MAP.keys()) # 遍历每个 sn 并判断是否都有 hole all_have_holes = False color_list = [] # 新增：用于保存所有 sn 的 color count_assemble = 0 for sn in all_sn_list: if direction and config.CAMERA_CONFIG_MAP[sn].get("direction") != direction: logger.debug(f"设备 {sn} 方向不匹配，跳过") continue start_time = time.time() # 开始计时 logger.info(f"开始处理设备 {sn} 的图片计算") pic_info = simple.enhanced_pic(sn, type) points = pic_info.get("point") if points is None or len(points) < 300: logger.warning(f"设备 {sn} 点云数据不足") all_have_holes = True break # 提取 color 并加入列表 color = pic_info.get("color") if color is not None: color_list.append(color) logger.info(f"picBitCloudy处理完成，结果: {all_have_holes}") # 返回结果中包含 color 列表 return jsonify({ "message": all_have_holes, "lack": all_have_holes, "colors": color_list, # 添加 color 列表 "status": "OK" }) @app.route("/api/picCompute", methods=['POST']) def picCompute(): logger.info("接收到单设备图片计算请求") # 输入参数 # sn:设备编号 # type:类型 try: # 获取 JSON 数据 data = request.get_json() if not data or 'sn' not in data: logger.error("缺少必需参数: 'sn'") return jsonify({ "message": "", "status": "ERROR", "error": "Missing required parameter: 'sn'" }), 400 sn = data['sn'] type = data['type'] logger.info(f"开始处理设备 {sn} 类型 {type} 的图片计算") template = config.TEMPLATE_CONFIG_MAP.get(type) if not template: logger.error(f"缺少模板配置: {type}") return jsonify({ "message": "", "status": "ERROR", "error": "Missing required parameter: 'type'" }), 400 pic_info = simple.enhanced_pic(sn, type) real_holes, count = image.detect_large_holes(pic_info["point"], sn, type) result = len(real_holes) != 0 logger.info(f"设备 {sn} 图片计算完成，检测到空洞: {result}") return jsonify({"message": result, "lack": result, "status": "OK"}) except Exception as e: logger.error(f"图片计算失败: {str(e)}") return jsonify({"message": "", "status": "ERROR", "error": f"Failed to get TIFF paths: {str(e)}"}), 500 return jsonify({"message": True, "lack": True, "status": "OK"}) @app.errorhandler(Exception) def handle_exception(e): # 处理所有未被捕获的异常 logger.error(f"未捕获的异常: {str(e)}") return jsonify({ "message": "", "status": "ERROR", "error": str(e) }), 500 if __name__ == "__main__": from waitress import serve logger.info("服务启动中...") print("Serving on http://0.0.0.0:5000") logger.info("服务启动成功，监听地址: http://0.0.0.0:5000") serve(app, host='0.0.0.0', port=5000)