autograder.question

  1"""
  2A single question (test case) for an assignment.
  3"""
  4
  5import abc
  6import functools
  7import numbers
  8import traceback
  9import typing
 10
 11import edq.util.serial
 12import edq.util.time
 13
 14import autograder.util.invoke
 15import autograder.util.math
 16
 17DEFAULT_TIMEOUT_SEC: float = 60
 18""" Default timeout for grading a question. """
 19
 20class AutograderFailError(RuntimeError):
 21    """
 22    This error indicates that fail() has been called on a question
 23    and execution should be stopped.
 24    """
 25
 26class AutograderHardFailError(RuntimeError):
 27    """
 28    This error indicates that hard_fail() has been called on a question
 29    and execution should be stopped for all questions in the assignment.
 30    """
 31
 32class GradedQuestion(edq.util.serial.DictConverter):
 33    """
 34    The result of a question being graded with a submission.
 35    """
 36
 37    def __init__(self,
 38            name: str,
 39            max_points: float,
 40            score: float = 0,
 41            message: str = '',
 42            hard_fail: bool = False,
 43            skipped: bool = False,
 44            grading_start_time: typing.Union[edq.util.time.Timestamp, int, None] = None,
 45            grading_end_time: typing.Union[edq.util.time.Timestamp, int, None] = None,
 46            **kwargs: typing.Any) -> None:
 47        self.name: str = name
 48        """ The name of the question. """
 49
 50        self.max_points: float = max_points
 51        """ The max points possible for this question. """
 52
 53        self.score: float = score
 54        """ The score earned for this question. """
 55
 56        self.message: str = message
 57        """ A message/feedback for the student. """
 58
 59        self.hard_fail = hard_fail
 60        """ Whether this question triggered a hard fail during grading. """
 61
 62        self.skipped = skipped
 63        """ Whether this question was skipped during grading. """
 64
 65        # Default the grading time to deal with situations where the grader throws an exception.
 66        now = edq.util.time.Timestamp.now()
 67
 68        if (grading_start_time is None):
 69            grading_start_time = now
 70
 71        self.grading_start_time: typing.Any = edq.util.time.Timestamp(grading_start_time)
 72        """ When grading started. """
 73
 74        if (grading_end_time is None):
 75            grading_end_time = now
 76
 77        self.grading_end_time: typing.Any = edq.util.time.Timestamp(grading_end_time)
 78        """ When grading ended. """
 79
 80    def scoring_report(self, prefix: str = '', precision: int = 2) -> str:
 81        """
 82        Get a string that represents the scoring for this question.
 83        """
 84
 85        if ((prefix != '') and (not prefix.endswith(' '))):
 86            prefix += ' '
 87
 88        score_str = autograder.util.math.number_to_str(self.score, precision = precision)
 89        max_points_str = autograder.util.math.number_to_str(self.max_points, precision = precision)
 90
 91        lines = [f"{prefix}{self.name}: {score_str} / {max_points_str}"]
 92        if (self.message != ''):
 93            for line in self.message.split("\n"):
 94                lines.append(prefix + '   ' + line)
 95
 96        return "\n".join(lines)
 97
 98    def __eq__(self, other: typing.Any) -> bool:
 99        return self.equals(other)
100
101    def equals(self, other: typing.Any, ignore_messages: bool = False, **kwargs: typing.Any) -> bool:
102        """ Check another graded question for equality. """
103
104        if (not isinstance(other, GradedQuestion)):
105            return False
106
107        if (
108                (self.name != other.name)
109                or (self.max_points != other.max_points)
110                or (self.score != other.score)
111                or (self.hard_fail != other.hard_fail)
112                or (self.skipped != other.skipped)):
113            return False
114
115        if (ignore_messages):
116            return True
117
118        return self.message == other.message
119
120class Question:
121    """
122    Questions are grade-able portions of an assignment.
123    They can also be thought of as "test cases".
124    Note that all scoring is in ints.
125    """
126
127    def __init__(self,
128            max_points: float = 0,
129            name: typing.Union[str, None] = None,
130            timeout: typing.Union[float, None] = DEFAULT_TIMEOUT_SEC,
131            ) -> None:
132        if (name is None):
133            name = type(self).__name__
134
135        self.name: str = name
136        """
137        The name of this question.
138        Defaults to the name of the question class.
139        """
140
141        if ((not isinstance(max_points, numbers.Real)) or (max_points < 0)):
142            raise ValueError("max_points must be a real number, got '{max_points}' (type: '{type(max_points)}).")
143
144        self.max_points: float = max_points
145        """ The maximum number of points possible for this question (does not include extra credit). """
146
147        self._timeout: typing.Union[float, None] = timeout
148        """ The number of seconds allowed when grading this question. """
149
150        # Create the base scoring artifact.
151        self.result: GradedQuestion = GradedQuestion(name = self.name, max_points = self.max_points)
152        """
153        The result of grading this question.
154        A default/empty one is created on construction and is added to during the grading process.
155        """
156
157    @abc.abstractmethod
158    def score_question(self, submission: typing.Any, **kwargs: typing.Any) -> None:
159        """
160        Assign an actual score to this question.
161        The implementer has full access to instance variables.
162        However, users should generally just call the grading methods to manipulate the result.
163        """
164
165    def grade(self, submission: typing.Any,
166            additional_data: typing.Union[typing.Dict[str, typing.Any], None] = None,
167            show_exceptions: bool = False) -> GradedQuestion:
168        """
169        Invoke the scoring method using a timeout and cleanup.
170        Return the graded question.
171        """
172
173        if (additional_data is None):
174            additional_data = {}
175
176        helper = functools.partial(self._score_helper, submission,
177                additional_data = additional_data)
178
179        self._internal_grade(helper, show_exceptions)
180
181        return self.result
182
183    def _internal_grade(self, helper: typing.Callable, show_exceptions: bool) -> None:
184        """
185        Handle the internal process for grading a question.
186        """
187
188        try:
189            success, value = autograder.util.invoke.with_timeout(self._timeout, helper)
190        except Exception:
191            if (show_exceptions):
192                traceback.print_exc()
193
194            self.set_result(0, "Raised an exception: " + traceback.format_exc())
195            return
196
197        if (not success):
198            if (value is None):
199                self.set_result(0, f"Timeout ({self._timeout} seconds).")
200            else:
201                self.set_result(0, f"Error during execution: '{value}'.")
202
203            return
204
205        # Because we use the helper method, we can only get None back if there was an error.
206        if (value is None):
207            self.set_result(0, "Error running scoring.")
208            return
209
210        self.result = value
211
212    def _score_helper(self, submission: typing.Any,
213            additional_data: typing.Union[typing.Dict[str, typing.Any], None] = None,
214            ) -> GradedQuestion:
215        """
216        Score the question, but make sure to return the result so
217        multiprocessing can properly pass it back.
218        """
219
220        if (additional_data is None):
221            additional_data = {}
222
223        self.result = GradedQuestion(name = self.name, max_points = self.max_points)
224
225        self.result.grading_start_time = edq.util.time.Timestamp.now()
226
227        try:
228            self.score_question(submission, **additional_data)
229        except AutograderFailError:
230            # The question has been failed, no additional output is required.
231            pass
232        except AutograderHardFailError:
233            # The question has been failed hard, signal to stop grading.
234            self.result.hard_fail = True
235
236        self.result.grading_end_time = edq.util.time.Timestamp.now()
237
238        return self.result
239
240    def get_last_result(self) -> GradedQuestion:
241        """ Get the current grading result. """
242
243        return self.result
244
245    def get_score(self) -> float:
246        """ Get the current assigned score for this grading. """
247
248        return self.result.score
249
250    # Grading functions.
251
252    def check_not_implemented(self, value: typing.Any) -> bool:
253        """
254        Check if the given value is a marker for not implemented.
255        By default, this checked for a NotImplemented value.
256        Children may override this to define their own not implemented values.
257        """
258
259        if (value is None):
260            self.fail("None returned.")
261
262        if (isinstance(value, type(NotImplemented))):
263            self.fail("NotImplemented returned.")
264
265        return False
266
267    def set_result(self, score: float, message: str) -> None:
268        """ Set the score and message for the current grading (overrides any other values). """
269
270        self.result.score = score
271        self.result.message = message
272
273    def set_score(self, score: float) -> None:
274        """ Set the score for the current grading (overrides any other values). """
275
276        self.result.score = score
277
278    def set_message(self, message: str) -> None:
279        """ Set the message for the current grading (overrides any other values). """
280
281        self.result.message = message
282
283    def fail(self, message: str) -> None:
284        """
285        Immediately fail this question, no partial credit.
286        """
287
288        self.set_result(0, message)
289        raise AutograderFailError()
290
291    def hard_fail(self, message: str) -> None:
292        """
293        Immediately hard fail this question, no partial credit.
294        Grading will be stopped for the rest of the assignment.
295        """
296
297        self.set_result(0, message)
298        raise AutograderHardFailError()
299
300    def full_credit(self, message: str = '') -> None:
301        """ Assign full credit for this question. """
302
303        self.set_score(self.max_points)
304
305        if (message != ''):
306            self.set_message(message)
307
308    def add_score(self, add_score: float) -> None:
309        """ Add the given score to the current score for this question. """
310
311        self.result.score += add_score
312
313    def add_message(self, message: str, add_score: float = 0) -> None:
314        """ Add the given message (and optional score) to the current grading for this question. """
315
316        if (self.result.message != ''):
317            self.result.message += "\n"
318
319        self.result.message += str(message)
320        self.result.score += add_score
321
322    def cap_score(self) -> None:
323        """
324        Cap the current score so it is in [0, self.max_points].
325        """
326
327        self.result.score = max(0, min(self.max_points, self.result.score))