-
Notifications
You must be signed in to change notification settings - Fork 0
/
QR.py
559 lines (468 loc) · 35.3 KB
/
QR.py
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
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
class QR:
def __init__(self, n, k, message):
self.format_ = 0
self.encoded_format = (self.format_ << 10) + self.qr_check_format(self.format_ << 10)
self.gf_exp = [0] * 512
self.gf_log = [0] * 256
self.prim = 0x11d
self.n = n
self.k = k
self.message = message
self.init_tables(prim=self.prim)
self.mesecc = self.rs_encode_msg([ord(x) for x in message], n-k)
self.mesecc_temp = self.mesecc
self.corrupt_mesecc = self.rs_encode_msg([ord(x) for x in message], n-k)
self.corrupt_mesecc[0] = 0
self.corrupt_mesecc[1] = 2
self.corrupt_mesecc[2] = 2
self.corrupt_mesecc[3] = 2
self.corrupt_mesecc[4] = 2
self.corrupt_mesecc[5] = 2
self.mesecc_temp_corrupt = self.corrupt_mesecc
self.corrected_message, self.corrected_ecc = self.rs_correct_msg(self.corrupt_mesecc, n-k, erase_pos=[0, 1, 2])
self.__str__()
# BCH error detection
def qr_check_format(self, fmt):
g = 0x537 # = 0b10100110111 in python 2.6+
for i in range(4,-1,-1):
if fmt & (1 << (i+10)):
fmt ^= g << i
return fmt
# BCH error correction
def hamming_weight(self, x):
weight = 0
while x > 0:
weight += x & 1
x >>= 1
return weight
def qr_decode_format(self, fmt):
best_fmt = -1
best_dist = 15
for test_fmt in range(0,32):
test_code = (test_fmt<<10) ^ self.qr_check_format(test_fmt<<10)
test_dist = self.hamming_weight(fmt ^ test_code)
if test_dist < best_dist:
best_dist = test_dist
best_fmt = test_fmt
elif test_dist == best_dist:
best_fmt = -1
return best_fmt
# finite field arithmetic
# addition and subtraction
def gf_add(self, x, y):
return x ^ y
def gf_sub(self, x, y):
return x ^ y # in binary galois field, subtraction is just the same as addition (since we mod 2)
# multiplication
def cl_mul(self, x,y):
'''Bitwise carry-less multiplication on integers'''
z = 0
i = 0
while (y>>i) > 0:
if y & (1<<i):
z ^= x<<i
i += 1
return z
# Galois Field multiplication with modular reduction
def gf_mult_noLUT(self, x, y, prim=0):
'''Multiplication in Galois Fields without using a precomputed look-up table (and thus it's slower)
by using the standard carry-less multiplication + modular reduction using an irreducible prime polynomial'''
### Define bitwise carry-less operations as inner functions ###
def cl_mult(self, x,y):
'''Bitwise carry-less multiplication on integers'''
z = 0
i = 0
while (y>>i) > 0:
if y & (1<<i):
z ^= x<<i
i += 1
return z
def bit_length(self, n):
'''Compute the position of the most significant bit (1) of an integer. Equivalent to int.bit_length()'''
bits = 0
while n >> bits: bits += 1
return bits
def cl_div(self, dividend, divisor=None):
'''Bitwise carry-less long division on integers and returns the remainder'''
# Compute the position of the most significant bit for each integers
dl1 = self.bit_length(dividend)
dl2 = self.bit_length(divisor)
# If the dividend is smaller than the divisor, just exit
if dl1 < dl2:
return dividend
# Else, align the most significant 1 of the divisor to the most significant 1 of the dividend (by shifting the divisor)
for i in range(dl1-dl2,-1,-1):
# Check that the dividend is divisible (useless for the first iteration but important for the next ones)
if dividend & (1 << i+dl2-1):
# If divisible, then shift the divisor to align the most significant bits and XOR (carry-less subtraction)
dividend ^= divisor << i
return dividend
### Main GF multiplication routine ###
# Multiply the gf numbers
result = self.cl_mult(x,y)
# Then do a modular reduction (ie, remainder from the division) with an irreducible primitive polynomial so that it stays inside GF bounds
if prim > 0:
result = self.cl_div(result, prim)
return result
# Russian Peasant Multiplication algorithm
def gf_mult_noLUT(self, x, y, prim=0, field_charac_full=256, carryless=True):
'''Galois Field integer multiplication using Russian Peasant Multiplication algorithm (faster than the standard multiplication + modular reduction).
If prim is 0 and carryless=False, then the function produces the result for a standard integers multiplication (no carry-less arithmetics nor modular reduction).'''
r = 0
while y: # while y is above 0
if y & 1: r = r ^ x if carryless else r + x # y is odd, then add the corresponding x to r (the sum of all x's corresponding to odd y's will give the final product). Note that since we're in GF(2), the addition is in fact an XOR (very important because in GF(2) the multiplication and additions are carry-less, thus it changes the result!).
y = y >> 1 # equivalent to y // 2
x = x << 1 # equivalent to x*2
if prim > 0 and x & field_charac_full: x = x ^ prim # GF modulo: if x >= 256 then apply modular reduction using the primitive polynomial (we just subtract, but since the primitive number can be above 256 then we directly XOR).
return r
# Multiplication with logarithms
# gf_exp = [0] * 512 # Create list of 512 elements. In Python 2.6+, consider using bytearray
# gf_log = [0] * 256
def init_tables(self, prim=0x11d):
'''Precompute the logarithm and anti-log tables for faster computation later, using the provided primitive polynomial.'''
# prim is the primitive (binary) polynomial. Since it's a polynomial in the binary sense,
# it's only in fact a single galois field value between 0 and 255, and not a list of gf values.
# global gf_exp, gf_log
# self.gf_exp = [0] * 512 # anti-log (exponential) table
# self.gf_log = [0] * 256 # log table
# For each possible value in the galois field 2^8, we will pre-compute the logarithm and anti-logarithm (exponential) of this value
x = 1
for i in range(0, 255):
self.gf_exp[i] = x # compute anti-log for this value and store it in a table
self.gf_log[x] = i # compute log at the same time
x = self.gf_mult_noLUT(x, 2, prim)
# If you use only generator==2 or a power of 2, you can use the following which is faster than gf_mult_noLUT():
#x <<= 1 # multiply by 2 (change 1 by another number y to multiply by a power of 2^y)
#if x & 0x100: # similar to x >= 256, but a lot faster (because 0x100 == 256)
#x ^= prim # substract the primary polynomial to the current value (instead of 255, so that we get a unique set made of coprime numbers), this is the core of the tables generation
# Optimization: double the size of the anti-log table so that we don't need to mod 255 to
# stay inside the bounds (because we will mainly use this table for the multiplication of two GF numbers, no more).
for i in range(255, 512):
self.gf_exp[i] = self.gf_exp[i - 255]
return [self.gf_log, self.gf_exp]
# The gf_exp table is oversized in order to simplify the multiplication function. This way, we don't have to check to make sure that gf_log[x] + gf_log[y] is within the table size
def gf_mul(self, x,y):
if x==0 or y==0:
return 0
return self.gf_exp[self.gf_log[x] + self.gf_log[y]] # should be gf_exp[(gf_log[x]+gf_log[y])%255] if gf_exp wasn't oversized
# Division
def gf_div(self, x,y):
if y==0:
raise ZeroDivisionError()
if x==0:
return 0
return self.gf_exp[(self.gf_log[x] + 255 - self.gf_log[y]) % 255]
# Power and Inverse
def gf_pow(self, x, power):
return self.gf_exp[(self.gf_log[x] * power) % 255]
def gf_inverse(self, x):
return self.gf_exp[255 - self.gf_log[x]] # gf_inverse(x) == gf_div(1, x)
# Polynomials (multiplies a polynomial by a scalar)
def gf_poly_scale(self, p,x):
r = [0] * len(p)
for i in range(0, len(p)):
r[i] = self.gf_mul(p[i], x)
return r
# Adds two polynomials (XOR)
def gf_poly_add(self, p,q):
r = [0] * max(len(p),len(q))
for i in range(0,len(p)):
r[i+len(r)-len(p)] = p[i]
for i in range(0,len(q)):
r[i+len(r)-len(q)] ^= q[i]
return r
# Multiplies two polynomials
def gf_poly_mul(self, p,q):
'''Multiply two polynomials, inside Galois Field'''
# Pre-allocate the result array
r = [0] * (len(p)+len(q)-1)
# Compute the polynomial multiplication (just like the outer product of two vectors,
# we multiply each coefficients of p with all coefficients of q)
for j in range(0, len(q)):
for i in range(0, len(p)):
r[i+j] ^= self.gf_mul(p[i], q[j]) # equivalent to: r[i + j] = gf_add(r[i+j], gf_mul(p[i], q[j]))
# -- you can see it's your usual polynomial multiplication
return r
# Horner's method, evaluate a polynomial at a particular value of x, producing a scalar result.
def gf_poly_eval(self, poly, x):
'''Evaluates a polynomial in GF(2^p) given the value for x. This is based on Horner's scheme for maximum efficiency.'''
y = poly[0]
for i in range(1, len(poly)):
y = self.gf_mul(y, x) ^ poly[i]
return y
# Exception management
class ReedSolomonError(Exception):
pass
# try:
# raise ReedSolomonError("Some error message")
# except ReedSolomonError as e:
# pass # do something here
# raise ReedSolomonError("Some error message")
# RS generator polynomial
def rs_generator_poly(self, nsym):
'''Generate an irreducible generator polynomial (necessary to encode a message into Reed-Solomon)'''
g = [1]
for i in range(0, nsym):
g = self.gf_poly_mul(g, [1, self.gf_pow(2, i)])
return g
# Polynomial division
def gf_poly_div(self, dividend, divisor):
'''Fast polynomial division by using Extended Synthetic Division and optimized for GF(2^p) computations
(doesn't work with standard polynomials outside of this galois field, see the Wikipedia article for generic algorithm).'''
# CAUTION: this function expects polynomials to follow the opposite convention at decoding:
# the terms must go from the biggest to lowest degree (while most other functions here expect
# a list from lowest to biggest degree). eg: 1 + 2x + 5x^2 = [5, 2, 1], NOT [1, 2, 5]
msg_out = list(dividend) # Copy the dividend
#normalizer = divisor[0] # precomputing for performance
for i in range(0, len(dividend) - (len(divisor)-1)):
#msg_out[i] /= normalizer # for general polynomial division (when polynomials are non-monic), the usual way of using
# synthetic division is to divide the divisor g(x) with its leading coefficient, but not needed here.
coef = msg_out[i] # precaching
if coef != 0: # log(0) is undefined, so we need to avoid that case explicitly (and it's also a good optimization).
for j in range(1, len(divisor)): # in synthetic division, we always skip the first coefficient of the divisior,
# because it's only used to normalize the dividend coefficient
if divisor[j] != 0: # log(0) is undefined
msg_out[i + j] ^= self.gf_mul(divisor[j], coef) # equivalent to the more mathematically correct
# (but xoring directly is faster): msg_out[i + j] += -divisor[j] * coef
# The resulting msg_out contains both the quotient and the remainder, the remainder being the size of the divisor
# (the remainder has necessarily the same degree as the divisor -- not length but degree == length-1 -- since it's
# what we couldn't divide from the dividend), so we compute the index where this separation is, and return the quotient and remainder.
separator = -(len(divisor)-1)
return msg_out[:separator], msg_out[separator:] # return quotient, remainder.
# Encoding main function
def rs_encode_msg(self, msg_in, nsym):
'''Reed-Solomon main encoding function, using polynomial division (algorithm Extended Synthetic Division)'''
if (len(msg_in) + nsym) > 255: raise ValueError("Message is too long (%i when max is 255)" % (len(msg_in)+nsym))
gen = self.rs_generator_poly(nsym)
# Init msg_out with the values inside msg_in and pad with len(gen)-1 bytes (which is the number of ecc symbols).
msg_out = [0] * (len(msg_in) + len(gen)-1)
# Initializing the Synthetic Division with the dividend (= input message polynomial)
msg_out[:len(msg_in)] = msg_in
# Synthetic division main loop
for i in range(len(msg_in)):
# Note that it's msg_out here, not msg_in. Thus, we reuse the updated value at each iteration
# (this is how Synthetic Division works: instead of storing in a temporary register the intermediate values,
# we directly commit them to the output).
coef = msg_out[i]
# log(0) is undefined, so we need to manually check for this case. There's no need to check
# the divisor here because we know it can't be 0 since we generated it.
if coef != 0:
# in synthetic division, we always skip the first coefficient of the divisior, because it's only used to normalize the dividend coefficient (which is here useless since the divisor, the generator polynomial, is always monic)
for j in range(1, len(gen)):
msg_out[i+j] ^= self.gf_mul(gen[j], coef) # equivalent to msg_out[i+j] += gf_mul(gen[j], coef)
# At this point, the Extended Synthetic Divison is done, msg_out contains the quotient in msg_out[:len(msg_in)]
# and the remainder in msg_out[len(msg_in):]. Here for RS encoding, we don't need the quotient but only the remainder
# (which represents the RS code), so we can just overwrite the quotient with the input message, so that we get
# our complete codeword composed of the message + code.
msg_out[:len(msg_in)] = msg_in
return msg_out
# Syndrome calculation
def rs_calc_syndromes(self, msg, nsym):
'''Given the received codeword msg and the number of error correcting symbols (nsym), computes the syndromes polynomial.
Mathematically, it's essentially equivalent to a Fourrier Transform (Chien search being the inverse).
'''
# Note the "[0] +" : we add a 0 coefficient for the lowest degree (the constant). This effectively shifts the syndrome, and will shift every computations depending on the syndromes (such as the errors locator polynomial, errors evaluator polynomial, etc. but not the errors positions).
# This is not necessary, you can adapt subsequent computations to start from 0 instead of skipping the first iteration (ie, the often seen range(1, n-k+1)),
synd = [0] * nsym
for i in range(0, nsym):
synd[i] = self.gf_poly_eval(msg, self.gf_pow(2,i))
return [0] + synd # pad with one 0 for mathematical precision (else we can end up with weird calculations sometimes)
# automate corruption of at least one character
def rs_check(self, msg, nsym):
'''Returns true if the message + ecc has no error or false otherwise (may not always catch a wrong decoding or a wrong message, particularly if there are too many errors -- above the Singleton bound --, but it usually does)'''
return ( max(self.rs_calc_syndromes(msg, nsym)) == 0 )
# Erasure correction
def rs_find_errata_locator(self, e_pos):
'''Compute the erasures/errors/errata locator polynomial from the erasures/errors/errata positions
(the positions must be relative to the x coefficient, eg: "hello worldxxxxxxxxx" is tampered to "h_ll_ worldxxxxxxxxx"
with xxxxxxxxx being the ecc of length n-k=9, here the string positions are [1, 4], but the coefficients are reversed
since the ecc characters are placed as the first coefficients of the polynomial, thus the coefficients of the
erased characters are n-1 - [1, 4] = [18, 15] = erasures_loc to be specified as an argument.'''
e_loc = [1] # just to init because we will multiply, so it must be 1 so that the multiplication starts correctly without nulling any term
# erasures_loc = product(1 - x*alpha**i) for i in erasures_pos and where alpha is the alpha chosen to evaluate polynomials.
for i in e_pos:
e_loc = self.gf_poly_mul( e_loc, self.gf_poly_add([1], [self.gf_pow(2, i), 0]) )
return e_loc
def rs_find_error_evaluator(self, synd, err_loc, nsym):
'''Compute the error (or erasures if you supply sigma=erasures locator polynomial, or errata) evaluator polynomial Omega
from the syndrome and the error/erasures/errata locator Sigma.'''
# Omega(x) = [ Synd(x) * Error_loc(x) ] mod x^(n-k+1)
_, remainder = self.gf_poly_div( self.gf_poly_mul(synd, err_loc), ([1] + [0]*(nsym+1)) ) # first multiply syndromes * errata_locator, then do a
# polynomial division to truncate the polynomial to the
# required length
# Faster way that is equivalent
#remainder = gf_poly_mul(synd, err_loc) # first multiply the syndromes with the errata locator polynomial
#remainder = remainder[len(remainder)-(nsym+1):] # then slice the list to truncate it (which represents the polynomial), which
# is equivalent to dividing by a polynomial of the length we want
return remainder
# Forney algorithm
def rs_correct_errata(self, msg_in, synd, err_pos): # err_pos is a list of the positions of the errors/erasures/errata
'''Forney algorithm, computes the values (error magnitude) to correct the input message.'''
# calculate errata locator polynomial to correct both errors and erasures (by combining the errors positions given by the error locator polynomial found by BM with the erasures positions given by caller)
coef_pos = [len(msg_in) - 1 - p for p in err_pos] # need to convert the positions to coefficients degrees for the errata locator algo to work (eg: instead of [0, 1, 2] it will become [len(msg)-1, len(msg)-2, len(msg) -3])
err_loc = self.rs_find_errata_locator(coef_pos)
# calculate errata evaluator polynomial (often called Omega or Gamma in academic papers)
err_eval = self.rs_find_error_evaluator(synd[::-1], err_loc, len(err_loc)-1)[::-1]
# Second part of Chien search to get the error location polynomial X from the error positions in err_pos (the roots of the error locator polynomial, ie, where it evaluates to 0)
X = [] # will store the position of the errors
for i in range(0, len(coef_pos)):
l = 255 - coef_pos[i]
X.append( self.gf_pow(2, -l) )
# Forney algorithm: compute the magnitudes
E = [0] * (len(msg_in)) # will store the values that need to be corrected (substracted) to the message containing errors. This is sometimes called the error magnitude polynomial.
Xlength = len(X)
for i, Xi in enumerate(X):
Xi_inv = self.gf_inverse(Xi)
# Compute the formal derivative of the error locator polynomial (see Blahut, Algebraic codes for data transmission, pp 196-197).
# the formal derivative of the errata locator is used as the denominator of the Forney Algorithm, which simply says that the ith error value is given by error_evaluator(gf_inverse(Xi)) / error_locator_derivative(gf_inverse(Xi)). See Blahut, Algebraic codes for data transmission, pp 196-197.
err_loc_prime_tmp = []
for j in range(0, Xlength):
if j != i:
err_loc_prime_tmp.append( self.gf_sub(1, self.gf_mul(Xi_inv, X[j])) )
# compute the product, which is the denominator of the Forney algorithm (errata locator derivative)
err_loc_prime = 1
for coef in err_loc_prime_tmp:
err_loc_prime = self.gf_mul(err_loc_prime, coef)
# equivalent to: err_loc_prime = functools.reduce(gf_mul, err_loc_prime_tmp, 1)
# Compute y (evaluation of the errata evaluator polynomial)
# This is a more faithful translation of the theoretical equation contrary to the old forney method. Here it is an exact reproduction:
# Yl = omega(Xl.inverse()) / prod(1 - Xj*Xl.inverse()) for j in len(X)
y = self.gf_poly_eval(err_eval[::-1], Xi_inv) # numerator of the Forney algorithm (errata evaluator evaluated)
y = self.gf_mul(self.gf_pow(Xi, 1), y)
# Check: err_loc_prime (the divisor) should not be zero.
if err_loc_prime == 0:
raise self.ReedSolomonError("Could not find error magnitude") # Could not find error magnitude
# Compute the magnitude
magnitude = self.gf_div(y, err_loc_prime) # magnitude value of the error, calculated by the Forney algorithm (an equation in fact): dividing the errata evaluator with the errata locator derivative gives us the errata magnitude (ie, value to repair) the ith symbol
E[err_pos[i]] = magnitude # store the magnitude for this error into the magnitude polynomial
# Apply the correction of values to get our message corrected! (note that the ecc bytes also gets corrected!)
# (this isn't the Forney algorithm, we just apply the result of decoding here)
msg_in = self.gf_poly_add(msg_in, E) # equivalent to Ci = Ri - Ei where Ci is the correct message, Ri the received (senseword) message, and Ei the errata magnitudes (minus is replaced by XOR since it's equivalent in GF(2^p)). So in fact here we substract from the received message the errors magnitude, which logically corrects the value to what it should be.
return msg_in
# Error correction
def rs_find_error_locator(self, synd, nsym, erase_loc=None, erase_count=0):
'''Find error/errata locator and evaluator polynomials with Berlekamp-Massey algorithm'''
# The idea is that BM will iteratively estimate the error locator polynomial.
# To do this, it will compute a Discrepancy term called Delta, which will tell us if the error locator polynomial needs an update or not
# (hence why it's called discrepancy: it tells us when we are getting off board from the correct value).
# Init the polynomials
if erase_loc: # if the erasure locator polynomial is supplied, we init with its value, so that we include erasures in the final locator polynomial
err_loc = list(erase_loc)
old_loc = list(erase_loc)
else:
err_loc = [1] # This is the main variable we want to fill, also called Sigma in other notations or more formally the errors/errata locator polynomial.
old_loc = [1] # BM is an iterative algorithm, and we need the errata locator polynomial of the previous iteration in order to update other necessary variables.
#L = 0 # update flag variable, not needed here because we use an alternative equivalent way of checking if update is needed (but using the flag could potentially be faster depending on if using length(list) is taking linear time in your language, here in Python it's constant so it's as fast.
# Fix the syndrome shifting: when computing the syndrome, some implementations may prepend a 0 coefficient for the lowest degree term (the constant). This is a case of syndrome shifting, thus the syndrome will be bigger than the number of ecc symbols (I don't know what purpose serves this shifting). If that's the case, then we need to account for the syndrome shifting when we use the syndrome such as inside BM, by skipping those prepended coefficients.
# Another way to detect the shifting is to detect the 0 coefficients: by definition, a syndrome does not contain any 0 coefficient (except if there are no errors/erasures, in this case they are all 0). This however doesn't work with the modified Forney syndrome, which set to 0 the coefficients corresponding to erasures, leaving only the coefficients corresponding to errors.
synd_shift = len(synd) - nsym
for i in range(0, nsym-erase_count): # generally: nsym-erase_count == len(synd), except when you input a partial erase_loc and using the full syndrome instead of the Forney syndrome, in which case nsym-erase_count is more correct (len(synd) will fail badly with IndexError).
if erase_loc: # if an erasures locator polynomial was provided to init the errors locator polynomial, then we must skip the FIRST erase_count iterations (not the last iterations, this is very important!)
K = erase_count+i+synd_shift
else: # if erasures locator is not provided, then either there's no erasures to account or we use the Forney syndromes, so we don't need to use erase_count nor erase_loc (the erasures have been trimmed out of the Forney syndromes).
K = i+synd_shift
# Compute the discrepancy Delta
# Here is the close-to-the-books operation to compute the discrepancy Delta: it's a simple polynomial multiplication of error locator with the syndromes, and then we get the Kth element.
#delta = gf_poly_mul(err_loc[::-1], synd)[K] # theoretically it should be gf_poly_add(synd[::-1], [1])[::-1] instead of just synd, but it seems it's not absolutely necessary to correctly decode.
# But this can be optimized: since we only need the Kth element, we don't need to compute the polynomial multiplication for any other element but the Kth. Thus to optimize, we compute the polymul only at the item we need, skipping the rest (avoiding a nested loop, thus we are linear time instead of quadratic).
# This optimization is actually described in several figures of the book "Algebraic codes for data transmission", Blahut, Richard E., 2003, Cambridge university press.
delta = synd[K]
for j in range(1, len(err_loc)):
delta ^= self.gf_mul(err_loc[-(j+1)], synd[K - j]) # delta is also called discrepancy. Here we do a partial polynomial multiplication (ie, we compute the polynomial multiplication only for the term of degree K). Should be equivalent to brownanrs.polynomial.mul_at().
#print "delta", K, delta, list(gf_poly_mul(err_loc[::-1], synd)) # debugline
# Shift polynomials to compute the next degree
old_loc = old_loc + [0]
# Iteratively estimate the errata locator and evaluator polynomials
if delta != 0: # Update only if there's a discrepancy
if len(old_loc) > len(err_loc): # Rule B (rule A is implicitly defined because rule A just says that we skip any modification for this iteration)
#if 2*L <= K+erase_count: # equivalent to len(old_loc) > len(err_loc), as long as L is correctly computed
# Computing errata locator polynomial Sigma
new_loc = self.gf_poly_scale(old_loc, delta)
old_loc = self.gf_poly_scale(err_loc, self.gf_inverse(delta)) # effectively we are doing err_loc * 1/delta = err_loc // delta
err_loc = new_loc
# Update the update flag
#L = K - L # the update flag L is tricky: in Blahut's schema, it's mandatory to use `L = K - L - erase_count` (and indeed in a previous draft of this function, if you forgot to do `- erase_count` it would lead to correcting only 2*(errors+erasures) <= (n-k) instead of 2*errors+erasures <= (n-k)), but in this latest draft, this will lead to a wrong decoding in some cases where it should correctly decode! Thus you should try with and without `- erase_count` to update L on your own implementation and see which one works OK without producing wrong decoding failures.
# Update with the discrepancy
err_loc = self.gf_poly_add(err_loc, self.gf_poly_scale(old_loc, delta))
# Check if the result is correct, that there's not too many errors to correct
while len(err_loc) and err_loc[0] == 0: del err_loc[0] # drop leading 0s, else errs will not be of the correct size
errs = len(err_loc) - 1
if (errs-erase_count) * 2 + erase_count > nsym:
raise self.ReedSolomonError("Too many errors to correct") # too many errors to correct
return err_loc
def rs_find_errors(self, err_loc, nmess): # nmess is len(msg_in)
'''Find the roots (ie, where evaluation = zero) of error polynomial by brute-force trial, this is a sort of Chien's search
(but less efficient, Chien's search is a way to evaluate the polynomial such that each evaluation only takes constant time).'''
errs = len(err_loc) - 1
err_pos = []
for i in range(nmess): # normally we should try all 2^8 possible values, but here we optimize to just check the interesting symbols
if self.gf_poly_eval(err_loc, self.gf_pow(2, i)) == 0: # It's a 0? Bingo, it's a root of the error locator polynomial,
# in other terms this is the location of an error
err_pos.append(nmess - 1 - i)
# Sanity check: the number of errors/errata positions found should be exactly the same as the length of the errata locator polynomial
if len(err_pos) != errs:
# couldn't find error locations
raise self.ReedSolomonError("Too many (or few) errors found by Chien Search for the errata locator polynomial!")
return err_pos
# Error and erasure correction
def rs_forney_syndromes(self, synd, pos, nmess):
# Compute Forney syndromes, which computes a modified syndromes to compute only errors (erasures are trimmed out). Do not confuse this with Forney algorithm, which allows to correct the message based on the location of errors.
erase_pos_reversed = [nmess-1-p for p in pos] # prepare the coefficient degree positions (instead of the erasures positions)
# Optimized method, all operations are inlined
fsynd = list(synd[1:]) # make a copy and trim the first coefficient which is always 0 by definition
for i in range(0, len(pos)):
x = self.gf_pow(2, erase_pos_reversed[i])
for j in range(0, len(fsynd) - 1):
fsynd[j] = self.gf_mul(fsynd[j], x) ^ fsynd[j + 1]
# Equivalent, theoretical way of computing the modified Forney syndromes: fsynd = (erase_loc * synd) % x^(n-k)
# See Shao, H. M., Truong, T. K., Deutsch, L. J., & Reed, I. S. (1986, April). A single chip VLSI Reed-Solomon decoder. In Acoustics, Speech, and Signal Processing, IEEE International Conference on ICASSP'86. (Vol. 11, pp. 2151-2154). IEEE.ISO 690
#erase_loc = rs_find_errata_locator(erase_pos_reversed, generator=generator) # computing the erasures locator polynomial
#fsynd = gf_poly_mul(erase_loc[::-1], synd[1:]) # then multiply with the syndrome to get the untrimmed forney syndrome
#fsynd = fsynd[len(pos):] # then trim the first erase_pos coefficients which are useless. Seems to be not necessary, but this reduces the computation time later in BM (thus it's an optimization).
return fsynd
def rs_correct_msg(self, msg_in, nsym, erase_pos=None):
'''Reed-Solomon main decoding function'''
if len(msg_in) > 255: # can't decode, message is too big
raise ValueError("Message is too long (%i when max is 255)" % len(msg_in))
msg_out = list(msg_in) # copy of message
# erasures: set them to null bytes for easier decoding (but this is not necessary, they will be corrected anyway, but debugging will be easier with null bytes because the error locator polynomial values will only depend on the errors locations, not their values)
if erase_pos is None:
erase_pos = []
else:
for e_pos in erase_pos:
msg_out[e_pos] = 0
# check if there are too many erasures to correct (beyond the Singleton bound)
if len(erase_pos) > nsym: raise self.ReedSolomonError("Too many erasures to correct")
# prepare the syndrome polynomial using only errors (ie: errors = characters that were either replaced by null byte
# or changed to another character, but we don't know their positions)
synd = self.rs_calc_syndromes(msg_out, nsym)
# check if there's any error/erasure in the input codeword. If not (all syndromes coefficients are 0), then just return the message as-is.
if max(synd) == 0:
return msg_out[:-nsym], msg_out[-nsym:] # no errors
# compute the Forney syndromes, which hide the erasures from the original syndrome (so that BM will just have to deal with errors, not erasures)
fsynd = self.rs_forney_syndromes(synd, erase_pos, len(msg_out))
# compute the error locator polynomial using Berlekamp-Massey
err_loc = self.rs_find_error_locator(fsynd, nsym, erase_count=len(erase_pos))
# locate the message errors using Chien search (or brute-force search)
err_pos = self.rs_find_errors(err_loc[::-1] , len(msg_out))
if err_pos is None:
raise self.ReedSolomonError("Could not locate error") # error location failed
# Find errors values and apply them to correct the message
# compute errata evaluator and errata magnitude polynomials, then correct errors and erasures
msg_out = self.rs_correct_errata(msg_out, synd, (erase_pos + err_pos)) # note that we here use the original syndrome, not the forney syndrome
# (because we will correct both errors and erasures, so we need the full syndrome)
# check if the final message is fully repaired
synd = self.rs_calc_syndromes(msg_out, nsym)
if max(synd) > 0:
raise self.ReedSolomonError("Could not correct message") # message could not be repaired
# return the successfully decoded message
return msg_out[:-nsym], msg_out[-nsym:] # also return the corrected ecc block so that the user can check()
def __str__(self) -> str:
return print(
"Original: " + str(self.mesecc),
"\nCorrupted: " + str(self.corrupt_mesecc),
"\nRepaired: " + str(self.corrected_message + self.corrected_ecc),
"\n" + ''.join([chr(x) for x in self.corrected_message]),
"\n\nGalois Field Exponential:\n" + str(self.gf_exp),
"\n\nGalois Field Logarithm Table:\n" + str(self.gf_log)
)