-
Notifications
You must be signed in to change notification settings - Fork 1
/
app.py
executable file
·157 lines (143 loc) · 6.06 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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
import config
from flask import Flask, request, render_template, render_template_string, Response, make_response, send_from_directory, redirect
from flask_sqlalchemy import SQLAlchemy, get_debug_queries
from temp_log import *
import json
from time import mktime, gmtime
from datetime import datetime, timedelta
from math import isnan, isinf
app = Flask("Thermometry", static_url_path="/static")
app.debug = False
app.config['SQLALCHEMY_DATABASE_URI'] = config.db
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = config.track_modifications
app.config['SQLALCHEMY_RECORD_QUERIES'] = True
db.init_app(app)
@app.after_request
def after_request(response):
if app.debug:
for query in get_debug_queries():
print("Query: %s\n\tParameters: %s\n\tDuration: %fs\n\tContext: %s\n" % (query.statement, query.parameters, query.duration, query.context))
return response
@app.route('/')
def index():
return redirect('/static/index.html', code=301)
@app.route('/<fridge_name>/<data_type>/', defaults={'sensor': None}, methods=['GET', 'POST'])
@app.route('/<fridge_name>/<data_type>/<sensor>')
def get_data(fridge_name, data_type, sensor):
# First pull up the relevant fridge
try:
fridge = Fridges.get_fridge(fridge_name)
except KeyError:
r = Response(render_template_string("Unable to find fridge {{fridge}}.", fridge=fridge_name))
r.status_code = 404
return r
# Are we looking for a supplementary field?
if data_type != 'data':
sup = fridge.supplementary.filter_by(name=data_type)
sup = sup.first()
if sup is None:
r = Response(render_template_string('Unable to find supplementary set {{suppl}}.', suppl=data_type))
r.status_code = 404
return r
source = sup
else:
sup = None
source = fridge
# Check whether we are trying to put new data in to the database
if request.method == 'POST':
return Response(insert_data(fridge, sup))
if 'count' in request.args:
try:
count = int(request.args['count'])
except ValueError:
count = 1000
else:
count = 1000
if any(x in ['legacy', 'single'] for x in request.args): # We only support fridges for this type of request
data_raw = fridge[-count:]
data = []
for row in data_raw:
data.append((row['Time'],
0, row["Four_K_RuO"],
0, row["Still_RuO"],
0, row["50mK_RuO"],
0, row["MC_CMN"],
0, row["MC_PT"]))
if 'single' in request.args:
data[-1] = data[-1] + (0, row["four_K_PT_front"])
if 'legacy' in request.args:
return Response(render_template("data_legacy.csv", data=data), mimetype='text/plain')
else:
return Response(render_template("data_single.csv", data=data), mimetype='text/plain')
elif 'sensors' in request.args:
data = [{'name': x.display_name, 'column_name': x.column_name} for x in source.sensors.filter_by(visible=1).order_by("view_order")]
elif 'summary' in request.args:
data_raw = source[-30:]
data = []
for row in data_raw:
if 'MC_Pt' in row: # Leiden Fridges
if row['MC_Pt'] > 20000 and row['MC_Pt'] < 500000:
data.append(row['MC_Pt'])
elif row['MC_Speer'] is not None and row['MC_Speer'] < 4000:
data.append(row['MC_Speer'])
else:
data.append(row['Four_K_RuO'])
elif 'MC' in row: # Bluefors Fridges
if 'Probe' in row and not isnan(row['Probe']):
data.append(row['Probe'])
elif isnan(row['MC']) or row['MC'] > 80.000:
if 'Still' in row and row['Still'] < 1.200:
data.append(0.007)
else:
data.append(row['Still'])
else:
data.append(row['MC'])
elif 'ProbeTemp' in row: # NMR Cryostat
data.append(float(row['ProbeTemp'] if row['ProbeTemp'] else 0)*1000)
else: # Also leiden
if row['Four_K_Pt'] is None or row['Four_K_Pt'] < 20000:
data.append(row['Four_K_RuO'])
else:
data.append(row['Four_K_Pt'])
elif 'current' in request.args:
data = source[-1]
data = dict(data)
data['Time'] = data['Time'].ctime()
for key in data.keys():
if isinstance(data[key], float) and (isinf(data[key]) or isnan(data[key])):
data[key] = None
else:
if 'hourly' in request.args:
try:
if not sensor:
raise KeyError("Sensor not found")
data_raw = list(source.hourly_avg(sensor))
except KeyError:
return Response("Sensor not found.")
elif 'start' in request.args or 'stop' in request.args:
start = datetime.fromtimestamp(float(request.args['start'])/1000)
stop = datetime.fromtimestamp(float(request.args['stop'])/1000)
stop += timedelta(1)
data_raw = list(source.range(start, stop))
else:
data_raw = list(source[-count:])
data = []
if not data_raw:
return Response("No data returned")
if sensor:
data = [(mktime(x['Time'].replace(tzinfo=None).timetuple())*1000, x[sensor] if x[sensor] and 0 < x[sensor] < 1000000 else None) for x in data_raw]
else:
return Response("Sensor not found.")
response = make_response(json.dumps(data))
response.mimetype = "application/json"
response.headers['Access-Control-Allow-Origin'] = "*"
return response
def insert_data(fridge, sup=None):
data = {x: y for x, y in request.form.items()}
if 'Time' in data:
data['Time'] = datetime.fromtimestamp(float(data['Time']))
if sup is not None:
sup.append(**data)
else:
fridge.append(**data)
return 'OK'