server_develop/src_code/test/test_src/multi_thread_test.py

# -*- coding: utf-8 -*-
# @Time : 2022/3/3 16:13
# @Author : MaochengHu
# @Email : wojiaohumaocheng@gmail.com
# @File : multi_thread_test.py
# @Project : server_develop


from multiprocessing import Pool
from threading import Thread
import os
from server_utils.image_convert import path2base64
from server_utils.read_properties import get_properties_dict
import requests
from multiprocessing import Process
headers = {"Content-Type": "application/json"}
properties_dict = get_properties_dict()
inference_address = properties_dict.get("inference_address")


def single_test(input_source=os.path.join(os.getcwd(), "../test_images/test1.jpg")):
    while True:
        request_url = "{}/predictions/yolov5".format(inference_address)
        image_base64 = path2base64(input_source)
        payload = {"data": image_base64,
                   "output_methods": str(["pd_xywh_json"]),
                   "base64_bool": True}

        req = requests.post(url=request_url, data=payload)
        print(eval(req.text))






if __name__ == '__main__':

    for i in range(3):
        t = Process(target=single_test)
        t.start()