miwytt commited on
Commit
1a0cfff
1 Parent(s): 51a4a09

Merge files

Browse files
Files changed (1) hide show
  1. app.py +166 -2
app.py CHANGED
@@ -1,8 +1,172 @@
1
- import src.translation_direction_detection.detector as tdd
2
  from nmtscore import NMTScorer
3
  import gradio as gr
4
 
5
- detector = tdd.TranslationDirectionDetector(NMTScorer("m2m100_418M"))
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6
 
7
  def translate_direction(text1, lang1, text2, lang2):
8
  lang_to_code = {"English": 'en',
 
 
1
  from nmtscore import NMTScorer
2
  import gradio as gr
3
 
4
+ from dataclasses import dataclass
5
+ from typing import List, Union, Optional
6
+
7
+ from scipy.special import softmax
8
+ from scipy.stats import permutation_test
9
+
10
+
11
+ @dataclass
12
+ class TranslationDirectionResult:
13
+ sentence1: Union[str, List[str]]
14
+ sentence2: Union[str, List[str]]
15
+ lang1: str
16
+ lang2: str
17
+ raw_prob_1_to_2: float
18
+ raw_prob_2_to_1: float
19
+ pvalue: Optional[float] = None
20
+
21
+ @property
22
+ def num_sentences(self):
23
+ return len(self.sentence1) if isinstance(self.sentence1, list) else 1
24
+
25
+ @property
26
+ def prob_1_to_2(self):
27
+ return softmax([self.raw_prob_1_to_2, self.raw_prob_2_to_1])[0]
28
+
29
+ @property
30
+ def prob_2_to_1(self):
31
+ return softmax([self.raw_prob_1_to_2, self.raw_prob_2_to_1])[1]
32
+
33
+ @property
34
+ def predicted_direction(self) -> str:
35
+ if self.raw_prob_1_to_2 >= self.raw_prob_2_to_1:
36
+ return self.lang1 + '→' + self.lang2
37
+ else:
38
+ return self.lang2 + '→' + self.lang1
39
+
40
+ def __str__(self):
41
+ s = f"""\
42
+ Predicted direction: {self.predicted_direction}
43
+ {self.num_sentences} sentence pair{"s" if self.num_sentences > 1 else ""}
44
+ {self.lang1}→{self.lang2}: {self.prob_1_to_2:.3f}
45
+ {self.lang2}→{self.lang1}: {self.prob_2_to_1:.3f}"""
46
+ if self.pvalue is not None:
47
+ s += f"\np-value: {self.pvalue}\n"
48
+ return s
49
+
50
+
51
+ class TranslationDirectionDetector:
52
+
53
+ def __init__(self, scorer: NMTScorer = None, use_normalization: bool = False):
54
+ self.scorer = scorer or NMTScorer()
55
+ self.use_normalization = use_normalization
56
+
57
+ def detect(self,
58
+ sentence1: Union[str, List[str]],
59
+ sentence2: Union[str, List[str]],
60
+ lang1: str,
61
+ lang2: str,
62
+ return_pvalue: bool = False,
63
+ pvalue_n_resamples: int = 9999,
64
+ score_kwargs: dict = None
65
+ ) -> TranslationDirectionResult:
66
+ if isinstance(sentence1, list) and isinstance(sentence2, list):
67
+ if len(sentence1) != len(sentence2):
68
+ raise ValueError("Lists sentence1 and sentence2 must have same length")
69
+ if len(sentence1) == 0:
70
+ raise ValueError("Lists sentence1 and sentence2 must not be empty")
71
+ if len(sentence1) == 1 and return_pvalue:
72
+ raise ValueError("return_pvalue=True requires the documents to have multiple sentences")
73
+ if lang1 == lang2:
74
+ raise ValueError("lang1 and lang2 must be different")
75
+
76
+ prob_1_to_2 = self.scorer.score_direct(
77
+ sentence2, sentence1,
78
+ lang2, lang1,
79
+ normalize=self.use_normalization,
80
+ both_directions=False,
81
+ score_kwargs=score_kwargs
82
+ )
83
+ prob_2_to_1 = self.scorer.score_direct(
84
+ sentence1, sentence2,
85
+ lang1, lang2,
86
+ normalize=self.use_normalization,
87
+ both_directions=False,
88
+ score_kwargs=score_kwargs
89
+ )
90
+ pvalue = None
91
+
92
+ if isinstance(sentence1, list): # document-level
93
+ # Compute the average probability per target token, across the complete document
94
+ # 1. Convert probabilities back to log probabilities
95
+ log_prob_1_to_2 = np.log2(np.array(prob_1_to_2))
96
+ log_prob_2_to_1 = np.log2(np.array(prob_2_to_1))
97
+ # 2. Reverse the sentence-level length normalization
98
+ sentence1_lengths = np.array([self._get_sentence_length(s) for s in sentence1])
99
+ sentence2_lengths = np.array([self._get_sentence_length(s) for s in sentence2])
100
+ log_prob_1_to_2 = sentence2_lengths * log_prob_1_to_2
101
+ log_prob_2_to_1 = sentence1_lengths * log_prob_2_to_1
102
+ # 4. Sum up the log probabilities across the document
103
+ total_log_prob_1_to_2 = log_prob_1_to_2.sum()
104
+ total_log_prob_2_to_1 = log_prob_2_to_1.sum()
105
+ # 3. Document-level length normalization
106
+ avg_log_prob_1_to_2 = total_log_prob_1_to_2 / sum(sentence2_lengths)
107
+ avg_log_prob_2_to_1 = total_log_prob_2_to_1 / sum(sentence1_lengths)
108
+ # 4. Convert back to probabilities
109
+ prob_1_to_2 = 2 ** avg_log_prob_1_to_2
110
+ prob_2_to_1 = 2 ** avg_log_prob_2_to_1
111
+
112
+ if return_pvalue:
113
+ x = np.vstack([log_prob_1_to_2, sentence2_lengths]).T
114
+ y = np.vstack([log_prob_2_to_1, sentence1_lengths]).T
115
+ result = permutation_test(
116
+ data=(x, y),
117
+ statistic=self._statistic_token_mean,
118
+ permutation_type="samples",
119
+ n_resamples=pvalue_n_resamples,
120
+ )
121
+ pvalue = result.pvalue
122
+ else:
123
+ if return_pvalue:
124
+ raise ValueError("return_pvalue=True requires sentence1 and sentence2 to be lists of sentences")
125
+
126
+ return TranslationDirectionResult(
127
+ sentence1=sentence1,
128
+ sentence2=sentence2,
129
+ lang1=lang1,
130
+ lang2=lang2,
131
+ raw_prob_1_to_2=prob_1_to_2,
132
+ raw_prob_2_to_1=prob_2_to_1,
133
+ pvalue=pvalue,
134
+ )
135
+
136
+ def _get_sentence_length(self, sentence: str) -> int:
137
+ tokens = self.scorer.model.tokenizer.tokenize(sentence)
138
+ return len(tokens)
139
+
140
+ @staticmethod
141
+ def _statistic_token_mean(x: np.ndarray, y: np.ndarray, axis: int = -1) -> float:
142
+ """
143
+ Statistic for scipy.stats.permutation_test
144
+
145
+ :param x: Matrix of shape (2 x num_sentences). The first row contains the unnormalized log probability
146
+ for lang1→lang2, the second row contains the sentence lengths in lang2.
147
+ :param y: Same as x, but for lang2→lang1
148
+ :return: Difference between lang1→lang2 and lang2→lang1
149
+ """
150
+ if axis != -1:
151
+ raise NotImplementedError("Only axis=-1 is supported")
152
+ # Add batch dim
153
+ if x.ndim == 2:
154
+ x = x[np.newaxis, ...]
155
+ y = y[np.newaxis, ...]
156
+ # Sum up the log probabilities across the document
157
+ total_log_prob_1_to_2 = x[:, 0].sum(axis=axis)
158
+ total_log_prob_2_to_1 = y[:, 0].sum(axis=axis)
159
+ # Document-level length normalization
160
+ avg_log_prob_1_to_2 = total_log_prob_1_to_2 / x[:, 1].sum(axis=axis)
161
+ avg_log_prob_2_to_1 = total_log_prob_2_to_1 / y[:, 1].sum(axis=axis)
162
+ # Convert to probabilities
163
+ prob_1_to_2 = 2 ** avg_log_prob_1_to_2
164
+ prob_2_to_1 = 2 ** avg_log_prob_2_to_1
165
+ # Compute difference
166
+ return prob_1_to_2 - prob_2_to_1
167
+
168
+
169
+ detector = TranslationDirectionDetector(NMTScorer("m2m100_418M"))
170
 
171
  def translate_direction(text1, lang1, text2, lang2):
172
  lang_to_code = {"English": 'en',