| """ |
| Step 2: Reed-Solomon erasure coding over the verified GF(256) field. |
| |
| Split data into k systematic shards + m parity shards (n = k+m). ANY k of the n |
| shards reconstruct the original -- so up to m shards can be lost or corrupted and |
| the data still survives. This is the honest sense of "small pieces that together |
| rebuild the whole": redundancy, not shrinkage (n shards total n/k x the data). |
| |
| All coefficients come from a Cauchy matrix over GF(256) (every square submatrix |
| invertible => any-k-of-n recovery). Every multiply is the GF(256) multiply that |
| the neural LOG/EXP units were verified bit-exact against (65536/65536), so the |
| fast table path here is the proven-equivalent of the neural coding core. |
| """ |
| from __future__ import annotations |
| from .gf256 import EXP, LOG, gf_mul |
|
|
| MUL = [[gf_mul(a, b) for b in range(256)] for a in range(256)] |
|
|
|
|
| def gf_inv(a: int) -> int: |
| return EXP[255 - LOG[a]] if a else 0 |
|
|
|
|
| def cauchy(k: int, m: int): |
| return [[gf_inv((k + i) ^ j) for j in range(k)] for i in range(m)] |
|
|
|
|
| def _gen_row(r: int, k: int, C): |
| if r < k: |
| row = [0] * k |
| row[r] = 1 |
| return row |
| return C[r - k] |
|
|
|
|
| def encode(data: bytes, k: int, m: int): |
| """-> (shards[n], shard_len, orig_len). shards[:k] are systematic data.""" |
| orig = len(data) |
| L = (orig + k - 1) // k |
| d = data + bytes(L * k - orig) |
| shards = [bytearray(d[j * L:(j + 1) * L]) for j in range(k)] |
| C = cauchy(k, m) |
| for i in range(m): |
| P = bytearray(L) |
| Ci = C[i] |
| for j in range(k): |
| MC = MUL[Ci[j]] |
| Dj = shards[j] |
| for b in range(L): |
| P[b] ^= MC[Dj[b]] |
| shards.append(P) |
| return shards, L, orig |
|
|
|
|
| def _invert(M, k): |
| A = [list(M[i]) + [1 if j == i else 0 for j in range(k)] for i in range(k)] |
| for col in range(k): |
| piv = col |
| while A[piv][col] == 0: |
| piv += 1 |
| A[col], A[piv] = A[piv], A[col] |
| inv = gf_inv(A[col][col]) |
| A[col] = [MUL[inv][x] for x in A[col]] |
| for r in range(k): |
| if r != col and A[r][col]: |
| f = A[r][col] |
| A[r] = [A[r][x] ^ MUL[f][A[col][x]] for x in range(2 * k)] |
| return [A[i][k:] for i in range(k)] |
|
|
|
|
| def decode(present: dict, k: int, m: int, L: int, orig: int) -> bytes: |
| """present: {shard_index: bytes}; needs >= k entries.""" |
| idxs = sorted(present.keys())[:k] |
| C = cauchy(k, m) |
| Minv = _invert([_gen_row(r, k, C) for r in idxs], k) |
| data_shards = [bytearray(L) for _ in range(k)] |
| for j in range(k): |
| row = Minv[j] |
| Dj = data_shards[j] |
| for t, r in enumerate(idxs): |
| c = row[t] |
| if not c: |
| continue |
| MC = MUL[c] |
| src = present[r] |
| for b in range(L): |
| Dj[b] ^= MC[src[b]] |
| return b"".join(bytes(s) for s in data_shards)[:orig] |
|
|
|
|
| def neural_parity_equiv(net_log, net_exp, k=4, m=2, L=16) -> bool: |
| """Compute one parity shard with the NEURAL gf_mul and confirm it matches the |
| golden coding -- i.e. the verified units drive the erasure code bit-exactly.""" |
| import os |
| from .gf256 import gf_mul_neural |
| data = os.urandom(L * k) |
| shards, _, _ = encode(data, k, m) |
| C = cauchy(k, m) |
| Dsh = [data[j * L:(j + 1) * L] for j in range(k)] |
| P = bytearray(L) |
| for j in range(k): |
| for b in range(L): |
| P[b] ^= gf_mul_neural(C[0][j], Dsh[j][b], net_log, net_exp) |
| return bytes(P) == bytes(shards[k]) |
|
|
