adapter.py was a 3,493-line monolith. Split it into focused modules with clear separation of responsibilities, bringing it down to ~857 lines: - Module-level helpers: hooks, classify, pickers, commands, setup, defaults, secrets - Frame-handler mixins: inbound, tool_frames, push_frames, media_frames, picker_frames, channel_frames, query_frames - mixin_base: IrisAdapterBase (declaration-only base for shared attrs) - adapter.py now holds only IrisAdapter (the composition of the 7 mixins + BasePlatformAdapter), register(), and test-facing re-exports The mixins come before BasePlatformAdapter in the MRO so their methods override the base; super() calls (e.g. send_image) still resolve to BasePlatformAdapter. No circular imports; dispatch.py and http_server.py (instance-method callers) are unaffected. Ruff complexity ceilings (PLR0911/0912/0913/0915) restored to Ruff's built-in defaults (12/50/6/5) instead of "just above the current maxima", which ratchets the bar down as code grows. The existing genuinely-complex functions (frame builders mirroring the wire schema, the QR matrix builder, the dispatch table) carry an explicit `# noqa: PLR09xx` marking them as reviewed, frozen exceptions; new code is held to the default ceilings. All 125 tests green (94 test_android + 31 test_android_http); no new ruff errors introduced.
495 lines
16 KiB
Python
495 lines
16 KiB
Python
"""Pure-stdlib QR encoder (ISO/IEC 18004) + terminal renderer.
|
||
|
||
Scope is deliberately minimal — we only ever encode ASCII pairing URLs
|
||
(``iris://pair?...``):
|
||
|
||
- **Mode:** byte mode only (no alphanumeric/numeric/kanji paths).
|
||
- **Error correction:** level **M** (15 %); auto-fallback to **L** if the
|
||
payload doesn't fit at M within the version cap.
|
||
- **Versions:** 1–10, auto-selected (smallest version whose capacity fits).
|
||
Payloads that don't fit v10-L raise :class:`QrTooLongError`.
|
||
|
||
No third-party imports (no ``qrcode``/``segno``/``Pillow``) — the plugin's
|
||
zero-new-dep rule. No I/O, no module-level mutable state, fully unit-testable.
|
||
|
||
Public API:
|
||
|
||
- :func:`qr_matrix` — encode *data* (ASCII) into a module matrix
|
||
(``True`` = dark) including the 4-module quiet zone.
|
||
- :func:`render_qr` — render *data* as a terminal QR using Unicode
|
||
half-blocks; returns ``""`` (not an exception) when the payload is too long.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
__all__ = ["QrTooLongError", "qr_matrix", "render_qr"]
|
||
|
||
|
||
class QrTooLongError(ValueError):
|
||
"""Raised when *data* doesn't fit in any supported version (1–10)."""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GF(256) arithmetic (polynomial 0x11D)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_GF_EXP = [0] * 512
|
||
_GF_LOG = [0] * 256
|
||
_x = 1
|
||
for _i in range(255):
|
||
_GF_EXP[_i] = _x
|
||
_GF_LOG[_x] = _i
|
||
_x <<= 1
|
||
if _x & 0x100:
|
||
_x ^= 0x11D
|
||
for _i in range(255, 512):
|
||
_GF_EXP[_i] = _GF_EXP[_i - 255]
|
||
|
||
|
||
def _gf_mul(a: int, b: int) -> int:
|
||
if a == 0 or b == 0:
|
||
return 0
|
||
return _GF_EXP[_GF_LOG[a] + _GF_LOG[b]]
|
||
|
||
|
||
def _rs_generator_poly(degree: int) -> list[int]:
|
||
"""Generator polynomial of *degree* (big-endian, leading coeff first)."""
|
||
poly = [1]
|
||
for i in range(degree):
|
||
new = [0] * (len(poly) + 1)
|
||
for k, coef in enumerate(poly):
|
||
new[k] ^= coef # x * coef
|
||
new[k + 1] ^= _gf_mul(coef, _GF_EXP[i])
|
||
poly = new
|
||
return poly
|
||
|
||
|
||
def _rs_encode(data: list[int], ec_len: int) -> list[int]:
|
||
"""Reed–Solomon error-correction codewords for *data*."""
|
||
gen = _rs_generator_poly(ec_len)
|
||
buf = list(data) + [0] * ec_len
|
||
for i in range(len(data)):
|
||
coef = buf[i]
|
||
if coef:
|
||
for j in range(1, len(gen)):
|
||
buf[i + j] ^= _gf_mul(gen[j], coef)
|
||
return buf[len(data) :]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Block structure (version, EC level) -> (ec_per_block, [(count, data_cw), ...])
|
||
#
|
||
# Source: ISO/IEC 18004 Table 9 (cross-checked against the reference encoder).
|
||
# Only levels L and M are needed (M primary, L fallback).
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_BLOCK_TABLE: dict[tuple[int, str], tuple[int, list[tuple[int, int]]]] = {
|
||
(1, "L"): (7, [(1, 19)]),
|
||
(1, "M"): (10, [(1, 16)]),
|
||
(2, "L"): (10, [(1, 34)]),
|
||
(2, "M"): (16, [(1, 28)]),
|
||
(3, "L"): (15, [(1, 55)]),
|
||
(3, "M"): (26, [(1, 44)]),
|
||
(4, "L"): (20, [(1, 80)]),
|
||
(4, "M"): (18, [(2, 32)]),
|
||
(5, "L"): (26, [(1, 108)]),
|
||
(5, "M"): (24, [(2, 43)]),
|
||
(6, "L"): (18, [(2, 68)]),
|
||
(6, "M"): (16, [(4, 27)]),
|
||
(7, "L"): (20, [(2, 78)]),
|
||
(7, "M"): (18, [(4, 31)]),
|
||
(8, "L"): (24, [(2, 97)]),
|
||
(8, "M"): (22, [(2, 38), (2, 39)]),
|
||
(9, "L"): (30, [(2, 116)]),
|
||
(9, "M"): (22, [(3, 36), (2, 37)]),
|
||
(10, "L"): (18, [(2, 68), (2, 69)]),
|
||
(10, "M"): (26, [(4, 43), (1, 44)]),
|
||
}
|
||
|
||
# Alignment-pattern centre coordinates per version (v1 has none).
|
||
_ALIGNMENT: dict[int, list[int]] = {
|
||
1: [],
|
||
2: [6, 18],
|
||
3: [6, 22],
|
||
4: [6, 26],
|
||
5: [6, 30],
|
||
6: [6, 34],
|
||
7: [6, 22, 38],
|
||
8: [6, 24, 42],
|
||
9: [6, 26, 46],
|
||
10: [6, 28, 50],
|
||
}
|
||
|
||
# EC level -> 2-bit format-info code (ISO/IEC 18004 Table 17).
|
||
_EC_FORMAT_BITS = {"L": 0b01, "M": 0b00}
|
||
|
||
_MIN_VERSION, _MAX_VERSION = 1, 10
|
||
_QUIET = 4
|
||
|
||
|
||
def _data_capacity(version: int, level: str) -> int:
|
||
"""Max payload bytes in byte mode for (version, level)."""
|
||
_, groups = _BLOCK_TABLE[(version, level)]
|
||
data_bits = sum(count * data_cw for count, data_cw in groups) * 8
|
||
# mode indicator (4) + char count (8 for v1-9, 16 for v10) + terminator (4)
|
||
count_bits = 16 if version >= 10 else 8
|
||
return (data_bits - 4 - count_bits - 4) // 8
|
||
|
||
|
||
def _select_version(data: bytes) -> tuple[int, str]:
|
||
for level in ("M", "L"):
|
||
for version in range(_MIN_VERSION, _MAX_VERSION + 1):
|
||
if len(data) <= _data_capacity(version, level):
|
||
return version, level
|
||
raise QrTooLongError(f"payload of {len(data)} bytes exceeds v{_MAX_VERSION}-L capacity")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Data encoding (byte mode)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _encode_data(data: bytes, version: int, level: str) -> list[int]:
|
||
"""Return the full codeword stream (data + EC), interleaved per spec."""
|
||
_, groups = _BLOCK_TABLE[(version, level)]
|
||
ec_per_block = _BLOCK_TABLE[(version, level)][0]
|
||
total_data_cw = sum(count * data_cw for count, data_cw in groups)
|
||
|
||
bits: list[int] = []
|
||
|
||
def put(value: int, width: int) -> None:
|
||
for i in range(width - 1, -1, -1):
|
||
bits.append((value >> i) & 1)
|
||
|
||
put(0b0100, 4) # byte mode
|
||
put(len(data), 16 if version >= 10 else 8) # char count
|
||
for byte in data:
|
||
put(byte, 8)
|
||
# terminator (up to 4 zero bits)
|
||
capacity_bits = total_data_cw * 8
|
||
put(0, min(4, capacity_bits - len(bits)))
|
||
# pad to byte boundary
|
||
if len(bits) % 8:
|
||
put(0, 8 - len(bits) % 8)
|
||
# pad bytes 0xEC / 0x11
|
||
pad_bytes = [0xEC, 0x11]
|
||
pi = 0
|
||
while len(bits) < capacity_bits:
|
||
put(pad_bytes[pi % 2], 8)
|
||
pi += 1
|
||
|
||
data_cw = [int("".join(map(str, bits[i : i + 8])), 2) for i in range(0, len(bits), 8)]
|
||
|
||
# Split into blocks, compute EC per block.
|
||
blocks: list[list[int]] = []
|
||
ec_blocks: list[list[int]] = []
|
||
idx = 0
|
||
for count, data_cw_len in groups:
|
||
for _ in range(count):
|
||
block = data_cw[idx : idx + data_cw_len]
|
||
idx += data_cw_len
|
||
blocks.append(block)
|
||
ec_blocks.append(_rs_encode(block, ec_per_block))
|
||
|
||
# Interleave data codewords, then EC codewords (ISO/IEC 18004 §8.6.3).
|
||
out: list[int] = []
|
||
max_data = max(len(b) for b in blocks)
|
||
for i in range(max_data):
|
||
for b in blocks:
|
||
if i < len(b):
|
||
out.append(b[i])
|
||
max_ec = max(len(b) for b in ec_blocks)
|
||
for i in range(max_ec):
|
||
for b in ec_blocks:
|
||
if i < len(b):
|
||
out.append(b[i])
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Matrix construction
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _bch(data: int, shift: int, generator: int) -> int:
|
||
"""BCH codeword: *data* shifted left by *shift*, the low *shift* bits
|
||
filled with the remainder of the division by *generator*."""
|
||
d = data << shift
|
||
g_len = generator.bit_length()
|
||
while d.bit_length() >= g_len:
|
||
d ^= generator << (d.bit_length() - g_len)
|
||
return (data << shift) | d
|
||
|
||
|
||
def _format_info(level: str, mask: int) -> int:
|
||
"""15-bit format info (BCH(15,5)) XORed with 0x5412."""
|
||
data = (_EC_FORMAT_BITS[level] << 3) | mask
|
||
return _bch(data, 10, 0x537) ^ 0x5412
|
||
|
||
|
||
def _version_info(version: int) -> int:
|
||
"""18-bit version info (BCH(18,6)); only for v7+."""
|
||
return _bch(version, 12, 0x1F25)
|
||
|
||
|
||
def _build_matrix(version: int, level: str, codewords: list[int], mask: int) -> list[list[bool]]: # noqa: PLR0912,PLR0915
|
||
size = 17 + 4 * version
|
||
# matrix[r][c] = dark; reserved[r][c] = function module (not data)
|
||
matrix = [[False] * size for _ in range(size)]
|
||
reserved = [[False] * size for _ in range(size)]
|
||
|
||
def set_module(r: int, c: int, dark: bool) -> None:
|
||
matrix[r][c] = dark
|
||
reserved[r][c] = True
|
||
|
||
# Finder patterns + separators (three corners).
|
||
for fr, fc in ((0, 0), (0, size - 7), (size - 7, 0)):
|
||
for r in range(-1, 8):
|
||
for c in range(-1, 8):
|
||
rr, cc = fr + r, fc + c
|
||
if not (0 <= rr < size and 0 <= cc < size):
|
||
continue
|
||
if 0 <= r <= 6 and 0 <= c <= 6:
|
||
# Canonical finder: 7x7 border dark, 5x5 white, 3x3 dark centre.
|
||
ring = max(abs(r - 3), abs(c - 3))
|
||
set_module(rr, cc, ring in (0, 1, 3))
|
||
else:
|
||
set_module(rr, cc, False) # separator
|
||
|
||
# Timing patterns.
|
||
for i in range(8, size - 8):
|
||
dark = i % 2 == 0
|
||
if not reserved[6][i]:
|
||
set_module(6, i, dark)
|
||
if not reserved[i][6]:
|
||
set_module(i, 6, dark)
|
||
|
||
# Alignment patterns (v2+), skipping those overlapping finders.
|
||
positions = _ALIGNMENT[version]
|
||
if len(positions) > 1:
|
||
for r in positions:
|
||
for c in positions:
|
||
# Skip the three corners that share a finder pattern.
|
||
if (
|
||
(r == positions[0] and c == positions[0])
|
||
or (r == positions[0] and c == positions[-1])
|
||
or (r == positions[-1] and c == positions[0])
|
||
):
|
||
continue
|
||
for dr in range(-2, 3):
|
||
for dc in range(-2, 3):
|
||
ring = max(abs(dr), abs(dc))
|
||
dark = ring != 1
|
||
set_module(r + dr, c + dc, dark)
|
||
|
||
# Dark module (always dark) at (4*version + 9, 8).
|
||
set_module(4 * version + 9, 8, True)
|
||
|
||
# Reserve format-info regions (filled after masking).
|
||
for i in range(9):
|
||
if not reserved[8][i]:
|
||
reserved[8][i] = True
|
||
if not reserved[i][8]:
|
||
reserved[i][8] = True
|
||
for i in range(8):
|
||
reserved[8][size - 1 - i] = True
|
||
reserved[size - 1 - i][8] = True
|
||
# (8,8) handled above; mark the remaining format cells.
|
||
reserved[8][8] = True
|
||
|
||
# Reserve version-info regions (v7+).
|
||
if version >= 7:
|
||
vinfo = _version_info(version)
|
||
for i in range(18):
|
||
bit = (vinfo >> i) & 1
|
||
# Two 3x6 blocks: top-left and bottom-right corners.
|
||
r, c = size - 11 + (i % 3), i // 3
|
||
set_module(r, c, bool(bit))
|
||
r, c = i // 3, size - 11 + (i % 3)
|
||
set_module(r, c, bool(bit))
|
||
|
||
# Place data codewords in the zig-zag, applying the mask. Start at the
|
||
# bottom-right and traverse column pairs bottom-to-top, then top-to-bottom.
|
||
bit_index = 0
|
||
total_bits = len(codewords) * 8
|
||
inc = -1
|
||
row = size - 1
|
||
for col in range(size - 1, 0, -2):
|
||
if col <= 6:
|
||
col -= 1 # skip the vertical timing column
|
||
while True:
|
||
for c in (col, col - 1):
|
||
if not reserved[row][c]:
|
||
bit = 0
|
||
if bit_index < total_bits:
|
||
bit = (codewords[bit_index // 8] >> (7 - bit_index % 8)) & 1
|
||
bit_index += 1
|
||
if _mask_bit(mask, row, c):
|
||
bit ^= 1
|
||
matrix[row][c] = bool(bit)
|
||
row += inc
|
||
if row < 0 or row >= size:
|
||
row -= inc
|
||
inc = -inc
|
||
break
|
||
|
||
# Write format info (after masking, unmasked).
|
||
fmt = _format_info(level, mask)
|
||
for i in range(15):
|
||
bit = bool((fmt >> i) & 1)
|
||
# Vertical copy (column 8).
|
||
if i < 6:
|
||
set_module(i, 8, bit)
|
||
elif i < 8:
|
||
set_module(i + 1, 8, bit)
|
||
else:
|
||
set_module(size - 15 + i, 8, bit)
|
||
# Horizontal copy (row 8).
|
||
if i < 8:
|
||
set_module(8, size - i - 1, bit)
|
||
elif i < 9:
|
||
set_module(8, 15 - i, bit)
|
||
else:
|
||
set_module(8, 15 - i - 1, bit)
|
||
|
||
return matrix
|
||
|
||
|
||
def _mask_bit(mask: int, r: int, c: int) -> bool: # noqa: PLR0911
|
||
if mask == 0:
|
||
return (r + c) % 2 == 0
|
||
if mask == 1:
|
||
return r % 2 == 0
|
||
if mask == 2:
|
||
return c % 3 == 0
|
||
if mask == 3:
|
||
return (r + c) % 3 == 0
|
||
if mask == 4:
|
||
return (r // 2 + c // 3) % 2 == 0
|
||
if mask == 5:
|
||
return (r * c) % 2 + (r * c) % 3 == 0
|
||
if mask == 6:
|
||
return ((r * c) % 2 + (r * c) % 3) % 2 == 0
|
||
if mask == 7:
|
||
return ((r + c) % 2 + (r * c) % 3) % 2 == 0
|
||
raise ValueError(f"invalid mask {mask}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Penalty scoring (ISO/IEC 18004 §8.8.2)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _penalty(matrix: list[list[bool]]) -> int:
|
||
size = len(matrix)
|
||
total = 0
|
||
|
||
# N1: runs of >= 5 same-colour in rows and columns.
|
||
for line in _all_lines(matrix):
|
||
run = 1
|
||
for i in range(1, len(line)):
|
||
if line[i] == line[i - 1]:
|
||
run += 1
|
||
else:
|
||
if run >= 5:
|
||
total += 3 + (run - 5)
|
||
run = 1
|
||
if run >= 5:
|
||
total += 3 + (run - 5)
|
||
|
||
# N2: 2x2 blocks of same colour.
|
||
for r in range(size - 1):
|
||
for c in range(size - 1):
|
||
v = matrix[r][c]
|
||
if v == matrix[r][c + 1] == matrix[r + 1][c] == matrix[r + 1][c + 1]:
|
||
total += 3
|
||
|
||
# N3: 10111010000 / 00001011101 patterns (with 4 light on one side).
|
||
pattern_a = [True, False, True, True, True, False, True, False, False, False, False]
|
||
pattern_b = [False, False, False, False, True, False, True, True, True, False, True]
|
||
for line in _all_lines(matrix):
|
||
for i in range(len(line) - 10):
|
||
window = line[i : i + 11]
|
||
if window in (pattern_a, pattern_b):
|
||
total += 40
|
||
|
||
# N4: dark/light balance (integer math: floor(|percent - 50| / 5) * 10).
|
||
dark = sum(cell for line in matrix for cell in line)
|
||
total += 10 * (abs(20 * dark - 10 * size * size) // (5 * size * size))
|
||
|
||
return total
|
||
|
||
|
||
def _all_lines(matrix: list[list[bool]]):
|
||
size = len(matrix)
|
||
for r in range(size):
|
||
yield matrix[r]
|
||
for c in range(size):
|
||
yield [matrix[r][c] for r in range(size)]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Public API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def qr_matrix(data: str) -> list[list[bool]]:
|
||
"""Encode *data* (ASCII) into a module matrix (``True`` = dark).
|
||
|
||
Includes the 4-module quiet zone. Raises :class:`QrTooLongError` when the
|
||
payload doesn't fit in versions 1–10.
|
||
"""
|
||
payload = data.encode("ascii")
|
||
version, level = _select_version(payload)
|
||
codewords = _encode_data(payload, version, level)
|
||
|
||
best = _build_matrix(version, level, codewords, 0)
|
||
best_penalty = _penalty(best)
|
||
for mask in range(1, 8):
|
||
m = _build_matrix(version, level, codewords, mask)
|
||
p = _penalty(m)
|
||
if p < best_penalty:
|
||
best, best_penalty = m, p
|
||
|
||
size = len(best)
|
||
return (
|
||
[[False] * (size + 2 * _QUIET) for _ in range(_QUIET)]
|
||
+ [[False] * _QUIET + row + [False] * _QUIET for row in best]
|
||
+ [[False] * (size + 2 * _QUIET) for _ in range(_QUIET)]
|
||
)
|
||
|
||
|
||
def render_qr(data: str) -> str:
|
||
"""Render *data* as a terminal QR using Unicode half-blocks.
|
||
|
||
Returns ``""`` (not an exception) when the payload is too long. Pair
|
||
consecutive module rows into one character row: both dark → ``█``, top
|
||
dark → ``▀``, bottom dark → ``▄``, both light → space. No ANSI colours or
|
||
cursor tricks — survives ``less``, log files, and copy-paste.
|
||
"""
|
||
try:
|
||
matrix = qr_matrix(data)
|
||
except QrTooLongError:
|
||
return ""
|
||
|
||
height = len(matrix)
|
||
width = len(matrix[0])
|
||
if height % 2:
|
||
matrix = matrix + [[False] * width]
|
||
|
||
lines: list[str] = []
|
||
for r in range(0, len(matrix), 2):
|
||
chars: list[str] = []
|
||
for c in range(width):
|
||
top, bottom = matrix[r][c], matrix[r + 1][c]
|
||
if top and bottom:
|
||
chars.append("█")
|
||
elif top:
|
||
chars.append("▀")
|
||
elif bottom:
|
||
chars.append("▄")
|
||
else:
|
||
chars.append(" ")
|
||
lines.append("".join(chars))
|
||
return "\n".join(lines)
|