-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathapp.py
77 lines (64 loc) · 2.23 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
from flask import Flask, request, jsonify
import base64
import subprocess
import uuid
import os
from functools import lru_cache
app = Flask(__name__)
def getRandomID(length=8):
return uuid.uuid4().hex[:length]
@app.route('/image-sync', methods=['POST'])
def sync_ocr():
data = request.get_json(force=True)
bytes_base64 = data['image_data'].encode()
image_data = base64.b64decode(bytes_base64)
ID=getRandomID()
pathToImages=os.path.join('images',ID)
open(pathToImages, 'wb').write(image_data)
command = ['tesseract', pathToImages, 'stdout', '--psm', '1', '--oem', '1', 'quiet']
process = subprocess.run(command, stdout=subprocess.PIPE)
text=process.stdout.decode('utf-8')
out=jsonify({'text':text})
return out
class TaskManager:
def __init__(self):
self.tasks={}
def create(self,ID,item):
self.tasks[ID]=item
def find_finished(self,ID):
if ID in self.tasks and self.tasks[ID].poll()==0:
return True
return False
def get(self, ID):
process = self.tasks.pop(ID)
out = process.communicate()[0].decode('utf-8')
return out
# # alternative version to cached the result
# @lru_cache(maxsize=None)
# def get(self, ID):
# process = self.tasks[ID]
# out = process.communicate()[0].decode('utf-8')
# return out
taskManager=TaskManager()
@app.route('/image', methods=['POST'])
def async_ocr_post():
data = request.get_json(force=True)
bytes_base64 = data['image_data'].encode()
image_data = base64.b64decode(bytes_base64)
ID = getRandomID()
pathToImages=os.path.join('images',ID)
open(pathToImages, 'wb').write(image_data)
command = ['tesseract', pathToImages, 'stdout', '--psm', '1', '--oem', '1', 'quiet']
process = subprocess.Popen(command, stdout=subprocess.PIPE)
taskManager.create(ID,process)
out=jsonify({'task_id':ID})
return out
@app.route('/image', methods=['GET'])
def async_ocr_get():
data = request.get_json(force=True)
ID = data['task_id']
text=None
if taskManager.find_finished(ID):
text=taskManager.get(ID)
out = jsonify({'task_id': text})
return out