-
Notifications
You must be signed in to change notification settings - Fork 3
/
opencpu.py
81 lines (64 loc) · 2.49 KB
/
opencpu.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
78
79
80
81
from io import BytesIO
import json
from typing import Any, Dict, List, Optional
from flask import current_app, Response
import pandas
import requests
class OpenCPUException(Exception):
def __init__(self, msg: str, method: str, response):
super().__init__(self, msg)
self.method = method
self.response = response
@property
def error_response(self):
return Response(self.response.content, status=self.response.status_code,
mimetype='text/plain')
def r_json_to_pandas(data: List[Dict[str, Any]]) -> pandas.DataFrame:
df = pandas.DataFrame(data=data)
if '_row' in df:
df = df.set_index('_row')
return df
def opencpu_request(method: str, files: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None, return_type='csv'):
files = files or {}
params = params or {}
params = {
k: json.dumps(v) for k, v in params.items()
}
url = current_app.config['OPENCPU_API_ROOT'] + '/viime/R/' + method
if return_type == 'csv':
url += '/csv?row.names=true'
elif return_type == 'png':
url += '/png'
elif return_type == 'json':
url += '/json'
else:
raise Exception('Unknown return type')
try:
resp = requests.post(url, files=files, data=params)
except requests.exceptions.RequestException as e:
raise OpenCPUException('Error connecting to OpenCPU server', method, e.response)
if not resp.ok:
current_app.logger.error(resp.content)
raise OpenCPUException(f'OpenCPU error calling {method}', method, resp)
result = resp.content
if return_type == 'csv':
result = pandas.read_csv(BytesIO(resp.content), index_col=0)
elif return_type == 'json':
result = json.loads(result)
return result
def process_table(method: str, table: pandas.DataFrame, params: Optional[Dict[str, Any]] = None):
files = {
'table': ('table.csv', table.to_csv().encode())
}
result = opencpu_request(method, files=files, params=params)
return Response(result.to_csv(), mimetype='text/csv')
def generate_image(method: str, table: pandas.DataFrame, params: Optional[Dict[str, Any]] = None):
files = {
'table': ('table.csv', table.to_csv().encode())
}
return opencpu_request(method, files=files, params=params, return_type='png')
if __name__ == '__main__':
import sys
table = pandas.read_csv(sys.argv[2], index_col=0)
print(process_table(sys.argv[1], table))