wuenlp_tools.models.suspense.listbased

  1import os
  2import shutil
  3from concurrent.futures import ProcessPoolExecutor
  4from pathlib import Path
  5from typing import Literal, Type, List, Optional
  6
  7from loguru import logger
  8
  9from wuenlp.impl.UIMANLPStructs import UIMAAnnotation, UIMAParagraph, UIMADocument, UIMASystemScene
 10from wuenlp.impl.uima import UIMASpan
 11
 12from wuenlp_tools.pipeline import PipelineStep, PipelineProcessor
 13from .extended_wordlists import all_lists_conceptnet, all_lists_julian, all_lists_w2v
 14from .wordlists import all_lists_base
 15
 16from .. import AnnoType
 17
 18
 19class WordList(object):
 20    def __init__(self, wordlist: List, case_sensitive: bool = False):
 21        self.wordlist = wordlist if case_sensitive else [word.lower() for word in wordlist]
 22        self.case_sensitive = case_sensitive
 23
 24    def __contains__(self, item):
 25        if not self.case_sensitive and item is not None:
 26            item = item.lower()
 27        return item in self.wordlist
 28
 29
 30def process_doc(doc: UIMADocument, unit_type: Optional[Type[UIMASpan]] = UIMAParagraph, extend: Optional[str] = None,
 31                case_sensitive: bool = False, anno_type: AnnoType = AnnoType.DANGER):
 32    assert unit_type is not None, "unit_type must be specified"
 33    assert extend in [None, "julian", "conceptnet",
 34                      "word2vec"], "extend must be one of [None, 'julian', 'conceptnet', 'word2vec']"
 35
 36    if extend == "conceptnet" and case_sensitive:
 37        logger.warning(
 38            "Case sensitive conceptnet wordlists are not available. Using case insensitive wordlists instead.")
 39        case_sensitive = False
 40
 41    word_list_map = {
 42        "julian": {t: WordList(wordlist, case_sensitive=case_sensitive) for t, wordlist in all_lists_julian.items()},
 43        "conceptnet": {t: WordList(wordlist, case_sensitive=False) for t, wordlist in
 44                       all_lists_conceptnet.items()},
 45        "word2vec": {t: WordList(wordlist, case_sensitive=case_sensitive) for t, wordlist in all_lists_w2v.items()},
 46        None: {t: WordList(wordlist, case_sensitive=case_sensitive) for t, wordlist in all_lists_base.items()}
 47    }
 48
 49    all_lists = word_list_map[extend]
 50
 51    logger.info(f"Processing {doc.path}")
 52
 53    annos = doc._get_annos_of_type(unit_type)
 54
 55    if not annos:
 56        logger.warning(f"No {unit_type.__name__} annotations found in {doc.path}")
 57        return
 58
 59    for anno in annos:
 60        lemmas = [tok.lemma for tok in anno.tokens]
 61
 62        if anno_type == AnnoType.DANGER:
 63            f_type = "overall danger"
 64        elif anno_type == AnnoType.FEAR:
 65            f_type = "fear"
 66        else:
 67            raise ValueError(f"Unknown annotation type: {anno_type}")
 68        matches = [word for word in lemmas if word in all_lists[f_type]]
 69        score = len(matches) / len(lemmas)
 70        score_signals = matches
 71
 72        feature_name = "danger" if anno_type == AnnoType.DANGER else "fear"
 73
 74        anno.additional_features[f"{feature_name}_score"] = score
 75        anno.additional_features[f"{feature_name}_score_signals"] = score_signals
 76
 77    average_score = sum(anno.additional_features[f"{feature_name}_score"] for anno in annos) / (len(
 78        annos) or 1)
 79
 80    for anno in annos:
 81        condition = check_condition(anno, average_score, feature_name)
 82        anno.additional_features[
 83            "has_dangerous_situation" if anno_type == AnnoType.DANGER else "has_fear_description"] = int(condition)
 84
 85    return doc
 86
 87
 88def check_condition(anno, average_score, feature_name):
 89    condition = anno.additional_features[f"{feature_name}_score"] > average_score
 90    return condition
 91
 92
 93class WordlistDangerProcessor(PipelineProcessor):
 94    def __call__(self, doc: UIMADocument, unit_type: Optional[Type[UIMASpan]], overwrite: Optional[bool] = False,
 95                 **kwargs):
 96        return process_doc(doc, unit_type=unit_type, extend="julian", case_sensitive=False, anno_type=AnnoType.DANGER)
 97
 98
 99class WordlistFearProcessor(PipelineProcessor):
100    def __call__(self, doc: UIMADocument, unit_type: Optional[Type[UIMASpan]], overwrite: Optional[bool] = False,
101                 **kwargs) -> UIMADocument:
102        return process_doc(doc, unit_type=unit_type, extend="julian", case_sensitive=False, anno_type=AnnoType.FEAR)
103
104
105WordlistDangerAnnotator = PipelineStep("Wordlist Danger Annotator", WordlistDangerProcessor(),
106                                       unit_type=UIMASystemScene)
107WordlistFearAnnotator = PipelineStep("Wordlist Fear Annotator", WordlistFearProcessor(),
108                                     unit_type=UIMASystemScene)
109
110if __name__ == '__main__':
111    from wuenlp_tools.pipeline import Pipeline
112    from wuenlp_tools.models.preprocess import Preprocessor
113    from wuenlp_tools.models.scenes.segmentation import BERTSceneSegmenter
114
115    pipeline = Pipeline(
116        steps=[Preprocessor, BERTSceneSegmenter, WordlistDangerAnnotator, WordlistFearAnnotator],
117    )
118
119    doc = pipeline("Das ist ein Test")