# CodeQ: an online programming tutor. # Copyright (C) 2015 UL FRI # # This program is free software: you can redistribute it and/or modify it under # the terms of the GNU Affero 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 Affero General Public License for more # details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import json import operator import os.path import threading from db.models import CodeqUser, Problem from db.util import make_identifier import monkey.rules import prolog.engine from prolog.util import used_predicates import server import server.user_session from server.problems import get_facts, load_file, load_language, load_problem, solutions_for_problems __all__ = ['PrologSession'] def format_prolog_output(reply, output): messages = [text for text in map(operator.itemgetter(1), output)] # When an engine is destroyed, a nested data object has the actual query result. event = reply['event'] if event == 'destroy': reply = reply['data'] event = reply['event'] if event == 'success': messages.append(prolog.engine.pretty_vars(reply['data'][0])) return messages, 'ok', True and reply['more'] if event == 'failure': messages.append('false') return messages, 'ok', False if event == 'error': # Remove potential module name (engine ID) from the error message. messages.append('error: ' + reply['data'].replace("'{}':".format(reply['id']), '')) return messages, 'error', False return messages, 'ok', False # TODO: is it possible to reach this return statement? class PrologSession(server.LanguageSession): """Abstracts a Prolog session. Only public methods are available to the outside world due to the use of multiprocessing managers. Therefore prefix any private methods with an underscore (_). No properties are accessible; use getters and setters instead. Values are passed by value instead of by reference (deep copy!). """ def __init__(self): self._access_lock = threading.Lock() self._engine_id = None self._sent_hints = [] def run(self, code): with self._access_lock: if self._engine_id is not None: prolog.engine.destroy(self._engine_id) self._engine_id = None engine_id, output = prolog.engine.create(code=code) if not engine_id: raise Exception('System error: could not create a prolog engine') self._engine_id = engine_id messages = [text for text in map(operator.itemgetter(1), output)] status = 'error' if 'error' in map(operator.itemgetter(0), output) else 'ok' return messages, status, False def query(self, query): with self._access_lock: if self._engine_id is None: return ['Prolog is not running'], 'error', False try: return format_prolog_output(*prolog.engine.ask(self._engine_id, query)) except Exception as e: return [str(e)], 'error', False def step(self): with self._access_lock: if self._engine_id is None: return ['Prolog is not running'], 'error', False try: return format_prolog_output(*prolog.engine.next(self._engine_id)) except Exception as e: return [str(e)], 'error', False def destroy(self): # this method was previously named: end() """Stops the Prolog engine.""" with self._access_lock: if self._engine_id is not None: prolog.engine.destroy(self._engine_id) self._engine_id = None return [], 'ok', False def __del__(self): # no locking needed if GC is removing us, as there cannot be any concurrent access by definition if hasattr(self, '_engine_id') and (self._engine_id is not None): prolog.engine.destroy(self._engine_id) self._engine_id = None # TODO remove this def hint(self, sid, problem_id, program): return [] def test(self, sid, problem_id, program): session = server.user_session.get_session_by_id(sid) problem = Problem.get(id=problem_id) language_module = load_language(problem.language, 'common') problem_module = load_problem(problem.language, problem.group, problem.identifier, 'common') aux_code = self._aux_code(session.get_uid(), problem, program) # experiment support for allowing none/automatic/manual/all hints # descriptor: {'id': 'hints', 'group': 'none|automatic|manual|all'} allowed_hints = 'all' for experiment in session.get_experiments(): if experiment.get('id') == 'hints': allowed_hints = experiment.get('group') break # check if the program is correct n_correct, n_all, msgs = problem_module.test(program, aux_code=aux_code) if n_correct == n_all: session.update_solution(problem_id, solution=program, done=True) else: hints = [] # syntax check if not hints and hasattr(language_module, 'check_syntax'): hints = language_module.check_syntax(program, aux_code=aux_code) # manually defined problem-specific hints if not hints and hasattr(problem_module, 'hint') and \ allowed_hints in ('all', 'manual'): hints = problem_module.hint(program, aux_code=aux_code) # automatic hints if not hints and allowed_hints in ('all', 'automatic'): bug_data = load_file(problem.language, problem.group, problem.identifier, 'bugs.json') if bug_data is not None: bugs = json.loads(bug_data) hints = monkey.rules.suggest(bugs, program) # generic language hints (style etc.) if not hints and hasattr(language_module, 'hint'): hints = language_module.hint(program, aux_code=aux_code) if hints: msgs.extend(hints) self._instantiate_and_save_hints(language_module, problem_module, msgs) return msgs # Return a string with definitions of aux. predicates used in [program] # (but only those that the user has already solved) and any required facts. def _aux_code(self, user_id, problem, program): problem_module = load_problem(problem.language, problem.group, problem.identifier, 'common') used_predicate_identifiers = {make_identifier(name) for name in used_predicates(program)} # FIXME this currently loads *all* solved predicates, since monkey.fix can # add goals that require other predicates, which causes the modified # program to fail if the predicate is not loaded (because it did not appear # in the initial program). dependencies = {p for p in CodeqUser.solved_problems(user_id, problem.language) if p[1] != problem.identifier} # TODO and p[1] in used_predicate_identifiers} return ('\n' + solutions_for_problems(problem.language, dependencies) + '\n' + get_facts(problem.language, problem_module)) # Add hint parameters (such as message index) based on hint class. Append # the finalized hints to the list of sent hints. def _instantiate_and_save_hints(self, language_mod, problem_mod, hints): with self._access_lock: for hint in hints: for mod in [language_mod, problem_mod]: if hasattr(mod, 'hint_type') and hint['id'] in mod.hint_type: hint_type = mod.hint_type[hint['id']] hint_type.instantiate(hint, self._sent_hints) self._sent_hints.extend(hints) def run_for_user(self, user_id, problem_id, program, query): """A "shorthand" method to start a Prolog session, load correct solutions of all user's solved problems and the given program, and ask a query. """ problem = Problem.get(id=problem_id) aux_code = self._aux_code(user_id, problem, program) messages, status, have_more = self.run(program+aux_code) if status == 'ok': more_messages, status, have_more = self.query(query) messages.extend(more_messages) return messages, status, have_more server.language_session_handlers['prolog'] = lambda user_session, problem_id, language_identifier, group_identifier, problem_identifier: PrologSession()