Skip to content

correction: Drop redundant BCH decoding paths #150

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
BenWestgate merged 1 commit into from
Oct 19, 2026
Merged

correction: Drop redundant BCH decoding paths #150

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
143 changes: 37 additions & 106 deletions src/codex32/correction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,28 +376,15 @@ def _small_locator(sequence: list[int], maximum: int) -> list[int] | None:
return [1, *coefficients] if len(coefficients) <= maximum else None


def _locator_poly(
syndromes: list[int],
erasure_poly: list[int],
maximum: int | None,
erasure_products: tuple[tuple[int, ...], ...] | None = None,
) -> list[int] | None:
def _locator_poly(syndromes: list[int], erasure_poly: list[int], maximum: int | None) -> list[int] | None:
"""Locate substitutions from Forney syndromes, which hide the known erasures."""
degree = len(erasure_poly) - 1
if degree == 2 and erasure_products is not None:
first, second = erasure_products
modified = [
syndromes[index - 2] ^ first[syndromes[index]] ^ second[syndromes[index - 1]]
for index in range(2, len(syndromes))
]
else:
modified = []
for output_index in range(degree, len(syndromes)):
value = syndromes[output_index - degree]
products = erasure_products
for index, coefficient in enumerate(erasure_poly[:-1]):
syndrome = syndromes[output_index - index]
value ^= _gf1024_mul(coefficient, syndrome) if products is None else products[index][syndrome]
modified.append(value)
modified = []
for output_index in range(degree, len(syndromes)):
value = syndromes[output_index - degree]
for index, coefficient in enumerate(erasure_poly[:-1]):
value ^= _gf1024_mul(coefficient, syndromes[output_index - index])
modified.append(value)
if maximum is not None and maximum <= 2:
return _small_locator(modified, maximum)
coefficients, _adjustment = _synthesize_rec(list(reversed(modified)))
Expand Down Expand Up @@ -425,13 +412,17 @@ def _bch_syndrome_corrections(
word_length: int,
max_substitutions: int | None,
erasure_state: tuple[tuple[int, ...], list[int]] | None = None,
erasure_products: tuple[tuple[int, ...], ...] | None = None,
) -> list[tuple[int, int]] | None:
"""Fill erasures and locate substitutions: locator, Chien search, then Forney.

GF(32) values that clear these 8 syndromes also clear their conjugates, which are the
remaining roots of the generator, so every returned correction reaches the target.
"""
word_roots = _word_roots(spec, word_length)
erasure_roots, erasure_poly = (
_erasure_state(spec, word_length, tuple(erasure_indices)) if erasure_state is None else erasure_state
)
locator = _locator_poly(syndromes, erasure_poly, max_substitutions, erasure_products)
locator = _locator_poly(syndromes, erasure_poly, max_substitutions)
if locator is None or max_substitutions is not None and len(locator) - 1 > max_substitutions:
return None
errors = [
Expand Down Expand Up @@ -465,45 +456,30 @@ def _repair_body(
alignment: tuple[int, tuple[tuple[int, ...], ...]],
erasure_indices: Sequence[int] | None,
erasure_state: tuple[tuple[int, ...], list[int]] | None,
erasure_products: tuple[tuple[int, ...], ...] | None,
packed_syndromes: int | None = None,
) -> tuple[list[int], list[tuple[int, int]]] | None:
erasures = (
tuple(index for index, value in enumerate(reversed(body)) if value < 0)
[index for index, value in enumerate(reversed(body)) if value < 0]
if erasure_indices is None
else erasure_indices
else list(erasure_indices)
)
result = (
_bch_syndrome_corrections(
spec,
list(erasures),
(
_aligned_syndromes(alignment, body, len(spec.roots))
if packed_syndromes is None
else [(packed_syndromes >> (10 * i)) & 1023 for i in range(len(spec.roots))]
),
len(body),
max_substitutions,
erasure_state,
erasure_products,
if len(erasures) <= len(spec.roots):
syndromes = (
_aligned_syndromes(alignment, body, len(spec.roots))
if packed_syndromes is None
else [(packed_syndromes >> (10 * i)) & 1023 for i in range(len(spec.roots))]
)
if len(erasures) <= len(spec.roots)
else None
)
if result is None and max_substitutions is not None:
return None
zeroed = [max(value, 0) for value in body]
residue: list[int] | None = None
if result is not None:
residue = _residue(spec, hrp, zeroed)
if not _corrections_reach_target(spec, residue, result):
result = None
if result is None and max_substitutions is None:
residue = _residue(spec, hrp, zeroed) if residue is None else residue
result = _linear_error_corrections(spec, list(erasures), residue)
if result is not None and not _corrections_reach_target(spec, residue, result):
result = None
return None if result is None else (zeroed, result)
result = _bch_syndrome_corrections(
spec, erasures, syndromes, len(body), max_substitutions, erasure_state
)
elif max_substitutions is None:
# More erasures than BCH roots: solve the residue directly for erasures only.
result = _linear_error_corrections(
spec, erasures, _residue(spec, hrp, [max(value, 0) for value in body])
)
else:
result = None
return None if result is None else ([max(value, 0) for value in body], result)


def _capture_volume(mutable_symbols: int, erasures: int, substitutions: int) -> int:
Expand All @@ -520,7 +496,6 @@ class _FixedCorrector:
__slots__ = (
"alignment",
"erasure_indices",
"erasure_products",
"erasure_state",
"hrp",
"max_substitutions",
Expand Down Expand Up @@ -551,7 +526,6 @@ def __init__(
self.max_substitutions = max_substitutions
self.mutable_start = mutable_start
self.erasure_indices: tuple[int, ...] | None = None
self.erasure_products: tuple[tuple[int, ...], ...] | None = None
self.erasure_state: tuple[tuple[int, ...], list[int]] | None = None

def correct(
Expand All @@ -569,9 +543,6 @@ def correct(
return None
self.erasure_indices = tuple(erasure_indices)
self.erasure_state = _erasure_state(self.spec, len(values), self.erasure_indices)
# Alignment locations change frequently. Building 2048 products per
# pair costs more than the handful of GF(1024) multiplies per call.
self.erasure_products = None
repair = _repair_body(
self.spec,
self.hrp,
Expand All @@ -580,7 +551,6 @@ def correct(
self.alignment,
erasure_indices,
self.erasure_state if erasure_indices is not None else None,
self.erasure_products if erasure_indices is not None else None,
packed_syndromes,
)
if repair is None:
Expand Down Expand Up @@ -674,45 +644,6 @@ def _linear_error_corrections(
return None if solution is None else list(zip(erasure_indices, solution))


def _error_corrections(
spec: _Spec,
erasure_indices: list[int],
residue: list[int],
word_length: int | None = None,
max_substitutions: int | None = None,
) -> list[tuple[int, int]] | None:
length = spec.period if word_length is None else word_length
bch = None
if len(erasure_indices) <= len(spec.roots):
bch = _bch_syndrome_corrections(
spec,
erasure_indices,
list(_syndromes(spec, residue, target=True)),
length,
max_substitutions,
)
if bch is not None and _corrections_reach_target(spec, residue, bch):
return bch
if max_substitutions is not None and max_substitutions > 0:
return None
if len(erasure_indices) > len(spec.generator):
return None
linear = _linear_error_corrections(spec, erasure_indices, residue)
return linear if linear is not None and _corrections_reach_target(spec, residue, linear) else None


def _corrections_reach_target(spec: _Spec, residue: list[int], corrections: list[tuple[int, int]]) -> bool:
count = max((index for index, _addend in corrections), default=-1) + 1
powers = _poly_powers(spec.generator, count)
corrected = list(residue)
for reverse_index, addend in corrections:
corrected = _poly_sum(
corrected,
[_gf32_multiply(addend, value) for value in powers[reverse_index]],
)
return corrected == list(reversed(spec.target))


def _correct_fixed(
damaged_text: str,
*,
Expand All @@ -725,8 +656,6 @@ def _correct_fixed(
hrp = suspected_profile.value if isinstance(suspected_profile, Profile) else suspected_profile.lower()
try:
uppercase = _validate_single_case_ascii(damaged_text)
except TypeError:
raise
except CodexError:
return None
prefix = f"{hrp}1"
Expand Down Expand Up @@ -910,8 +839,6 @@ def correct_worksheet_residue(
try:
_validate_single_case_ascii(residue)
values = list(reversed(_chars_to_u5(residue)))
except TypeError:
raise
except CodexError as error:
raise InvalidCorrectionInput(str(error)) from error
if len(values) == len(_SHORT_SPEC.generator):
Expand All @@ -931,7 +858,11 @@ def correct_worksheet_residue(
raise InvalidCorrectionInput("erasure indices must be distinct")
if any(index < 0 or index >= spec.period for index in indices):
raise InvalidCorrectionInput(f"erasure indices must be between 0 and {spec.period - 1}")
result = _error_corrections(spec, indices, values)
if len(indices) <= len(spec.roots):
syndromes = list(_syndromes(spec, values, target=True))
result = _bch_syndrome_corrections(spec, indices, syndromes, spec.period, None)
else:
result = _linear_error_corrections(spec, indices, values)
if result is None:
return None
return tuple(WorksheetCorrection(index, CHARSET[addend]) for index, addend in sorted(result))
Expand Down
Loading