submit_notification (bool): True if the submit notification should be added
"""
try:
html = self.lcp.get_html()
# If we cannot construct the problem HTML,
# then generate an error message instead.
except Exception as err: # pylint: disable=broad-except
html = self.handle_problem_html_error(err)
html = self.remove_tags_from_html(html)
_ = self.runtime.service(self, "i18n").ugettext
# Enable/Disable Submit button if should_enable_submit_button returns True/False.
submit_button = self.submit_button_name()
submit_button_submitting = self.submit_button_submitting_name()
should_enable_submit_button = self.should_enable_submit_button()
submit_disabled_ctas = None
if not should_enable_submit_button:
cta_service = self.runtime.service(self, "call_to_action")
if cta_service:
submit_disabled_ctas = cta_service.get_ctas(self, 'capa_submit_disabled')
content = {
'name': self.display_name_with_default,
'html': smart_str(html),
'weight': self.weight,
}
# If demand hints are available, emit hint button and div.
demand_hints = self.lcp.tree.xpath("//problem/demandhint/hint")
demand_hint_possible, should_enable_next_hint = self._should_enable_demand_hint(demand_hints=demand_hints)
answer_notification_type, answer_notification_message = self._get_answer_notification(
render_notifications=submit_notification)
save_message = None
if self.has_saved_answers:
save_message = _(
"Your answers were previously saved. Click '{button_name}' to grade them."
).format(button_name=self.submit_button_name())
context = {
'problem': content,
'id': str(self.location),
'short_id': self.location.html_id(),
'submit_button': submit_button,
'submit_button_submitting': submit_button_submitting,
'should_enable_submit_button': should_enable_submit_button,
'reset_button': self.should_show_reset_button(),
'save_button': self.should_show_save_button(),
'answer_available': self.answer_available(),
'attempts_used': self.attempts,
'attempts_allowed': self.max_attempts,
'demand_hint_possible': demand_hint_possible,
'should_enable_next_hint': should_enable_next_hint,
'answer_notification_type': answer_notification_type,
'answer_notification_message': answer_notification_message,
'has_saved_answers': self.has_saved_answers,
'save_message': save_message,
'submit_disabled_cta': submit_disabled_ctas[0] if submit_disabled_ctas else None,
}
html = self.runtime.service(self, 'mako').render_template('problem.html', context)
if encapsulate:
html = HTML('
{html}
').format(
id=self.location.html_id(), ajax_url=self.ajax_url, html=HTML(html)
)
# Now do all the substitutions which the LMS module_render normally does, but
# we need to do here explicitly since we can get called for our HTML via AJAX
html = self.runtime.service(self, "replace_urls").replace_urls(html)
return html
def _get_answer_notification(self, render_notifications):
"""
Generate the answer notification type and message from the current problem status.
Arguments:
render_notifications (bool): If false the method will return an None for type and message
"""
answer_notification_message = None
answer_notification_type = None
if render_notifications:
progress = self.get_progress()
id_list = list(self.lcp.correct_map.keys())
# Show only a generic message if hiding correctness
if not self.correctness_available():
answer_notification_type = 'submitted'
elif len(id_list) == 1:
# Only one answer available
answer_notification_type = self.lcp.correct_map.get_correctness(id_list[0])
elif len(id_list) > 1:
# Check the multiple answers that are available
answer_notification_type = self.lcp.correct_map.get_correctness(id_list[0])
for answer_id in id_list[1:]:
if self.lcp.correct_map.get_correctness(answer_id) != answer_notification_type:
# There is at least 1 of the following combinations of correctness states
# Correct and incorrect, Correct and partially correct, or Incorrect and partially correct
# which all should have a message type of Partially Correct
answer_notification_type = 'partially-correct'
break
# Build the notification message based on the notification type and translate it.
ungettext = self.runtime.service(self, "i18n").ungettext
_ = self.runtime.service(self, "i18n").ugettext
if answer_notification_type == 'incorrect':
if progress is not None:
answer_notification_message = ungettext(
"Incorrect ({progress} point)",
"Incorrect ({progress} points)",
progress.frac()[1]
).format(progress=str(progress))
else:
answer_notification_message = _('Incorrect')
elif answer_notification_type == 'correct':
if progress is not None:
answer_notification_message = ungettext(
"Correct ({progress} point)",
"Correct ({progress} points)",
progress.frac()[1]
).format(progress=str(progress))
else:
answer_notification_message = _('Correct')
elif answer_notification_type == 'partially-correct':
if progress is not None:
answer_notification_message = ungettext(
"Partially correct ({progress} point)",
"Partially correct ({progress} points)",
progress.frac()[1]
).format(progress=str(progress))
else:
answer_notification_message = _('Partially Correct')
elif answer_notification_type == 'submitted':
answer_notification_message = _("Answer submitted.")
return answer_notification_type, answer_notification_message
def remove_tags_from_html(self, html):
"""
The capa xml includes many tags such as
or which are not
meant to be part of the client html. We strip them all and return the resulting html.
"""
tags = ['demandhint', 'choicehint', 'optionhint', 'stringhint', 'numerichint', 'optionhint',
'correcthint', 'regexphint', 'additional_answer', 'stringequalhint', 'compoundhint',
'stringequalhint']
for tag in tags:
html = re.sub(fr'<{tag}.*?>.*?', '', html, flags=re.DOTALL) # xss-lint: disable=python-interpolate-html # lint-amnesty, pylint: disable=line-too-long
# Some of these tags span multiple lines
# Note: could probably speed this up by calling sub() once with a big regex
# vs. simply calling sub() many times as we have here.
return html
def hint_button(self, data):
"""
Hint button handler, returns new html using hint_index from the client.
"""
hint_index = int(data['hint_index'])
return self.get_demand_hint(hint_index)
def used_all_attempts(self):
""" All attempts have been used """
return self.max_attempts is not None and self.attempts >= self.max_attempts
def is_past_due(self):
"""
Is it now past this problem's due date, including grace period?
"""
return (self.close_date is not None and
datetime.datetime.now(utc) > self.close_date)
def closed(self):
"""
Is the student still allowed to submit answers?
"""
if self.used_all_attempts():
return True
if self.is_past_due():
return True
return False
def is_submitted(self):
"""
Used to decide to show or hide RESET or CHECK buttons.
Means that student submitted problem and nothing more.
Problem can be completely wrong.
Pressing RESET button makes this function to return False.
"""
# used by conditional module
return self.lcp.done
def is_attempted(self):
"""
Has the problem been attempted?
used by conditional module
"""
return self.attempts > 0
def is_correct(self):
"""
True iff full points
"""
# self.score is initialized in self.lcp but in this method is accessed before self.lcp so just call it first.
self.lcp # pylint: disable=pointless-statement
return self.score.raw_earned == self.score.raw_possible
def answer_available(self):
"""
Is the user allowed to see an answer?
"""
user_is_staff = self.runtime.service(self, 'user').get_current_user().opt_attrs.get(ATTR_KEY_USER_IS_STAFF)
if not self.correctness_available():
# If correctness is being withheld, then don't show answers either.
return False
elif self.showanswer == '':
return False
elif self.showanswer == SHOWANSWER.NEVER:
return False
elif user_is_staff:
# This is after the 'never' check because admins can see the answer
# unless the problem explicitly prevents it
return True
elif self.showanswer == SHOWANSWER.ATTEMPTED:
return self.is_attempted() or self.is_past_due()
elif self.showanswer == SHOWANSWER.ANSWERED:
# NOTE: this is slightly different from 'attempted' -- resetting the problems
# makes lcp.done False, but leaves attempts unchanged.
return self.is_correct()
elif self.showanswer == SHOWANSWER.CLOSED:
return self.closed()
elif self.showanswer == SHOWANSWER.FINISHED:
return self.closed() or self.is_correct()
elif self.showanswer == SHOWANSWER.CORRECT_OR_PAST_DUE:
return self.is_correct() or self.is_past_due()
elif self.showanswer == SHOWANSWER.PAST_DUE:
return self.is_past_due()
elif self.showanswer == SHOWANSWER.AFTER_SOME_NUMBER_OF_ATTEMPTS:
required_attempts = self.attempts_before_showanswer_button
if self.max_attempts and required_attempts >= self.max_attempts:
required_attempts = self.max_attempts
return self.attempts >= required_attempts
elif self.showanswer == SHOWANSWER.ALWAYS:
return True
elif self.showanswer == SHOWANSWER.AFTER_ALL_ATTEMPTS:
return self.used_all_attempts()
elif self.showanswer == SHOWANSWER.AFTER_ALL_ATTEMPTS_OR_CORRECT:
return self.used_all_attempts() or self.is_correct()
elif self.showanswer == SHOWANSWER.ATTEMPTED_NO_PAST_DUE:
return self.is_attempted()
return False
def correctness_available(self):
"""
Is the user allowed to see whether she's answered correctly?
Limits access to the correct/incorrect flags, messages, and problem score.
"""
user_is_staff = self.runtime.service(self, 'user').get_current_user().opt_attrs.get(ATTR_KEY_USER_IS_STAFF)
return ShowCorrectness.correctness_available(
show_correctness=self.show_correctness,
due_date=self.close_date,
has_staff_access=user_is_staff,
)
def update_score(self, data):
"""
Delivers grading response (e.g. from asynchronous code checking) to
the capa problem, so its score can be updated
'data' must have a key 'response' which is a string that contains the
grader's response
No ajax return is needed. Return empty dict.
"""
queuekey = data['queuekey']
score_msg = data['xqueue_body']
self.lcp.update_score(score_msg, queuekey)
self.set_state_from_lcp()
self.set_score(self.score_from_lcp(self.lcp))
self.publish_grade(grader_response=True)
return {} # No AJAX return is needed
def handle_ungraded_response(self, data):
"""
Delivers a response from the XQueue to the capa problem
The score of the problem will not be updated
Args:
- data (dict) must contain keys:
queuekey - a key specific to this response
xqueue_body - the body of the response
Returns:
empty dictionary
No ajax return is needed, so an empty dict is returned
"""
queuekey = data['queuekey']
score_msg = data['xqueue_body']
# pass along the xqueue message to the problem
self.lcp.ungraded_response(score_msg, queuekey)
self.set_state_from_lcp()
return {}
def handle_input_ajax(self, data):
"""
Handle ajax calls meant for a particular input in the problem
Args:
- data (dict) - data that should be passed to the input
Returns:
- dict containing the response from the input
"""
response = self.lcp.handle_input_ajax(data)
# save any state changes that may occur
self.set_state_from_lcp()
return response
def get_answer(self, _data):
"""
For the "show answer" button.
Returns the answers and rendered "correct status span" HTML:
{'answers' : answers, 'correct_status_html': correct_status_span_html}.
The "correct status span" HTML is injected beside the correct answers
for radio button and checkmark problems, so that there is a visual
indication of the correct answers that is not solely based on color
(and also screen reader text).
"""
event_info = {}
event_info['problem_id'] = str(self.location)
self.track_function_unmask('showanswer', event_info)
if not self.answer_available(): # lint-amnesty, pylint: disable=no-else-raise
raise NotFoundError('Answer is not available')
else:
answers = self.lcp.get_question_answers()
self.set_state_from_lcp()
# answers (eg ) may have embedded images
# but be careful, some problems are using non-string answer dicts
new_answers = {}
for answer_id in answers:
try:
answer_content = self.runtime.service(self, "replace_urls").replace_urls(answers[answer_id])
new_answer = {answer_id: answer_content}
except TypeError:
log.debug('Unable to perform URL substitution on answers[%s]: %s',
answer_id, answers[answer_id])
new_answer = {answer_id: answers[answer_id]}
new_answers.update(new_answer)
return {
'answers': new_answers,
'correct_status_html': self.runtime.service(self, 'mako').render_template(
'status_span.html',
{'status': Status('correct', self.runtime.service(self, "i18n").ugettext)}
)
}
# Figure out if we should move these to capa_problem?
def get_problem(self, _data):
"""
Return results of get_problem_html, as a simple dict for json-ing.
{ 'html': }
Used if we want to reconfirm we have the right thing e.g. after
several AJAX calls.
"""
return {'html': self.get_problem_html(encapsulate=False, submit_notification=True)}
@staticmethod
def make_dict_of_responses(data):
"""
Make dictionary of student responses (aka "answers")
`data` is POST dictionary (webob.multidict.MultiDict).
The `data` dict has keys of the form 'x_y', which are mapped
to key 'y' in the returned dict. For example,
'input_1_2_3' would be mapped to '1_2_3' in the returned dict.
Some inputs always expect a list in the returned dict
(e.g. checkbox inputs). The convention is that
keys in the `data` dict that end with '[]' will always
have list values in the returned dict.
For example, if the `data` dict contains {'input_1[]': 'test' }
then the output dict would contain {'1': ['test'] }
(the value is a list).
Some other inputs such as ChoiceTextInput expect a dict of values in the returned
dict If the key ends with '{}' then we will assume that the value is a json
encoded dict and deserialize it.
For example, if the `data` dict contains {'input_1{}': '{"1_2_1": 1}'}
then the output dict would contain {'1': {"1_2_1": 1} }
(the value is a dictionary)
Raises an exception if:
-A key in the `data` dictionary does not contain at least one underscore
(e.g. "input" is invalid, but "input_1" is valid)
-Two keys end up with the same name in the returned dict.
(e.g. 'input_1' and 'input_1[]', which both get mapped to 'input_1'
in the returned dict)
"""
answers = {}
# webob.multidict.MultiDict is a view of a list of tuples,
# so it will return a multi-value key once for each value.
# We only want to consider each key a single time, so we use set(data.keys())
for key in set(data.keys()):
# e.g. input_resistor_1 ==> resistor_1
_, _, name = key.partition('_')
# If key has no underscores, then partition
# will return (key, '', '')
# We detect this and raise an error
if not name: # lint-amnesty, pylint: disable=no-else-raise
raise ValueError(f"{key} must contain at least one underscore")
else:
# This allows for answers which require more than one value for
# the same form input (e.g. checkbox inputs). The convention is that
# if the name ends with '[]' (which looks like an array), then the
# answer will be an array.
# if the name ends with '{}' (Which looks like a dict),
# then the answer will be a dict
is_list_key = name.endswith('[]')
is_dict_key = name.endswith('{}')
name = name[:-2] if is_list_key or is_dict_key else name
if is_list_key:
val = data.getall(key)
elif is_dict_key:
try:
val = json.loads(data[key])
# If the submission wasn't deserializable, raise an error.
except(KeyError, ValueError):
raise ValueError( # lint-amnesty, pylint: disable=raise-missing-from
f"Invalid submission: {data[key]} for {key}"
)
else:
val = data[key]
# If the name already exists, then we don't want
# to override it. Raise an error instead
if name in answers: # lint-amnesty, pylint: disable=no-else-raise
raise ValueError(f"Key {name} already exists in answers dict")
else:
answers[name] = val
return answers
def publish_grade(self, score=None, only_if_higher=None, **kwargs):
"""
Publishes the student's current grade to the system as an event
"""
if not score:
score = self.score
event = {
'value': score.raw_earned,
'max_value': score.raw_possible,
'only_if_higher': only_if_higher,
}
if kwargs.get('grader_response'):
event['grader_response'] = kwargs['grader_response']
self.runtime.publish(self, 'grade', event)
return {'grade': self.score.raw_earned, 'max_grade': self.score.raw_possible}
# pylint: disable=too-many-statements
def submit_problem(self, data, override_time=False):
"""
Checks whether answers to a problem are correct
Returns a map of correct/incorrect answers:
{'success' : 'correct' | 'incorrect' | AJAX alert msg string,
'contents' : html}
"""
event_info = {}
event_info['state'] = self.lcp.get_state()
event_info['problem_id'] = str(self.location)
self.lcp.has_saved_answers = False
answers = self.make_dict_of_responses(data)
answers_without_files = convert_files_to_filenames(answers)
event_info['answers'] = answers_without_files
metric_name = 'capa.check_problem.{}'.format # lint-amnesty, pylint: disable=unused-variable
# Can override current time
current_time = datetime.datetime.now(utc)
if override_time is not False:
current_time = override_time
_ = self.runtime.service(self, "i18n").ugettext
# Too late. Cannot submit
if self.closed():
log.error(
'ProblemClosedError: Problem %s, close date: %s, due:%s, is_past_due: %s, attempts: %s/%s,',
str(self.location),
self.close_date,
self.due,
self.is_past_due(),
self.attempts,
self.max_attempts,
)
event_info['failure'] = 'closed'
self.track_function_unmask('problem_check_fail', event_info)
raise NotFoundError(_("Problem is closed."))
# Problem submitted. Student should reset before checking again
if self.done and self.rerandomize == RANDOMIZATION.ALWAYS:
event_info['failure'] = 'unreset'
self.track_function_unmask('problem_check_fail', event_info)
raise NotFoundError(_("Problem must be reset before it can be submitted again."))
# Problem queued. Students must wait a specified waittime before they are allowed to submit
# IDEA: consider stealing code from below: pretty-print of seconds, cueing of time remaining
if self.lcp.is_queued():
prev_submit_time = self.lcp.get_recentmost_queuetime()
xqueue_service = self.runtime.service(self, 'xqueue')
waittime_between_requests = xqueue_service.waittime if xqueue_service else 0
if (current_time - prev_submit_time).total_seconds() < waittime_between_requests:
msg = _("You must wait at least {wait} seconds between submissions.").format(
wait=waittime_between_requests)
return {'success': msg, 'html': ''}
# Wait time between resets: check if is too soon for submission.
if self.last_submission_time is not None and self.submission_wait_seconds not in [0, None]:
seconds_since_submission = (current_time - self.last_submission_time).total_seconds()
if seconds_since_submission < self.submission_wait_seconds:
remaining_secs = int(self.submission_wait_seconds - seconds_since_submission)
msg = _('You must wait at least {wait_secs} between submissions. {remaining_secs} remaining.').format(
wait_secs=self.pretty_print_seconds(self.submission_wait_seconds),
remaining_secs=self.pretty_print_seconds(remaining_secs))
return {
'success': msg,
'html': ''
}
try:
# expose the attempt number to a potential python custom grader
# self.lcp.context['attempt'] refers to the attempt number (1-based)
self.lcp.context['attempt'] = self.attempts + 1
correct_map = self.lcp.grade_answers(answers)
# self.attempts refers to the number of attempts that did not
# raise an error (0-based)
self.attempts = self.attempts + 1
self.lcp.done = True
self.set_state_from_lcp()
self.set_score(self.score_from_lcp(self.lcp))
self.set_last_submission_time()
except (StudentInputError, ResponseError, LoncapaProblemError) as inst:
if self.debug:
log.warning(
"StudentInputError in capa_module:problem_check",
exc_info=True
)
# Save the user's state before failing
self.set_state_from_lcp()
self.set_score(self.score_from_lcp(self.lcp))
# If the user is a staff member, include
# the full exception, including traceback,
# in the response
user_is_staff = self.runtime.service(self, 'user').get_current_user().opt_attrs.get(ATTR_KEY_USER_IS_STAFF)
if user_is_staff:
msg = f"Staff debug info: {traceback.format_exc()}"
# Otherwise, display just an error message,
# without a stack trace
else:
full_error = inst.args[0]
try:
# only return the error value of the exception
msg = full_error.split("\\n")[-2].split(": ", 1)[1]
except IndexError:
msg = full_error
return {'success': msg}
except Exception as err:
# Save the user's state before failing
self.set_state_from_lcp()
self.set_score(self.score_from_lcp(self.lcp))
if self.debug:
msg = f"Error checking problem: {str(err)}"
msg += f'\nTraceback:\n{traceback.format_exc()}'
return {'success': msg}
raise
published_grade = self.publish_grade()
# success = correct if ALL questions in this problem are correct
success = 'correct'
for answer_id in correct_map:
if not correct_map.is_correct(answer_id):
success = 'incorrect'
# NOTE: We are logging both full grading and queued-grading submissions. In the latter,
# 'success' will always be incorrect
event_info['grade'] = published_grade['grade']
event_info['max_grade'] = published_grade['max_grade']
event_info['correct_map'] = correct_map.get_dict()
event_info['success'] = success
event_info['attempts'] = self.attempts
event_info['submission'] = self.get_submission_metadata_safe(answers_without_files, correct_map)
self.track_function_unmask('problem_check', event_info)
# render problem into HTML
html = self.get_problem_html(encapsulate=False, submit_notification=True)
# Withhold success indicator if hiding correctness
if not self.correctness_available():
success = 'submitted'
return {
'success': success,
'contents': html
}
# pylint: enable=too-many-statements
def track_function_unmask(self, title, event_info):
"""
All calls to runtime.track_function route through here so that the
choice names can be unmasked.
"""
# Do the unmask translates on a copy of event_info,
# avoiding problems where an event_info is unmasked twice.
event_unmasked = copy.deepcopy(event_info)
self.unmask_event(event_unmasked)
self.runtime.publish(self, title, event_unmasked)
def unmask_event(self, event_info):
"""
Translates in-place the event_info to account for masking
and adds information about permutation options in force.
"""
# answers is like: {u'i4x-Stanford-CS99-problem-dada976e76f34c24bc8415039dee1300_2_1': u'mask_0'}
# Each response values has an answer_id which matches the key in answers.
for response in self.lcp.responders.values():
# Un-mask choice names in event_info for masked responses.
if response.has_mask():
# We don't assume much about the structure of event_info,
# but check for the existence of the things we need to un-mask.
# Look for answers/id
answer = event_info.get('answers', {}).get(response.answer_id)
if answer is not None:
event_info['answers'][response.answer_id] = response.unmask_name(answer)
# Look for state/student_answers/id
answer = event_info.get('state', {}).get('student_answers', {}).get(response.answer_id)
if answer is not None:
event_info['state']['student_answers'][response.answer_id] = response.unmask_name(answer)
# Look for old_state/student_answers/id -- parallel to the above case, happens on reset
answer = event_info.get('old_state', {}).get('student_answers', {}).get(response.answer_id)
if answer is not None:
event_info['old_state']['student_answers'][response.answer_id] = response.unmask_name(answer)
# Add 'permutation' to event_info for permuted responses.
permutation_option = None
if response.has_shuffle():
permutation_option = 'shuffle'
elif response.has_answerpool():
permutation_option = 'answerpool'
if permutation_option is not None:
# Add permutation record tuple: (one of:'shuffle'/'answerpool', [as-displayed list])
if 'permutation' not in event_info:
event_info['permutation'] = {}
event_info['permutation'][response.answer_id] = (permutation_option, response.unmask_order())
def pretty_print_seconds(self, num_seconds):
"""
Returns time duration nicely formated, e.g. "3 minutes 4 seconds"
"""
# Here _ is the N variant ungettext that does pluralization with a 3-arg call
ungettext = self.runtime.service(self, "i18n").ungettext
hours = num_seconds // 3600
sub_hour = num_seconds % 3600
minutes = sub_hour // 60
seconds = sub_hour % 60
display = ""
if hours > 0:
display += ungettext("{num_hour} hour", "{num_hour} hours", hours).format(num_hour=hours)
if minutes > 0:
if display != "":
display += " "
# translators: "minute" refers to a minute of time
display += ungettext("{num_minute} minute", "{num_minute} minutes", minutes).format(num_minute=minutes)
# Taking care to make "0 seconds" instead of "" for 0 time
if seconds > 0 or (hours == 0 and minutes == 0):
if display != "":
display += " "
# translators: "second" refers to a second of time
display += ungettext("{num_second} second", "{num_second} seconds", seconds).format(num_second=seconds)
return display
def get_submission_metadata_safe(self, answers, correct_map):
"""
Ensures that no exceptions are thrown while generating input metadata summaries. Returns the
summary if it is successfully created, otherwise an empty dictionary.
"""
try:
return self.get_submission_metadata(answers, correct_map)
except Exception: # pylint: disable=broad-except
# NOTE: The above process requires deep inspection of capa structures that may break for some
# uncommon problem types. Ensure that it does not prevent answer submission in those
# cases. Any occurrences of errors in this block should be investigated and resolved.
log.exception('Unable to gather submission metadata, it will not be included in the event.')
return {}
def get_submission_metadata(self, answers, correct_map):
"""
Return a map of inputs to their corresponding summarized metadata.
Returns:
A map whose keys are a unique identifier for the input (in this case a capa input_id) and
whose values are:
question (str): Is the prompt that was presented to the student. It corresponds to the
label of the input.
answer (mixed): Is the answer the student provided. This may be a rich structure,
however it must be json serializable.
response_type (str): The XML tag of the capa response type.
input_type (str): The XML tag of the capa input type.
correct (bool): Whether or not the provided answer is correct. Will be an empty
string if correctness could not be determined.
variant (str): In some cases the same question can have several different variants.
This string should uniquely identify the variant of the question that was answered.
In the capa context this corresponds to the `seed`.
This function attempts to be very conservative and make very few assumptions about the structure
of the problem. If problem related metadata cannot be located it should be replaced with empty
strings ''.
"""
input_metadata = {}
for input_id, internal_answer in answers.items():
answer_input = self.lcp.inputs.get(input_id)
if answer_input is None:
log.warning('Input id %s is not mapped to an input type.', input_id)
answer_response = None
for responder in self.lcp.responders.values():
if input_id in responder.answer_ids:
answer_response = responder
if answer_response is None:
log.warning('Answer responder could not be found for input_id %s.', input_id)
user_visible_answer = internal_answer
if hasattr(answer_input, 'get_user_visible_answer'):
user_visible_answer = answer_input.get_user_visible_answer(internal_answer)
# If this problem has rerandomize enabled, then it will generate N variants of the
# question, one per unique seed value. In this case we would like to know which
# variant was selected. Ideally it would be nice to have the exact question that
# was presented to the user, with values interpolated etc, but that can be done
# later if necessary.
variant = ''
if self.rerandomize != RANDOMIZATION.NEVER:
variant = self.get_seed()
is_correct = correct_map.is_correct(input_id)
if is_correct is None:
is_correct = ''
response_data = getattr(answer_input, 'response_data', {})
input_metadata[input_id] = {
'question': response_data.get('label', ''),
'answer': user_visible_answer,
'response_type': getattr(getattr(answer_response, 'xml', None), 'tag', ''),
'input_type': getattr(answer_input, 'tag', ''),
'correct': is_correct,
'variant': variant,
'group_label': response_data.get('group_label', ''),
}
return input_metadata
def save_problem(self, data):
"""
Save the passed in answers.
Returns a dict { 'success' : bool, 'msg' : message }
The message is informative on success, and an error message on failure.
"""
event_info = {}
event_info['state'] = self.lcp.get_state()
event_info['problem_id'] = str(self.location)
answers = self.make_dict_of_responses(data)
event_info['answers'] = answers
_ = self.runtime.service(self, "i18n").ugettext
# Too late. Cannot submit
if self.closed() and not self.max_attempts == 0:
event_info['failure'] = 'closed'
self.track_function_unmask('save_problem_fail', event_info)
return {
'success': False,
# pylint: disable=line-too-long
# Translators: 'closed' means the problem's due date has passed. You may no longer attempt to solve the problem.
'msg': _("Problem is closed."),
# pylint: enable=line-too-long
}
# Problem submitted. Student should reset before saving
# again.
if self.done and self.rerandomize == RANDOMIZATION.ALWAYS:
event_info['failure'] = 'done'
self.track_function_unmask('save_problem_fail', event_info)
return {
'success': False,
'msg': _("Problem needs to be reset prior to save.")
}
self.lcp.student_answers = answers
self.lcp.has_saved_answers = True
self.set_state_from_lcp()
self.set_score(self.score_from_lcp(self.lcp))
self.track_function_unmask('save_problem_success', event_info)
msg = _("Your answers have been saved.")
if not self.max_attempts == 0:
msg = _(
"Your answers have been saved but not graded. Click '{button_name}' to grade them."
).format(button_name=self.submit_button_name())
return {
'success': True,
'msg': msg,
'html': self.get_problem_html(encapsulate=False)
}
def reset_problem(self, _data):
"""
Changes problem state to unfinished -- removes student answers,
Causes problem to rerender itself if randomization is enabled.
Returns a dictionary of the form:
{'success': True/False,
'html': Problem HTML string }
If an error occurs, the dictionary will also have an
`error` key containing an error message.
"""
event_info = {}
event_info['old_state'] = self.lcp.get_state()
event_info['problem_id'] = str(self.location)
_ = self.runtime.service(self, "i18n").ugettext
if self.closed():
event_info['failure'] = 'closed'
self.track_function_unmask('reset_problem_fail', event_info)
return {
'success': False,
# pylint: disable=line-too-long
# Translators: 'closed' means the problem's due date has passed. You may no longer attempt to solve the problem.
'msg': _("You cannot select Reset for a problem that is closed."),
# pylint: enable=line-too-long
}
if not self.is_submitted():
event_info['failure'] = 'not_done'
self.track_function_unmask('reset_problem_fail', event_info)
return {
'success': False,
'msg': _("You must submit an answer before you can select Reset."),
}
if self.is_submitted() and self.rerandomize in [RANDOMIZATION.ALWAYS, RANDOMIZATION.ONRESET]:
# Reset random number generator seed.
self.choose_new_seed()
# Generate a new problem with either the previous seed or a new seed
self.lcp = self.new_lcp(None)
# Pull in the new problem seed
self.set_state_from_lcp()
self.set_score(self.score_from_lcp(self.lcp))
# Grade may have changed, so publish new value
self.publish_grade()
event_info['new_state'] = self.lcp.get_state()
self.track_function_unmask('reset_problem', event_info)
return {
'success': True,
'html': self.get_problem_html(encapsulate=False),
}
# ScorableXBlockMixin methods
def rescore(self, only_if_higher=False):
"""
Checks whether the existing answers to a problem are correct.
This is called when the correct answer to a problem has been changed,
and the grade should be re-evaluated.
If only_if_higher is True, the answer and grade are updated
only if the resulting score is higher than before.
Returns a dict with one key:
{'success' : 'correct' | 'incorrect' | AJAX alert msg string }
Raises NotFoundError if called on a problem that has not yet been
answered, or NotImplementedError if it's a problem that cannot be rescored.
Returns the error messages for exceptions occurring while performing
the rescoring, rather than throwing them.
"""
event_info = {'state': self.lcp.get_state(), 'problem_id': str(self.location)}
_ = self.runtime.service(self, "i18n").ugettext
if not self.lcp.supports_rescoring():
event_info['failure'] = 'unsupported'
self.track_function_unmask('problem_rescore_fail', event_info)
# pylint: disable=line-too-long
# Translators: 'rescoring' refers to the act of re-submitting a student's solution so it can get a new score.
raise NotImplementedError(_("Problem's definition does not support rescoring."))
# pylint: enable=line-too-long
if not self.done:
event_info['failure'] = 'unanswered'
self.track_function_unmask('problem_rescore_fail', event_info)
raise NotFoundError(_("Problem must be answered before it can be graded again."))
# get old score, for comparison:
orig_score = self.get_score()
event_info['orig_score'] = orig_score.raw_earned
event_info['orig_total'] = orig_score.raw_possible
try:
self.update_correctness()
calculated_score = self.calculate_score()
except (StudentInputError, ResponseError, LoncapaProblemError) as inst: # lint-amnesty, pylint: disable=unused-variable
log.warning("Input error in capa_module:problem_rescore", exc_info=True)
event_info['failure'] = 'input_error'
self.track_function_unmask('problem_rescore_fail', event_info)
raise
except Exception:
event_info['failure'] = 'unexpected'
self.track_function_unmask('problem_rescore_fail', event_info)
raise
# rescoring should have no effect on attempts, so don't
# need to increment here, or mark done. Just save.
self.set_state_from_lcp()
self.publish_grade(score=calculated_score, only_if_higher=only_if_higher)
event_info['new_score'] = calculated_score.raw_earned
event_info['new_total'] = calculated_score.raw_possible
# success = correct if ALL questions in this problem are correct
success = 'correct'
for answer_id in self.lcp.correct_map:
if not self.lcp.correct_map.is_correct(answer_id):
success = 'incorrect'
# NOTE: We are logging both full grading and queued-grading submissions. In the latter,
# 'success' will always be incorrect
event_info['correct_map'] = self.lcp.correct_map.get_dict()
event_info['success'] = success
event_info['attempts'] = self.attempts
self.track_function_unmask('problem_rescore', event_info)
def has_submitted_answer(self):
return self.done
def set_score(self, score):
"""
Sets the internal score for the problem. This is not derived directly
from the internal LCP in keeping with the ScorableXBlock spec.
"""
self.score = score
def get_score(self):
"""
Returns the score currently set on the block.
"""
return self.score
def update_correctness(self):
"""
Updates correct map of the LCP.
Operates by creating a new correctness map based on the current
state of the LCP, and updating the old correctness map of the LCP.
"""
# Make sure that the attempt number is always at least 1 for grading purposes,
# even if the number of attempts have been reset and this problem is regraded.
self.lcp.context['attempt'] = max(self.attempts, 1)
new_correct_map = self.lcp.get_grade_from_current_answers(None)
self.lcp.correct_map.update(new_correct_map)
def calculate_score(self):
"""
Returns the score calculated from the current problem state.
"""
new_score = self.lcp.calculate_score()
return Score(raw_earned=new_score['score'], raw_possible=new_score['total'])
def score_from_lcp(self, lcp):
"""
Returns the score associated with the correctness map
currently stored by the LCP.
"""
lcp_score = lcp.calculate_score()
return Score(raw_earned=lcp_score['score'], raw_possible=lcp_score['total'])
class ComplexEncoder(json.JSONEncoder):
"""
Extend the JSON encoder to correctly handle complex numbers
"""
def default(self, obj): # lint-amnesty, pylint: disable=arguments-differ, method-hidden
"""
Print a nicely formatted complex number, or default to the JSON encoder
"""
if isinstance(obj, complex):
return f"{obj.real:.7g}{obj.imag:+.7g}*j"
return json.JSONEncoder.default(self, obj)
def randomization_bin(seed, problem_id):
"""
Pick a randomization bin for the problem given the user's seed and a problem id.
We do this because we only want e.g. 20 randomizations of a problem to make analytics
interesting. To avoid having sets of students that always get the same problems,
we'll combine the system's per-student seed with the problem id in picking the bin.
"""
r_hash = hashlib.sha1()
r_hash.update(str(seed).encode())
r_hash.update(str(problem_id).encode())
# get the first few digits of the hash, convert to an int, then mod.
return int(r_hash.hexdigest()[:7], 16) % NUM_RANDOMIZATION_BINS