Source code for idp_server.rest

# Copyright 2019 Ingmar Dasseville, Pierre Carbonnelle
#
# This file is part of Interactive_Consultant.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""
This module implements the IDP-Z3 web server

To profile it, set with_profiling to True
"""

with_png = False
with_profiling = False

from contextlib import redirect_stdout
import io
import os
import threading
import traceback

from flask import Flask, g, send_from_directory  # g is required for pyinstrument
from flask_cors import CORS
from flask_restful import Resource, Api, reqparse

from idp_engine import IDP
from idp_engine.utils import log
from .State import State
from .Inferences import explain, abstract
from .IO import Output, metaJSON

from typing import Dict

if with_profiling:
    from pyinstrument import Profiler

if with_png:
    # library to generate call graph, for documentation purposes
    from pycallgraph2 import PyCallGraph
    from pycallgraph2.output import GraphvizOutput
    from pycallgraph2 import GlobbingFilter
    from pycallgraph2 import Config
    config = Config(max_depth=8)
    config.trace_filter = GlobbingFilter(
        exclude=['arpeggio.*', 'ast.*', 'flask*', 'json.*', 'pycallgraph2.*',
                'textx.*', 'werkzeug.*', 'z3.*']
        )


current_dir = os.path.dirname(os.path.realpath(__file__))
parent_dir = os.path.join(current_dir, os.pardir)

static_file_dir = os.path.join(current_dir, 'static')
examples_file_dir = os.path.join(current_dir, 'examples')
docs_file_dir = os.path.join(parent_dir, 'docs')
app = Flask(__name__)
CORS(app)

api = Api(app)

parser = reqparse.RequestParser()
parser.add_argument('method', type=str, help='Method to execute')
parser.add_argument('code', type=str, help='Code')
parser.add_argument('active', type=str, help='Three-valued structure')
parser.add_argument('previous_active', type=str, help='Previous input by user')
parser.add_argument('expanded', type=str, action='append', help='list of expanded symbols')
parser.add_argument('symbol', type=str, help='Symbol to explain or optimize')
parser.add_argument('value', type=str, help='Value to explain')
parser.add_argument('minimize', type=bool, help='True -> minimize ; False -> maximize')


[docs]class HelloWorld(Resource): def get(self): import time time.sleep(100) return {'hello': 'world'}
z3lock = threading.Lock() idps: Dict[str, IDP] = {} # {code_string : idp}
[docs]def idpOf(code): """ Function to retrieve an IDP object for IDP code. If the object doesn't exist yet, we create it. `idps` is a dict which contains an IDP object for each IDP code. This way, easy caching can be achieved. :arg code: the IDP code. :returns IDP: the IDP object. """ global idps if code in idps: return idps[code] else: idp = IDP.parse(code) if 20 < len(idps): # remove oldest entry, to prevent memory overflow idps = {k: v for k, v in list(idps.items())[1:]} idps[code] = idp return idp
[docs]class run(Resource): """ Class which handles the run. <<Explanation of what the run is here.>> :arg Resource: <<explanation of resource>> """
[docs] def post(self): """ Method to run an IDP program with a procedure block. :returns stdout. """ log("start /run") with z3lock: try: args = parser.parse_args() try: idp = idpOf(args['code']) # capture stdout, print() with io.StringIO() as buf, redirect_stdout(buf): try: idp.execute() except Exception as exc: print(exc) out = buf.getvalue() return out except Exception as exc: traceback.print_exc() return str(exc) except Exception as exc: traceback.print_exc() return str(exc)
[docs]class meta(Resource): """ Class which handles the meta. <<Explanation of what the meta is here.>> :arg Resource: <<explanation of resource>> """
[docs] def post(self): """ Method to export the metaJSON from the resource. :returns metaJSON: a json string containing the meta. """ log("start /meta") with z3lock: try: args = parser.parse_args() try: idp = idpOf(args['code']) state = State.make(idp, "{}", "{}") out = metaJSON(state) out["propagated"] = Output(state).fill(state) return out except Exception as exc: traceback.print_exc() return str(exc) except Exception as exc: traceback.print_exc() return str(exc)
[docs]class metaWithGraph(meta): # subclass that generates call graphs
[docs] def post(self): graphviz = GraphvizOutput() graphviz.output_file = 'docs/meta.png' with PyCallGraph(output=graphviz, config=config): return super().post()
[docs]class eval(Resource): def post(self): log("start /eval") with z3lock: try: if with_profiling: g.profiler = Profiler() g.profiler.start() args = parser.parse_args() idp = idpOf(args['code']) method = args['method'] given_json = args['active'] previous_active = args.get('previous_active', None) expanded = tuple([]) if args['expanded'] is None else tuple(args['expanded']) field = args.get('field', None) state = State.make(idp, previous_active, given_json) out = {} if method == "propagate": if field: # field is the applied symbol for which a dropdown is open state = state.get_range(field) out = Output(state).fill(state) if method == "modelexpand": generator = state.expand(max=1, complete=False) state.assignments = list(generator)[0] out = Output(state).fill(state) if method == "explain": out = explain(state, args['value']) if method == "minimize": state = state.optimize(args['symbol'], args['minimize'], complete=False) out = Output(state).fill(state) if method == "abstract": if args['symbol'] != "": # theory to explain ? newTheory = (str(IDP.parse(args['code']).vocabulary) + "theory {\n" + args['symbol'] + "\n}\n" ) idpModel = IDP.parse(newTheory) expanded = {} # for expr in idpModel.subtences.values(): # expanded.update(expr.collect_symbols()) expanded = tuple(expanded.keys()) state = State.make(idpModel, "") out = abstract(state, given_json) log("end /eval " + method) if with_profiling: g.profiler.stop() print(g.profiler.output_text(unicode=True, color=True)) return out except Exception as exc: if with_profiling: g.profiler.stop() traceback.print_exc() return str(exc)
[docs]class evalWithGraph(eval): # subcclass that generates call graphs def post(self): args = parser.parse_args() method = args['method'] graphviz = GraphvizOutput() graphviz.output_file = 'docs/' + method + '.png' with PyCallGraph(output=graphviz, config=config): return super().post()
@app.route('/', methods=['GET']) def serve_dir_directory_index(): return send_from_directory(static_file_dir, 'index.html') @app.route('/IDE', methods=['GET']) def serve_IDE(): return send_from_directory(static_file_dir, 'index.html') @app.route('/<path:path>', methods=['GET']) def serve_file_in_dir(path): if not os.path.isfile(os.path.join(static_file_dir, path)): path = os.path.join(path, 'index.html') return send_from_directory(static_file_dir, path) @app.route('/examples/<path:path>', methods=['GET']) def serve_examples_file(path): if not os.path.isfile(os.path.join(examples_file_dir, path)): return "file not found: " + path return send_from_directory(examples_file_dir, path) # local help files @app.route('/docs/<path:path>', methods=['GET']) def serve_docs_file(path): if not os.path.isfile(os.path.join(docs_file_dir, path)): return "file not found: " + path return send_from_directory(docs_file_dir, path) api.add_resource(HelloWorld, '/test') if with_png: api.add_resource(metaWithGraph, '/meta') api.add_resource(evalWithGraph, '/eval') else: api.add_resource(run, '/run') api.add_resource(meta, '/meta') api.add_resource(eval, '/eval') if __name__ == '__main__': app.run(debug=True)