|
|
from dataclasses import dataclass
|
|
|
from typing import Dict, List, Sequence
|
|
|
import stanza
|
|
|
import transformers
|
|
|
import json
|
|
|
import streamlit as st
|
|
|
from .web_utilities import st_cache_data_if, st_cache_resource_if, supported_cache
|
|
|
from .anonymize import add_space_to_comma_endpoint, change_name_patient_abbreviations
|
|
|
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
|
class SentenceBoundary:
|
|
|
text: str
|
|
|
prefix: str
|
|
|
|
|
|
def __str__(self):
|
|
|
return self.prefix + self.text
|
|
|
|
|
|
|
|
|
@dataclass
|
|
|
class SentenceBoundaries:
|
|
|
def __init__(self) -> None:
|
|
|
self._sentence_boundaries: List[SentenceBoundary] = []
|
|
|
|
|
|
@property
|
|
|
def sentence_boundaries(self):
|
|
|
return self._sentence_boundaries
|
|
|
|
|
|
def update_sentence_boundaries(
|
|
|
self, sentence_boundaries_list: List[SentenceBoundary]
|
|
|
):
|
|
|
self._sentence_boundaries = sentence_boundaries_list
|
|
|
return self
|
|
|
|
|
|
def from_doc(self, doc: stanza.Document):
|
|
|
start_idx = 0
|
|
|
for sent in doc.sentences:
|
|
|
self.sentence_boundaries.append(
|
|
|
SentenceBoundary(
|
|
|
text=sent.text,
|
|
|
prefix=doc.text[start_idx : sent.tokens[0].start_char],
|
|
|
)
|
|
|
)
|
|
|
start_idx = sent.tokens[-1].end_char
|
|
|
self.sentence_boundaries.append(
|
|
|
SentenceBoundary(text="", prefix=doc.text[start_idx:])
|
|
|
)
|
|
|
return self
|
|
|
|
|
|
@property
|
|
|
def nonempty_sentences(self) -> List[str]:
|
|
|
return [item.text for item in self.sentence_boundaries if item.text]
|
|
|
|
|
|
def map_sentence_boundaries(self, d: Dict[str, str]) -> List:
|
|
|
return SentenceBoundaries().update_sentence_boundaries(
|
|
|
[
|
|
|
SentenceBoundary(text=d.get(sb.text, sb.text), prefix=sb.prefix)
|
|
|
for sb in self.sentence_boundaries
|
|
|
]
|
|
|
)
|
|
|
|
|
|
def __str__(self) -> str:
|
|
|
return "".join(map(str, self.sentence_boundaries))
|
|
|
|
|
|
|
|
|
@st_cache_resource_if(supported_cache, max_entries=5, ttl=3600)
|
|
|
def minibatch(seq, size):
|
|
|
items = []
|
|
|
for x in seq:
|
|
|
items.append(x)
|
|
|
if len(items) >= size:
|
|
|
yield items
|
|
|
items = []
|
|
|
if items:
|
|
|
yield items
|
|
|
|
|
|
|
|
|
|
|
|
class Translator:
|
|
|
def __init__(self, source_lang: str, dest_lang: str, use_gpu: bool = False) -> None:
|
|
|
|
|
|
self.model_name = "Helsinki-NLP/opus-mt-" + source_lang + "-" + dest_lang
|
|
|
self.model = transformers.MarianMTModel.from_pretrained(self.model_name)
|
|
|
|
|
|
|
|
|
self.tokenizer = transformers.MarianTokenizer.from_pretrained(self.model_name)
|
|
|
self.sentencizer = stanza.Pipeline(
|
|
|
source_lang, processors="tokenize", verbose=False, use_gpu=use_gpu
|
|
|
)
|
|
|
|
|
|
def sentencize(self, texts: Sequence[str]) -> List[SentenceBoundaries]:
|
|
|
return [
|
|
|
SentenceBoundaries().from_doc(doc=self.sentencizer.process(text))
|
|
|
for text in texts
|
|
|
]
|
|
|
|
|
|
def translate(
|
|
|
self, texts: Sequence[str], batch_size: int = 10, truncation=True
|
|
|
) -> Sequence[str]:
|
|
|
if isinstance(texts, str):
|
|
|
raise ValueError("Expected a sequence of texts")
|
|
|
text_sentences = self.sentencize(texts)
|
|
|
translations = {
|
|
|
sent: None for text in text_sentences for sent in text.nonempty_sentences
|
|
|
}
|
|
|
|
|
|
for text_batch in minibatch(
|
|
|
sorted(translations, key=len, reverse=True), batch_size
|
|
|
):
|
|
|
tokens = self.tokenizer(
|
|
|
text_batch, return_tensors="pt", padding=True, truncation=truncation
|
|
|
)
|
|
|
|
|
|
|
|
|
translate_tokens = self.model.generate(**tokens)
|
|
|
translate_batch = [
|
|
|
self.tokenizer.decode(t, skip_special_tokens=True)
|
|
|
for t in translate_tokens
|
|
|
]
|
|
|
for text, translated in zip(text_batch, translate_batch):
|
|
|
translations[text] = translated
|
|
|
|
|
|
return [
|
|
|
str(text.map_sentence_boundaries(translations)) for text in text_sentences
|
|
|
]
|
|
|
|
|
|
|
|
|
|
|
|
@st_cache_data_if(supported_cache, max_entries=5, ttl=3600)
|
|
|
def translate_report(
|
|
|
Report, Last_name, First_name, _nlp, _marian_fr_en, dict_correction, abbreviation_dict
|
|
|
):
|
|
|
Report_name, list_replaced_abb_name = change_name_patient_abbreviations(
|
|
|
Report, Last_name, First_name, abbreviation_dict
|
|
|
)
|
|
|
MarianText_raw = translate_marian(Report_name, _nlp, _marian_fr_en)
|
|
|
MarianText_space = add_space_to_comma_endpoint(MarianText_raw, _nlp)
|
|
|
MarianText, list_replaced = correct_marian(
|
|
|
MarianText_space, dict_correction, Last_name, First_name
|
|
|
)
|
|
|
del MarianText_raw
|
|
|
del MarianText_space
|
|
|
return MarianText, list_replaced, list_replaced_abb_name
|
|
|
|
|
|
|
|
|
|
|
|
@st_cache_resource_if(supported_cache, max_entries=5, ttl=3600)
|
|
|
def translate_marian(Report_name, _nlp, _marian_fr_en):
|
|
|
list_of_sentence = []
|
|
|
for sentence in _nlp.process(Report_name).sentences:
|
|
|
list_of_sentence.append(sentence.text)
|
|
|
MarianText_raw = "\n".join(_marian_fr_en.translate(list_of_sentence))
|
|
|
del list_of_sentence
|
|
|
return MarianText_raw
|
|
|
|
|
|
|
|
|
@st_cache_data_if(supported_cache, max_entries=5, ttl=3600)
|
|
|
def correct_marian(MarianText_space, dict_correction, Last_name, First_name):
|
|
|
MarianText = MarianText_space
|
|
|
list_replaced = []
|
|
|
for key, value in dict_correction.items():
|
|
|
if key in MarianText:
|
|
|
list_replaced.append(
|
|
|
{
|
|
|
"name": Last_name,
|
|
|
"surname": First_name,
|
|
|
"type": "marian_correction",
|
|
|
"value": key,
|
|
|
"correction": value,
|
|
|
"lf_detected": True,
|
|
|
"manual_validation": True,
|
|
|
}
|
|
|
)
|
|
|
MarianText = MarianText.replace(key, value)
|
|
|
return MarianText, list_replaced
|
|
|
|
|
|
|
|
|
|
|
|
@st_cache_resource_if(supported_cache, max_entries=5, ttl=3600)
|
|
|
def get_translation_dict_correction():
|
|
|
dict_correction_FRspec = {
|
|
|
"PC": "head circumference",
|
|
|
"palatine slot": "cleft palate",
|
|
|
"ASD": "autism",
|
|
|
"ADHD": "attention deficit hyperactivity disorder",
|
|
|
"IUGR": "intrauterin growth retardation",
|
|
|
"QI": "IQ ",
|
|
|
"QIT": "FSIQ ",
|
|
|
"ITQ": "FSIQ ",
|
|
|
"DS": "SD",
|
|
|
"FOP": "patent foramen ovale",
|
|
|
"PFO": "patent foramen ovale",
|
|
|
"ARCF": "fetal distress",
|
|
|
"\n": " ",
|
|
|
"associated": "with",
|
|
|
"Mr.": "Mr",
|
|
|
"Mrs.": "Mrs",
|
|
|
}
|
|
|
|
|
|
dict_correction = {}
|
|
|
for key, value in dict_correction_FRspec.items():
|
|
|
dict_correction[" " + key + " "] = " " + value + " "
|
|
|
|
|
|
with open("data/hp_fr_en_translated_marian_review_lwg.json", "r") as outfile:
|
|
|
hpo_translated = json.load(outfile)
|
|
|
|
|
|
for key, value in hpo_translated.items():
|
|
|
dict_correction[" " + key + " "] = " " + value + " "
|
|
|
|
|
|
with open("data/fr_abbreviations_translation.json", "r") as outfile:
|
|
|
hpo_translated_abbreviations = json.load(outfile)
|
|
|
|
|
|
for key, value in hpo_translated_abbreviations.items():
|
|
|
dict_correction[" " + key + " "] = " " + value + " "
|
|
|
|
|
|
del hpo_translated
|
|
|
del hpo_translated_abbreviations
|
|
|
return dict_correction
|
|
|
|
|
|
|
|
|
|
