# coding=utf-8 import operator import threading import prolog.engine import server import server.user_session from db.models import CodeqUser, Problem from . import 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._problem_id = -1 self._sent_hints = [] def run(self, code): with self._access_lock: if self._engine_id is not None: prolog.engine.stop(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 self._problem_id = -1 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 def hint(self, sid, problem_id, program): session = server.user_session.get_session_by_id(sid) p = Problem.get(id=problem_id) language_module = problems.load_language(p.language, 'common') problem_module = problems.load_problem(p.language, p.group, p.identifier, 'common') solved_problems = [pp for pp in CodeqUser.solved_problems(session.get_uid(), p.language) if pp != (p.group, p.identifier)] hints = [] # check if the program is already correct passed, _ = problem_module.test(program, solved_problems) if passed: hints = [{'id': 'program_already_correct'}] if not hints and hasattr(language_module, 'hint'): hints = language_module.hint(program, solved_problems) if not hints and hasattr(problem_module, 'hint'): hints = problem_module.hint(program, solved_problems) if not hints: hints = [{'id': 'no_hint'}] self._instantiate_and_save_hints(language_module, problem_module, hints) return hints def test(self, sid, problem_id, program): session = server.user_session.get_session_by_id(sid) p = Problem.get(id=problem_id) language_module = problems.load_language(p.language, 'common') problem_module = problems.load_problem(p.language, p.group, p.identifier, 'common') solved_problems = [pp for pp in CodeqUser.solved_problems(session.get_uid(), p.language) if pp != (p.group, p.identifier)] try: passed, hints = problem_module.test(program, solved_problems) if passed: session.update_solution(problem_id, done=True) except AttributeError as ex: hints = [{'id': 'system_error', 'args': {'message': 'test function does not exist'}}] self._instantiate_and_save_hints(language_module, problem_module, hints) return hints # 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. """ p = Problem.get(id=problem_id) problem_module = problems.load_problem(p.language, p.group, p.identifier, 'common') solved_problems = [pp for pp in CodeqUser.solved_problems(user_id, p.language) if pp != (p.group, p.identifier)] other_solutions = problems.solutions_for_problems(p.language, solved_problems) problem_facts = problems.get_facts(p.language, problem_module) or '' code = program + '\n' + other_solutions + '\n' + problem_facts messages, status, have_more = self.run(code) if status == 'ok': more_messages, status, have_more = self.query(query) messages.extend(more_messages) self._problem_id = problem_id return messages, status, have_more server.language_session_handlers['prolog'] = lambda user_session, problem_id, language_identifier, group_identifier, problem_identifier: PrologSession()