convert_need_interface.py 29 KB

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