diff --git a/mathrace_interaction/mathrace_interaction/journal_reader.py b/mathrace_interaction/mathrace_interaction/journal_reader.py new file mode 100644 index 0000000..a8f00ab --- /dev/null +++ b/mathrace_interaction/mathrace_interaction/journal_reader.py @@ -0,0 +1,1047 @@ +# Copyright (C) 2024 by the Turing @ DMF authors +# +# This file is part of Turing @ DMF. +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Read a mathrace journal.""" + +import abc +import datetime +import sys +import types +import typing + +from mathrace_interaction.determine_journal_version import determine_journal_version + +TuringDict: typing.TypeAlias = dict[str, typing.Any] + + +class AbstractJournalReader(abc.ABC): + """ + An abstract class representing a reader of a mathrace journal. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + + Attributes + ---------- + _journal_stream + The I/O stream that reads the journal generated by mathrace or simdis, provided as input. + _journal_stream_context + The context of _journal_stream. Only valid after entering in this context, and before exiting from it. + _race_name + Name of the race, provided as input. + _race_date + Date of the race, provided as input. + """ + + def __init__(self, journal_stream: typing.TextIO, race_name: str, race_date: datetime.datetime) -> None: + self._journal_stream = journal_stream + self._journal_stream_context: typing.TextIO | None = None + self._race_name = race_name + self._race_date = race_date + + def __enter__(self) -> typing.Self: + """Enter the journal I/O stream context.""" + self._journal_stream_context = self._journal_stream.__enter__() + return self + + def __exit__( + self, exception_type: type[BaseException] | None, + exception_value: BaseException | None, + traceback: types.TracebackType | None + ) -> None: + """Exit the journal I/O stream context.""" + self._journal_stream.__exit__(exception_type, exception_value, traceback) + self._journal_stream_context = None + + def read(self) -> TuringDict: + """Read the mathrace journal, and convert it into a dictionary compatible with turing.""" + # Prepare a dictionary to store the output + turing_dict: TuringDict = dict() + + # Prepare a further dictionary for fields which are collected by mathrace, but not read in by turing + turing_dict["mathrace_only"] = dict() + + # Set name and date + turing_dict["nome"] = self._race_name + turing_dict["inizio"] = self._race_date.isoformat() + + # The first line must contain the initialization of the file + first_line = self._read_line() + if first_line != "--- 001 inizializzazione simulatore": + raise RuntimeError(f"Invalid first line {first_line}") + del first_line + + # The second section contains the definition of the race + self._read_race_definition_section(turing_dict) + + # The third section contains the definition of the questions + self._read_questions_definition_section(turing_dict) + + # The fourth section contains the definition of the teams + self._read_teams_definition_section(turing_dict) + + # The fifth section contains all race events + self._read_race_events_section(turing_dict) + + # The final line must contain the finalization of the file + final_line = self._read_line() + if final_line != "--- 999 fine simulatore": + raise RuntimeError(f"Invalid final line {final_line}") + del final_line + + # There must be no further lines in the stream + try: + extra_line = self._read_line() + except StopIteration: + pass + else: + raise RuntimeError(f"Journal contains extra line {extra_line} after race end") + + # Return the populated dictionary + return turing_dict + + def _read_line(self) -> str: + """Read one line from the journal stream.""" + line, _, _ = self._read_line_with_positions() + return line + + def _read_line_with_positions(self) -> tuple[str, int, int]: + """Read one line from the journal stream, returning the stream positions before and after the operation.""" + stream = self._journal_stream_context + assert stream is not None + before = stream.tell() + line = "#" # mock line to start the while loop + while line.startswith("#"): + line = stream.readline() + if len(line) == 0: + raise StopIteration() + else: + return line.strip("\n"), before, stream.tell() + + def _reset_stream_to_position(self, position: int) -> None: + """Reset stream to a specific position.""" + stream = self._journal_stream_context + assert stream is not None + stream.seek(position) + + @abc.abstractmethod + def _read_race_definition_section(self, turing_dict: TuringDict) -> None: + """Read the race definition section.""" + pass + + @abc.abstractmethod + def _read_questions_definition_section(self, turing_dict: TuringDict) -> None: + """Read the questions definition section.""" + pass + + @abc.abstractmethod + def _read_teams_definition_section(self, turing_dict: TuringDict) -> None: + """Read the teams definition section.""" + pass + + @abc.abstractmethod + def _read_race_events_section(self, turing_dict: TuringDict) -> None: + """Read all race events.""" + pass + + +class JournalReaderR5539(AbstractJournalReader): + """ + A class representing a reader of a mathrace journal, version r5539. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + + Class Attributes + ---------------- + RACE_DEFINITION + The race setup code associated to the race definition. + QUESTION_DEFINITION + The race setup code associated to the question definition. + RACE_START + The race event code associated to the start of the race. + JOLLY_SELECTION + The race event code associated to jolly selection by a team. + ANSWER_SUBMISSION + The race event code associated to answer submission by a team. + JOLLY_TIMEOUT + The race event code associated to jolly timeout. + TIMER_UPDATE + The race event code associated to a timer update. + RACE_SUSPENDED + The race event code associated to a race suspension. + RACE_RESUMED + The race event code associated to a race resumption. + RACE_END + The race event code associated to the end of the race. + MANUAL_BONUS + The race event code associated to the manual addition of a bonus. + """ + + # Race setup codes + RACE_DEFINITION = "003" + QUESTION_DEFINITION = "004" + + # Race event codes + RACE_START = "002" + JOLLY_SELECTION = "010" + ANSWER_SUBMISSION = "011" + JOLLY_TIMEOUT = "021" + TIMER_UPDATE = "022" + RACE_SUSPENDED = "027" + RACE_RESUMED = "028" + RACE_END = "029" + MANUAL_BONUS = "091" + + def _read_race_definition_section(self, turing_dict: TuringDict) -> None: + """Read the race definition section.""" + line = self._read_line() + if not line.startswith("--- "): + raise RuntimeError( + f'Invalid line {line} in race definition: it does not start with "--- "') + + # Discard comments at the end of the line + if " -- squadre:" in line: + line, _ = line.split(" -- squadre:") + elif " squadre:" in line: + line, _ = line.split(" squadre:") + + # Process the line + self._process_race_definition_line(line, turing_dict) + + def _process_race_definition_line(self, line: str, turing_dict: TuringDict) -> None: + """Process the race definition line.""" + if not line[4:].startswith(self.RACE_DEFINITION): + raise RuntimeError( + f"Invalid line {line} in race definition: it does not start with " + f'"--- {self.RACE_DEFINITION}"') + + # Race definition is split in 10 parts + race_def_split = line[8:].split(" ") + if len(race_def_split) != 10: + raise RuntimeError( + f"Invalid line {line} in race definition: it does not contain the expected number of parts") + + # Determine the participating teams from the first part + self._process_race_definition_num_teams_entry(line, race_def_split[0], turing_dict) + + # Determine the number of questions from the second entry + self._process_race_definition_num_questions_entry(line, race_def_split[1], turing_dict) + + # Determine the initial score from the third entry + self._process_race_definition_initial_score_entry(line, race_def_split[2], turing_dict) + + # Determine the bonus cardinality from the fourth entry + self._process_race_definition_bonus_cardinality_entry(line, race_def_split[3], turing_dict) + + # Determine the superbonus cardinality from the fifth entry + self._process_race_definition_superbonus_cardinality_entry(line, race_def_split[4], turing_dict) + + # Determine the value of n and k from the sixth entry + self._process_race_definition_n_k_blocco_entry(line, race_def_split[5], turing_dict) + + # Determine the value of the alternative k parameter from the seventh entry + self._process_race_definition_alternative_k_blocco_entry(line, race_def_split[6], turing_dict) + + # Determine the value of the race type from the eighth entry + self._process_race_definition_race_type_entry(line, race_def_split[7], turing_dict) + + # Determine the total time of the race from the ninth entry + self._process_race_definition_total_time_entry(line, race_def_split[8], turing_dict) + + # Determine the deadline time for question score periodic increase from the tenth entry + self._process_race_definition_deadline_score_increase_entry(line, race_def_split[9], turing_dict) + + # mathrace does not store the cutoff, hence leave it unset in turing + turing_dict["cutoff"] = None + + def _process_race_definition_num_teams_entry( + self, line: str, num_teams: str, turing_dict: TuringDict + ) -> None: + """Process the number of teams entry in the race definition line.""" + turing_dict["mathrace_only"]["num_teams"] = num_teams + + def _process_race_definition_num_questions_entry( + self, line: str, num_questions: str, turing_dict: TuringDict + ) -> None: + """Process the number of questions entry in the race definition line.""" + turing_dict["num_problemi"] = num_questions + + def _process_race_definition_initial_score_entry( + self, line: str, initial_score: str, turing_dict: TuringDict + ) -> None: + """Process the initial score entry in the race definition line.""" + turing_dict["mathrace_only"]["initial_score"] = initial_score + expected_score = int(turing_dict["num_problemi"]) * 10 + if int(initial_score) != expected_score: + raise RuntimeError( + f"Invalid line {line} in race definition: the expected score is {expected_score}, " + f"but the race definition contains {initial_score}. This is not compatible with turing, " + f"since it does not allow to change the initial score") + + def _process_race_definition_bonus_cardinality_entry( + self, line: str, bonus_cardinality: str, turing_dict: TuringDict + ) -> None: + """Process the bonus cardinality entry in the race definition line.""" + turing_dict["mathrace_only"]["bonus_cardinality"] = bonus_cardinality + if int(bonus_cardinality) != 10: + raise RuntimeError( + f"Invalid line {line} in race definition: the expected bonus cardinality is 10, " + f"but the race definition contains {bonus_cardinality}. This is not compatible with this " + "version, which has hardcoded bonus points") + else: + turing_dict["fixed_bonus"] = "20,15,10,8,6,5,4,3,2,1" + + def _process_race_definition_superbonus_cardinality_entry( + self, line: str, superbonus_cardinality: str, turing_dict: TuringDict + ) -> None: + """Process the superbonus cardinality entry in the race definition line.""" + turing_dict["mathrace_only"]["superbonus_cardinality"] = superbonus_cardinality + if int(superbonus_cardinality) != 6: + raise RuntimeError( + f"Invalid line {line} in race definition: the expected superbonus cardinality is 6, " + f"but the race definition contains {superbonus_cardinality}. This is not compatible with this " + "version, which has hardcoded superbonus points") + else: + turing_dict["super_mega_bonus"] = "100,60,40,30,20,10" + + def _process_race_definition_n_k_blocco_entry( + self, line: str, n_k_blocco: str, turing_dict: TuringDict + ) -> None: + """Process the value of n in the race definition line.""" + turing_dict["n_blocco"] = n_k_blocco + turing_dict["k_blocco"] = "1" + + def _process_race_definition_alternative_k_blocco_entry( + self, line: str, alternative_k_blocco: str, turing_dict: TuringDict + ) -> None: + """Process the alternative k parameter entry in the race definition line.""" + turing_dict["mathrace_only"]["alternative_k_blocco"] = alternative_k_blocco + if int(alternative_k_blocco) != 1: + raise RuntimeError( + f"Invalid line {line} in race definition: the expected alternative k is 1, " + f"but the race definition contains {alternative_k_blocco}. This is not compatible with this " + "turing, which has hardcoded this value to 1") + + def _process_race_definition_race_type_entry( + self, line: str, race_type: str, turing_dict: TuringDict + ) -> None: + """Process the race type entry in the race definition line.""" + turing_dict["mathrace_only"]["race_type"] = race_type + if int(race_type) != 1: + raise RuntimeError( + f"Invalid line {line} in race definition: the expected race type is 1, " + f"but the race definition contains {race_type}. This is not compatible with this " + "turing, which has only supports a single race type") + + def _process_race_definition_total_time_entry( + self, line: str, total_time: str, turing_dict: TuringDict + ) -> None: + """Process the total time of the race in the race definition line.""" + turing_dict["durata"] = total_time + + def _process_race_definition_deadline_score_increase_entry( + self, line: str, deadline_score_increase: str, turing_dict: TuringDict + ) -> None: + """Process the deadline time for question score periodic increase in the race definition line.""" + turing_dict["durata_blocco"] = deadline_score_increase + + def _read_questions_definition_section(self, turing_dict: TuringDict) -> None: + """Read the questions definition section.""" + # Define first a default configuration in which every question has a score of 20 + questions: list[dict[str, str]] = list() + num_questions = int(turing_dict["num_problemi"]) + for q in range(num_questions): + questions.append({ + "problema": str(q + 1), "nome": f"Problema {q + 1}", + "risposta": str(int(True)), "punteggio": "20" + }) + turing_dict["soluzioni"] = questions + # Update the default initialization based on values read from file + line, before, _ = self._read_line_with_positions() + while line.startswith(f"--- {self.QUESTION_DEFINITION}"): + self._process_question_definition_line(line, turing_dict) + line, before, _ = self._read_line_with_positions() + # The line that caused the while loop to break was not a team definition line, so we need to + # reset the stream to the previous line + self._reset_stream_to_position(before) + + def _process_question_definition_line(self, line: str, turing_dict: TuringDict) -> None: + """Process a question definition line.""" + question_def = line[8:] + if "quesito " not in question_def: + raise RuntimeError( + f"Invalid line {line} in question definition: it does not contain the word quesito") + question_def, _ = question_def.split(" quesito") + question_id_str, question_score = question_def.split(" ") + question_id = int(question_id_str) - 1 + assert turing_dict["soluzioni"][question_id]["problema"] == question_id_str + turing_dict["soluzioni"][question_id]["punteggio"] = question_score + + def _read_teams_definition_section(self, turing_dict: TuringDict) -> None: + """Read the teams definition section.""" + # This format does not really have a teams definition section, so initialize a default set of teams + num_teams = int(turing_dict["mathrace_only"]["num_teams"]) + turing_dict["squadre"] = [ + {"nome": f"Squadra {t + 1}", "num": str(t + 1), "ospite": False} for t in range(num_teams)] + + def _read_race_events_section(self, turing_dict: TuringDict) -> None: + """Read all race events.""" + # Process the race start event first + line = self._read_line() + self._process_race_start_line(line, turing_dict) + # Allocate a mathrace only storage for manual corrections and timestamp offset + turing_dict["mathrace_only"]["manual_bonuses"] = list() + turing_dict["mathrace_only"]["timestamp_offset"] = "" + # Process the remaining race events until the race end one + turing_dict["eventi"] = list() + while True: + line = self._read_line() + try: + self._process_race_event_line(line, turing_dict) + except StopIteration: + break + + def _process_race_event_line(self, line: str, turing_dict: TuringDict) -> None: + """Process a race event line.""" + timestamp_str, event_type, event_content = line.split(" ", maxsplit=2) + if event_type == self.JOLLY_SELECTION: + self._process_jolly_selection_event(timestamp_str, event_content, turing_dict) + elif event_type == self.ANSWER_SUBMISSION: + self._process_answer_submission_event(timestamp_str, event_content, turing_dict) + elif event_type == self.JOLLY_TIMEOUT: + self._process_jolly_timeout_event(timestamp_str, event_content, turing_dict) + elif event_type == self.TIMER_UPDATE: + self._process_timer_update_event(timestamp_str, event_content, turing_dict) + elif event_type == self.RACE_SUSPENDED: + self._process_race_suspended_event(timestamp_str, event_content, turing_dict) + elif event_type == self.RACE_RESUMED: + self._process_race_resumed_event(timestamp_str, event_content, turing_dict) + elif event_type == self.RACE_END: + self._process_race_end_event(timestamp_str, event_content, turing_dict) + elif event_type == self.MANUAL_BONUS: + self._process_manual_bonus_event(timestamp_str, event_content, turing_dict) + else: + raise RuntimeError(f"Invalid line {line} in race events: unhandled event type {event_type}") + + def _process_race_start_line(self, line: str, turing_dict: TuringDict) -> None: + """Ensure that the race start line is correctly formatted.""" + if not line.endswith(f"{self.RACE_START} inizio gara"): + raise RuntimeError(f"Invalid line {line} in race event: it does not contain the race start") + + def _process_jolly_selection_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a jolly selection event.""" + # Allow jolly to be selected even before the offset is computed, since setting it with + # a slightly wrong timestamp does not affect the overall score of the race + event_datetime = self._convert_timestamp_to_datetime(timestamp_str, False, turing_dict) + # Determine mathrace event ID, if available + event_mathrace_id = self._determine_mathrace_event_id(event_content) + # Process the event content + team_id, question_id, _ = event_content.split(" ", maxsplit=2) + # Append to output dictionary + turing_dict["eventi"].append({ + "subclass": "Jolly", "orario": event_datetime.isoformat(), + "squadra_id" : team_id, "problema" : question_id, "mathrace_id": event_mathrace_id + }) + + def _process_answer_submission_event( + self, timestamp_str: str, event_content: str, turing_dict: TuringDict + ) -> None: + """Process an answer submission event.""" + # Answer submission requires a strict datetime, including time stamp offset, because + # slightly different times may end up affecting the overall team score + event_datetime = self._convert_timestamp_to_datetime(timestamp_str, True, turing_dict) + # Determine mathrace event ID, if available + event_mathrace_id = self._determine_mathrace_event_id(event_content) + # Process the event content + team_id, question_id, answer, _ = event_content.split(" ", maxsplit=3) + # Append to output dictionary + turing_dict["eventi"].append({ + "subclass": "Consegna", "orario": event_datetime.isoformat(), + "squadra_id" : team_id, "problema" : question_id, "risposta": answer, "mathrace_id": event_mathrace_id + }) + + def _process_jolly_timeout_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a jolly timeout event.""" + event_datetime = self._convert_timestamp_to_datetime(timestamp_str, True, turing_dict) + turing_dict["mathrace_only"]["jolly_timeout"] = event_datetime.isoformat() + + def _process_timer_update_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """ + Process a timer update event. + + Timer update events are not propagated to turing, with the exception of the first event that + is used to compute mathrace timer offset. + """ + if turing_dict["mathrace_only"]["timestamp_offset"] == "": + # Compute timestamp offset the first time a timer update event is trigger + if not event_content.startswith("aggiorna punteggio esercizi"): + raise RuntimeError(f"Invalid event content {event_content} in timer update event") + turing_dict["mathrace_only"]["timestamp_offset"] = str(60 - int(timestamp_str)) + else: + # Assume that the offset is constant throughout the race, and do nothing + pass + + def _process_race_suspended_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a race suspended event. Currently ignored.""" + pass + + def _process_race_resumed_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a race resumed event. Currently ignored.""" + pass + + def _process_race_end_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a race end event to stop iterating through the file.""" + raise StopIteration() + + def _process_manual_bonus_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a manual bonus event.""" + event_datetime = self._convert_timestamp_to_datetime(timestamp_str, True, turing_dict) + turing_dict["mathrace_only"]["manual_bonuses"].append({ + "orario": event_datetime, "motivazione": event_content}) + + def _convert_timestamp_to_datetime( + self, timestamp_str: str, strict: bool, turing_dict: TuringDict + ) -> datetime.datetime: + """ + Convert a timestamp into a date and time. + + The strict flag controls the behavior when the value of timestamp_offset is empty (i.e., uninitialized). + If strict is enabled, an empty value of timestamp_offset causes an error. + If strict is disabled, an empty value of timestamp_offset is considered as zero. + """ + timestamp_offset = turing_dict["mathrace_only"]["timestamp_offset"] + if timestamp_offset != "": + timestamp = int(timestamp_str) + int(timestamp_offset) + else: + if strict: + raise RuntimeError( + f"Cannot convert {timestamp_str} to date and time because of empty timestamp offset") + else: + timestamp = int(timestamp_str) + return self._race_date + datetime.timedelta(seconds=timestamp) + + def _determine_mathrace_event_id(self, event_content: str) -> str: + """Determine mathrace event ID. This version does not store them, so a placeholder is returned.""" + return "-1" + + +class JournalReaderR11167(JournalReaderR5539): + """ + A class representing a reader of a mathrace journal, version r11167. + + This version introduced some code changes in race events + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + + Class Attributes + ---------------- + RACE_START + The race event code associated to the start of the race (changed from r5539). + JOLLY_SELECTION + The race event code associated to jolly selection by a team (changed from r5539). + ANSWER_SUBMISSION + The race event code associated to answer submission by a team (changed from r5539). + JOLLY_TIMEOUT + The race event code associated to jolly timeout (changed from r5539). + TIMER_UPDATE + The race event code associated to a timer update (changed from r5539). + RACE_SUSPENDED + The race event code associated to a race suspension (changed from r5539). + RACE_RESUMED + The race event code associated to a race resumption (changed from r5539). + RACE_END + The race event code associated to the end of the race (changed from r5539). + MANUAL_BONUS + The race event code associated to the manual addition of a bonus (changed from r5539). + """ + + # Race event codes + RACE_START = "200" + JOLLY_SELECTION = "120" + ANSWER_SUBMISSION = "110" + JOLLY_TIMEOUT = "121" + TIMER_UPDATE = "101" + RACE_SUSPENDED = "201" + RACE_RESUMED = "202" + RACE_END = "210" + MANUAL_BONUS = "130" + + +class JournalReaderR11184(JournalReaderR11167): + """ + A class representing a reader of a mathrace journal, version r11184. + + This version introduced protocol numbers in some race events. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + """ + + def _determine_mathrace_event_id(self, event_content: str) -> str: + """Determine mathrace event ID, based on it protocol number.""" + if " PROT:" not in event_content: + raise RuntimeError( + f"Cannot determine protocol number from {event_content}") + else: + _, after_prot = event_content.split(" PROT:") + prot, _ = after_prot.split(" ", maxsplit=1) + return prot + + +class JournalReaderR11189(JournalReaderR11184): + """ + A class representing a reader of a mathrace journal, version r11189. + + This version added a further timer event. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + + Class Attributes + ---------------- + TIMER_UPDATE_OTHER_TIMER + The race event code associated to a timer update of the second timer. + """ + + TIMER_UPDATE_OTHER_TIMER = "901" + + def _process_race_event_line(self, line: str, turing_dict: TuringDict) -> None: + """Process a race event line. Ignore any event assiocated to the update of the second timer.""" + timestamp_str, event_type, event_content = line.split(" ", maxsplit=2) + if event_type == self.TIMER_UPDATE_OTHER_TIMER: + # Ignore the event + pass + else: + super()._process_race_event_line(line, turing_dict) + + +class JournalReaderR17497(JournalReaderR11189): + """ + A class representing a reader of a mathrace journal, version r17497. + + This version added support for non-default k. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + """ + + def _process_race_definition_n_k_blocco_entry( + self, line: str, n_k_blocco: str, turing_dict: TuringDict + ) -> None: + """Process the value of n and k in the race definition line.""" + if "." in n_k_blocco: + n_blocco, k_blocco = n_k_blocco.split(".") + turing_dict["n_blocco"] = n_blocco + turing_dict["k_blocco"] = k_blocco + else: + super()._process_race_definition_n_k_blocco_entry(line, n_k_blocco, turing_dict) + + +class JournalReaderR17505(JournalReaderR17497): + """ + A class representing a reader of a mathrace journal, version r17505. + + This version added team definition as a race code. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + + Class Attributes + ---------------- + TEAM_DEFINITION + The race setup code associated to the definition of a team. + """ + + TEAM_DEFINITION = "005" + + def _read_teams_definition_section(self, turing_dict: TuringDict) -> None: + """Read the teams definition section.""" + # Call the parent initialization to assign default values to any team which may not have been provided + super()._read_teams_definition_section(turing_dict) + teams = turing_dict["squadre"] + # Update the default initialization based on values read from file + line, before, _ = self._read_line_with_positions() + while line.startswith(f"--- {self.TEAM_DEFINITION}"): + team_def = line[8:] + team_id_str, team_guest_status, team_name = team_def.split(" ", maxsplit=2) + team_id = int(team_id_str) - 1 + teams[team_id]["nome"] = team_name + teams[team_id]["ospite"] = True if team_guest_status == "1" else False + line, before, _ = self._read_line_with_positions() + # The line that caused the while loop to break was not a team definition line, so we need to + # reset the stream to the previous line + self._reset_stream_to_position(before) + + + +class JournalReaderR17548(JournalReaderR17505): + """ + A class representing a reader of a mathrace journal, version r17548. + + This version added an alternative race definition. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + + Class Attributes + ---------------- + RACE_DEFINITION_ALTERNATIVE + The race setup code associated to the alternative race definition. + """ + + RACE_DEFINITION_ALTERNATIVE = "002" + + def _process_race_definition_line(self, line: str, turing_dict: TuringDict) -> None: + """Process the race definition line, include the alternative race definition.""" + # Try with the standard race definition first + if line[4:].startswith(self.RACE_DEFINITION): + super()._process_race_definition_line(line, turing_dict) + return + + # Otherwise, continue with the alternative format + if not line[4:].startswith(self.RACE_DEFINITION_ALTERNATIVE): + raise RuntimeError( + f"Invalid line {line} in race definition: it does not start with " + f'"--- {self.RACE_DEFINITION}" or "--- {self.RACE_DEFINITION_ALTERNATIVE}"') + + # Race definition is split in 4 parts + race_def_split = line[8:].split(" ") + if len(race_def_split) != 4: + raise RuntimeError( + f"Invalid line {line} in alternative race definition: it does not contain the expected " + "number of parts") + + # Determine the participating teams from the first part. This may also include the initial score + self._process_alternative_race_definition_num_teams_entry(line, race_def_split[0], turing_dict) + + # Determine the number of questions from the second entry + self._process_alternative_race_definition_num_questions_entry(line, race_def_split[1], turing_dict) + + # Determine the value of n, k and alternative k from the third entry + self._process_alternative_race_definition_n_k_altk_blocco_entry(line, race_def_split[2], turing_dict) + + # Determine the total time of the race, and the deadline time for question score periodic increase, + # from the fourth entry + self._process_alternative_race_definition_total_time_deadline_score_increase_entry( + line, race_def_split[3], turing_dict) + + # Bonus and superbonus cardinality are not reported in this format, and delayed to the next two lines + turing_dict["mathrace_only"]["bonus_cardinality"] = "N/A" + turing_dict["mathrace_only"]["superbonus_cardinality"] = "N/A" + + # mathrace does not store the cutoff, hence leave it unset in turing + turing_dict["cutoff"] = None + + def _process_alternative_race_definition_num_teams_entry( + self, line: str, num_teams_alternative: str, turing_dict: TuringDict + ) -> None: + """Process the number of teams entry in the alternative race definition line.""" + # Split the input string into at most three substrings + if ":" in num_teams_alternative: + num_teams_all, initial_score = num_teams_alternative.split(":") + else: + num_teams_all = num_teams_alternative + initial_score = "" + if "+" in num_teams_all: + num_teams_nonguests, num_teams_guests = num_teams_all.split("+") + else: + num_teams_nonguests = num_teams_all + num_teams_guests = "0" + # Store the results in the mathrace only section + turing_dict["mathrace_only"]["num_teams"] = str(int(num_teams_nonguests) + int(num_teams_guests)) + turing_dict["mathrace_only"]["num_teams_alternative"] = num_teams_alternative + turing_dict["mathrace_only"]["num_teams_nonguests"] = num_teams_nonguests + turing_dict["mathrace_only"]["num_teams_guests"] = num_teams_guests + # Initialize the initial score, if possible + if initial_score != "": + turing_dict["mathrace_only"]["initial_score"] = initial_score + else: + # It is not possible to set the initial score, because we do not know yet the number of + # questions in the race. Delay the initialization until the second entry gets read + turing_dict["mathrace_only"]["initial_score"] = "N/A" + + def _process_alternative_race_definition_num_questions_entry( + self, line: str, num_questions_alternative: str, turing_dict: TuringDict + ) -> None: + """Process the number of questions entry in the alternative race definition line.""" + if ":" in num_questions_alternative: + num_questions, default_score = num_questions_alternative.split(":") + else: + num_questions = num_questions_alternative + # Store the results in the dictionary + turing_dict["mathrace_only"]["num_questions_alternative"] = num_questions_alternative + turing_dict["num_problemi"] = num_questions + turing_dict["mathrace_only"]["default_score"] = num_questions_alternative + # It is now possible to set the initial score, if it calculation was delayed when reading the first entry + initial_score = turing_dict["mathrace_only"]["initial_score"] + if initial_score == "N/A": + initial_score == str(int(num_questions) * 10) + else: + expected_score = int(num_questions) * 10 + if int(initial_score) != expected_score: + raise RuntimeError( + f"Invalid line {line} in race definition: the expected score is {expected_score}, " + f"but the race definition contains {initial_score}. This is not compatible with turing, " + f"since it does not allow to change the initial score") + + def _process_alternative_race_definition_n_k_altk_blocco_entry( + self, line: str, n_k_altk_blocco: str, turing_dict: TuringDict + ) -> None: + """Process the value of n, k and alternative k in the alternative race definition line.""" + if ";" in n_k_altk_blocco: + n_k_blocco, alternative_k_blocco = n_k_altk_blocco.split(";") + else: + n_k_blocco = n_k_altk_blocco + alternative_k_blocco = "1" + if "." in n_k_blocco: + n_blocco, k_blocco = n_k_blocco.split(".") + else: + n_blocco = n_k_blocco + k_blocco = "1" + # Store n_blocco and k_blocco + turing_dict["n_blocco"] = n_blocco + turing_dict["k_blocco"] = k_blocco + # Call parent implementation to store alternative_k_blocco, which also carries out consistency checks + super()._process_race_definition_alternative_k_blocco_entry(line, alternative_k_blocco, turing_dict) + + def _process_alternative_race_definition_total_time_deadline_score_increase_entry( + self, line: str, times: str, turing_dict: TuringDict + ) -> None: + """Process the total time of the race in the race definition line.""" + if "-" not in times: + raise RuntimeError( + f"Invalid line {line} in race definition: it does not contain the opreator -") + turing_dict["durata"], turing_dict["durata_blocco"] = times.split("-") + + +class JournalReaderR20642(JournalReaderR17548): + """ + A class representing a reader of a mathrace journal, version r20642. + + This version added bonus and superbonus race codes. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + + Class Attributes + ---------------- + BONUS_DEFINITION + The race setup code associated to bonus definition. + SUPERBONUS_DEFINITION + The race setup code associated to superbonus definition. + """ + + BONUS_DEFINITION = "011" + SUPERBONUS_DEFINITION = "012" + + def _read_race_definition_section(self, turing_dict: TuringDict) -> None: + """Read the race definition section, including two further lines for bonus and superbonus definition.""" + # Read the first line as in previous versions + super()._read_race_definition_section(turing_dict) + + # Read next the bonus and superbonus definition, if available + for (code, mathrace_key, turing_key) in ( + (self.BONUS_DEFINITION, "bonus_cardinality", "fixed_bonus"), + (self.SUPERBONUS_DEFINITION, "superbonus_cardinality", "super_mega_bonus") + ): + line, before, _ = self._read_line_with_positions() + if not line.startswith(f"--- {code}"): + # Bonus definition was not found: reset the stream to the previous line, because it has now read + # a line that belongs to the next section + self._reset_stream_to_position(before) + else: + self._process_bonus_or_superbonus_definition_line(line, turing_dict, mathrace_key, turing_key) + + def _process_bonus_or_superbonus_definition_line( + self, line: str, turing_dict: TuringDict, mathrace_key: str, turing_key: str + ) -> None: + """Process the bonus or superbonus definition line.""" + if "definizione dei" in line: + line, _ = line[8:].split("definizione dei") + else: + line = line[8:] + bonus_split = line.split(" ") + bonus_values_cardinality = bonus_split[0] + bonus_cardinality = turing_dict["mathrace_only"][mathrace_key] + if bonus_cardinality == "N/A": + # Read in all numbers in the line, except for the first one containing the maximum cardinality + bouns_values_end = -1 + turing_dict["mathrace_only"][mathrace_key] = bonus_values_cardinality + else: + if int(bonus_values_cardinality) < int(bonus_cardinality): + raise RuntimeError(f"Invalid line {line} in race definition: not enough values to read") + bouns_values_end = int(bonus_cardinality) + 1 + turing_dict[turing_key] = ",".join(bonus_split[1:bouns_values_end]) + + +class JournalReaderR20644(JournalReaderR20642): + """ + A class representing a reader of a mathrace journal, version r20644. + + This version prints human readable timestamps. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + """ + + def _process_jolly_selection_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a jolly selection event, preprocessing the timestamp.""" + return super()._process_jolly_selection_event( + self._convert_timestamp_to_number_of_seconds(timestamp_str), event_content, turing_dict) + + def _process_answer_submission_event( + self, timestamp_str: str, event_content: str, turing_dict: TuringDict + ) -> None: + """Process an answer submission event, preprocessing the timestamp.""" + return super()._process_answer_submission_event( + self._convert_timestamp_to_number_of_seconds(timestamp_str), event_content, turing_dict) + + def _process_jolly_timeout_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a jolly timeout event, preprocessing the timestamp.""" + return super()._process_jolly_timeout_event( + self._convert_timestamp_to_number_of_seconds(timestamp_str), event_content, turing_dict) + + def _process_timer_update_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a timer update event, preprocessing the timestamp.""" + return super()._process_timer_update_event( + self._convert_timestamp_to_number_of_seconds(timestamp_str), event_content, turing_dict) + + def _process_manual_bonus_event(self, timestamp_str: str, event_content: str, turing_dict: TuringDict) -> None: + """Process a manual bonus event, preprocessing the timestamp.""" + return super()._process_manual_bonus_event( + self._convert_timestamp_to_number_of_seconds(timestamp_str), event_content, turing_dict) + + def _convert_timestamp_to_number_of_seconds(self, timestamp_str: str) -> str: + """Convert a timestamp of the form hh:mm:ss.msec to an integer number of seconds.""" + return str(int(sum(x * float(t) for x, t in zip([1, 60, 3600], reversed(timestamp_str.split(":")))))) + +class JournalReaderR25013(JournalReaderR20644): + """ + A class representing a reader of a mathrace journal, version r25013. + + This version added an extra field to the question definition, but with a placeholder value + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + """ + + def _process_question_definition_line(self, line: str, turing_dict: TuringDict) -> None: + """Process a question definition line, discarding the placeholder for the exact answer.""" + question_def = line[8:] + if "quesito " not in question_def: + raise RuntimeError( + f"Invalid line {line} in question definition: it does not contain the word quesito") + question_def, _ = question_def.split(" quesito") + question_id_str, question_score, question_exact_answer = question_def.split(" ") + if question_exact_answer != "0000": + raise RuntimeError( + f"Invalid line {line} in question definition: it does not contain the expected placeholder") + question_id = int(question_id_str) - 1 + assert turing_dict["soluzioni"][question_id]["problema"] == question_id_str + turing_dict["soluzioni"][question_id]["punteggio"] = question_score + + +def journal_reader( + journal_stream: typing.TextIO, race_name: str, race_date: datetime.datetime +) -> AbstractJournalReader: + """ + Read a mathrace journal. + + Parameters + ---------- + journal_stream + The I/O stream that reads the journal generated by mathrace or simdis. + The I/O stream is typically generated by open(). + race_name + Name of the race. + race_date + Date of the race. + """ + # Determine the version of the mathrace journal + version = determine_journal_version(journal_stream) + # The previous called has consumed the stream: reset it back to the beginning + journal_stream.seek(0) + # Return an object of the class corresponding to the detected version + journal_reader_class = getattr(sys.modules[__name__], f"JournalReader{version.capitalize()}") + return journal_reader_class(journal_stream, race_name, race_date) # type: ignore[no-any-return] diff --git a/mathrace_interaction/pyproject.toml b/mathrace_interaction/pyproject.toml index 4c0e16b..8654f0c 100644 --- a/mathrace_interaction/pyproject.toml +++ b/mathrace_interaction/pyproject.toml @@ -13,7 +13,7 @@ maintainers = [ ] description = "Interfacing mathrace and turing" license = {text = "GNU Affero General Public License v3 or later (AGPLv3+)"} -requires-python = ">=3.10" +requires-python = ">=3.11" classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", @@ -24,7 +24,6 @@ classifiers = [ "Operating System :: MacOS :: MacOS X", "Programming Language :: Python", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Topic :: Software Development :: Libraries :: Python Modules" diff --git a/mathrace_interaction/tests/unit/test_journal_reader.py b/mathrace_interaction/tests/unit/test_journal_reader.py new file mode 100644 index 0000000..fc8d69a --- /dev/null +++ b/mathrace_interaction/tests/unit/test_journal_reader.py @@ -0,0 +1,19 @@ +# Copyright (C) 2024 by the Turing @ DMF authors +# +# This file is part of Turing @ DMF. +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Test mathrace_interaction.journal_reader.""" + +import datetime +import pathlib + +from mathrace_interaction.journal_reader import journal_reader + + +def test_journal_reader_success(data_journals: list[pathlib.Path]) -> None: + """Test that journal_reader runs successfully on all journals in the data directory.""" + for journal in data_journals: + journal_date = datetime.datetime(int(journal.parent.name), 1, 1, tzinfo=datetime.timezone.utc) + with journal_reader(open(journal), journal.name, journal_date) as journal_stream: + journal_stream.read()