example-apps/PDD/pcb-defect-detection/tools/application.py

   1
   2 import config
   3 from flask_sslify import SSLify
   4 from flask import Flask, request, jsonify, Response
   5 from flask_cors import CORS
   6
   7 import json
   8 import requests
   9 import os
  10 import os.path
  11 from os import path
  12 import base64
  13 import sys
  14
  15 app = Flask(__name__)
  16 CORS(app)
  17 sslify = SSLify(app)
  18 app.config['JSON_AS_ASCII'] = False
  19 app.config['INPUT_IMAGE_PATH'] = '/usr/app/input_image/'
  20 app.config['OUTPUT_IMAGE_PATH'] = '/usr/app/output_image/'
  21
  22 app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024
  23 ALLOWED_EXTENSIONS = set(['png', 'jpg', 'jpeg'])
  24 count = 0
  25 listOfPath = []
  26 listOfImages = []
  27 mock_func = 0
  28 # result_image_path = app.config['OUTPUT_IMAGE_PATH']
  29
  30 listOfPath = ["beijinglab/camera1", "beijinglab/camera2",
  31               "beijinglab/camera3",
  32               "shenzhenlab/camera1", "shenzhenlab/camera2",
  33               "shenzhenlab/camera3",
  34               "shanghailab/camera1", "shanghailab/camera2",
  35               "shanghailab/camera3"]
  36
  37 def path_exist(image_path):
  38     print('image path:',image_path)
  39
  40     pathexist = 0
  41     for imagepath in listOfPath:
  42         print('image path in list:', imagepath)
  43         if (image_path == imagepath):
  44             print('image path match')
  45             pathexist = 1
  46             break
  47
  48     return pathexist
  49
  50 def get_images(abs_image_path):
  51     """
  52     preview images
  53
  54     :return: result image
  55     """
  56     del listOfImages[:]
  57
  58     arr = os.listdir(abs_image_path)
  59     for file in arr:
  60         print('image name:',abs_image_path + file)
  61
  62     for image in arr:
  63         ## as base64 string
  64         with open(abs_image_path + image, "rb") as img_file:
  65             jpeg_bin = base64.b64encode(img_file.read())
  66             listOfImages.append(jpeg_bin)
  67
  68     response = {'image01': listOfImages[0],
  69                 'image02': listOfImages[1],
  70                 'image03': listOfImages[2],}
  71                 #'image04': listOfImages[3]}
  72                # 'image05': listOfImages[4],}
  73
  74     return response
  75
  76 def detection(input_image_path, output_image_path):
  77     """
  78     detection
  79
  80     :return: detection status
  81     """
  82     cmd = 'cd tools' + ' && python inference.py --data_dir=' + \
  83           input_image_path + ' --save_dir=' + output_image_path + ' --GPU=0'
  84     print(cmd)
  85
  86     os.system(cmd)
  87     return 1
  88
  89
  90 @app.route('/v1/pcb/preview/<EdgeLoc>/<cameraNum>', methods=['GET'])
  91 def preview_image(EdgeLoc, cameraNum):
  92     """
  93     preview images
  94
  95     :return: result image
  96     """
  97     app.logger.info("Received message from ClientIP [" + request.remote_addr
  98                     + "] Operation [" + request.method + "]" +
  99                     " Resource [" + request.url + "]")
 100
 101     image_path = EdgeLoc + '/' + cameraNum
 102     pathexist = path_exist(image_path)
 103     if pathexist == 0:
 104         response = {'image' : 'null'}
 105         print ('image path not exist:',image_path)
 106         return jsonify(response)
 107
 108     abs_image_path = app.config['INPUT_IMAGE_PATH'] + image_path + '/'
 109     print('abs image path:',abs_image_path)
 110     response = get_images(abs_image_path)
 111
 112     return jsonify(response)
 113
 114
 115 @app.route('/v1/pcb/resultimage', methods=['GET'])
 116 def result_image():
 117     """
 118     preview images
 119
 120     :return: result image
 121     """
 122     app.logger.info("Received message from ClientIP [" + request.remote_addr
 123                     + "] Operation [" + request.method + "]" +
 124                     " Resource [" + request.url + "]")
 125
 126     print('result image path global:',result_image_path)
 127     response = get_images(result_image_path)
 128
 129     return jsonify(response)
 130
 131
 132 @app.route('/v1/pcb/detection/<EdgeLoc>/<cameraNum>', methods=['GET'])
 133 def detect_image(EdgeLoc, cameraNum):
 134     """
 135     detect images
 136
 137     :return: success or failure
 138     """
 139     app.logger.info("Received message from ClientIP [" + request.remote_addr
 140                     + "] Operation [" + request.method + "]" +
 141                     " Resource [" + request.url + "]")
 142     global result_image_path
 143
 144     image_path = EdgeLoc + '/' + cameraNum
 145     pathexist = path_exist(image_path)
 146     if pathexist == 0:
 147         response = {'responce': 'failure'}
 148         print ('image path not exist:',image_path)
 149         return jsonify(response)
 150
 151     input_image_path = app.config['INPUT_IMAGE_PATH'] + image_path + '/'
 152     print('input image path:', input_image_path)
 153
 154     output_image_path = app.config['OUTPUT_IMAGE_PATH'] + image_path + '/'
 155     print('out image path:', output_image_path)
 156
 157     ret = detection(input_image_path, output_image_path)
 158     if (ret == 0):
 159         response = {'responce': 'failure'}
 160         print('detectio algo failed ')
 161         return jsonify(response)
 162
 163     result_image_path = output_image_path
 164     print('result image path:', result_image_path)
 165
 166     response = {'responce': 'success'}
 167     print('detection sucess')
 168     return jsonify(response)
 169
 170
 171 def start_server(handler):
 172     app.logger.addHandler(handler)
 173     if config.ssl_enabled:
 174         context = (config.ssl_certfilepath, config.ssl_keyfilepath)
 175         app.run(host=config.server_address, debug=True, ssl_context=context,
 176                 threaded=True, port=config.server_port)
 177     else:
 178         app.run(host=config.server_address, debug=True, threaded=True,
 179                 port=config.server_port)