File: lm.py

package info (click to toggle)
python-pynlpl 1.2.9-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,900 kB
  • sloc: python: 25,677; sh: 73; makefile: 3
file content (295 lines) | stat: -rw-r--r-- 11,646 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
#---------------------------------------------------------------
# PyNLPl - Language Models
#   by Maarten van Gompel, ILK, Universiteit van Tilburg
#   http://ilk.uvt.nl/~mvgompel
#   proycon AT anaproy DOT nl
#
#   Licensed under GPLv3
#
#----------------------------------------------------------------

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import io
import math
import sys

from pynlpl.statistics import FrequencyList, product
from pynlpl.textprocessors import Windower

if sys.version < '3':
    from codecs import getwriter
    stderr = getwriter('utf-8')(sys.stderr)
    stdout = getwriter('utf-8')(sys.stdout)
else:
    stderr = sys.stderr
    stdout = sys.stdout


class SimpleLanguageModel:
    """This is a simple unsmoothed language model. This class can both hold and compute the model."""

    def __init__(self, n=2, casesensitive = True, beginmarker = "<begin>", endmarker = "<end>"):
        self.casesensitive = casesensitive
        self.freqlistN = FrequencyList(None, self.casesensitive)
        self.freqlistNm1 = FrequencyList(None, self.casesensitive)

        assert isinstance(n,int) and n >= 2
        self.n = n
        self.beginmarker = beginmarker
        self.endmarker = endmarker
        self.sentences = 0

        if self.beginmarker:
            self._begingram = tuple([self.beginmarker] * (n-1))
        if self.endmarker:
            self._endgram = tuple([self.endmarker] * (n-1))

    def append(self, sentence):
        if isinstance(sentence, str) or isinstance(sentence, unicode):
            sentence = sentence.strip().split(' ')
        self.sentences += 1
        for ngram in Windower(sentence,self.n, self.beginmarker, self.endmarker):
            self.freqlistN.count(ngram)
        for ngram in Windower(sentence,self.n-1, self.beginmarker, self.endmarker):
            self.freqlistNm1.count(ngram)


    def load(self, filename):
        self.freqlistN = FrequencyList(None, self.casesensitive)
        self.freqlistNm1 = FrequencyList(None, self.casesensitive)
        f = io.open(filename,'r',encoding='utf-8')
        mode = False
        for line in f.readlines():
            line = line.strip()
            if line:
                if not mode:
                    if line != "[simplelanguagemodel]":
                        raise Exception("File is not a SimpleLanguageModel")
                    else:
                        mode = 1
                elif mode == 1:
                    if line[:2] == 'n=':
                        self.n = int(line[2:])
                    elif line[:12] == 'beginmarker=':
                        self.beginmarker = line[12:]
                    elif line[:10] == 'endmarker=':
                        self.endmarker = line[10:]
                    elif line[:10] == 'sentences=':
                        self.sentences = int(line[10:])
                    elif line[:14] == 'casesensitive=':
                        self.casesensitive = bool(int(line[14:]))
                        self.freqlistN = FrequencyList(None, self.casesensitive)
                        self.freqlistNm1 = FrequencyList(None, self.casesensitive)
                    elif line == "[freqlistN]":
                        mode = 2
                    else:
                        raise Exception("Syntax error in language model file: ", line)
                elif mode == 2:
                    if line == "[freqlistNm1]":
                        mode = 3
                    else:
                        try:
                            type, count = line.split("\t")
                            self.freqlistN.count(type.split(' '),int(count))
                        except:
                            print("Warning, could not parse line whilst loading frequency list: ", line,file=stderr)
                elif mode == 3:
                        try:
                            type, count = line.split("\t")
                            self.freqlistNm1.count(type.split(' '),int(count))
                        except:
                            print("Warning, could not parse line whilst loading frequency list: ", line,file=stderr)

        if self.beginmarker:
            self._begingram = [self.beginmarker] * (self.n-1)
        if self.endmarker:
            self._endgram = [self.endmarker] * (self.n-1)


    def save(self, filename):
        f = io.open(filename,'w',encoding='utf-8')
        f.write("[simplelanguagemodel]\n")
        f.write("n="+str(self.n)+"\n")
        f.write("sentences="+str(self.sentences)+"\n")
        f.write("beginmarker="+self.beginmarker+"\n")
        f.write("endmarker="+self.endmarker+"\n")
        f.write("casesensitive="+str(int(self.casesensitive))+"\n")
        f.write("\n")
        f.write("[freqlistN]\n")
        for line in self.freqlistN.output():
            f.write(line+"\n")
        f.write("[freqlistNm1]\n")
        for line in self.freqlistNm1.output():
            f.write(line+"\n")
        f.close()


    def scoresentence(self, sentence):
        return product([self[x] for x in Windower(sentence, self.n, self.beginmarker, self.endmarker)])


    def __getitem__(self, ngram):
        assert len(ngram) == self.n

        nm1gram = ngram[:-1]

        if (self.beginmarker and nm1gram == self._begingram) or (self.endmarker and nm1gram == self._endgram):
            return self.freqlistN[ngram] / float(self.sentences)
        else:
            return self.freqlistN[ngram] / float(self.freqlistNm1[nm1gram])


class ARPALanguageModel(object):

    """Full back-off language model, loaded from file in ARPA format.

    This class does not build the model but allows you to use a pre-computed one.
    You can use the tool ngram-count from for instance SRILM to actually build the model.

    """

    class NgramsProbs(object):
        """Store Ngrams with their probabilities and backoffs.

        This class is used in order to abstract the physical storage layout,
        and enable memory/speed tradeoffs.

        """

        def __init__(self, data, mode='simple', delim=' '):
            """Create an ngrams storage with the given method:

            'simple' method is a Python dictionary (quick, takes much memory).
            'trie' method is more space-efficient (~35% reduction) but slower.
            data is a dictionary of ngram-tuple => (probability, backoff).
            delim is the strings which converts ngrams between tuple and
            unicode string (for saving in trie mode).

            """
            self.delim = delim
            self.mode = mode
            if mode == 'simple':
                self._data = data
            elif mode == 'trie':
                import marisa_trie
                self._data = marisa_trie.RecordTrie("@dd", [(self.delim.join(k), v) for k, v in data.items()])
            else:
                raise ValueError("mode {} is not supported for NgramsProbs".format(mode))

        def prob(self, ngram):
            """Return probability of given ngram tuple"""
            return self._data[ngram][0] if self.mode == 'simple' else self._data[self.delim.join(ngram)][0][0]

        def backoff(self, ngram):
            """Return backoff value of a given ngram tuple"""
            return self._data[ngram][1] if self.mode == 'simple' else self._data[self.delim.join(ngram)][0][1]

        def __len__(self):
            return len(self._data)


    def __init__(self, filename, encoding='utf-8', encoder=None, base_e=True, dounknown=True, debug=False, mode='simple'):
        # parameters
        self.encoder = (lambda x: x) if encoder is None else encoder
        self.base_e = base_e
        self.dounknown = dounknown
        self.debug = debug
        self.mode = mode
        # other attributes
        self.total = {}

        data = {}

        with io.open(filename, 'rt', encoding=encoding) as f:
            order = None
            for line in f:
                line = line.strip()
                if line == '\\data\\':
                    order = 0
                elif line == '\\end\\':
                    break
                elif line.startswith('\\') and line.endswith(':'):
                    for i in range(1, 10):
                        if line == '\\{}-grams:'.format(i):
                            order = i
                            break
                    else:
                        raise ValueError("Order of n-gram is not supported!")
                elif line:
                    if order == 0:  # still in \data\ section
                        if line.startswith('ngram'):
                            n = int(line[6])
                            v = int(line[8:])
                            self.total[n] = v
                    elif order > 0:
                        fields = line.split('\t')
                        logprob = float(fields[0])
                        if base_e:  # * log(10) does log10 to log_e conversion
                            logprob *= math.log(10)
                        ngram = self.encoder(tuple(fields[1].split()))
                        if len(fields) > 2:
                            backoffprob = float(fields[2])
                            if base_e:  # * log(10) does log10 to log_e conversion
                                backoffprob *= math.log(10)
                            if self.debug:
                                msg = "Adding to LM: {}\t{}\t{}"
                                print(msg.format(ngram, logprob, backoffprob), file=stderr)
                        else:
                            backoffprob = 0.0
                            if self.debug:
                                msg = "Adding to LM: {}\t{}"
                                print(msg.format(ngram, logprob), file=stderr)
                        data[ngram] = (logprob, backoffprob)
                    elif self.debug:
                        print("Unable to parse ARPA LM line: " + line, file=stderr)
        self.order = order
        self.ngrams = self.NgramsProbs(data, mode)

    def score(self, data, history=None):
        result = 0
        for word in data:
            result += self.scoreword(word, history)
            if history:
                history += (word,)
            else:
                history = (word,)
        return result

    def scoreword(self, word, history=None):
        if isinstance(word, str) or (sys.version < '3' and isinstance(word, unicode)):
            word = (word,)

        if history:
            lookup = history + word
        else:
            lookup = word

        if len(lookup) > self.order:
            lookup = lookup[-self.order:]

        try:
            return self.ngrams.prob(lookup)
        except KeyError:  # not found, back off
            if not history:
                if self.dounknown:
                    try:
                        return self.ngrams.prob(('<unk>',))
                    except KeyError:
                        msg = "Word {} not found. And no history specified and model has no <unk>."
                        raise KeyError(msg.format(word))
                else:
                    msg = "Word {} not found. And no history specified."
                    raise KeyError(msg.format(word))
            else:
                try:
                    backoffweight = self.ngrams.backoff(history)
                except KeyError:
                    backoffweight = 0  # backoff weight will be 0 if not found
                return backoffweight + self.scoreword(word, history[1:])

    def __len__(self):
        return len(self.ngrams)