feature: on-off-tasks (#11)

0d766a4 3 months ago

24.4 kB

from typing import Dict, List, Any
from measures.VocabularyAnalyser import VocabularyAnalyser
from measures.OnTaskAnalyser import OnTaskAnalyser, ON_TASK_LABEL
from scipy.special import softmax
from collections import Counter, defaultdict
import numpy as np
import weakref
import re
import nltk
from nltk.corpus import stopwords
nltk.download('stopwords')

from utils import clean_str, clean_str_nopunct
import torch
from utils import MultiHeadModel, BertInputBuilder, get_num_words, MATH_PREFIXES, MATH_WORDS, plural_to_singular

import transformers
from transformers import BertTokenizer, BertForSequenceClassification
from transformers.utils import logging
from pathlib import Path

transformers.logging.set_verbosity_debug()

UPTAKE_MODEL = 'ddemszky/uptake-model'
REASONING_MODEL = 'ddemszky/student-reasoning'
QUESTION_MODEL = 'ddemszky/question-detection'
FOCUSING_QUESTION_MODEL = 'ddemszky/focusing-questions'


class Utterance:
    def __init__(self, speaker, text, uid=None,
                 transcript=None, starttime=None, endtime=None, **kwargs):
        self.speaker = speaker
        self.text = text
        self.uid = uid
        self.starttime = starttime
        self.endtime = endtime
        self.transcript = weakref.ref(transcript) if transcript else None
        self.props = kwargs
        self.role = None
        self.word_count = self.get_num_words()
        self.timestamp = [starttime, endtime]
        if starttime is not None and endtime is not None:                   
            self.unit_measure = endtime - starttime
        else:
            self.unit_measure = None
        self.aggregate_unit_measure = endtime  
        self.num_math_terms = None
        self.math_terms = None
        self.vocabulary_terms = None
        self.on_task = None

        # moments
        self.uptake = None
        self.reasoning = None
        self.question = None
        self.focusing_question = None 

    def get_clean_text(self, remove_punct=False):
        if remove_punct:
            return clean_str_nopunct(self.text)
        return clean_str(self.text)

    def get_num_words(self):
        return get_num_words(self.text)

    def to_dict(self):
        return {
            'speaker': self.speaker,
            'text': self.text,
            'uid': self.uid,
            'starttime': self.starttime,
            'endtime': self.endtime,
            'uptake': self.uptake,
            'reasoning': self.reasoning,
            'question':  self.question,
            'focusingQuestion': self.focusing_question,
            'numMathTerms': self.num_math_terms,
            'mathTerms': self.math_terms,
            'vocabularyTerms': self.vocabulary_terms,
            'vocabularyMatches': self.vocabulary_matches,
            **self.props
        }

    def to_talk_timeline_dict(self):
        return{
            'speaker': self.speaker,
            'text': self.text,
            'uid': self.uid,
            'role': self.role,
            'timestamp': self.timestamp,
            'moments': {
                'reasoning': True if self.reasoning else False, 
                'questioning': True if self.question else False, 
                'uptake': True if self.uptake else False, 
                'focusingQuestion': True if self.focusing_question else False,
                # Legacy alias for onTask below. On/off-task is not a talk moment;
                # this key is kept only so already-stored analyses and older
                # frontends keep working, and is retired in the UI.
                'mathWord': bool(self.on_task) and self.on_task.get('label') == ON_TASK_LABEL
            },
            # On/off-task classification: a measure in its own right, alongside
            # vocabulary. None means not classified (teacher utterance or empty text).
            'onTask': None if self.on_task is None else self.on_task.get('label') == ON_TASK_LABEL,
            'unitMeasure': self.unit_measure,
            'aggregateUnitMeasure': self.aggregate_unit_measure,
            'wordCount': self.word_count,
            'numMathTerms': self.num_math_terms,
            'mathTerms': self.math_terms,
            'vocabularyTerms': self.vocabulary_terms,
            'vocabularyMatches': self.vocabulary_matches
        }

    def __repr__(self):
        return f"Utterance(speaker='{self.speaker}'," \
               f"text='{self.text}', uid={self.uid}," \
               f"starttime={self.starttime}, endtime={self.endtime}, props={self.props})"

class Transcript:
    def __init__(self, **kwargs):
        self.utterances = []
        self.params = kwargs

    def add_utterance(self, utterance):
        utterance.transcript = weakref.ref(self)
        self.utterances.append(utterance)

    def get_idx(self, idx):
        if idx >= len(self.utterances):
            return None
        return self.utterances[idx]

    def get_uid(self, uid):
        for utt in self.utterances:
            if utt.uid == uid:
                return utt
        return None

    def length(self):
        return len(self.utterances)

    def update_utterance_roles(self, uptake_speaker):
        for utt in self.utterances:
            if (utt.speaker == uptake_speaker):
                utt.role = 'teacher'
            else:
                utt.role = 'student'

    def get_talk_distribution_and_length(self, uptake_speaker):
        if ((uptake_speaker is None)):
            return None
        teacher_words = 0
        teacher_utt_count = 0
        student_words = 0
        student_utt_count = 0
        for utt in self.utterances:
            if (utt.speaker == uptake_speaker):
                utt.role = 'teacher'
                teacher_words += utt.get_num_words()
                teacher_utt_count += 1
            else:
                utt.role = 'student'
                student_words += utt.get_num_words()
                student_utt_count += 1
        if teacher_words + student_words > 0:         
            teacher_percentage = round(
                (teacher_words / (teacher_words + student_words)) * 100)
            student_percentage = 100 - teacher_percentage
        else:
            teacher_percentage = student_percentage = 0
        avg_teacher_length = teacher_words / teacher_utt_count if teacher_utt_count > 0 else 0
        avg_student_length = student_words / student_utt_count if student_utt_count > 0 else 0
        return {'teacher': teacher_percentage, 'student': student_percentage}, {'teacher': avg_teacher_length, 'student': avg_student_length}

    def get_vocabulary_table(self):
        """
        Build a summary of matched vocabulary terms by utterance role.
        Returns a dict of the form:
        {
          'add': {
             'teacher': {'count': 2, 'utterances': [1,5]},
             'student': {'count': 1, 'utterances': [2]}
          },
          'divide': {
             'teacher': {...},
             'student': {...}
          }
        }
        """
        vocab_summary = defaultdict(lambda: {"teacher": {"count": 0, "utterances": []},
                                             "student": {"count": 0, "utterances": []}})

        for utt in self.utterances:
            role = utt.role if utt.role in ("teacher", "student") else "student"  # default
            if utt.vocabulary_terms:
                for term in utt.vocabulary_terms:
                    vocab_summary[term][role]["count"] += 1
                    vocab_summary[term][role]["utterances"].append(utt.uid)

        return vocab_summary


    def get_word_clouds(self):
        # Initialize dictionaries
        teacher_dict = Counter()
        student_dict = Counter()
        uptake_teacher_dict = Counter()
        stop_words = stopwords.words('english')

        # Go through the utterances
        for utt in self.utterances:
            # Get clean text
            clean_text = utt.get_clean_text(remove_punct=True)
            words = clean_text.split()
            words = [word for word in words if word not in stop_words and word not in ['inaudible', 'crosstalk']]

            # Handle uptake case
            if utt.role == 'teacher' and utt.uptake == 1:
                uptake_teacher_dict.update(words)

            general_text = ' '.join(words)
            # Replace math terms with empty strings
            for math_term in utt.math_terms:
                general_text = general_text.replace(math_term, '')
                general_text = general_text.replace('  ', ' ')

            general_words = general_text.split()
            # Update the appropriate dictionary
            if utt.role == 'teacher':
                teacher_dict.update(general_words)
            else:
                student_dict.update(general_words)

        # Sorting and trimming dictionaries
        dict_list = dict_to_list(teacher_dict, 'general') + dict_to_list(student_dict, 'general')
        uptake_dict_list = dict_to_list(uptake_teacher_dict, 'teacher')
        teacher_dict_list = dict_to_list(teacher_dict, 'general')
        student_dict_list = dict_to_list(student_dict, 'general')

        sorted_dict_list = sorted(dict_list, key=lambda x: x['value'], reverse=True)
        sorted_uptake_dict_list = sorted(uptake_dict_list, key=lambda x: x['value'], reverse=True)
        sorted_teacher_dict_list = sorted(teacher_dict_list, key=lambda x: x['value'], reverse=True)
        sorted_student_dict_list = sorted(student_dict_list, key=lambda x: x['value'], reverse=True)
        return sorted_dict_list[:50], sorted_uptake_dict_list[:50], sorted_teacher_dict_list[:50], sorted_student_dict_list[:50]

    def get_talk_timeline(self):
        return [utterance.to_talk_timeline_dict() for utterance in self.utterances]
    
    def calculate_aggregate_word_count(self):
        unit_measures = [utt.unit_measure for utt in self.utterances]
        if None in unit_measures:
            aggregate_word_count = 0
            for utt in self.utterances: 
                aggregate_word_count += utt.get_num_words()
                utt.unit_measure = utt.get_num_words()
                utt.aggregate_unit_measure = aggregate_word_count


    def to_dict(self):
        return {
            'utterances': [utterance.to_dict() for utterance in self.utterances],
            **self.params
        }

    def __repr__(self):
        return f"Transcript(utterances={self.utterances}, custom_params={self.params})"

class QuestionModel:
    def __init__(self, device, tokenizer, input_builder, max_length=300, path=QUESTION_MODEL):
        print("Loading models...")
        self.device = device
        self.tokenizer = tokenizer
        self.input_builder = input_builder
        self.max_length = max_length
        self.model = MultiHeadModel.from_pretrained(
            path, head2size={"is_question": 2})
        self.model.to(self.device)

    def run_inference(self, transcript):
        self.model.eval()
        with torch.no_grad():
            for i, utt in enumerate(transcript.utterances):
                if "?" in utt.text:
                    utt.question = 1
                else:
                    text = utt.get_clean_text(remove_punct=True)
                    instance = self.input_builder.build_inputs([], text,
                                                               max_length=self.max_length,
                                                               input_str=True)
                    output = self.get_prediction(instance)
                    # print(output)
                    utt.question = np.argmax(
                        output["is_question_logits"][0].tolist())

    def get_prediction(self, instance):
        instance["attention_mask"] = [[1] * len(instance["input_ids"])]
        for key in ["input_ids", "token_type_ids", "attention_mask"]:
            instance[key] = torch.tensor(
                instance[key]).unsqueeze(0)  # Batch size = 1
            instance[key].to(self.device)

        output = self.model(input_ids=instance["input_ids"],
                            attention_mask=instance["attention_mask"],
                            token_type_ids=instance["token_type_ids"],
                            return_pooler_output=False)
        return output

class ReasoningModel:
    def __init__(self, device, tokenizer, input_builder, max_length=128, path=REASONING_MODEL):
        print("Loading models...")
        self.device = device
        self.tokenizer = tokenizer
        self.input_builder = input_builder
        self.max_length = max_length
        self.model = BertForSequenceClassification.from_pretrained(path)
        self.model.to(self.device)

    def run_inference(self, transcript, min_num_words=8, uptake_speaker=None):
        self.model.eval()
        with torch.no_grad():
            for i, utt in enumerate(transcript.utterances):
                if utt.get_num_words() >= min_num_words and utt.speaker != uptake_speaker:
                    instance = self.input_builder.build_inputs([], utt.text,
                                                               max_length=self.max_length,
                                                               input_str=True)
                    output = self.get_prediction(instance)
                    utt.reasoning = np.argmax(output["logits"][0].tolist())

    def get_prediction(self, instance):
        instance["attention_mask"] = [[1] * len(instance["input_ids"])]
        for key in ["input_ids", "token_type_ids", "attention_mask"]:
            instance[key] = torch.tensor(
                instance[key]).unsqueeze(0)  # Batch size = 1
            instance[key].to(self.device)

        output = self.model(input_ids=instance["input_ids"],
                            attention_mask=instance["attention_mask"],
                            token_type_ids=instance["token_type_ids"])
        return output

class UptakeModel:
    def __init__(self, device, tokenizer, input_builder, max_length=120, path=UPTAKE_MODEL):
        print("Loading models...")
        self.device = device
        self.tokenizer = tokenizer
        self.input_builder = input_builder
        self.max_length = max_length
        self.model = MultiHeadModel.from_pretrained(path, head2size={"nsp": 2})
        self.model.to(self.device)

    def run_inference(self, transcript, min_prev_words, uptake_speaker=None):
        self.model.eval()
        prev_num_words = 0
        prev_utt = None
        with torch.no_grad():
            for i, utt in enumerate(transcript.utterances):
                if ((uptake_speaker is None) or (utt.speaker == uptake_speaker)) and (prev_num_words >= min_prev_words):
                    textA = prev_utt.get_clean_text(remove_punct=False)
                    textB = utt.get_clean_text(remove_punct=False)
                    instance = self.input_builder.build_inputs([textA], textB,
                                                               max_length=self.max_length,
                                                               input_str=True)
                    output = self.get_prediction(instance)

                    utt.uptake = int(
                        softmax(output["nsp_logits"][0].tolist())[1] > .8)
                prev_num_words = utt.get_num_words()
                prev_utt = utt

    def get_prediction(self, instance):
        instance["attention_mask"] = [[1] * len(instance["input_ids"])]
        for key in ["input_ids", "token_type_ids", "attention_mask"]:
            instance[key] = torch.tensor(
                instance[key]).unsqueeze(0)  # Batch size = 1
            instance[key].to(self.device)

        output = self.model(input_ids=instance["input_ids"],
                            attention_mask=instance["attention_mask"],
                            token_type_ids=instance["token_type_ids"],
                            return_pooler_output=False)
        return output

class FocusingQuestionModel:
    def __init__(self, device, tokenizer, input_builder, max_length=128, path=FOCUSING_QUESTION_MODEL):
        print("Loading models...")
        self.device = device
        self.tokenizer = tokenizer
        self.input_builder = input_builder
        self.model = BertForSequenceClassification.from_pretrained(path)
        self.model.to(self.device)
        self.max_length = max_length

    def run_inference(self, transcript, min_focusing_words=0, uptake_speaker=None):
        self.model.eval()
        with torch.no_grad():
            for i, utt in enumerate(transcript.utterances):
                if utt.speaker != uptake_speaker or uptake_speaker is None:
                    utt.focusing_question = None
                    continue
                if utt.get_num_words() < min_focusing_words:
                    utt.focusing_question = None
                    continue
                instance = self.input_builder.build_inputs([], utt.text, max_length=self.max_length, input_str=True)
                output = self.get_prediction(instance)
                utt.focusing_question = np.argmax(output["logits"][0].tolist())

    def get_prediction(self, instance):
        instance["attention_mask"] = [[1] * len(instance["input_ids"])]
        for key in ["input_ids", "token_type_ids", "attention_mask"]:
            instance[key] = torch.tensor(
                instance[key]).unsqueeze(0)  # Batch size = 1
            instance[key].to(self.device)

        output = self.model(input_ids=instance["input_ids"],
                            attention_mask=instance["attention_mask"],
                            token_type_ids=instance["token_type_ids"])
        return output   

def dict_to_list(d, category):
    combined_dict = Counter()
    for word, count in d.items():
        singular_word = plural_to_singular(word)
        combined_dict[singular_word] += count
    return [{'text': word, 'value': count, 'category': category} for word, count in combined_dict.items()]

def load_math_terms():
    math_regexes = []
    math_terms_dict = {}
    for term in MATH_PREFIXES:
        math_terms_dict[rf"\b{term}(s|es|d|ed)?\b"] = term
        math_regexes.append(rf"\b{term}(s|es|d|ed)?\b")

    for term in MATH_WORDS:
        if not term in MATH_PREFIXES:
            math_terms_dict[rf"\b{term}\b"] = term
            math_regexes.append(rf"\b{term}\b")
    return math_regexes, math_terms_dict

def run_math_density(transcript):
    math_regexes, math_terms_dict = load_math_terms()
    sorted_regexes = sorted(math_regexes, key=len, reverse=True)
    teacher_math_word_cloud = {}
    student_math_word_cloud = {}

    for i, utt in enumerate(transcript.utterances):
        text = utt.get_clean_text(remove_punct=True)
        num_matches = 0
        matched_positions = set()
        match_list = set()
        for regex in sorted_regexes:
            matches = list(re.finditer(regex, text, re.IGNORECASE))
            # Filter out matches that share positions with longer terms
            matches = [match for match in matches if not any(match.start() in range(existing[0], existing[1]) for existing in matched_positions)]
            if len(matches) > 0:
                if utt.role == "teacher":
                    if math_terms_dict[regex] not in teacher_math_word_cloud:
                        teacher_math_word_cloud[math_terms_dict[regex]] = 0
                    teacher_math_word_cloud[math_terms_dict[regex]] += len(matches)
                else:
                    if math_terms_dict[regex] not in student_math_word_cloud:
                        student_math_word_cloud[math_terms_dict[regex]] = 0
                    student_math_word_cloud[math_terms_dict[regex]] += len(matches)
            for match in matches:
                match_list.add(match.group())
                matched_positions.add((match.start(), match.end()))
            num_matches += len(matches)
        utt.num_math_terms = num_matches
        utt.math_terms = list(match_list)

    # Initialize lists
    teacher_dict_list = []
    student_dict_list = []
    dict_list = []

    # Process teacher_math_word_cloud
    teacher_dict_list = dict_to_list(teacher_math_word_cloud, 'math')
    dict_list.extend(teacher_dict_list)

    # Process student_math_word_cloud
    student_dict_list = dict_to_list(student_math_word_cloud, 'math')
    dict_list.extend(student_dict_list)

    # Sort the lists
    sorted_dict_list = sorted(dict_list, key=lambda x: x['value'], reverse=True)
    sorted_teacher_dict_list = sorted(teacher_dict_list, key=lambda x: x['value'], reverse=True)
    sorted_student_dict_list = sorted(student_dict_list, key=lambda x: x['value'], reverse=True)

    # Return the sorted lists
    return sorted_dict_list[:50], sorted_teacher_dict_list[:50], sorted_student_dict_list[:50]

class EndpointHandler():
    def __init__(self, path="."):
        print("Loading models...")
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
        self.input_builder = BertInputBuilder(tokenizer=self.tokenizer)

    def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
       data args:
            inputs (:obj: `list`):
            List of dicts, where each dict represents an utterance; each utterance object must have a `speaker`,
            `text` and `uid`and can include list of custom properties
            parameters (:obj: `dict`)
       Return:
            A :obj:`list` | `dict`: will be serialized and returned
        """
        # get inputs
        utterances = data.pop("inputs", data)
        params = data.pop("parameters", None)

        transcript = Transcript(filename=params.pop("filename", None))
        for utt in utterances:
            transcript.add_utterance(Utterance(**utt))

        print("Running inference on %d examples..." % transcript.length())
        logging.set_verbosity_info()
        # Uptake
        uptake_model = UptakeModel(
            self.device, self.tokenizer, self.input_builder)
        uptake_speaker = params.pop("uptake_speaker", None)
        uptake_model.run_inference(transcript, min_prev_words=params['uptake_min_num_words'],
                                   uptake_speaker=uptake_speaker)
        del uptake_model
        
        # Reasoning
        reasoning_model = ReasoningModel(
            self.device, self.tokenizer, self.input_builder)
        reasoning_model.run_inference(transcript, uptake_speaker=uptake_speaker)
        del reasoning_model
        
        # Question
        question_model = QuestionModel(
            self.device, self.tokenizer, self.input_builder)
        question_model.run_inference(transcript)
        del question_model
        
        # Focusing Question
        focusing_question_model = FocusingQuestionModel(
            self.device, self.tokenizer, self.input_builder)
        focusing_question_model.run_inference(transcript, uptake_speaker=uptake_speaker)
        del focusing_question_model

        glossary_path = Path(__file__).resolve().parent / "glossary.csv"
        vocabulary_analyser = VocabularyAnalyser(str(glossary_path))
        vocabulary_analyser.run_analysis(transcript)
        del vocabulary_analyser

        on_task_analyser = OnTaskAnalyser()
        on_task_analyser.run_analysis(transcript, uptake_speaker=uptake_speaker)
        del on_task_analyser

        transcript.update_utterance_roles(uptake_speaker)
        sorted_math_cloud, teacher_math_cloud, student_math_cloud = run_math_density(transcript)
        transcript.calculate_aggregate_word_count()
        return_dict = {'talkDistribution': None, 'talkLength': None, 'talkMoments': None, 'studentTopWords': None, 'teacherTopWords': None}
        talk_dist, talk_len = transcript.get_talk_distribution_and_length(uptake_speaker)
        return_dict['talkDistribution'] = talk_dist
        return_dict['talkLength'] = talk_len
        talk_moments = transcript.get_talk_timeline()
        return_dict['talkMoments'] = talk_moments
        word_cloud, uptake_word_cloud, teacher_general_cloud, student_general_cloud = transcript.get_word_clouds()
        teacher_cloud = teacher_math_cloud + teacher_general_cloud
        student_cloud = student_math_cloud + student_general_cloud
        return_dict['teacherTopWords'] = teacher_cloud
        return_dict['studentTopWords'] = student_cloud
        return_dict['vocabularyTable'] = transcript.get_vocabulary_table()

        return return_dict

	from typing import Dict, List, Any
	from measures.VocabularyAnalyser import VocabularyAnalyser
	from measures.OnTaskAnalyser import OnTaskAnalyser, ON_TASK_LABEL
	from scipy.special import softmax
	from collections import Counter, defaultdict
	import numpy as np
	import weakref
	import re
	import nltk
	from nltk.corpus import stopwords
	nltk.download('stopwords')

	from utils import clean_str, clean_str_nopunct
	import torch
	from utils import MultiHeadModel, BertInputBuilder, get_num_words, MATH_PREFIXES, MATH_WORDS, plural_to_singular

	import transformers
	from transformers import BertTokenizer, BertForSequenceClassification
	from transformers.utils import logging
	from pathlib import Path

	transformers.logging.set_verbosity_debug()

	UPTAKE_MODEL = 'ddemszky/uptake-model'
	REASONING_MODEL = 'ddemszky/student-reasoning'
	QUESTION_MODEL = 'ddemszky/question-detection'
	FOCUSING_QUESTION_MODEL = 'ddemszky/focusing-questions'


	class Utterance:
	def __init__(self, speaker, text, uid=None,
	transcript=None, starttime=None, endtime=None, **kwargs):
	self.speaker = speaker
	self.text = text
	self.uid = uid
	self.starttime = starttime
	self.endtime = endtime
	self.transcript = weakref.ref(transcript) if transcript else None
	self.props = kwargs
	self.role = None
	self.word_count = self.get_num_words()
	self.timestamp = [starttime, endtime]
	if starttime is not None and endtime is not None:
	self.unit_measure = endtime - starttime
	else:
	self.unit_measure = None
	self.aggregate_unit_measure = endtime
	self.num_math_terms = None
	self.math_terms = None
	self.vocabulary_terms = None
	self.on_task = None

	# moments
	self.uptake = None
	self.reasoning = None
	self.question = None
	self.focusing_question = None

	def get_clean_text(self, remove_punct=False):
	if remove_punct:
	return clean_str_nopunct(self.text)
	return clean_str(self.text)

	def get_num_words(self):
	return get_num_words(self.text)

	def to_dict(self):
	return {
	'speaker': self.speaker,
	'text': self.text,
	'uid': self.uid,
	'starttime': self.starttime,
	'endtime': self.endtime,
	'uptake': self.uptake,
	'reasoning': self.reasoning,
	'question': self.question,
	'focusingQuestion': self.focusing_question,
	'numMathTerms': self.num_math_terms,
	'mathTerms': self.math_terms,
	'vocabularyTerms': self.vocabulary_terms,
	'vocabularyMatches': self.vocabulary_matches,
	**self.props
	}

	def to_talk_timeline_dict(self):
	return{
	'speaker': self.speaker,
	'text': self.text,
	'uid': self.uid,
	'role': self.role,
	'timestamp': self.timestamp,
	'moments': {
	'reasoning': True if self.reasoning else False,
	'questioning': True if self.question else False,
	'uptake': True if self.uptake else False,
	'focusingQuestion': True if self.focusing_question else False,
	# Legacy alias for onTask below. On/off-task is not a talk moment;
	# this key is kept only so already-stored analyses and older
	# frontends keep working, and is retired in the UI.
	'mathWord': bool(self.on_task) and self.on_task.get('label') == ON_TASK_LABEL
	},
	# On/off-task classification: a measure in its own right, alongside
	# vocabulary. None means not classified (teacher utterance or empty text).
	'onTask': None if self.on_task is None else self.on_task.get('label') == ON_TASK_LABEL,
	'unitMeasure': self.unit_measure,
	'aggregateUnitMeasure': self.aggregate_unit_measure,
	'wordCount': self.word_count,
	'numMathTerms': self.num_math_terms,
	'mathTerms': self.math_terms,
	'vocabularyTerms': self.vocabulary_terms,
	'vocabularyMatches': self.vocabulary_matches
	}

	def __repr__(self):
	return f"Utterance(speaker='{self.speaker}'," \
	f"text='{self.text}', uid={self.uid}," \
	f"starttime={self.starttime}, endtime={self.endtime}, props={self.props})"

	class Transcript:
	def __init__(self, **kwargs):
	self.utterances = []
	self.params = kwargs

	def add_utterance(self, utterance):
	utterance.transcript = weakref.ref(self)
	self.utterances.append(utterance)

	def get_idx(self, idx):
	if idx >= len(self.utterances):
	return None
	return self.utterances[idx]

	def get_uid(self, uid):
	for utt in self.utterances:
	if utt.uid == uid:
	return utt
	return None

	def length(self):
	return len(self.utterances)

	def update_utterance_roles(self, uptake_speaker):
	for utt in self.utterances:
	if (utt.speaker == uptake_speaker):
	utt.role = 'teacher'
	else:
	utt.role = 'student'

	def get_talk_distribution_and_length(self, uptake_speaker):
	if ((uptake_speaker is None)):
	return None
	teacher_words = 0
	teacher_utt_count = 0
	student_words = 0
	student_utt_count = 0
	for utt in self.utterances:
	if (utt.speaker == uptake_speaker):
	utt.role = 'teacher'
	teacher_words += utt.get_num_words()
	teacher_utt_count += 1
	else:
	utt.role = 'student'
	student_words += utt.get_num_words()
	student_utt_count += 1
	if teacher_words + student_words > 0:
	teacher_percentage = round(
	(teacher_words / (teacher_words + student_words)) * 100)
	student_percentage = 100 - teacher_percentage
	else:
	teacher_percentage = student_percentage = 0
	avg_teacher_length = teacher_words / teacher_utt_count if teacher_utt_count > 0 else 0
	avg_student_length = student_words / student_utt_count if student_utt_count > 0 else 0
	return {'teacher': teacher_percentage, 'student': student_percentage}, {'teacher': avg_teacher_length, 'student': avg_student_length}

	def get_vocabulary_table(self):
	"""
	Build a summary of matched vocabulary terms by utterance role.
	Returns a dict of the form:
	{
	'add': {
	'teacher': {'count': 2, 'utterances': [1,5]},
	'student': {'count': 1, 'utterances': [2]}
	},
	'divide': {
	'teacher': {...},
	'student': {...}
	}
	}
	"""
	vocab_summary = defaultdict(lambda: {"teacher": {"count": 0, "utterances": []},
	"student": {"count": 0, "utterances": []}})

	for utt in self.utterances:
	role = utt.role if utt.role in ("teacher", "student") else "student" # default
	if utt.vocabulary_terms:
	for term in utt.vocabulary_terms:
	vocab_summary[term][role]["count"] += 1
	vocab_summary[term][role]["utterances"].append(utt.uid)

	return vocab_summary


	def get_word_clouds(self):
	# Initialize dictionaries
	teacher_dict = Counter()
	student_dict = Counter()
	uptake_teacher_dict = Counter()
	stop_words = stopwords.words('english')

	# Go through the utterances
	for utt in self.utterances:
	# Get clean text
	clean_text = utt.get_clean_text(remove_punct=True)
	words = clean_text.split()
	words = [word for word in words if word not in stop_words and word not in ['inaudible', 'crosstalk']]

	# Handle uptake case
	if utt.role == 'teacher' and utt.uptake == 1:
	uptake_teacher_dict.update(words)

	general_text = ' '.join(words)
	# Replace math terms with empty strings
	for math_term in utt.math_terms:
	general_text = general_text.replace(math_term, '')
	general_text = general_text.replace(' ', ' ')

	general_words = general_text.split()
	# Update the appropriate dictionary
	if utt.role == 'teacher':
	teacher_dict.update(general_words)
	else:
	student_dict.update(general_words)

	# Sorting and trimming dictionaries
	dict_list = dict_to_list(teacher_dict, 'general') + dict_to_list(student_dict, 'general')
	uptake_dict_list = dict_to_list(uptake_teacher_dict, 'teacher')
	teacher_dict_list = dict_to_list(teacher_dict, 'general')
	student_dict_list = dict_to_list(student_dict, 'general')

	sorted_dict_list = sorted(dict_list, key=lambda x: x['value'], reverse=True)
	sorted_uptake_dict_list = sorted(uptake_dict_list, key=lambda x: x['value'], reverse=True)
	sorted_teacher_dict_list = sorted(teacher_dict_list, key=lambda x: x['value'], reverse=True)
	sorted_student_dict_list = sorted(student_dict_list, key=lambda x: x['value'], reverse=True)
	return sorted_dict_list[:50], sorted_uptake_dict_list[:50], sorted_teacher_dict_list[:50], sorted_student_dict_list[:50]

	def get_talk_timeline(self):
	return [utterance.to_talk_timeline_dict() for utterance in self.utterances]

	def calculate_aggregate_word_count(self):
	unit_measures = [utt.unit_measure for utt in self.utterances]
	if None in unit_measures:
	aggregate_word_count = 0
	for utt in self.utterances:
	aggregate_word_count += utt.get_num_words()
	utt.unit_measure = utt.get_num_words()
	utt.aggregate_unit_measure = aggregate_word_count


	def to_dict(self):
	return {
	'utterances': [utterance.to_dict() for utterance in self.utterances],
	**self.params
	}

	def __repr__(self):
	return f"Transcript(utterances={self.utterances}, custom_params={self.params})"

	class QuestionModel:
	def __init__(self, device, tokenizer, input_builder, max_length=300, path=QUESTION_MODEL):
	print("Loading models...")
	self.device = device
	self.tokenizer = tokenizer
	self.input_builder = input_builder
	self.max_length = max_length
	self.model = MultiHeadModel.from_pretrained(
	path, head2size={"is_question": 2})
	self.model.to(self.device)

	def run_inference(self, transcript):
	self.model.eval()
	with torch.no_grad():
	for i, utt in enumerate(transcript.utterances):
	if "?" in utt.text:
	utt.question = 1
	else:
	text = utt.get_clean_text(remove_punct=True)
	instance = self.input_builder.build_inputs([], text,
	max_length=self.max_length,
	input_str=True)
	output = self.get_prediction(instance)
	# print(output)
	utt.question = np.argmax(
	output["is_question_logits"][0].tolist())

	def get_prediction(self, instance):
	instance["attention_mask"] = [[1] * len(instance["input_ids"])]
	for key in ["input_ids", "token_type_ids", "attention_mask"]:
	instance[key] = torch.tensor(
	instance[key]).unsqueeze(0) # Batch size = 1
	instance[key].to(self.device)

	output = self.model(input_ids=instance["input_ids"],
	attention_mask=instance["attention_mask"],
	token_type_ids=instance["token_type_ids"],
	return_pooler_output=False)
	return output

	class ReasoningModel:
	def __init__(self, device, tokenizer, input_builder, max_length=128, path=REASONING_MODEL):
	print("Loading models...")
	self.device = device
	self.tokenizer = tokenizer
	self.input_builder = input_builder
	self.max_length = max_length
	self.model = BertForSequenceClassification.from_pretrained(path)
	self.model.to(self.device)

	def run_inference(self, transcript, min_num_words=8, uptake_speaker=None):
	self.model.eval()
	with torch.no_grad():
	for i, utt in enumerate(transcript.utterances):
	if utt.get_num_words() >= min_num_words and utt.speaker != uptake_speaker:
	instance = self.input_builder.build_inputs([], utt.text,
	max_length=self.max_length,
	input_str=True)
	output = self.get_prediction(instance)
	utt.reasoning = np.argmax(output["logits"][0].tolist())

	def get_prediction(self, instance):
	instance["attention_mask"] = [[1] * len(instance["input_ids"])]
	for key in ["input_ids", "token_type_ids", "attention_mask"]:
	instance[key] = torch.tensor(
	instance[key]).unsqueeze(0) # Batch size = 1
	instance[key].to(self.device)

	output = self.model(input_ids=instance["input_ids"],
	attention_mask=instance["attention_mask"],
	token_type_ids=instance["token_type_ids"])
	return output

	class UptakeModel:
	def __init__(self, device, tokenizer, input_builder, max_length=120, path=UPTAKE_MODEL):
	print("Loading models...")
	self.device = device
	self.tokenizer = tokenizer
	self.input_builder = input_builder
	self.max_length = max_length
	self.model = MultiHeadModel.from_pretrained(path, head2size={"nsp": 2})
	self.model.to(self.device)

	def run_inference(self, transcript, min_prev_words, uptake_speaker=None):
	self.model.eval()
	prev_num_words = 0
	prev_utt = None
	with torch.no_grad():
	for i, utt in enumerate(transcript.utterances):
	if ((uptake_speaker is None) or (utt.speaker == uptake_speaker)) and (prev_num_words >= min_prev_words):
	textA = prev_utt.get_clean_text(remove_punct=False)
	textB = utt.get_clean_text(remove_punct=False)
	instance = self.input_builder.build_inputs([textA], textB,
	max_length=self.max_length,
	input_str=True)
	output = self.get_prediction(instance)

	utt.uptake = int(
	softmax(output["nsp_logits"][0].tolist())[1] > .8)
	prev_num_words = utt.get_num_words()
	prev_utt = utt

	def get_prediction(self, instance):
	instance["attention_mask"] = [[1] * len(instance["input_ids"])]
	for key in ["input_ids", "token_type_ids", "attention_mask"]:
	instance[key] = torch.tensor(
	instance[key]).unsqueeze(0) # Batch size = 1
	instance[key].to(self.device)

	output = self.model(input_ids=instance["input_ids"],
	attention_mask=instance["attention_mask"],
	token_type_ids=instance["token_type_ids"],
	return_pooler_output=False)
	return output

	class FocusingQuestionModel:
	def __init__(self, device, tokenizer, input_builder, max_length=128, path=FOCUSING_QUESTION_MODEL):
	print("Loading models...")
	self.device = device
	self.tokenizer = tokenizer
	self.input_builder = input_builder
	self.model = BertForSequenceClassification.from_pretrained(path)
	self.model.to(self.device)
	self.max_length = max_length

	def run_inference(self, transcript, min_focusing_words=0, uptake_speaker=None):
	self.model.eval()
	with torch.no_grad():
	for i, utt in enumerate(transcript.utterances):
	if utt.speaker != uptake_speaker or uptake_speaker is None:
	utt.focusing_question = None
	continue
	if utt.get_num_words() < min_focusing_words:
	utt.focusing_question = None
	continue
	instance = self.input_builder.build_inputs([], utt.text, max_length=self.max_length, input_str=True)
	output = self.get_prediction(instance)
	utt.focusing_question = np.argmax(output["logits"][0].tolist())

	def get_prediction(self, instance):
	instance["attention_mask"] = [[1] * len(instance["input_ids"])]
	for key in ["input_ids", "token_type_ids", "attention_mask"]:
	instance[key] = torch.tensor(
	instance[key]).unsqueeze(0) # Batch size = 1
	instance[key].to(self.device)

	output = self.model(input_ids=instance["input_ids"],
	attention_mask=instance["attention_mask"],
	token_type_ids=instance["token_type_ids"])
	return output

	def dict_to_list(d, category):
	combined_dict = Counter()
	for word, count in d.items():
	singular_word = plural_to_singular(word)
	combined_dict[singular_word] += count
	return [{'text': word, 'value': count, 'category': category} for word, count in combined_dict.items()]

	def load_math_terms():
	math_regexes = []
	math_terms_dict = {}
	for term in MATH_PREFIXES:
	math_terms_dict[rf"\b{term}(s\|es\|d\|ed)?\b"] = term
	math_regexes.append(rf"\b{term}(s\|es\|d\|ed)?\b")

	for term in MATH_WORDS:
	if not term in MATH_PREFIXES:
	math_terms_dict[rf"\b{term}\b"] = term
	math_regexes.append(rf"\b{term}\b")
	return math_regexes, math_terms_dict

	def run_math_density(transcript):
	math_regexes, math_terms_dict = load_math_terms()
	sorted_regexes = sorted(math_regexes, key=len, reverse=True)
	teacher_math_word_cloud = {}
	student_math_word_cloud = {}

	for i, utt in enumerate(transcript.utterances):
	text = utt.get_clean_text(remove_punct=True)
	num_matches = 0
	matched_positions = set()
	match_list = set()
	for regex in sorted_regexes:
	matches = list(re.finditer(regex, text, re.IGNORECASE))
	# Filter out matches that share positions with longer terms
	matches = [match for match in matches if not any(match.start() in range(existing[0], existing[1]) for existing in matched_positions)]
	if len(matches) > 0:
	if utt.role == "teacher":
	if math_terms_dict[regex] not in teacher_math_word_cloud:
	teacher_math_word_cloud[math_terms_dict[regex]] = 0
	teacher_math_word_cloud[math_terms_dict[regex]] += len(matches)
	else:
	if math_terms_dict[regex] not in student_math_word_cloud:
	student_math_word_cloud[math_terms_dict[regex]] = 0
	student_math_word_cloud[math_terms_dict[regex]] += len(matches)
	for match in matches:
	match_list.add(match.group())
	matched_positions.add((match.start(), match.end()))
	num_matches += len(matches)
	utt.num_math_terms = num_matches
	utt.math_terms = list(match_list)

	# Initialize lists
	teacher_dict_list = []
	student_dict_list = []
	dict_list = []

	# Process teacher_math_word_cloud
	teacher_dict_list = dict_to_list(teacher_math_word_cloud, 'math')
	dict_list.extend(teacher_dict_list)

	# Process student_math_word_cloud
	student_dict_list = dict_to_list(student_math_word_cloud, 'math')
	dict_list.extend(student_dict_list)

	# Sort the lists
	sorted_dict_list = sorted(dict_list, key=lambda x: x['value'], reverse=True)
	sorted_teacher_dict_list = sorted(teacher_dict_list, key=lambda x: x['value'], reverse=True)
	sorted_student_dict_list = sorted(student_dict_list, key=lambda x: x['value'], reverse=True)

	# Return the sorted lists
	return sorted_dict_list[:50], sorted_teacher_dict_list[:50], sorted_student_dict_list[:50]

	class EndpointHandler():
	def __init__(self, path="."):
	print("Loading models...")
	self.device = "cuda" if torch.cuda.is_available() else "cpu"
	self.tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
	self.input_builder = BertInputBuilder(tokenizer=self.tokenizer)

	def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
	"""
	data args:
	inputs (:obj: `list`):
	List of dicts, where each dict represents an utterance; each utterance object must have a `speaker`,
	`text` and `uid`and can include list of custom properties
	parameters (:obj: `dict`)
	Return:
	A :obj:`list` \| `dict`: will be serialized and returned
	"""
	# get inputs
	utterances = data.pop("inputs", data)
	params = data.pop("parameters", None)

	transcript = Transcript(filename=params.pop("filename", None))
	for utt in utterances:
	transcript.add_utterance(Utterance(**utt))

	print("Running inference on %d examples..." % transcript.length())
	logging.set_verbosity_info()
	# Uptake
	uptake_model = UptakeModel(
	self.device, self.tokenizer, self.input_builder)
	uptake_speaker = params.pop("uptake_speaker", None)
	uptake_model.run_inference(transcript, min_prev_words=params['uptake_min_num_words'],
	uptake_speaker=uptake_speaker)
	del uptake_model

	# Reasoning
	reasoning_model = ReasoningModel(
	self.device, self.tokenizer, self.input_builder)
	reasoning_model.run_inference(transcript, uptake_speaker=uptake_speaker)
	del reasoning_model

	# Question
	question_model = QuestionModel(
	self.device, self.tokenizer, self.input_builder)
	question_model.run_inference(transcript)
	del question_model

	# Focusing Question
	focusing_question_model = FocusingQuestionModel(
	self.device, self.tokenizer, self.input_builder)
	focusing_question_model.run_inference(transcript, uptake_speaker=uptake_speaker)
	del focusing_question_model

	glossary_path = Path(__file__).resolve().parent / "glossary.csv"
	vocabulary_analyser = VocabularyAnalyser(str(glossary_path))
	vocabulary_analyser.run_analysis(transcript)
	del vocabulary_analyser

	on_task_analyser = OnTaskAnalyser()
	on_task_analyser.run_analysis(transcript, uptake_speaker=uptake_speaker)
	del on_task_analyser

	transcript.update_utterance_roles(uptake_speaker)
	sorted_math_cloud, teacher_math_cloud, student_math_cloud = run_math_density(transcript)
	transcript.calculate_aggregate_word_count()
	return_dict = {'talkDistribution': None, 'talkLength': None, 'talkMoments': None, 'studentTopWords': None, 'teacherTopWords': None}
	talk_dist, talk_len = transcript.get_talk_distribution_and_length(uptake_speaker)
	return_dict['talkDistribution'] = talk_dist
	return_dict['talkLength'] = talk_len
	talk_moments = transcript.get_talk_timeline()
	return_dict['talkMoments'] = talk_moments
	word_cloud, uptake_word_cloud, teacher_general_cloud, student_general_cloud = transcript.get_word_clouds()
	teacher_cloud = teacher_math_cloud + teacher_general_cloud
	student_cloud = student_math_cloud + student_general_cloud
	return_dict['teacherTopWords'] = teacher_cloud
	return_dict['studentTopWords'] = student_cloud
	return_dict['vocabularyTable'] = transcript.get_vocabulary_table()

	return return_dict