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")
import time

num = 1
t_time = time.time()
def single_test(input_source=os.path.join(os.getcwd(), "../test_images/test1.jpg")):
    global  num

    while True:
        request_url = "http://0.0.0.0:8080/predictions/yolov5"
        image_base64 = path2base64(input_source)

        payload = {"data": image_base64,
                   "output_methods": str(["pd_xywh_json"]),
                   "base64_bool": True}
        s_time = time.time()
        req = requests.post(url=request_url, data=payload)
        # print(eval(req.text))
        num += 1
        print(f"sub {time.time()-s_time} --- {num}")
        print(f"total {time.time() - t_time} -- {num}")





if __name__ == '__main__':

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