diff --git a/perguntations/app.py b/perguntations/app.py index 325e630..f020652 100644 --- a/perguntations/app.py +++ b/perguntations/app.py @@ -20,7 +20,7 @@ from sqlalchemy.orm import sessionmaker # this project from perguntations.models import Student, Test, Question from perguntations.tools import load_yaml -from perguntations.test import TestFactory, TestFactoryException +from perguntations.testfactory import TestFactory, TestFactoryException logger = logging.getLogger(__name__) @@ -182,22 +182,16 @@ class App(): testconf.update(conf) # start test factory - logger.info('Making test factory...') + logger.info('Running test factory...') try: self.testfactory = TestFactory(testconf) except TestFactoryException as exc: logger.critical(exc) raise AppException('Failed to create test factory!') from exc - logger.info('Test factory ready. No errors found.') - # ------------------------------------------------------------------------ def _pregenerate_tests(self, num): event_loop = asyncio.get_event_loop() - # for _ in range(num): - # test = event_loop.run_until_complete(self.testfactory.generate()) - # self.pregenerated_tests.append(test) - self.pregenerated_tests += [ event_loop.run_until_complete(self.testfactory.generate()) for _ in range(num)] diff --git a/perguntations/main.py b/perguntations/main.py index 38ea58e..8a7c813 100644 --- a/perguntations/main.py +++ b/perguntations/main.py @@ -99,13 +99,12 @@ def get_logger_config(debug=False): }, }, } - default_config['loggers'].update({ - f'{APP_NAME}.{module}': { - 'handlers': ['default'], - 'level': level, - 'propagate': False, - } for module in ['app', 'models', 'factory', 'questions', - 'test', 'tools']}) + + modules = ['app', 'models', 'questions', 'test', 'testfactory', 'tools'] + logger = {'handlers': ['default'], 'level': level, 'propagate': False} + + default_config['loggers'].update({f'{APP_NAME}.{module}': logger + for module in modules}) return load_yaml(config_file, default=default_config) diff --git a/perguntations/test.py b/perguntations/test.py index 2445ee1..7c963b8 100644 --- a/perguntations/test.py +++ b/perguntations/test.py @@ -1,300 +1,16 @@ ''' -TestFactory - generates tests for students Test - instances of this class are individual tests ''' - # python standard library -from os import path -import random from datetime import datetime import logging -import re -from typing import Any, Dict - -# this project -from perguntations.questions import QFactory, QuestionException -from perguntations.tools import load_yaml # Logger configuration logger = logging.getLogger(__name__) # ============================================================================ -class TestFactoryException(Exception): - '''exception raised in this module''' - - -# ============================================================================ -class TestFactory(dict): - ''' - Each instance of TestFactory() is a test generator. - For example, if we want to serve two different tests, then we need two - instances of TestFactory(), one for each test. - ''' - - # ------------------------------------------------------------------------ - def __init__(self, conf: Dict[str, Any]) -> None: - ''' - Loads configuration from yaml file, then overrides some configurations - using the conf argument. - Base questions are added to a pool of questions factories. - ''' - - # --- set test defaults and then use given configuration - super().__init__({ # defaults - 'title': '', - 'show_points': True, - 'scale': None, # or [0, 20] - 'duration': 0, # 0=infinite - 'autosubmit': False, - 'debug': False, - 'show_ref': False, - }) - self.update(conf) - - # --- perform sanity checks and normalize the test questions - self.sanity_checks() - logger.info('Sanity checks PASSED.') - - # --- find refs of all questions used in the test - qrefs = {r for qq in self['questions'] for r in qq['ref']} - logger.info('Declared %d questions (each test uses %d).', - len(qrefs), len(self["questions"])) - - # --- for review, we are done. no factories needed - if self['review']: - logger.info('Review mode. No questions loaded. No factories.') - return - - # --- load and build question factories - self.question_factory = {} - - counter = 1 - for file in self["files"]: - fullpath = path.normpath(path.join(self["questions_dir"], file)) - (dirname, filename) = path.split(fullpath) - - logger.info('Loading "%s"...', fullpath) - questions = load_yaml(fullpath) # , default=[]) - - for i, question in enumerate(questions): - # make sure every question in the file is a dictionary - if not isinstance(question, dict): - msg = f'Question {i} in {file} is not a dictionary' - raise TestFactoryException(msg) - - # check if ref is missing, then set to '/path/file.yaml:3' - if 'ref' not in question: - question['ref'] = f'{file}:{i:04}' - logger.warning('Missing ref set to "%s"', question["ref"]) - - # check for duplicate refs - if question['ref'] in self.question_factory: - other = self.question_factory[question['ref']] - otherfile = path.join(other.question['path'], - other.question['filename']) - msg = (f'Duplicate reference "{question["ref"]}" in files ' - f'"{otherfile}" and "{fullpath}".') - raise TestFactoryException(msg) - - # make factory only for the questions used in the test - if question['ref'] in qrefs: - question.setdefault('type', 'information') - question.update({ - 'filename': filename, - 'path': dirname, - 'index': i # position in the file, 0 based - }) - - self.question_factory[question['ref']] = QFactory(question) - - # check if all the questions can be correctly generated - try: - self.question_factory[question['ref']].generate() - except Exception as exc: - msg = f'Failed to generate "{question["ref"]}"' - raise TestFactoryException(msg) from exc - else: - logger.info('%4d. "%s" Ok.', counter, question["ref"]) - counter += 1 - - qmissing = qrefs.difference(set(self.question_factory.keys())) - if qmissing: - raise TestFactoryException(f'Could not find questions {qmissing}.') - - # ------------------------------------------------------------------------ - def check_test_ref(self) -> None: - '''Test must have a `ref`''' - if 'ref' not in self: - raise TestFactoryException('Missing "ref" in configuration!') - if not re.match(r'^[a-zA-Z0-9_-]+$', self['ref']): - raise TestFactoryException('Test "ref" can only contain the ' - 'characters a-zA-Z0-9_-') - - def check_missing_database(self) -> None: - '''Test must have a database''' - if 'database' not in self: - raise TestFactoryException('Missing "database" in configuration') - if not path.isfile(path.expanduser(self['database'])): - msg = f'Database "{self["database"]}" not found!' - raise TestFactoryException(msg) - - def check_missing_answers_directory(self) -> None: - '''Test must have a answers directory''' - if 'answers_dir' not in self: - msg = 'Missing "answers_dir" in configuration' - raise TestFactoryException(msg) - - def check_answers_directory_writable(self) -> None: - '''Answers directory must be writable''' - testfile = path.join(path.expanduser(self['answers_dir']), 'REMOVE-ME') - try: - with open(testfile, 'w') as file: - file.write('You can safely remove this file.') - except OSError as exc: - msg = f'Cannot write answers to directory "{self["answers_dir"]}"' - raise TestFactoryException(msg) from exc - - def check_questions_directory(self) -> None: - '''Check if questions directory is missing or not accessible.''' - if 'questions_dir' not in self: - logger.warning('Missing "questions_dir". Using "%s"', - path.abspath(path.curdir)) - self['questions_dir'] = path.curdir - elif not path.isdir(path.expanduser(self['questions_dir'])): - raise TestFactoryException(f'Can\'t find questions directory ' - f'"{self["questions_dir"]}"') - - def check_import_files(self) -> None: - '''Check if there are files to import (with questions)''' - if 'files' not in self: - msg = ('Missing "files" in configuration with the list of ' - 'question files to import!') - raise TestFactoryException(msg) - - if isinstance(self['files'], str): - self['files'] = [self['files']] - - def check_question_list(self) -> None: - '''normalize question list''' - if 'questions' not in self: - raise TestFactoryException('Missing "questions" in configuration') - - for i, question in enumerate(self['questions']): - # normalize question to a dict and ref to a list of references - if isinstance(question, str): # e.g., - some_ref - question = {'ref': [question]} # becomes - ref: [some_ref] - elif isinstance(question, dict) and isinstance(question['ref'], str): - question['ref'] = [question['ref']] - elif isinstance(question, list): - question = {'ref': [str(a) for a in question]} - - self['questions'][i] = question - - def check_missing_title(self) -> None: - '''Warns if title is missing''' - if not self['title']: - logger.warning('Title is undefined!') - - def check_grade_scaling(self) -> None: - '''Just informs the scale limits''' - if 'scale_points' in self: - msg = ('*** DEPRECATION WARNING: *** scale_points, scale_min, ' - 'scale_max were replaced by "scale: [min, max]".') - logger.warning(msg) - self['scale'] = [self['scale_min'], self['scale_max']] - - - # ------------------------------------------------------------------------ - def sanity_checks(self) -> None: - ''' - Checks for valid keys and sets default values. - Also checks if some files and directories exist - ''' - self.check_test_ref() - self.check_missing_database() - self.check_missing_answers_directory() - self.check_answers_directory_writable() - self.check_questions_directory() - self.check_import_files() - self.check_question_list() - self.check_missing_title() - self.check_grade_scaling() - - # ------------------------------------------------------------------------ - async def generate(self): - ''' - Given a dictionary with a student dict {'name':'john', 'number': 123} - returns instance of Test() for that particular student - ''' - - # make list of questions - questions = [] - qnum = 1 # track question number - nerr = 0 # count errors during questions generation - - for qlist in self['questions']: - # choose list of question variants - choose = qlist.get('choose', 1) - qrefs = random.sample(qlist['ref'], k=choose) - - for qref in qrefs: - # generate instance of question - try: - question = await self.question_factory[qref].gen_async() - except QuestionException: - logger.error('Can\'t generate question "%s". Skipping.', qref) - nerr += 1 - continue - - # some defaults - if question['type'] in ('information', 'success', 'warning', - 'alert'): - question['points'] = qlist.get('points', 0.0) - else: - question['points'] = qlist.get('points', 1.0) - question['number'] = qnum # counter for non informative panels - qnum += 1 - - questions.append(question) - - # setup scale - total_points = sum(q['points'] for q in questions) - - if total_points > 0: - # normalize question points to scale - if self['scale'] is not None: - scale_min, scale_max = self['scale'] - for question in questions: - question['points'] *= (scale_max - scale_min) / total_points - else: - self['scale'] = [0, total_points] - else: - logger.warning('Total points is **ZERO**.') - if self['scale'] is None: - self['scale'] = [0, 20] # default - - if nerr > 0: - logger.error('%s errors found!', nerr) - - # copy these from the test configuratoin to each test instance - inherit = {'ref', 'title', 'database', 'answers_dir', - 'questions_dir', 'files', - 'duration', 'autosubmit', - 'scale', 'show_points', - 'show_ref', 'debug', } - # NOT INCLUDED: testfile, allow_all, review - - return Test({'questions': questions, **{k:self[k] for k in inherit}}) - - # ------------------------------------------------------------------------ - def __repr__(self): - testsettings = '\n'.join(f' {k:14s}: {v}' for k, v in self.items()) - return '{\n' + testsettings + '\n}' - - -# ============================================================================ class Test(dict): ''' Each instance Test() is a concrete test of a single student. @@ -335,7 +51,6 @@ class Test(dict): ''' for ref, ans in answers_dict.items(): self['questions'][ref].set_answer(ans) - # self['questions'][ref]['answer'] = ans # ------------------------------------------------------------------------ async def correct(self) -> float: diff --git a/perguntations/testfactory.py b/perguntations/testfactory.py new file mode 100644 index 0000000..cbebdc0 --- /dev/null +++ b/perguntations/testfactory.py @@ -0,0 +1,295 @@ +''' +TestFactory - generates tests for students +''' + +# python standard library +from os import path +import random +import logging +import re +from typing import Any, Dict + +# this project +from perguntations.questions import QFactory, QuestionException +from perguntations.test import Test +from perguntations.tools import load_yaml + +# Logger configuration +logger = logging.getLogger(__name__) + + +# ============================================================================ +class TestFactoryException(Exception): + '''exception raised in this module''' + + +# ============================================================================ +class TestFactory(dict): + ''' + Each instance of TestFactory() is a test generator. + For example, if we want to serve two different tests, then we need two + instances of TestFactory(), one for each test. + ''' + + # ------------------------------------------------------------------------ + def __init__(self, conf: Dict[str, Any]) -> None: + ''' + Loads configuration from yaml file, then overrides some configurations + using the conf argument. + Base questions are added to a pool of questions factories. + ''' + + # --- set test defaults and then use given configuration + super().__init__({ # defaults + 'title': '', + 'show_points': True, + 'scale': None, # or [0, 20] + 'duration': 0, # 0=infinite + 'autosubmit': False, + 'debug': False, + 'show_ref': False, + }) + self.update(conf) + + # --- for review, we are done. no factories needed + if self['review']: + logger.info('Review mode. No questions loaded. No factories.') + return + + # --- perform sanity checks and normalize the test questions + self.sanity_checks() + logger.info('Sanity checks PASSED.') + + # --- find refs of all questions used in the test + qrefs = {r for qq in self['questions'] for r in qq['ref']} + logger.info('Declared %d questions (each test uses %d).', + len(qrefs), len(self["questions"])) + + # --- load and build question factories + self.question_factory = {} + + counter = 1 + for file in self["files"]: + fullpath = path.normpath(path.join(self["questions_dir"], file)) + (dirname, filename) = path.split(fullpath) + + logger.info('Loading "%s"...', fullpath) + questions = load_yaml(fullpath) # , default=[]) + + for i, question in enumerate(questions): + # make sure every question in the file is a dictionary + if not isinstance(question, dict): + msg = f'Question {i} in {file} is not a dictionary' + raise TestFactoryException(msg) + + # check if ref is missing, then set to '/path/file.yaml:3' + if 'ref' not in question: + question['ref'] = f'{file}:{i:04}' + logger.warning('Missing ref set to "%s"', question["ref"]) + + # check for duplicate refs + if question['ref'] in self.question_factory: + other = self.question_factory[question['ref']] + otherfile = path.join(other.question['path'], + other.question['filename']) + msg = (f'Duplicate reference "{question["ref"]}" in files ' + f'"{otherfile}" and "{fullpath}".') + raise TestFactoryException(msg) + + # make factory only for the questions used in the test + if question['ref'] in qrefs: + question.setdefault('type', 'information') + question.update({ + 'filename': filename, + 'path': dirname, + 'index': i # position in the file, 0 based + }) + + self.question_factory[question['ref']] = QFactory(question) + + # check if all the questions can be correctly generated + try: + self.question_factory[question['ref']].generate() + except Exception as exc: + msg = f'Failed to generate "{question["ref"]}"' + raise TestFactoryException(msg) from exc + else: + logger.info('%4d. "%s" Ok.', counter, question["ref"]) + counter += 1 + + qmissing = qrefs.difference(set(self.question_factory.keys())) + if qmissing: + raise TestFactoryException(f'Could not find questions {qmissing}.') + + logger.info('Test factory ready. No errors found.') + + + # ------------------------------------------------------------------------ + def check_test_ref(self) -> None: + '''Test must have a `ref`''' + if 'ref' not in self: + raise TestFactoryException('Missing "ref" in configuration!') + if not re.match(r'^[a-zA-Z0-9_-]+$', self['ref']): + raise TestFactoryException('Test "ref" can only contain the ' + 'characters a-zA-Z0-9_-') + + def check_missing_database(self) -> None: + '''Test must have a database''' + if 'database' not in self: + raise TestFactoryException('Missing "database" in configuration') + if not path.isfile(path.expanduser(self['database'])): + msg = f'Database "{self["database"]}" not found!' + raise TestFactoryException(msg) + + def check_missing_answers_directory(self) -> None: + '''Test must have a answers directory''' + if 'answers_dir' not in self: + msg = 'Missing "answers_dir" in configuration' + raise TestFactoryException(msg) + + def check_answers_directory_writable(self) -> None: + '''Answers directory must be writable''' + testfile = path.join(path.expanduser(self['answers_dir']), 'REMOVE-ME') + try: + with open(testfile, 'w') as file: + file.write('You can safely remove this file.') + except OSError as exc: + msg = f'Cannot write answers to directory "{self["answers_dir"]}"' + raise TestFactoryException(msg) from exc + + def check_questions_directory(self) -> None: + '''Check if questions directory is missing or not accessible.''' + if 'questions_dir' not in self: + logger.warning('Missing "questions_dir". Using "%s"', + path.abspath(path.curdir)) + self['questions_dir'] = path.curdir + elif not path.isdir(path.expanduser(self['questions_dir'])): + raise TestFactoryException(f'Can\'t find questions directory ' + f'"{self["questions_dir"]}"') + + def check_import_files(self) -> None: + '''Check if there are files to import (with questions)''' + if 'files' not in self: + msg = ('Missing "files" in configuration with the list of ' + 'question files to import!') + raise TestFactoryException(msg) + + if isinstance(self['files'], str): + self['files'] = [self['files']] + + def check_question_list(self) -> None: + '''normalize question list''' + if 'questions' not in self: + raise TestFactoryException('Missing "questions" in configuration') + + for i, question in enumerate(self['questions']): + # normalize question to a dict and ref to a list of references + if isinstance(question, str): # e.g., - some_ref + question = {'ref': [question]} # becomes - ref: [some_ref] + elif isinstance(question, dict) and isinstance(question['ref'], str): + question['ref'] = [question['ref']] + elif isinstance(question, list): + question = {'ref': [str(a) for a in question]} + + self['questions'][i] = question + + def check_missing_title(self) -> None: + '''Warns if title is missing''' + if not self['title']: + logger.warning('Title is undefined!') + + def check_grade_scaling(self) -> None: + '''Just informs the scale limits''' + if 'scale_points' in self: + msg = ('*** DEPRECATION WARNING: *** scale_points, scale_min, ' + 'scale_max were replaced by "scale: [min, max]".') + logger.warning(msg) + self['scale'] = [self['scale_min'], self['scale_max']] + + + # ------------------------------------------------------------------------ + def sanity_checks(self) -> None: + ''' + Checks for valid keys and sets default values. + Also checks if some files and directories exist + ''' + self.check_test_ref() + self.check_missing_database() + self.check_missing_answers_directory() + self.check_answers_directory_writable() + self.check_questions_directory() + self.check_import_files() + self.check_question_list() + self.check_missing_title() + self.check_grade_scaling() + + # ------------------------------------------------------------------------ + async def generate(self): + ''' + Given a dictionary with a student dict {'name':'john', 'number': 123} + returns instance of Test() for that particular student + ''' + + # make list of questions + questions = [] + qnum = 1 # track question number + nerr = 0 # count errors during questions generation + + for qlist in self['questions']: + # choose list of question variants + choose = qlist.get('choose', 1) + qrefs = random.sample(qlist['ref'], k=choose) + + for qref in qrefs: + # generate instance of question + try: + question = await self.question_factory[qref].gen_async() + except QuestionException: + logger.error('Can\'t generate question "%s". Skipping.', qref) + nerr += 1 + continue + + # some defaults + if question['type'] in ('information', 'success', 'warning', + 'alert'): + question['points'] = qlist.get('points', 0.0) + else: + question['points'] = qlist.get('points', 1.0) + question['number'] = qnum # counter for non informative panels + qnum += 1 + + questions.append(question) + + # setup scale + total_points = sum(q['points'] for q in questions) + + if total_points > 0: + # normalize question points to scale + if self['scale'] is not None: + scale_min, scale_max = self['scale'] + for question in questions: + question['points'] *= (scale_max - scale_min) / total_points + else: + self['scale'] = [0, total_points] + else: + logger.warning('Total points is **ZERO**.') + if self['scale'] is None: + self['scale'] = [0, 20] # default + + if nerr > 0: + logger.error('%s errors found!', nerr) + + # copy these from the test configuratoin to each test instance + inherit = {'ref', 'title', 'database', 'answers_dir', + 'questions_dir', 'files', + 'duration', 'autosubmit', + 'scale', 'show_points', + 'show_ref', 'debug', } + # NOT INCLUDED: testfile, allow_all, review + + return Test({'questions': questions, **{k:self[k] for k in inherit}}) + + # ------------------------------------------------------------------------ + def __repr__(self): + testsettings = '\n'.join(f' {k:14s}: {v}' for k, v in self.items()) + return '{\n' + testsettings + '\n}' -- libgit2 0.21.2