convert_need_interface.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785
  1. # encoding=utf8
  2. import base64
  3. import inspect
  4. import json
  5. import logging
  6. import multiprocessing
  7. import os
  8. import pickle
  9. import random
  10. import sys
  11. import time
  12. import uuid
  13. import zlib
  14. from queue import Queue
  15. import redis
  16. from werkzeug.exceptions import NotFound
  17. sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../")
  18. import traceback
  19. import requests
  20. from format_convert import _global
  21. from format_convert.utils import get_platform, get_sequential_data, judge_error_code, request_post, get_ip_port, \
  22. get_intranet_ip, get_logger, log, get_args_from_config
  23. from ocr.ocr_interface import ocr, OcrModels
  24. from otr.otr_interface import otr, OtrModels
  25. from format_convert.libreoffice_interface import office_convert
  26. import numpy as np
  27. MAX_COMPUTE = True
  28. if get_platform() == "Windows":
  29. FROM_REMOTE = False
  30. only_test_ocr = False
  31. if only_test_ocr:
  32. ip_port_flag = {}
  33. ip_port_dict = get_ip_port()
  34. for _k in ip_port_dict.keys():
  35. ip_port_flag.update({_k: {"ocr": 0,
  36. "otr": 0,
  37. "convert": 0,
  38. "office": 0
  39. }})
  40. _global.update({"ip_port_flag": ip_port_flag})
  41. ip_port_dict["http://127.0.0.1"]["ocr"] = ["17000"]
  42. ip_port_dict["http://127.0.0.1"]["otr"] = ["18000"]
  43. _global.update({"ip_port": ip_port_dict})
  44. else:
  45. FROM_REMOTE = True
  46. if MAX_COMPUTE:
  47. FROM_REMOTE = False
  48. # ip_port_dict = get_ip_port()
  49. # ip = 'http://127.0.0.1'
  50. # ocr_port_list = ip_port_dict.get(ip).get("ocr")
  51. # otr_port_list = ip_port_dict.get(ip).get("otr")
  52. lock = multiprocessing.RLock()
  53. # 连接redis数据库
  54. # redis_db = redis.StrictRedis(host='192.168.2.103', port='6379',
  55. # db=1, password='bidi123456', health_check_interval=300)
  56. redis_db = None
  57. def _interface(_dict, time_out=60, retry_times=3):
  58. try:
  59. # 重试
  60. model_type = _dict.get("model_type")
  61. while retry_times:
  62. ip_port = interface_pool(model_type)
  63. if judge_error_code(ip_port):
  64. return ip_port
  65. _url = ip_port + "/" + model_type
  66. # base64_stream = base64.b64encode(pickle.dumps(_dict))
  67. r = json.loads(request_post(_url, {"data": json.dumps(_dict),
  68. "model_type": model_type}, time_out=time_out))
  69. log("get _interface return")
  70. if type(r) == list:
  71. # 接口连不上换个端口重试
  72. if retry_times <= 1:
  73. return r
  74. else:
  75. retry_times -= 1
  76. log("retry post _interface... left times " + str(retry_times) + " " + model_type)
  77. continue
  78. if judge_error_code(r):
  79. return r
  80. return r
  81. break
  82. except TimeoutError:
  83. return [-5]
  84. except requests.exceptions.ConnectionError as e:
  85. return [-2]
  86. def from_office_interface(src_path, dest_path, target_format, retry_times=1, from_remote=FROM_REMOTE):
  87. try:
  88. # Win10跳出超时装饰器
  89. # if get_platform() == "Windows":
  90. # # origin_office_convert = office_convert.__wrapped__
  91. # # file_path = origin_office_convert(src_path, dest_path, target_format, retry_times)
  92. # file_path = office_convert(src_path, dest_path, target_format, retry_times)
  93. # else:
  94. # # 将装饰器包装为一个类,否则多进程Pickle会报错 it's not the same object as xxx 问题,
  95. # # timeout_decorator_obj = my_timeout_decorator.TimeoutClass(office_convert, 180, TimeoutError)
  96. # # file_path = timeout_decorator_obj.run(src_path, dest_path, target_format, retry_times)
  97. #
  98. # file_path = office_convert(src_path, dest_path, target_format, retry_times)
  99. if from_remote:
  100. # 重试
  101. retry_times_1 = 1
  102. retry_times_2 = 2
  103. while retry_times_1 and retry_times_2:
  104. # _ip = ip_pool("soffice", _random=True)
  105. # _port = port_pool("soffice", _random=True)
  106. # _ip = interface_ip_list[0]
  107. # _port = "16002"
  108. # _ip, _port = interface_pool("soffice")
  109. # ip_port = from_schedule_interface("office")
  110. ip_port = interface_pool_gunicorn("office")
  111. if judge_error_code(ip_port):
  112. return ip_port
  113. _url = ip_port + "/soffice"
  114. with open(src_path, "rb") as f:
  115. file_bytes = f.read()
  116. base64_stream = base64.b64encode(file_bytes)
  117. start_time = time.time()
  118. r = json.loads(request_post(_url, {"src_path": src_path,
  119. "dest_path": dest_path,
  120. "file": base64_stream,
  121. "target_format": target_format,
  122. "retry_times": retry_times}, time_out=25))
  123. log("get interface return")
  124. log("office use time " + str(time.time()-start_time))
  125. if type(r) == list:
  126. # 接口连不上换个端口重试
  127. if retry_times_1 <= 1:
  128. return r
  129. else:
  130. retry_times_1 -= 1
  131. log("retry post office_interface... left times " + str(retry_times_1))
  132. continue
  133. file_str = r.get("data")
  134. if judge_error_code(file_str):
  135. if retry_times_2 <= 1:
  136. return file_str
  137. else:
  138. retry_times_2 -= 1
  139. continue
  140. file_bytes = eval(file_str)
  141. uid1 = src_path.split(os.sep)[-1].split(".")[0]
  142. file_path = dest_path + uid1 + "." + target_format
  143. if not os.path.exists(os.path.dirname(file_path)):
  144. os.makedirs(os.path.dirname(file_path), mode=0o777)
  145. with open(file_path, "wb") as f:
  146. f.write(file_bytes)
  147. break
  148. else:
  149. file_path = office_convert(src_path, dest_path, target_format, retry_times)
  150. if judge_error_code(file_path):
  151. return file_path
  152. return file_path
  153. except TimeoutError:
  154. log("from_office_interface timeout error!")
  155. return [-5]
  156. except:
  157. log("from_office_interface error!")
  158. print("from_office_interface", traceback.print_exc())
  159. return [-1]
  160. def from_ocr_interface(image_stream, is_table=False, from_remote=FROM_REMOTE):
  161. log("into from_ocr_interface")
  162. try:
  163. base64_stream = base64.b64encode(image_stream)
  164. # 调用接口
  165. try:
  166. if from_remote:
  167. retry_times_1 = 3
  168. # 重试
  169. while retry_times_1:
  170. # _ip = ip_pool("ocr", _random=True)
  171. # _port = port_pool("ocr", _random=True)
  172. # if _ip == interface_ip_list[1]:
  173. # _port = ocr_port_list[0]
  174. # _ip, _port = interface_pool("ocr")
  175. # ip_port = _ip + ":" + _port
  176. # ip_port = from_schedule_interface("ocr")
  177. ip_port = interface_pool_gunicorn("ocr")
  178. if judge_error_code(ip_port):
  179. return ip_port
  180. _url = ip_port + "/ocr"
  181. r = json.loads(request_post(_url, {"data": base64_stream,
  182. "md5": _global.get("md5")},
  183. time_out=60))
  184. log("get interface return")
  185. if type(r) == list:
  186. # 接口连不上换个端口重试
  187. if retry_times_1 <= 1:
  188. if is_table:
  189. return r, r
  190. else:
  191. return r
  192. else:
  193. retry_times_1 -= 1
  194. log("retry post ocr_interface... left times " + str(retry_times_1))
  195. continue
  196. if judge_error_code(r):
  197. return r
  198. break
  199. else:
  200. if globals().get("global_ocr_model") is None:
  201. print("=========== init ocr model ===========")
  202. globals().update({"global_ocr_model": OcrModels().get_model()})
  203. r = ocr(data=base64_stream, ocr_model=globals().get("global_ocr_model"))
  204. except TimeoutError:
  205. if is_table:
  206. return [-5], [-5]
  207. else:
  208. return [-5]
  209. except requests.exceptions.ConnectionError as e:
  210. if is_table:
  211. return [-2], [-2]
  212. else:
  213. return [-2]
  214. _dict = r
  215. text_list = eval(_dict.get("text"))
  216. bbox_list = eval(_dict.get("bbox"))
  217. if text_list is None:
  218. text_list = []
  219. if bbox_list is None:
  220. bbox_list = []
  221. if is_table:
  222. return text_list, bbox_list
  223. else:
  224. if text_list and bbox_list:
  225. text = get_sequential_data(text_list, bbox_list, html=True)
  226. if judge_error_code(text):
  227. return text
  228. else:
  229. text = ""
  230. return text
  231. except Exception as e:
  232. log("from_ocr_interface error!")
  233. # print("from_ocr_interface", e, global_type)
  234. if is_table:
  235. return [-1], [-1]
  236. else:
  237. return [-1]
  238. def from_gpu_interface_flask(_dict, model_type, predictor_type):
  239. log("into from_gpu_interface")
  240. start_time = time.time()
  241. try:
  242. # 调用接口
  243. _dict.update({"predictor_type": predictor_type, "model_type": model_type})
  244. if model_type == "ocr":
  245. use_zlib = True
  246. else:
  247. use_zlib = False
  248. result = _interface(_dict, time_out=30, retry_times=2, use_zlib=use_zlib)
  249. log("from_gpu_interface finish size " + str(sys.getsizeof(_dict)) + " time " + str(time.time()-start_time))
  250. return result
  251. except Exception as e:
  252. log("from_gpu_interface error!")
  253. log("from_gpu_interface failed " + str(time.time()-start_time))
  254. traceback.print_exc()
  255. return [-2]
  256. def from_gpu_interface_redis(_dict, model_type, predictor_type):
  257. log("into from_gpu_interface")
  258. start_time = time.time()
  259. try:
  260. # 调用接口
  261. _uuid = uuid.uuid1().hex
  262. _dict.update({"predictor_type": predictor_type, "model_type": model_type,
  263. "uuid": _uuid})
  264. _time = time.time()
  265. log("pickle.dumps(_dict)" + str(_dict))
  266. redis_db.rpush("producer_"+model_type, pickle.dumps(_dict))
  267. log("producer_" + model_type + " len " + str(redis_db.llen("producer_" + model_type)))
  268. log("to producer_" + model_type + " time " + str(time.time()-_time))
  269. _time = time.time()
  270. time_out = 300
  271. while True:
  272. time.sleep(0.2)
  273. if time.time() - _time > time_out:
  274. raise Exception
  275. if redis_db.hexists("consumer_"+model_type, _uuid):
  276. time1 = time.time()
  277. result = redis_db.hget("consumer_"+model_type, _uuid)
  278. log("from consumer_"+model_type + " time " + str(time.time()-time1))
  279. break
  280. result = pickle.loads(result)
  281. log("from_gpu_interface finish - size " + str(sys.getsizeof(_dict)) + " - time " + str(time.time()-start_time))
  282. return result
  283. except Exception as e:
  284. log("from_gpu_interface error!")
  285. log("from_gpu_interface failed " + str(time.time()-start_time))
  286. traceback.print_exc()
  287. return [-2]
  288. # def from_gpu_flask_sm(_dict, model_type, predictor_type):
  289. # log("into from_gpu_share_memory")
  290. # start_time = time.time()
  291. # shm = None
  292. # try:
  293. # # 放入共享内存
  294. # _time = time.time()
  295. # np_data = _dict.get("inputs")
  296. # shm = to_share_memory(np_data)
  297. # log("data into share memory " + str(shm.name) + " " + str(time.time()-_time))
  298. #
  299. # # 调用接口
  300. # _time = time.time()
  301. # _dict.pop("inputs")
  302. # _dict.update({"predictor_type": predictor_type, "model_type": model_type,
  303. # "sm_name": shm.name, "sm_shape": np_data.shape,
  304. # "sm_dtype": str(np_data.dtype)})
  305. # result = _interface(_dict, time_out=30, retry_times=2)
  306. # log("_interface cost " + str(time.time()-_time))
  307. #
  308. # # 读取共享内存
  309. # _time = time.time()
  310. # sm_name = result.get("sm_name")
  311. # sm_shape = result.get("sm_shape")
  312. # sm_dtype = result.get("sm_dtype")
  313. # sm_dtype = get_np_type(sm_dtype)
  314. # if sm_name:
  315. # outputs = from_share_memory(sm_name, sm_shape, sm_dtype)
  316. # else:
  317. # log("from_share_memory failed!")
  318. # raise Exception
  319. # log("data from share memory " + sm_name + " " + str(time.time()-_time))
  320. #
  321. # log("from_gpu_interface finish - size " + str(sys.getsizeof(_dict)) + " - time " + str(time.time()-start_time))
  322. # return {"preds": outputs, "gpu_time": result.get("gpu_time")}
  323. # except Exception as e:
  324. # log("from_gpu_interface failed " + str(time.time()-start_time))
  325. # traceback.print_exc()
  326. # return [-2]
  327. # finally:
  328. # # del b # Unnecessary; merely emphasizing the array is no longer used
  329. # if shm:
  330. # try:
  331. # shm.close()
  332. # shm.unlink()
  333. # except FileNotFoundError:
  334. # log("share memory " + shm.name + " not exists!")
  335. # except Exception:
  336. # traceback.print_exc()
  337. #
  338. #
  339. # def from_gpu_share_memory(_dict, model_type, predictor_type):
  340. # log("into from_gpu_share_memory")
  341. # start_time = time.time()
  342. # try:
  343. # _dict.update({"model_type": model_type, "predictor_type": predictor_type})
  344. # outputs, gpu_time = share_memory_pool(_dict)
  345. # log("from_gpu_share_memory finish - size " + str(sys.getsizeof(_dict)) + " - time " + str(time.time()-start_time))
  346. # return {"preds": outputs, "gpu_time": float(gpu_time)}
  347. # except Exception as e:
  348. # log("from_gpu_interface failed " + str(time.time()-start_time))
  349. # traceback.print_exc()
  350. # return [-2]
  351. def from_otr_interface2(image_stream):
  352. log("into from_otr_interface")
  353. try:
  354. base64_stream = base64.b64encode(image_stream)
  355. # 调用接口
  356. try:
  357. if globals().get("global_otr_model") is None:
  358. globals().update({"global_otr_model": OtrModels().get_model()})
  359. print("=========== init otr model ===========")
  360. r = otr(data=base64_stream, otr_model=globals().get("global_otr_model"))
  361. except TimeoutError:
  362. return [-5], [-5], [-5], [-5], [-5]
  363. except requests.exceptions.ConnectionError as e:
  364. log("from_otr_interface")
  365. print("from_otr_interface", traceback.print_exc())
  366. return [-2], [-2], [-2], [-2], [-2]
  367. # 处理结果
  368. _dict = r
  369. points = eval(_dict.get("points"))
  370. split_lines = eval(_dict.get("split_lines"))
  371. bboxes = eval(_dict.get("bboxes"))
  372. outline_points = eval(_dict.get("outline_points"))
  373. lines = eval(_dict.get("lines"))
  374. # print("from_otr_interface len(bboxes)", len(bboxes))
  375. if points is None:
  376. points = []
  377. if split_lines is None:
  378. split_lines = []
  379. if bboxes is None:
  380. bboxes = []
  381. if outline_points is None:
  382. outline_points = []
  383. if lines is None:
  384. lines = []
  385. return points, split_lines, bboxes, outline_points, lines
  386. except Exception as e:
  387. log("from_otr_interface error!")
  388. print("from_otr_interface", traceback.print_exc())
  389. return [-1], [-1], [-1], [-1], [-1]
  390. def from_otr_interface(image_stream, is_from_pdf=False, from_remote=FROM_REMOTE):
  391. log("into from_otr_interface")
  392. try:
  393. base64_stream = base64.b64encode(image_stream)
  394. # 调用接口
  395. try:
  396. if from_remote:
  397. log("from remote")
  398. retry_times_1 = 3
  399. # 重试
  400. while retry_times_1:
  401. # _ip = ip_pool("otr", _random=True)
  402. # _port = port_pool("otr", _random=True)
  403. # if _ip == interface_ip_list[1]:
  404. # _port = otr_port_list[0]
  405. ip_port = interface_pool_gunicorn("otr")
  406. # ip_port = from_schedule_interface("otr")
  407. if judge_error_code(ip_port):
  408. return ip_port
  409. _url = ip_port + "/otr"
  410. r = json.loads(request_post(_url, {"data": base64_stream,
  411. "is_from_pdf": is_from_pdf,
  412. "md5": _global.get("md5")}, time_out=60))
  413. log("get interface return")
  414. if type(r) == list:
  415. # 接口连不上换个端口重试
  416. if retry_times_1 <= 1:
  417. return r
  418. else:
  419. retry_times_1 -= 1
  420. log("retry post otr_interface... left times " + str(retry_times_1))
  421. continue
  422. if judge_error_code(r):
  423. return r
  424. break
  425. else:
  426. log("from local")
  427. log("otr_model " + str(globals().get("global_otr_model")))
  428. if globals().get("global_otr_model") is None:
  429. print("=========== init otr model ===========")
  430. globals().update({"global_otr_model": OtrModels().get_model()})
  431. log("init finish")
  432. r = otr(data=base64_stream, otr_model=globals().get("global_otr_model"), is_from_pdf=is_from_pdf)
  433. # r = otr(data=base64_stream, otr_model=None, is_from_pdf=is_from_pdf)
  434. except TimeoutError:
  435. return [-5]
  436. except requests.exceptions.ConnectionError as e:
  437. log("from_otr_interface")
  438. print("from_otr_interface", traceback.print_exc())
  439. return [-2]
  440. # 处理结果
  441. _dict = r
  442. list_line = eval(_dict.get("list_line"))
  443. return list_line
  444. except Exception as e:
  445. log("from_otr_interface error!")
  446. print("from_otr_interface", traceback.print_exc())
  447. return [-1]
  448. # def from_schedule_interface(interface_type):
  449. # try:
  450. # _ip = "http://" + get_intranet_ip()
  451. # _port = ip_port_dict.get(_ip).get("schedule")[0]
  452. # _url = _ip + ":" + _port + "/schedule"
  453. # data = {"interface_type": interface_type}
  454. # result = json.loads(request_post(_url, data, time_out=10)).get("data")
  455. # if judge_error_code(result):
  456. # return result
  457. # _ip, _port = result
  458. # log("from_schedule_interface " + _ip + " " + _port)
  459. # return _ip + ":" + _port
  460. # except requests.exceptions.ConnectionError as e:
  461. # log("from_schedule_interface ConnectionError")
  462. # return [-2]
  463. # except:
  464. # log("from_schedule_interface error!")
  465. # traceback.print_exc()
  466. # return [-1]
  467. def interface_pool(interface_type, use_gunicorn=True):
  468. ip_port_flag = _global.get("ip_port_flag")
  469. ip_port_dict = _global.get("ip_port")
  470. try:
  471. if use_gunicorn:
  472. _ip = "http://127.0.0.1"
  473. _port = ip_port_dict.get(_ip).get(interface_type)[0]
  474. ip_port = _ip + ":" + str(_port)
  475. log(ip_port)
  476. return ip_port
  477. # 负载均衡, 选取ip
  478. interface_load_list = []
  479. for _ip in ip_port_flag.keys():
  480. if ip_port_dict.get(_ip).get(interface_type):
  481. load_scale = ip_port_flag.get(_ip).get(interface_type) / len(ip_port_dict.get(_ip).get(interface_type))
  482. interface_load_list.append([_ip, load_scale])
  483. if not interface_load_list:
  484. raise NotFound
  485. interface_load_list.sort(key=lambda x: x[-1])
  486. _ip = interface_load_list[0][0]
  487. # 负载均衡, 选取port
  488. ip_type_cnt = ip_port_flag.get(_ip).get(interface_type)
  489. ip_type_total = len(ip_port_dict.get(_ip).get(interface_type))
  490. if ip_type_cnt == 0:
  491. ip_type_cnt = random.randint(0, ip_type_total-1)
  492. port_index = ip_type_cnt % ip_type_total
  493. _port = ip_port_dict.get(_ip).get(interface_type)[port_index]
  494. # 更新flag
  495. current_flag = ip_type_cnt
  496. if current_flag >= 10000:
  497. ip_port_flag[_ip][interface_type] = 0
  498. else:
  499. ip_port_flag[_ip][interface_type] = current_flag + 1
  500. _global.update({"ip_port_flag": ip_port_flag})
  501. # log(str(_global.get("ip_port_flag")))
  502. ip_port = _ip + ":" + str(_port)
  503. log(ip_port)
  504. return ip_port
  505. except NotFound:
  506. log("cannot read ip from config! checkout config")
  507. return [-2]
  508. except:
  509. traceback.print_exc()
  510. return [-1]
  511. def interface_pool_gunicorn(interface_type):
  512. ip_flag_list = _global.get("ip_flag")
  513. ip_port_flag_dict = _global.get("ip_port_flag")
  514. ip_port_dict = _global.get("ip_port")
  515. try:
  516. if ip_flag_list is None or ip_port_dict is None or ip_port_flag_dict is None:
  517. raise NotFound
  518. if interface_type == "office":
  519. _ip = "http://127.0.0.1"
  520. # 选取端口
  521. port_list = ip_port_dict.get(_ip).get("MASTER").get(interface_type)
  522. ip_type_cnt = ip_port_flag_dict.get(_ip).get(interface_type)
  523. if ip_type_cnt == 0:
  524. _port = port_list[random.randint(0, len(port_list)-1)]
  525. else:
  526. _port = port_list[ip_type_cnt % len(port_list)]
  527. # 更新flag
  528. if ip_port_flag_dict.get(_ip).get(interface_type) >= 10000:
  529. ip_port_flag_dict[_ip][interface_type] = 0
  530. else:
  531. ip_port_flag_dict[_ip][interface_type] += 1
  532. _global.update({"ip_port_flag": ip_port_flag_dict})
  533. else:
  534. # 负载均衡, 选取ip
  535. ip_flag_list.sort(key=lambda x: x[1])
  536. if ip_flag_list[-1][1] == 0:
  537. ip_index = random.randint(0, len(ip_flag_list)-1)
  538. else:
  539. ip_index = 0
  540. _ip = ip_flag_list[ip_index][0]
  541. if "master" in _ip:
  542. port_index = 1
  543. else:
  544. port_index = 0
  545. _ip = _ip.split("_")[0]
  546. # 选取端口, 使用gunicorn则直接选第一个
  547. # _port = ip_port_dict.get(_ip).get("MASTER").get(interface_type)[0]
  548. log("_ip " + _ip)
  549. log("interface_type " + interface_type)
  550. port_list = get_args_from_config(ip_port_dict, _ip, interface_type)
  551. log("port_list" + str(port_list))
  552. if port_index >= len(port_list):
  553. port_index = 0
  554. _port = get_args_from_config(ip_port_dict, _ip, interface_type)[port_index][0]
  555. # 更新flag
  556. if ip_flag_list[ip_index][1] >= 10000:
  557. ip_flag_list[ip_index][1] = 0
  558. else:
  559. ip_flag_list[ip_index][1] += + 1
  560. _global.update({"ip_flag": ip_flag_list})
  561. ip_port = _ip + ":" + str(_port)
  562. log(ip_port)
  563. return ip_port
  564. except NotFound:
  565. log("ip_flag or ip_port_dict is None! checkout config")
  566. return [-2]
  567. except:
  568. traceback.print_exc()
  569. return [-1]
  570. # def share_memory_pool(args_dict):
  571. # np_data = args_dict.get("inputs")
  572. # _type = args_dict.get("model_type")
  573. # args_dict.update({"sm_shape": np_data.shape, "sm_dtype": str(np_data.dtype)})
  574. #
  575. # if _type == 'ocr':
  576. # port_list = ocr_port_list
  577. # elif _type == 'otr':
  578. # port_list = otr_port_list
  579. # else:
  580. # log("type error! only support ocr otr")
  581. # raise Exception
  582. #
  583. # # 循环判断是否有空的share memory
  584. # empty_sm_list = None
  585. # sm_list_name = ""
  586. # while empty_sm_list is None:
  587. # for p in port_list:
  588. # sm_list_name = "sml_"+_type+"_"+str(p)
  589. # sm_list = get_share_memory_list(sm_list_name)
  590. # if sm_list[0] == "0":
  591. # lock.acquire(timeout=0.1)
  592. # if sm_list[0] == "0":
  593. # sm_list[0] = "1"
  594. # sm_list[-1] = "0"
  595. # empty_sm_list = sm_list
  596. # break
  597. # else:
  598. # continue
  599. # lock.release()
  600. #
  601. # log(str(os.getppid()) + " empty_sm_list " + sm_list_name)
  602. #
  603. # # numpy放入共享内存
  604. # _time = time.time()
  605. # release_share_memory(get_share_memory("psm_" + str(os.getpid())))
  606. # shm = to_share_memory(np_data)
  607. # log("data into share memory " + str(shm.name) + " " + str(time.time()-_time))
  608. #
  609. # # 参数放入共享内存列表
  610. # empty_sm_list[1] = args_dict.get("md5")
  611. # empty_sm_list[2] = args_dict.get("model_type")
  612. # empty_sm_list[3] = args_dict.get("predictor_type")
  613. # empty_sm_list[4] = args_dict.get("args")
  614. # empty_sm_list[5] = str(shm.name)
  615. # empty_sm_list[6] = str(args_dict.get("sm_shape"))
  616. # empty_sm_list[7] = args_dict.get("sm_dtype")
  617. # empty_sm_list[-1] = "1"
  618. # # log("empty_sm_list[7] " + empty_sm_list[7])
  619. # close_share_memory_list(empty_sm_list)
  620. #
  621. # # 循环判断是否完成
  622. # finish_sm_list = get_share_memory_list(sm_list_name)
  623. # while True:
  624. # if finish_sm_list[-1] == "0":
  625. # break
  626. #
  627. # # 读取共享内存
  628. # _time = time.time()
  629. # sm_name = finish_sm_list[5]
  630. # sm_shape = finish_sm_list[6]
  631. # sm_shape = eval(sm_shape)
  632. # sm_dtype = finish_sm_list[7]
  633. # gpu_time = finish_sm_list[8]
  634. # sm_dtype = get_np_type(sm_dtype)
  635. # outputs = from_share_memory(sm_name, sm_shape, sm_dtype)
  636. # log(args_dict.get("model_type") + " " + args_dict.get("predictor_type") + " outputs " + str(outputs.shape))
  637. # log("data from share memory " + sm_name + " " + str(time.time()-_time))
  638. #
  639. # # 释放
  640. # release_share_memory(get_share_memory(sm_name))
  641. #
  642. # # 重置share memory list
  643. # finish_sm_list[-1] = "0"
  644. # finish_sm_list[0] = "0"
  645. #
  646. # close_share_memory_list(finish_sm_list)
  647. # return outputs, gpu_time
  648. # def interface_pool(interface_type):
  649. # try:
  650. # ip_port_dict = _global.get("ip_port")
  651. # ip_list = list(ip_port_dict.keys())
  652. # _ip = random.choice(ip_list)
  653. # if interface_type != 'office':
  654. # _port = ip_port_dict.get(_ip).get(interface_type)[0]
  655. # else:
  656. # _port = random.choice(ip_port_dict.get(_ip).get(interface_type))
  657. # log(_ip + ":" + _port)
  658. # return _ip + ":" + _port
  659. # except Exception as e:
  660. # traceback.print_exc()
  661. # return [-1]
  662. # def ip_pool(interface_type, _random=False):
  663. # ip_flag_name = interface_type + '_ip_flag'
  664. # ip_flag = globals().get(ip_flag_name)
  665. # if ip_flag is None:
  666. # if _random:
  667. # _r = random.randint(0, len(interface_ip_list)-1)
  668. # ip_flag = _r
  669. # globals().update({ip_flag_name: ip_flag})
  670. # ip_index = _r
  671. # else:
  672. # ip_flag = 0
  673. # globals().update({ip_flag_name: ip_flag})
  674. # ip_index = 0
  675. # else:
  676. # ip_index = ip_flag % len(interface_ip_list)
  677. # ip_flag += 1
  678. #
  679. # if ip_flag >= 10000:
  680. # ip_flag = 0
  681. # globals().update({ip_flag_name: ip_flag})
  682. #
  683. # log("ip_pool " + interface_type + " " + str(ip_flag) + " " + str(interface_ip_list[ip_index]))
  684. # return interface_ip_list[ip_index]
  685. #
  686. #
  687. # def port_pool(interface_type, _random=False):
  688. # port_flag_name = interface_type + '_port_flag'
  689. #
  690. # port_flag = globals().get(port_flag_name)
  691. # if port_flag is None:
  692. # if _random:
  693. # if interface_type == "ocr":
  694. # _r = random.randint(0, len(ocr_port_list)-1)
  695. # elif interface_type == "otr":
  696. # _r = random.randint(0, len(otr_port_list)-1)
  697. # else:
  698. # _r = random.randint(0, len(soffice_port_list)-1)
  699. # port_flag = _r
  700. # globals().update({port_flag_name: port_flag})
  701. # port_index = _r
  702. # else:
  703. # port_flag = 0
  704. # globals().update({port_flag_name: port_flag})
  705. # port_index = 0
  706. # else:
  707. # if interface_type == "ocr":
  708. # port_index = port_flag % len(ocr_port_list)
  709. # elif interface_type == "otr":
  710. # port_index = port_flag % len(otr_port_list)
  711. # else:
  712. # port_index = port_flag % len(soffice_port_list)
  713. # port_flag += 1
  714. #
  715. # if port_flag >= 10000:
  716. # port_flag = 0
  717. # globals().update({port_flag_name: port_flag})
  718. #
  719. # if interface_type == "ocr":
  720. # log("port_pool " + interface_type + " " + str(port_flag) + " " + ocr_port_list[port_index])
  721. # return ocr_port_list[port_index]
  722. # elif interface_type == "otr":
  723. # log("port_pool " + interface_type + " " + str(port_flag) + " " + otr_port_list[port_index])
  724. # return otr_port_list[port_index]
  725. # else:
  726. # log("port_pool " + interface_type + " " + str(port_flag) + " " + soffice_port_list[port_index])
  727. # return soffice_port_list[port_index]