Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import argparse
- import sqlite3
- from uuid import UUID
- from flask import g, json, Flask
- from blocks.config import config
- from blocks.log.sqlite import ANCESTORS_QUERY
- app = Flask(__name__)
- @app.route('/experiments', methods=['GET'])
- def experiments():
- """List the experiments"""
- experiments = get_db().execute(
- # NOT IN returns NULL if *any* value in the subquery is NULL
- """SELECT DISTINCT uuid FROM status WHERE uuid NOT IN (
- SELECT value FROM status WHERE (
- key = 'resumed_from' AND value IS NOT NULL
- )
- )""").fetchall()
- return json.jsonify({
- 'experiments': [UUID(bytes=bytes(e[0])) for e in experiments]
- })
- @app.route('/experiments/<uuid>', methods=['get'])
- def experiment_uuid(uuid):
- """Get the status of an experiment, and a list of all channels."""
- b_uuid = sqlite3.Binary(UUID(uuid).bytes)
- status = get_db().execute("SELECT key, value FROM status WHERE uuid = ?",
- (b_uuid,)).fetchall()
- entries = get_db().execute(
- ANCESTORS_QUERY +
- "SELECT DISTINCT key FROM entries WHERE uuid IN ancestors", (b_uuid,)
- ).fetchall()
- return json.jsonify({
- 'status': {row[0]: row[1] for row in status
- if not isinstance(row[1], (sqlite3.Binary, bytes))},
- 'entries': [row[0] for row in entries]
- })
- @app.route('/experiments/<uuid>/<key>', methods=['get'])
- def experiment_uuid_key(uuid, key):
- """Get the values of a channel."""
- b_uuid = sqlite3.Binary(UUID(uuid).bytes)
- # TODO Confirm whether this sort works across platforms
- entries = get_db().execute(
- ANCESTORS_QUERY + "SELECT time, MIN(value) FROM entries "
- "JOIN ancestors ON entries.uuid = ancestors.parent "
- "WHERE uuid IN ancestors AND key = ? GROUP BY time",
- (b_uuid, key)
- ).fetchall()
- iterations_done, values = zip(*entries)
- return json.jsonify({
- 'iterations_done': iterations_done,
- key: values
- })
- def get_db():
- db = getattr(g, '_database', None)
- if db is None:
- db = sqlite3.connect(args.log)
- return db
- @app.teardown_appcontext
- def close_connection(exception):
- db = getattr(g, '_database', None)
- if db is not None:
- db.close()
- if __name__ == "__main__":
- parser = argparse.ArgumentParser(description='Interface for Blocks log.')
- parser.add_argument('log', nargs='?', type=str, help='The log to load.',
- default=config.sqlite_database)
- args = parser.parse_args()
- app.run(debug=True)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement