Skip to content

Instantly share code, notes, and snippets.

@JamesSaxon
Last active April 23, 2021 02:38
Show Gist options
  • Save JamesSaxon/a5474f19f86a77350372e6f767c48573 to your computer and use it in GitHub Desktop.
Save JamesSaxon/a5474f19f86a77350372e6f767c48573 to your computer and use it in GitHub Desktop.
minimal user interactions and ndt7 test through vanilla wsgi
import time, sqlite3
from urllib.parse import parse_qs
import json
html = """
<html>
<body>
<p>
NDT7 LAN Speedtest:
<button id="ndt7-button" onclick="launch_ndt7()" >Start!</button>
<table id="ndt7-table" hidden >
<tr><td>&#8593;</td><td id='download-cell'></td><td>Mbps</td></tr></div>
<tr><td>&#8595;</td><td id='upload-cell'> </td><td>Mbps</td></tr></div>
</table>
</p>
How is your network performing?
<form id='feedback' action="" method="post">
Awful! <input name="perf" type="range" min="1" max="100" value="{}"> Great!<br>
<input type="hidden" name="dl" id="download-form" value="-1">
<input type="hidden" name="ul" id="upload-form" value="-1">
<input type="submit" value="Submit">
</form>
{}
<script type='text/javascript' src='ndt7-core.js'></script>
<script type='text/javascript' src='ndt7-run.js'></script>
</body>
</html>
"""
def get_and_save_post_data(env, sqlite_fname = "/var/www/scripts/wsgi.sqlite"):
con = sqlite3.connect(sqlite_fname)
# con.execute("CREATE TABLE IF NOT EXISTS wsgi (ts INT, perf INT, dl NUMBER, ul NUMBER);")
try:
request_body_size = int(env.get('CONTENT_LENGTH', 0))
except:
return {}
if not request_body_size: return {}
request_body = env['wsgi.input'].read(request_body_size)
parsed_data = parse_qs(request_body)
data = {k.decode('utf-8') : float(v[0]) for k, v in parsed_data.items()}
# data = json.loads(request_body) ## For json
now = int(time.time())
val = data["perf"]
dl = data["dl"]
ul = data["ul"]
con.execute(f"INSERT INTO wsgi VALUES ({now}, {val}, {dl}, {ul});")
con.commit()
con.close()
return data
def get_recent_results(N = 10, sqlite_fname = "/var/www/scripts/wsgi.sqlite"):
con = sqlite3.connect(sqlite_fname)
results = con.execute(f"SELECT ts, perf, dl, ul FROM wsgi ORDER BY ts DESC LIMIT {N};")
txt = '<p>'
txt+= '<table>'
txt+= '<tr><td>Timestamp</td><td>Subjective</td><td>Download [Mbps]</td><td>Upload [Mbps]</td>\n'
for row in results:
txt += '<tr><td>{}</td><td>{}</td><td>{}</td><td>{}</td>\n'.format(*row)
txt += "</table>\n</p>"
con.close()
return txt
def application(environ, start_response):
now = str(int(time.time()))
ts_txt = '<p>It is currently: {}</p>\n'.format(now)
data = get_and_save_post_data(environ)
value = data.get("perf", 50)
perf_txt = ""
if "perf" in data:
perf_txt = "<p>At {}: {}.</p>".format(now, value)
recent_txt = get_recent_results()
txt = perf_txt + recent_txt + ts_txt
start_response('200 OK', [('Content-type','text/html')])
return [html.format(value, txt).encode('UTF-8')]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment