convert_need_interface.py 39 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033
  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 cv2
  14. import redis
  15. from werkzeug.exceptions import NotFound
  16. sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../")
  17. from atc.atc_interface import AtcModels, atc
  18. from idc.idc_interface import IdcModels, idc
  19. from isr.isr_interface import IsrModels, isr
  20. import traceback
  21. import requests
  22. from format_convert import _global
  23. from format_convert.utils import get_platform, get_sequential_data, judge_error_code, request_post, get_ip_port, \
  24. get_intranet_ip, get_logger, log, get_args_from_config, get_using_ip
  25. from ocr.ocr_interface import ocr, OcrModels
  26. from otr.otr_interface import otr, OtrModels
  27. from format_convert.libreoffice_interface import office_convert
  28. import numpy as np
  29. MAX_COMPUTE = False
  30. if get_platform() == "Windows":
  31. FROM_REMOTE = False
  32. only_test_ocr = False
  33. if only_test_ocr:
  34. ip_port_flag = {}
  35. ip_port_dict = get_ip_port()
  36. for _k in ip_port_dict.keys():
  37. ip_port_flag.update({_k: {"ocr": 0,
  38. "otr": 0,
  39. "convert": 0,
  40. "office": 0
  41. }})
  42. _global.update({"ip_port_flag": ip_port_flag})
  43. ip_port_dict["http://127.0.0.1"]["ocr"] = ["17000"]
  44. ip_port_dict["http://127.0.0.1"]["otr"] = ["18000"]
  45. _global.update({"ip_port": ip_port_dict})
  46. else:
  47. FROM_REMOTE = True
  48. if MAX_COMPUTE:
  49. FROM_REMOTE = False
  50. # ip_port_dict = get_ip_port()
  51. # ip = 'http://127.0.0.1'
  52. # ocr_port_list = ip_port_dict.get(ip).get("ocr")
  53. # otr_port_list = ip_port_dict.get(ip).get("otr")
  54. lock = multiprocessing.RLock()
  55. # 连接redis数据库
  56. # redis_db = redis.StrictRedis(host='192.168.2.103', port='6379',
  57. # db=1, password='bidi123456', health_check_interval=300)
  58. redis_db = None
  59. def _interface(_dict, time_out=60, retry_times=3):
  60. try:
  61. # 重试
  62. model_type = _dict.get("model_type")
  63. while retry_times:
  64. ip_port = interface_pool(model_type)
  65. if judge_error_code(ip_port):
  66. return ip_port
  67. _url = ip_port + "/" + model_type
  68. # base64_stream = base64.b64encode(pickle.dumps(_dict))
  69. r = json.loads(request_post(_url, {"data": json.dumps(_dict),
  70. "model_type": model_type}, time_out=time_out))
  71. log("get _interface return")
  72. if type(r) == list:
  73. # 接口连不上换个端口重试
  74. if retry_times <= 1:
  75. return r
  76. else:
  77. retry_times -= 1
  78. log("retry post _interface... left times " + str(retry_times) + " " + model_type)
  79. continue
  80. if judge_error_code(r):
  81. return r
  82. return r
  83. break
  84. except TimeoutError:
  85. return [-5]
  86. except requests.exceptions.ConnectionError as e:
  87. return [-2]
  88. def from_office_interface(src_path, dest_path, target_format, retry_times=1, from_remote=FROM_REMOTE):
  89. try:
  90. # Win10跳出超时装饰器
  91. # if get_platform() == "Windows":
  92. # # origin_office_convert = office_convert.__wrapped__
  93. # # file_path = origin_office_convert(src_path, dest_path, target_format, retry_times)
  94. # file_path = office_convert(src_path, dest_path, target_format, retry_times)
  95. # else:
  96. # # 将装饰器包装为一个类,否则多进程Pickle会报错 it's not the same object as xxx 问题,
  97. # # timeout_decorator_obj = my_timeout_decorator.TimeoutClass(office_convert, 180, TimeoutError)
  98. # # file_path = timeout_decorator_obj.run(src_path, dest_path, target_format, retry_times)
  99. #
  100. # file_path = office_convert(src_path, dest_path, target_format, retry_times)
  101. if from_remote:
  102. # 重试
  103. retry_times_1 = 1
  104. retry_times_2 = 2
  105. while retry_times_1 and retry_times_2:
  106. # _ip = ip_pool("soffice", _random=True)
  107. # _port = port_pool("soffice", _random=True)
  108. # _ip = interface_ip_list[0]
  109. # _port = "16002"
  110. # _ip, _port = interface_pool("soffice")
  111. # ip_port = from_schedule_interface("office")
  112. ip_port = interface_pool_gunicorn("office")
  113. if judge_error_code(ip_port):
  114. return ip_port
  115. _url = ip_port + "/soffice"
  116. with open(src_path, "rb") as f:
  117. file_bytes = f.read()
  118. base64_stream = base64.b64encode(file_bytes)
  119. start_time = time.time()
  120. r = json.loads(request_post(_url, {"src_path": src_path,
  121. "dest_path": dest_path,
  122. "file": base64_stream,
  123. "target_format": target_format,
  124. "retry_times": retry_times}, time_out=25))
  125. log("get interface return")
  126. log("office use time " + str(time.time()-start_time))
  127. if type(r) == list:
  128. # 接口连不上换个端口重试
  129. if retry_times_1 <= 1:
  130. return r
  131. else:
  132. retry_times_1 -= 1
  133. log("retry post office_interface... left times " + str(retry_times_1))
  134. continue
  135. file_str = r.get("data")
  136. if judge_error_code(file_str):
  137. if retry_times_2 <= 1:
  138. return file_str
  139. else:
  140. retry_times_2 -= 1
  141. continue
  142. file_bytes = eval(file_str)
  143. uid1 = src_path.split(os.sep)[-1].split(".")[0]
  144. file_path = dest_path + uid1 + "." + target_format
  145. if not os.path.exists(os.path.dirname(file_path)):
  146. os.makedirs(os.path.dirname(file_path), mode=0o777)
  147. with open(file_path, "wb") as f:
  148. f.write(file_bytes)
  149. break
  150. else:
  151. file_path = office_convert(src_path, dest_path, target_format, retry_times)
  152. if judge_error_code(file_path):
  153. return file_path
  154. return file_path
  155. except TimeoutError:
  156. log("from_office_interface timeout error!")
  157. return [-5]
  158. except:
  159. log("from_office_interface error!")
  160. print("from_office_interface", traceback.print_exc())
  161. return [-1]
  162. def from_ocr_interface(image_stream, is_table=False, from_remote=FROM_REMOTE):
  163. log("into from_ocr_interface")
  164. try:
  165. base64_stream = base64.b64encode(image_stream)
  166. # 调用接口
  167. try:
  168. if from_remote:
  169. retry_times_1 = 3
  170. # 重试
  171. while retry_times_1:
  172. # _ip = ip_pool("ocr", _random=True)
  173. # _port = port_pool("ocr", _random=True)
  174. # if _ip == interface_ip_list[1]:
  175. # _port = ocr_port_list[0]
  176. # _ip, _port = interface_pool("ocr")
  177. # ip_port = _ip + ":" + _port
  178. # ip_port = from_schedule_interface("ocr")
  179. ip_port = interface_pool_gunicorn("ocr")
  180. if judge_error_code(ip_port):
  181. return ip_port
  182. _url = ip_port + "/ocr"
  183. r = json.loads(request_post(_url, {"data": base64_stream,
  184. "md5": _global.get("md5")},
  185. time_out=60))
  186. log("get interface return")
  187. if type(r) == list:
  188. # 接口连不上换个端口重试
  189. if retry_times_1 <= 1:
  190. if is_table:
  191. return r, r
  192. else:
  193. return r
  194. else:
  195. retry_times_1 -= 1
  196. log("retry post ocr_interface... left times " + str(retry_times_1))
  197. continue
  198. if judge_error_code(r):
  199. return r
  200. break
  201. else:
  202. if globals().get("global_ocr_model") is None:
  203. print("=========== init ocr model ===========")
  204. globals().update({"global_ocr_model": OcrModels().get_model()})
  205. r = ocr(data=base64_stream, ocr_model=globals().get("global_ocr_model"))
  206. except TimeoutError:
  207. if is_table:
  208. return [-5], [-5]
  209. else:
  210. return [-5]
  211. except requests.exceptions.ConnectionError as e:
  212. if is_table:
  213. return [-2], [-2]
  214. else:
  215. return [-2]
  216. _dict = r
  217. text_list = eval(_dict.get("text"))
  218. bbox_list = eval(_dict.get("bbox"))
  219. if text_list is None:
  220. text_list = []
  221. if bbox_list is None:
  222. bbox_list = []
  223. if is_table:
  224. return text_list, bbox_list
  225. else:
  226. if text_list and bbox_list:
  227. text = get_sequential_data(text_list, bbox_list, html=True)
  228. if judge_error_code(text):
  229. return text
  230. else:
  231. text = ""
  232. return text
  233. except Exception as e:
  234. log("from_ocr_interface error!")
  235. # print("from_ocr_interface", e, global_type)
  236. if is_table:
  237. return [-1], [-1]
  238. else:
  239. return [-1]
  240. def from_gpu_interface_flask(_dict, model_type, predictor_type):
  241. log("into from_gpu_interface")
  242. start_time = time.time()
  243. try:
  244. # 调用接口
  245. _dict.update({"predictor_type": predictor_type, "model_type": model_type})
  246. if model_type == "ocr":
  247. use_zlib = True
  248. else:
  249. use_zlib = False
  250. result = _interface(_dict, time_out=30, retry_times=2, use_zlib=use_zlib)
  251. log("from_gpu_interface finish size " + str(sys.getsizeof(_dict)) + " time " + str(time.time()-start_time))
  252. return result
  253. except Exception as e:
  254. log("from_gpu_interface error!")
  255. log("from_gpu_interface failed " + str(time.time()-start_time))
  256. traceback.print_exc()
  257. return [-2]
  258. def from_gpu_interface_redis(_dict, model_type, predictor_type):
  259. log("into from_gpu_interface")
  260. start_time = time.time()
  261. try:
  262. # 调用接口
  263. _uuid = uuid.uuid1().hex
  264. _dict.update({"predictor_type": predictor_type, "model_type": model_type,
  265. "uuid": _uuid})
  266. _time = time.time()
  267. log("pickle.dumps(_dict)" + str(_dict))
  268. redis_db.rpush("producer_"+model_type, pickle.dumps(_dict))
  269. log("producer_" + model_type + " len " + str(redis_db.llen("producer_" + model_type)))
  270. log("to producer_" + model_type + " time " + str(time.time()-_time))
  271. _time = time.time()
  272. time_out = 300
  273. while True:
  274. time.sleep(0.2)
  275. if time.time() - _time > time_out:
  276. raise Exception
  277. if redis_db.hexists("consumer_"+model_type, _uuid):
  278. time1 = time.time()
  279. result = redis_db.hget("consumer_"+model_type, _uuid)
  280. log("from consumer_"+model_type + " time " + str(time.time()-time1))
  281. break
  282. result = pickle.loads(result)
  283. log("from_gpu_interface finish - size " + str(sys.getsizeof(_dict)) + " - time " + str(time.time()-start_time))
  284. return result
  285. except Exception as e:
  286. log("from_gpu_interface error!")
  287. log("from_gpu_interface failed " + str(time.time()-start_time))
  288. traceback.print_exc()
  289. return [-2]
  290. # def from_gpu_flask_sm(_dict, model_type, predictor_type):
  291. # log("into from_gpu_share_memory")
  292. # start_time = time.time()
  293. # shm = None
  294. # try:
  295. # # 放入共享内存
  296. # _time = time.time()
  297. # np_data = _dict.get("inputs")
  298. # shm = to_share_memory(np_data)
  299. # log("data into share memory " + str(shm.name) + " " + str(time.time()-_time))
  300. #
  301. # # 调用接口
  302. # _time = time.time()
  303. # _dict.pop("inputs")
  304. # _dict.update({"predictor_type": predictor_type, "model_type": model_type,
  305. # "sm_name": shm.name, "sm_shape": np_data.shape,
  306. # "sm_dtype": str(np_data.dtype)})
  307. # result = _interface(_dict, time_out=30, retry_times=2)
  308. # log("_interface cost " + str(time.time()-_time))
  309. #
  310. # # 读取共享内存
  311. # _time = time.time()
  312. # sm_name = result.get("sm_name")
  313. # sm_shape = result.get("sm_shape")
  314. # sm_dtype = result.get("sm_dtype")
  315. # sm_dtype = get_np_type(sm_dtype)
  316. # if sm_name:
  317. # outputs = from_share_memory(sm_name, sm_shape, sm_dtype)
  318. # else:
  319. # log("from_share_memory failed!")
  320. # raise Exception
  321. # log("data from share memory " + sm_name + " " + str(time.time()-_time))
  322. #
  323. # log("from_gpu_interface finish - size " + str(sys.getsizeof(_dict)) + " - time " + str(time.time()-start_time))
  324. # return {"preds": outputs, "gpu_time": result.get("gpu_time")}
  325. # except Exception as e:
  326. # log("from_gpu_interface failed " + str(time.time()-start_time))
  327. # traceback.print_exc()
  328. # return [-2]
  329. # finally:
  330. # # del b # Unnecessary; merely emphasizing the array is no longer used
  331. # if shm:
  332. # try:
  333. # shm.close()
  334. # shm.unlink()
  335. # except FileNotFoundError:
  336. # log("share memory " + shm.name + " not exists!")
  337. # except Exception:
  338. # traceback.print_exc()
  339. #
  340. #
  341. # def from_gpu_share_memory(_dict, model_type, predictor_type):
  342. # log("into from_gpu_share_memory")
  343. # start_time = time.time()
  344. # try:
  345. # _dict.update({"model_type": model_type, "predictor_type": predictor_type})
  346. # outputs, gpu_time = share_memory_pool(_dict)
  347. # log("from_gpu_share_memory finish - size " + str(sys.getsizeof(_dict)) + " - time " + str(time.time()-start_time))
  348. # return {"preds": outputs, "gpu_time": float(gpu_time)}
  349. # except Exception as e:
  350. # log("from_gpu_interface failed " + str(time.time()-start_time))
  351. # traceback.print_exc()
  352. # return [-2]
  353. def from_otr_interface2(image_stream):
  354. log("into from_otr_interface")
  355. try:
  356. base64_stream = base64.b64encode(image_stream)
  357. # 调用接口
  358. try:
  359. if globals().get("global_otr_model") is None:
  360. globals().update({"global_otr_model": OtrModels().get_model()})
  361. print("=========== init otr model ===========")
  362. r = otr(data=base64_stream, otr_model=globals().get("global_otr_model"))
  363. except TimeoutError:
  364. return [-5], [-5], [-5], [-5], [-5]
  365. except requests.exceptions.ConnectionError as e:
  366. log("from_otr_interface")
  367. print("from_otr_interface", traceback.print_exc())
  368. return [-2], [-2], [-2], [-2], [-2]
  369. # 处理结果
  370. _dict = r
  371. points = eval(_dict.get("points"))
  372. split_lines = eval(_dict.get("split_lines"))
  373. bboxes = eval(_dict.get("bboxes"))
  374. outline_points = eval(_dict.get("outline_points"))
  375. lines = eval(_dict.get("lines"))
  376. # print("from_otr_interface len(bboxes)", len(bboxes))
  377. if points is None:
  378. points = []
  379. if split_lines is None:
  380. split_lines = []
  381. if bboxes is None:
  382. bboxes = []
  383. if outline_points is None:
  384. outline_points = []
  385. if lines is None:
  386. lines = []
  387. return points, split_lines, bboxes, outline_points, lines
  388. except Exception as e:
  389. log("from_otr_interface error!")
  390. print("from_otr_interface", traceback.print_exc())
  391. return [-1], [-1], [-1], [-1], [-1]
  392. def from_otr_interface(image_stream, is_from_pdf=False, from_remote=FROM_REMOTE):
  393. log("into from_otr_interface")
  394. try:
  395. base64_stream = base64.b64encode(image_stream)
  396. # 调用接口
  397. try:
  398. if from_remote:
  399. log("from remote")
  400. retry_times_1 = 3
  401. # 重试
  402. while retry_times_1:
  403. # _ip = ip_pool("otr", _random=True)
  404. # _port = port_pool("otr", _random=True)
  405. # if _ip == interface_ip_list[1]:
  406. # _port = otr_port_list[0]
  407. ip_port = interface_pool_gunicorn("otr")
  408. # ip_port = from_schedule_interface("otr")
  409. if judge_error_code(ip_port):
  410. return ip_port
  411. _url = ip_port + "/otr"
  412. r = json.loads(request_post(_url, {"data": base64_stream,
  413. "is_from_pdf": is_from_pdf,
  414. "md5": _global.get("md5")}, time_out=60))
  415. log("get interface return")
  416. if type(r) == list:
  417. # 接口连不上换个端口重试
  418. if retry_times_1 <= 1:
  419. return r
  420. else:
  421. retry_times_1 -= 1
  422. log("retry post otr_interface... left times " + str(retry_times_1))
  423. continue
  424. if judge_error_code(r):
  425. return r
  426. break
  427. else:
  428. log("from local")
  429. log("otr_model " + str(globals().get("global_otr_model")))
  430. if globals().get("global_otr_model") is None:
  431. print("=========== init otr model ===========")
  432. globals().update({"global_otr_model": OtrModels().get_model()})
  433. log("init finish")
  434. r = otr(data=base64_stream, otr_model=globals().get("global_otr_model"), is_from_pdf=is_from_pdf)
  435. # r = otr(data=base64_stream, otr_model=None, is_from_pdf=is_from_pdf)
  436. except TimeoutError:
  437. return [-5]
  438. except requests.exceptions.ConnectionError as e:
  439. log("from_otr_interface")
  440. print("from_otr_interface", traceback.print_exc())
  441. return [-2]
  442. # 处理结果
  443. _dict = r
  444. list_line = eval(_dict.get("list_line"))
  445. return list_line
  446. except Exception as e:
  447. log("from_otr_interface error!")
  448. print("from_otr_interface", traceback.print_exc())
  449. return [-1]
  450. def from_isr_interface(image_stream, from_remote=FROM_REMOTE):
  451. log("into from_isr_interface")
  452. # os.environ["CUDA_VISIBLE_DEVICES"] = "0"
  453. start_time = time.time()
  454. try:
  455. base64_stream = base64.b64encode(image_stream)
  456. # 调用接口
  457. try:
  458. if from_remote:
  459. retry_times_1 = 3
  460. # 重试
  461. while retry_times_1:
  462. ip_port = interface_pool_gunicorn("isr")
  463. if judge_error_code(ip_port):
  464. return ip_port
  465. _url = ip_port + "/isr"
  466. r = json.loads(request_post(_url, {"data": base64_stream,
  467. "md5": _global.get("md5")},
  468. time_out=60))
  469. log("get interface return")
  470. if type(r) == list:
  471. # 接口连不上换个端口重试
  472. if retry_times_1 <= 1:
  473. return r
  474. else:
  475. retry_times_1 -= 1
  476. log("retry post isr_interface... left times " + str(retry_times_1))
  477. continue
  478. if judge_error_code(r):
  479. return r
  480. break
  481. else:
  482. if globals().get("global_isr_model") is None:
  483. print("=========== init isr model ===========")
  484. isr_yolo_model, isr_model = IsrModels().get_model()
  485. globals().update({"global_isr_yolo_model": isr_yolo_model})
  486. globals().update({"global_isr_model": isr_model})
  487. r = isr(data=base64_stream,
  488. isr_yolo_model=globals().get("global_isr_yolo_model"),
  489. isr_model=globals().get("global_isr_model"))
  490. except TimeoutError:
  491. return [-5]
  492. except requests.exceptions.ConnectionError as e:
  493. return [-2]
  494. _dict = r
  495. if from_remote:
  496. image_string = _dict.get("image")
  497. if judge_error_code(image_string):
  498. return image_string
  499. # [1]代表检测不到印章,直接返回
  500. if isinstance(image_string, list) and image_string == [1]:
  501. return image_string
  502. image_base64 = image_string.encode("utf-8")
  503. image_bytes = base64.b64decode(image_base64)
  504. buffer = np.frombuffer(image_bytes, dtype=np.uint8)
  505. image_np = cv2.imdecode(buffer, 1)
  506. else:
  507. image_np = _dict.get("image")
  508. log("from_isr_interface cost time " + str(time.time()-start_time))
  509. return image_np
  510. except Exception as e:
  511. log("from_isr_interface error!")
  512. traceback.print_exc()
  513. return [-11]
  514. finally:
  515. # os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
  516. pass
  517. def from_idc_interface(image_stream, from_remote=FROM_REMOTE):
  518. log("into from_idc_interface")
  519. start_time = time.time()
  520. try:
  521. base64_stream = base64.b64encode(image_stream)
  522. # 调用接口
  523. try:
  524. if from_remote:
  525. retry_times_1 = 3
  526. # 重试
  527. while retry_times_1:
  528. ip_port = interface_pool_gunicorn("idc")
  529. if judge_error_code(ip_port):
  530. return ip_port
  531. _url = ip_port + "/idc"
  532. r = json.loads(request_post(_url, {"data": base64_stream,
  533. "md5": _global.get("md5")},
  534. time_out=60))
  535. log("get interface return")
  536. if type(r) == list:
  537. # 接口连不上换个端口重试
  538. if retry_times_1 <= 1:
  539. return r
  540. else:
  541. retry_times_1 -= 1
  542. log("retry post idc_interface... left times " + str(retry_times_1))
  543. continue
  544. if judge_error_code(r):
  545. return r
  546. break
  547. else:
  548. if globals().get("global_idc_model") is None:
  549. print("=========== init idc model ===========")
  550. idc_model = IdcModels().get_model()
  551. globals().update({"global_idc_model": idc_model})
  552. r = idc(data=base64_stream,
  553. model=globals().get("global_idc_model"))
  554. except TimeoutError:
  555. return [-5]
  556. except requests.exceptions.ConnectionError as e:
  557. return [-2]
  558. _dict = r
  559. angle = _dict.get("angle")
  560. log("from_idc_interface cost time " + str(time.time()-start_time))
  561. return angle
  562. except Exception as e:
  563. log("from_idc_interface error!")
  564. traceback.print_exc()
  565. return [-10]
  566. def from_atc_interface(text, from_remote=FROM_REMOTE):
  567. log("into from_atc_interface")
  568. start_time = time.time()
  569. try:
  570. # 调用接口
  571. try:
  572. if from_remote:
  573. retry_times_1 = 3
  574. # 重试
  575. while retry_times_1:
  576. ip_port = interface_pool_gunicorn("atc")
  577. if judge_error_code(ip_port):
  578. return ip_port
  579. _url = ip_port + "/atc"
  580. r = json.loads(request_post(_url, {"data": text,
  581. "md5": _global.get("md5")},
  582. time_out=60))
  583. log("get interface return")
  584. if type(r) == list:
  585. # 接口连不上换个端口重试
  586. if retry_times_1 <= 1:
  587. return r
  588. else:
  589. retry_times_1 -= 1
  590. log("retry post atc_interface... left times " + str(retry_times_1))
  591. continue
  592. if judge_error_code(r):
  593. return r
  594. break
  595. else:
  596. if globals().get("global_atc_model") is None:
  597. print("=========== init atc model ===========")
  598. atc_model = AtcModels().get_model()
  599. globals().update({"global_atc_model": atc_model})
  600. r = atc(data=text,
  601. model=globals().get("global_atc_model"))
  602. except TimeoutError:
  603. return [-5]
  604. except requests.exceptions.ConnectionError as e:
  605. return [-2]
  606. _dict = r
  607. classification = _dict.get("classification")
  608. log("from_atc_interface cost time " + str(time.time()-start_time))
  609. return classification
  610. except Exception as e:
  611. log("from_atc_interface error!")
  612. traceback.print_exc()
  613. return [-10]
  614. # def from_schedule_interface(interface_type):
  615. # try:
  616. # _ip = "http://" + get_intranet_ip()
  617. # _port = ip_port_dict.get(_ip).get("schedule")[0]
  618. # _url = _ip + ":" + _port + "/schedule"
  619. # data = {"interface_type": interface_type}
  620. # result = json.loads(request_post(_url, data, time_out=10)).get("data")
  621. # if judge_error_code(result):
  622. # return result
  623. # _ip, _port = result
  624. # log("from_schedule_interface " + _ip + " " + _port)
  625. # return _ip + ":" + _port
  626. # except requests.exceptions.ConnectionError as e:
  627. # log("from_schedule_interface ConnectionError")
  628. # return [-2]
  629. # except:
  630. # log("from_schedule_interface error!")
  631. # traceback.print_exc()
  632. # return [-1]
  633. def interface_pool(interface_type, use_gunicorn=True):
  634. ip_port_flag = _global.get("ip_port_flag")
  635. ip_port_dict = _global.get("ip_port")
  636. try:
  637. if use_gunicorn:
  638. _ip = "http://127.0.0.1"
  639. _port = ip_port_dict.get(_ip).get(interface_type)[0]
  640. ip_port = _ip + ":" + str(_port)
  641. log(ip_port)
  642. return ip_port
  643. # 负载均衡, 选取ip
  644. interface_load_list = []
  645. for _ip in ip_port_flag.keys():
  646. if ip_port_dict.get(_ip).get(interface_type):
  647. load_scale = ip_port_flag.get(_ip).get(interface_type) / len(ip_port_dict.get(_ip).get(interface_type))
  648. interface_load_list.append([_ip, load_scale])
  649. if not interface_load_list:
  650. raise NotFound
  651. interface_load_list.sort(key=lambda x: x[-1])
  652. _ip = interface_load_list[0][0]
  653. # 负载均衡, 选取port
  654. ip_type_cnt = ip_port_flag.get(_ip).get(interface_type)
  655. ip_type_total = len(ip_port_dict.get(_ip).get(interface_type))
  656. if ip_type_cnt == 0:
  657. ip_type_cnt = random.randint(0, ip_type_total-1)
  658. port_index = ip_type_cnt % ip_type_total
  659. _port = ip_port_dict.get(_ip).get(interface_type)[port_index]
  660. # 更新flag
  661. current_flag = ip_type_cnt
  662. if current_flag >= 10000:
  663. ip_port_flag[_ip][interface_type] = 0
  664. else:
  665. ip_port_flag[_ip][interface_type] = current_flag + 1
  666. _global.update({"ip_port_flag": ip_port_flag})
  667. # log(str(_global.get("ip_port_flag")))
  668. ip_port = _ip + ":" + str(_port)
  669. log(ip_port)
  670. return ip_port
  671. except NotFound:
  672. log("cannot read ip from config! checkout config")
  673. return [-2]
  674. except:
  675. traceback.print_exc()
  676. return [-1]
  677. def interface_pool_gunicorn(interface_type):
  678. ip_port_flag_dict = _global.get("ip_port_flag")
  679. ip_port_dict = _global.get("ip_port")
  680. try:
  681. if ip_port_dict is None or ip_port_flag_dict is None:
  682. raise NotFound
  683. # 负载均衡, 选取有该接口的ip
  684. min_cnt = 10000.
  685. interface_cnt = 0
  686. _ip = None
  687. port_list = []
  688. for key in ip_port_flag_dict.keys():
  689. temp_port_list = get_args_from_config(ip_port_dict, key, interface_type)
  690. if not temp_port_list:
  691. continue
  692. interface_cnt = ip_port_flag_dict.get(key).get(interface_type)
  693. if interface_cnt is not None and interface_cnt / len(temp_port_list[0]) < min_cnt:
  694. _ip = key
  695. min_cnt = interface_cnt / len(temp_port_list[0])
  696. port_list = temp_port_list[0]
  697. # 选取端口
  698. if interface_type == "office":
  699. # 刚开始随机,后续求余
  700. if min_cnt == 0:
  701. _port = port_list[random.randint(0, len(port_list)-1)]
  702. ip_port_flag_dict[_ip][interface_type] = int(_port[-2:])
  703. else:
  704. _port = port_list[interface_cnt % len(port_list)]
  705. else:
  706. # 使用gunicorn则直接选第一个
  707. _port = port_list[0]
  708. # 更新flag
  709. if ip_port_flag_dict.get(_ip).get(interface_type) >= 10000:
  710. ip_port_flag_dict[_ip][interface_type] = 0
  711. else:
  712. ip_port_flag_dict[_ip][interface_type] += 1
  713. _global.update({"ip_port_flag": ip_port_flag_dict})
  714. ip_port = _ip + ":" + str(_port)
  715. log(interface_type)
  716. log(ip_port)
  717. return ip_port
  718. except NotFound:
  719. log("ip_flag or ip_port_dict is None! checkout config")
  720. return [-2]
  721. except:
  722. traceback.print_exc()
  723. return [-1]
  724. def interface_pool_gunicorn_old(interface_type):
  725. ip_flag_list = _global.get("ip_flag")
  726. ip_port_flag_dict = _global.get("ip_port_flag")
  727. ip_port_dict = _global.get("ip_port")
  728. try:
  729. if ip_flag_list is None or ip_port_dict is None or ip_port_flag_dict is None:
  730. raise NotFound
  731. if interface_type == "office":
  732. # _ip = "http://127.0.0.1"
  733. _ip = get_using_ip()
  734. # 选取端口
  735. port_list = ip_port_dict.get(_ip).get("MASTER").get(interface_type)
  736. ip_type_cnt = ip_port_flag_dict.get(_ip).get(interface_type)
  737. if ip_type_cnt == 0:
  738. _port = port_list[random.randint(0, len(port_list)-1)]
  739. else:
  740. _port = port_list[ip_type_cnt % len(port_list)]
  741. # 更新flag
  742. if ip_port_flag_dict.get(_ip).get(interface_type) >= 10000:
  743. ip_port_flag_dict[_ip][interface_type] = 0
  744. else:
  745. ip_port_flag_dict[_ip][interface_type] += 1
  746. _global.update({"ip_port_flag": ip_port_flag_dict})
  747. else:
  748. # 负载均衡, 选取ip
  749. ip_flag_list.sort(key=lambda x: x[1])
  750. if ip_flag_list[-1][1] == 0:
  751. ip_index = random.randint(0, len(ip_flag_list)-1)
  752. else:
  753. ip_index = 0
  754. _ip = ip_flag_list[ip_index][0]
  755. if "master" in _ip:
  756. port_index = 1
  757. else:
  758. port_index = 0
  759. _ip = _ip.split("_")[0]
  760. # 选取端口, 使用gunicorn则直接选第一个
  761. # _port = ip_port_dict.get(_ip).get("MASTER").get(interface_type)[0]
  762. log("_ip " + _ip)
  763. log("interface_type " + interface_type)
  764. port_list = get_args_from_config(ip_port_dict, _ip, interface_type)
  765. log("port_list" + str(port_list))
  766. if port_index >= len(port_list):
  767. port_index = 0
  768. _port = port_list[port_index][0]
  769. # # 选取端口, 使用gunicorn则直接选第一个
  770. # _ip = _ip.split("_")[0]
  771. # port_list = get_args_from_config(ip_port_dict, _ip, interface_type)
  772. # if
  773. # print(port_list)
  774. # _port = port_list[0][0]
  775. # 更新flag
  776. if ip_flag_list[ip_index][1] >= 10000:
  777. ip_flag_list[ip_index][1] = 0
  778. else:
  779. ip_flag_list[ip_index][1] += + 1
  780. _global.update({"ip_flag": ip_flag_list})
  781. ip_port = _ip + ":" + str(_port)
  782. log(ip_port)
  783. return ip_port
  784. except NotFound:
  785. log("ip_flag or ip_port_dict is None! checkout config")
  786. return [-2]
  787. except:
  788. traceback.print_exc()
  789. return [-1]
  790. # def share_memory_pool(args_dict):
  791. # np_data = args_dict.get("inputs")
  792. # _type = args_dict.get("model_type")
  793. # args_dict.update({"sm_shape": np_data.shape, "sm_dtype": str(np_data.dtype)})
  794. #
  795. # if _type == 'ocr':
  796. # port_list = ocr_port_list
  797. # elif _type == 'otr':
  798. # port_list = otr_port_list
  799. # else:
  800. # log("type error! only support ocr otr")
  801. # raise Exception
  802. #
  803. # # 循环判断是否有空的share memory
  804. # empty_sm_list = None
  805. # sm_list_name = ""
  806. # while empty_sm_list is None:
  807. # for p in port_list:
  808. # sm_list_name = "sml_"+_type+"_"+str(p)
  809. # sm_list = get_share_memory_list(sm_list_name)
  810. # if sm_list[0] == "0":
  811. # lock.acquire(timeout=0.1)
  812. # if sm_list[0] == "0":
  813. # sm_list[0] = "1"
  814. # sm_list[-1] = "0"
  815. # empty_sm_list = sm_list
  816. # break
  817. # else:
  818. # continue
  819. # lock.release()
  820. #
  821. # log(str(os.getppid()) + " empty_sm_list " + sm_list_name)
  822. #
  823. # # numpy放入共享内存
  824. # _time = time.time()
  825. # release_share_memory(get_share_memory("psm_" + str(os.getpid())))
  826. # shm = to_share_memory(np_data)
  827. # log("data into share memory " + str(shm.name) + " " + str(time.time()-_time))
  828. #
  829. # # 参数放入共享内存列表
  830. # empty_sm_list[1] = args_dict.get("md5")
  831. # empty_sm_list[2] = args_dict.get("model_type")
  832. # empty_sm_list[3] = args_dict.get("predictor_type")
  833. # empty_sm_list[4] = args_dict.get("args")
  834. # empty_sm_list[5] = str(shm.name)
  835. # empty_sm_list[6] = str(args_dict.get("sm_shape"))
  836. # empty_sm_list[7] = args_dict.get("sm_dtype")
  837. # empty_sm_list[-1] = "1"
  838. # # log("empty_sm_list[7] " + empty_sm_list[7])
  839. # close_share_memory_list(empty_sm_list)
  840. #
  841. # # 循环判断是否完成
  842. # finish_sm_list = get_share_memory_list(sm_list_name)
  843. # while True:
  844. # if finish_sm_list[-1] == "0":
  845. # break
  846. #
  847. # # 读取共享内存
  848. # _time = time.time()
  849. # sm_name = finish_sm_list[5]
  850. # sm_shape = finish_sm_list[6]
  851. # sm_shape = eval(sm_shape)
  852. # sm_dtype = finish_sm_list[7]
  853. # gpu_time = finish_sm_list[8]
  854. # sm_dtype = get_np_type(sm_dtype)
  855. # outputs = from_share_memory(sm_name, sm_shape, sm_dtype)
  856. # log(args_dict.get("model_type") + " " + args_dict.get("predictor_type") + " outputs " + str(outputs.shape))
  857. # log("data from share memory " + sm_name + " " + str(time.time()-_time))
  858. #
  859. # # 释放
  860. # release_share_memory(get_share_memory(sm_name))
  861. #
  862. # # 重置share memory list
  863. # finish_sm_list[-1] = "0"
  864. # finish_sm_list[0] = "0"
  865. #
  866. # close_share_memory_list(finish_sm_list)
  867. # return outputs, gpu_time
  868. # def interface_pool(interface_type):
  869. # try:
  870. # ip_port_dict = _global.get("ip_port")
  871. # ip_list = list(ip_port_dict.keys())
  872. # _ip = random.choice(ip_list)
  873. # if interface_type != 'office':
  874. # _port = ip_port_dict.get(_ip).get(interface_type)[0]
  875. # else:
  876. # _port = random.choice(ip_port_dict.get(_ip).get(interface_type))
  877. # log(_ip + ":" + _port)
  878. # return _ip + ":" + _port
  879. # except Exception as e:
  880. # traceback.print_exc()
  881. # return [-1]
  882. # def ip_pool(interface_type, _random=False):
  883. # ip_flag_name = interface_type + '_ip_flag'
  884. # ip_flag = globals().get(ip_flag_name)
  885. # if ip_flag is None:
  886. # if _random:
  887. # _r = random.randint(0, len(interface_ip_list)-1)
  888. # ip_flag = _r
  889. # globals().update({ip_flag_name: ip_flag})
  890. # ip_index = _r
  891. # else:
  892. # ip_flag = 0
  893. # globals().update({ip_flag_name: ip_flag})
  894. # ip_index = 0
  895. # else:
  896. # ip_index = ip_flag % len(interface_ip_list)
  897. # ip_flag += 1
  898. #
  899. # if ip_flag >= 10000:
  900. # ip_flag = 0
  901. # globals().update({ip_flag_name: ip_flag})
  902. #
  903. # log("ip_pool " + interface_type + " " + str(ip_flag) + " " + str(interface_ip_list[ip_index]))
  904. # return interface_ip_list[ip_index]
  905. #
  906. #
  907. # def port_pool(interface_type, _random=False):
  908. # port_flag_name = interface_type + '_port_flag'
  909. #
  910. # port_flag = globals().get(port_flag_name)
  911. # if port_flag is None:
  912. # if _random:
  913. # if interface_type == "ocr":
  914. # _r = random.randint(0, len(ocr_port_list)-1)
  915. # elif interface_type == "otr":
  916. # _r = random.randint(0, len(otr_port_list)-1)
  917. # else:
  918. # _r = random.randint(0, len(soffice_port_list)-1)
  919. # port_flag = _r
  920. # globals().update({port_flag_name: port_flag})
  921. # port_index = _r
  922. # else:
  923. # port_flag = 0
  924. # globals().update({port_flag_name: port_flag})
  925. # port_index = 0
  926. # else:
  927. # if interface_type == "ocr":
  928. # port_index = port_flag % len(ocr_port_list)
  929. # elif interface_type == "otr":
  930. # port_index = port_flag % len(otr_port_list)
  931. # else:
  932. # port_index = port_flag % len(soffice_port_list)
  933. # port_flag += 1
  934. #
  935. # if port_flag >= 10000:
  936. # port_flag = 0
  937. # globals().update({port_flag_name: port_flag})
  938. #
  939. # if interface_type == "ocr":
  940. # log("port_pool " + interface_type + " " + str(port_flag) + " " + ocr_port_list[port_index])
  941. # return ocr_port_list[port_index]
  942. # elif interface_type == "otr":
  943. # log("port_pool " + interface_type + " " + str(port_flag) + " " + otr_port_list[port_index])
  944. # return otr_port_list[port_index]
  945. # else:
  946. # log("port_pool " + interface_type + " " + str(port_flag) + " " + soffice_port_list[port_index])
  947. # return soffice_port_list[port_index]
  948. if __name__ == "__main__":
  949. from format_convert.utils import set_flask_global
  950. _global._init()
  951. set_flask_global()
  952. for i in range(10):
  953. print("result", interface_pool_gunicorn("otr"))