afiqzudinhadi/zmk-steno-engine
1
0
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity Actions

Scaffolding: ZMK module structure, Kconfig, DTS, behavior driver

Browse source 
- zephyr/module.yml with dts_root for dt-bindings
- Kconfig: STENO_ENGINE, dict selection (Plover/Lapwing/test),
  MPHF toggle, Unicode modes, history size, multi-stroke timeout
- DTS behavior binding (one_param, steno key index)
- behavior_steno.c: chord accumulation, all-up detection,
  multi-stroke buffering with timeout, star undo, formatter pipeline,
  3-way dict dispatch (split/MPHF/simple trie)
- dict_embed.S: .incbin from generated header path
- steno_keys.h: 23-key layout, bit positions matching compiler
This commit is contained in:
afiqzudinhadi 2026-07-02 02:04:28 +08:00
parent 234bc8d731
commit 1ec204b845
33 changed files with 5443 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

553
tools/compile_mphf.py Executable file
View file

@ -0,0 +1,553 @@
#!/usr/bin/env python3
"""CHD MPHF dictionary compiler for steno engine.
Reads Plover JSON dictionaries and produces a compact binary for
embedded use (nRF52840, 462KB flash budget).
"""
import json
import struct
import math
import argparse
import sys
import os
import zlib
from collections import defaultdict
# ─── Steno stroke parsing ───
STENO_KEYS = {
'#': 0x00400000,
'S-': 0x00000001, 'T-': 0x00000002, 'K-': 0x00000004,
'P-': 0x00000008, 'W-': 0x00000010, 'H-': 0x00000020,
'R-': 0x00000040, 'A-': 0x00000080, 'O-': 0x00000100,
'*': 0x00000200, '-E': 0x00000400, '-U': 0x00000800,
'-F': 0x00001000, '-R': 0x00002000, '-P': 0x00004000,
'-B': 0x00008000, '-L': 0x00010000, '-G': 0x00020000,
'-T': 0x00040000, '-S': 0x00080000, '-D': 0x00100000,
'-Z': 0x00200000,
}
IMPLICIT_HYPHEN = set('AOEU*')
def parse_stroke(s):
result = 0
if '#' in s:
result |= STENO_KEYS['#']
s = s.replace('#', '')
has_hyphen = '-' in s
s_clean = s.replace('-', '')
if not has_hyphen and not any(c in IMPLICIT_HYPHEN for c in s_clean):
for c in s_clean:
key = c + '-'
if key in STENO_KEYS:
result |= STENO_KEYS[key]
return result
if has_hyphen:
hyphen_pos = s.index('-')
for i, c in enumerate(s):
if c == '-':
continue
if c in 'AO':
result |= STENO_KEYS[c + '-']
elif c in 'EU':
result |= STENO_KEYS['-' + c]
elif c == '*':
result |= STENO_KEYS['*']
elif i < hyphen_pos and (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
elif i > hyphen_pos and ('-' + c) in STENO_KEYS:
result |= STENO_KEYS['-' + c]
else:
past_vowels = False
for c in s_clean:
if c in 'AO':
result |= STENO_KEYS[c + '-']
past_vowels = True
elif c in 'EU':
result |= STENO_KEYS['-' + c]
past_vowels = True
elif c == '*':
result |= STENO_KEYS['*']
past_vowels = True
elif not past_vowels and (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
elif past_vowels and ('-' + c) in STENO_KEYS:
result |= STENO_KEYS['-' + c]
elif (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
return result
# ─── Hashing ───
def fnv1a_32(data: bytes) -> int:
"""FNV-1a 32-bit hash."""
h = 0x811c9dc5
for b in data:
h ^= b
h = (h * 0x01000193) & 0xFFFFFFFF
return h
def hash_key(key_bytes: bytes, seed: int) -> int:
"""Hash key with seed by prepending seed bytes."""
return fnv1a_32(struct.pack('<I', seed) + key_bytes)
# ─── Bit packing ───
class BitWriter:
def __init__(self):
self.data = bytearray()
self.bit_pos = 0 # total bits written
def write_bits(self, value, n_bits):
"""Write n_bits of value (LSB first)."""
for i in range(n_bits):
if self.bit_pos % 8 == 0:
self.data.append(0)
if value & (1 << i):
self.data[-1] |= (1 << (self.bit_pos % 8))
self.bit_pos += 1
def pad_to_alignment(self, alignment=4):
"""Pad to byte alignment."""
while len(self.data) % alignment != 0:
self.data.append(0)
self.bit_pos = len(self.data) * 8
def to_bytes(self):
return bytes(self.data)
class BitReader:
def __init__(self, data):
self.data = data
self.bit_pos = 0
def read_bits(self, n_bits):
value = 0
for i in range(n_bits):
byte_idx = self.bit_pos // 8
bit_idx = self.bit_pos % 8
if self.data[byte_idx] & (1 << bit_idx):
value |= (1 << i)
self.bit_pos += 1
return value
# ─── Key encoding ───
def encode_key(stroke_str):
"""Parse stroke string → key_bytes (each stroke as u32 LE, concatenated)."""
parts = stroke_str.split('/')
strokes = tuple(parse_stroke(s) for s in parts)
key_bytes = b''.join(struct.pack('<I', s) for s in strokes)
return strokes, key_bytes
# ─── Importance scoring ───
def score_entry(stroke_str, translation):
"""Lower score = more important = keep first."""
n_strokes = stroke_str.count('/') + 1
has_format = '{' in translation
return (n_strokes, has_format, len(translation), stroke_str)
# ─── CHD MPHF construction ───
def build_chd(keys_and_bytes, entry_count):
"""
Build CHD MPHF.
keys_and_bytes: list of (index, key_bytes) for each entry
entry_count: total number of entries
Returns: (displacements, slot_to_entry_idx, max_displacement)
displacements[bucket] = d value
slot_to_entry_idx[slot] = index into keys_and_bytes, or -1 if empty
"""
bucket_count = max(entry_count // 3, min(entry_count, 16))
# Assign keys to buckets
buckets = defaultdict(list)
for idx, (_, kb) in enumerate(keys_and_bytes):
b = hash_key(kb, 0) % bucket_count
buckets[b].append(idx)
# Sort buckets by size descending
sorted_buckets = sorted(buckets.items(), key=lambda x: len(x[1]), reverse=True)
displacements = [0] * bucket_count
occupied = set()
slot_to_entry = [-1] * entry_count
max_disp = 0
for bucket_id, members in sorted_buckets:
if not members:
continue
member_key_bytes = [(m, keys_and_bytes[m][1]) for m in members]
placed = False
for d in range(65536):
slots = []
collision = False
seen = set()
for _, kb in member_key_bytes:
slot = hash_key(kb, d + 1) % entry_count
if slot in occupied or slot in seen:
collision = True
break
seen.add(slot)
slots.append(slot)
if collision:
continue
# Place all members
for i, (m, _) in enumerate(member_key_bytes):
occupied.add(slots[i])
slot_to_entry[slots[i]] = m
displacements[bucket_id] = d
if d > max_disp:
max_disp = d
placed = True
break
if not placed:
print(f"FATAL: bucket {bucket_id} with {len(members)} keys failed after 65536 tries",
file=sys.stderr)
return None, None, None
return displacements, slot_to_entry, max_disp
# ─── Compilation ───
def compile_mphf(entries, max_size=None):
"""
entries: list of (stroke_str, translation) from JSON dict
max_size: max output size in bytes (default: 462*1024 = 473088)
Returns: bytes (the compiled binary) or None if can't fit
"""
if max_size is None:
max_size = 462 * 1024
# Sort by importance for potential trimming
entries_scored = sorted(entries, key=lambda e: score_entry(e[0], e[1]))
# Parse all keys, dedup by key_bytes (last wins for same key)
seen_keys = {}
for stroke_str, translation in entries_scored:
strokes, key_bytes = encode_key(stroke_str)
if key_bytes in seen_keys:
prev = seen_keys[key_bytes]
print(f" Dedup: '{stroke_str}'→'{translation}' collides with "
f"'{prev[2]}'→'{prev[1]}', keeping first", file=sys.stderr)
continue
entry = (key_bytes, translation, stroke_str, strokes)
seen_keys[key_bytes] = entry
keys_and_bytes = list(seen_keys.values())
# Iteratively trim if needed
while True:
entry_count = len(keys_and_bytes)
if entry_count == 0:
return None
# Build deduped string table
translations = [kb[1] for kb in keys_and_bytes]
unique_translations = sorted(set(translations))
trans_to_id = {t: i for i, t in enumerate(unique_translations)}
unique_count = len(unique_translations)
# Estimate size
bucket_count = max(entry_count // 3, min(entry_count, 16))
est_value_bits = max(1, math.ceil(math.log2(max(unique_count, 2))))
est_disp_bits = 16 # conservative
est_disp_bytes = (bucket_count * est_disp_bits + 7) // 8
est_disp_bytes = ((est_disp_bytes + 3) // 4) * 4
est_value_bytes = (entry_count * est_value_bits + 7) // 8
est_value_bytes = ((est_value_bytes + 3) // 4) * 4
est_fp_bytes = ((entry_count + 3) // 4) * 4
# String table (block-compressed)
string_data_raw = b''
string_offsets = []
for t in unique_translations:
string_offsets.append(len(string_data_raw))
string_data_raw += t.encode('utf-8') + b'\x00'
block_size = 4096
compressed_blocks = []
for i in range(0, len(string_data_raw), block_size):
block = string_data_raw[i:i + block_size]
compressed_blocks.append(zlib.compress(block, 9))
est_str_offsets = unique_count * 3 # u24 packed LE
est_block_dir = 2 + len(compressed_blocks) * 4 # u16 count + u32 offsets
est_str_data = sum(len(b) for b in compressed_blocks) + est_block_dir
# Prefix table
prefix_strokes = set()
for kb, trans, stroke_str, strokes in keys_and_bytes:
if len(strokes) > 1:
prefix_strokes.add(strokes[0])
prefix_list = sorted(prefix_strokes)
est_prefix = len(prefix_list) * 4
total_est = 32 + est_disp_bytes + est_value_bytes + est_fp_bytes + est_str_offsets + est_str_data + est_prefix
if total_est <= max_size:
break
# Trim: use ratio of overshoot to estimate how many entries to cut
overshoot_ratio = total_est / max_size
target_entries = int(entry_count / overshoot_ratio * 0.98) # 2% safety margin
trim_count = max(1, entry_count - target_entries)
keys_and_bytes = keys_and_bytes[:entry_count - trim_count]
print(f" Trimming to {len(keys_and_bytes)} entries (est {total_est} > {max_size})",
file=sys.stderr)
entry_count = len(keys_and_bytes)
bucket_count = max(entry_count // 3, min(entry_count, 16))
print(f" Building CHD MPHF: {entry_count} entries, {bucket_count} buckets...",
file=sys.stderr)
# Build CHD
chd_input = [(i, keys_and_bytes[i][0]) for i in range(entry_count)]
displacements, slot_to_entry, max_disp = build_chd(chd_input, entry_count)
if displacements is None:
return None
# Compute actual bit widths
disp_bits = max(1, math.ceil(math.log2(max(max_disp + 1, 2))))
value_bits = max(1, math.ceil(math.log2(max(unique_count, 2))))
prefix_count = len(prefix_list)
print(f" Max displacement: {max_disp}, disp_bits: {disp_bits}", file=sys.stderr)
print(f" Unique translations: {unique_count}, value_bits: {value_bits}", file=sys.stderr)
print(f" Prefix entries: {prefix_count}", file=sys.stderr)
# ─── Build binary ───
# Displacements section
disp_writer = BitWriter()
for d in displacements:
disp_writer.write_bits(d, disp_bits)
disp_writer.pad_to_alignment(4)
disp_section = disp_writer.to_bytes()
# Values section: slot → value_id
val_writer = BitWriter()
fingerprints = bytearray(entry_count)
for slot in range(entry_count):
entry_idx = slot_to_entry[slot]
if entry_idx >= 0:
kb, trans, stroke_str, strokes = keys_and_bytes[entry_idx]
val_id = trans_to_id[trans]
val_writer.write_bits(val_id, value_bits)
fingerprints[slot] = fnv1a_32(kb) & 0xFF
else:
val_writer.write_bits(0, value_bits)
fingerprints[slot] = 0
val_writer.pad_to_alignment(4)
val_section = val_writer.to_bytes()
# Fingerprints section
fp_section = bytes(fingerprints)
# Pad to 4-byte boundary
while len(fp_section) % 4 != 0:
fp_section += b'\x00'
# String offsets section (u24 packed LE — 3 bytes each, into raw/uncompressed table)
str_offsets_section = b''.join(struct.pack('<I', off)[:3] for off in string_offsets)
# String data section (block-compressed)
block_dir = struct.pack('<H', len(compressed_blocks))
block_offset = 0
for blk in compressed_blocks:
block_dir += struct.pack('<I', block_offset)
block_offset += len(blk)
str_data_section = block_dir + b''.join(compressed_blocks)
# Prefix table section
prefix_section = b''.join(struct.pack('<I', s) for s in prefix_list)
# Header (32 bytes):
# magic: u32, version: u16, flags: u16,
# entry_count: u32, bucket_count: u32, unique_count: u32,
# value_bits: u8, disp_bits: u8, prefix_count: u16,
# reserved0: u32, reserved1: u32
header = struct.pack('<IHHIIIBBHii',
0x4F4E5453, # magic "STNO"
2, # version
0x0001, # flags: bit 0 = block-compressed strings
entry_count, # entry_count
bucket_count, # bucket_count
unique_count, # unique_count
value_bits, # value_bits
disp_bits, # disp_bits
prefix_count, # prefix_count
0, # reserved0
0, # reserved1
)
assert len(header) == 32, f"Header is {len(header)} bytes, expected 32"
binary = header + disp_section + val_section + fp_section + str_offsets_section + str_data_section + prefix_section
# ─── Verification ───
print(f" Verifying all {entry_count} entries...", file=sys.stderr)
errors = 0
for entry_idx in range(entry_count):
kb, trans, stroke_str, strokes = keys_and_bytes[entry_idx]
# Lookup through MPHF
bucket = hash_key(kb, 0) % bucket_count
# Read displacement
disp_reader = BitReader(disp_section)
disp_reader.bit_pos = bucket * disp_bits
d = disp_reader.read_bits(disp_bits)
slot = hash_key(kb, d + 1) % entry_count
# Check fingerprint
expected_fp = fnv1a_32(kb) & 0xFF
if fingerprints[slot] != expected_fp:
print(f" VERIFY FAIL: fingerprint mismatch for '{stroke_str}' at slot {slot}: "
f"got {fingerprints[slot]}, expected {expected_fp}", file=sys.stderr)
errors += 1
continue
# Check value
val_reader = BitReader(val_section)
val_reader.bit_pos = slot * value_bits
val_id = val_reader.read_bits(value_bits)
# Resolve string from compressed table
off_bytes = str_offsets_section[val_id * 3:(val_id + 1) * 3]
str_off = off_bytes[0] | (off_bytes[1] << 8) | (off_bytes[2] << 16)
block_idx = str_off // block_size
in_block_off = str_off % block_size
raw_block = zlib.decompress(compressed_blocks[block_idx])
if b'\x00' in raw_block[in_block_off:]:
end = raw_block.index(b'\x00', in_block_off)
resolved = raw_block[in_block_off:end].decode('utf-8')
elif block_idx + 1 < len(compressed_blocks):
part1 = raw_block[in_block_off:]
next_block = zlib.decompress(compressed_blocks[block_idx + 1])
end = next_block.index(b'\x00')
resolved = (part1 + next_block[:end]).decode('utf-8')
else:
resolved = raw_block[in_block_off:].decode('utf-8')
if resolved != trans:
print(f" VERIFY FAIL: value mismatch for '{stroke_str}': "
f"got '{resolved}', expected '{trans}'", file=sys.stderr)
errors += 1
if errors:
print(f" VERIFICATION FAILED: {errors} errors", file=sys.stderr)
return None
print(f" Verification passed: all {entry_count} entries OK", file=sys.stderr)
# Check final size
if len(binary) > max_size:
print(f" WARNING: output {len(binary)} bytes exceeds max {max_size}", file=sys.stderr)
return binary, {
'entry_count': entry_count,
'bucket_count': bucket_count,
'unique_count': unique_count,
'value_bits': value_bits,
'disp_bits': disp_bits,
'max_displacement': max_disp,
'prefix_count': prefix_count,
'disp_section_bytes': len(disp_section),
'val_section_bytes': len(val_section),
'fp_section_bytes': len(fp_section),
'str_offsets_bytes': len(str_offsets_section),
'str_data_bytes': len(str_data_section),
'str_data_raw_bytes': len(string_data_raw),
'prefix_section_bytes': len(prefix_section),
'total_bytes': len(binary),
}
def print_stats(stats):
"""Print size breakdown statistics."""
total = stats['entry_count']
print(f"Entries: {stats['entry_count']}")
print(f"MPHF displacements: {stats['disp_section_bytes']/1024:.1f} KB "
f"({stats['bucket_count']} buckets, {stats['disp_bits']} bits each)")
print(f"Value array: {stats['val_section_bytes']/1024:.1f} KB "
f"({stats['entry_count']} entries, {stats['value_bits']} bits each)")
print(f"Fingerprints: {stats['fp_section_bytes']/1024:.1f} KB")
print(f"String offsets: {stats['str_offsets_bytes']/1024:.1f} KB "
f"({stats['unique_count']} unique x 3 bytes)")
print(f"String data: {stats['str_data_bytes']/1024:.1f} KB"
f" (compressed, {stats.get('str_data_raw_bytes', 0)/1024:.1f} KB raw)")
print(f"Prefix table: {stats['prefix_section_bytes']/1024:.1f} KB "
f"({stats['prefix_count']} entries x 4 bytes)")
print(f"Total: {stats['total_bytes']/1024:.1f} KB")
def main():
parser = argparse.ArgumentParser(description='Compile steno dictionary to MPHF binary format')
parser.add_argument('input', help='Input JSON dictionary (Plover format)')
parser.add_argument('output', help='Output binary file')
parser.add_argument('--max-size', type=int, default=462*1024,
help='Maximum output size in bytes (default: 473088 = 462KB)')
parser.add_argument('--max-entries', type=int, default=None,
help='Maximum number of entries')
parser.add_argument('--stats', action='store_true',
help='Print size breakdown statistics')
parser.add_argument('--verify', action='store_true', default=True,
help='Verify compiled dict (default: true)')
args = parser.parse_args()
# Load dictionary
with open(args.input) as f:
raw_dict = json.load(f)
print(f"Loaded {len(raw_dict)} entries from {args.input}", file=sys.stderr)
entries = list(raw_dict.items())
if args.max_entries is not None:
entries_scored = sorted(entries, key=lambda e: score_entry(e[0], e[1]))
entries = entries_scored[:args.max_entries]
print(f"Trimmed to {len(entries)} entries (--max-entries)", file=sys.stderr)
result = compile_mphf(entries, max_size=args.max_size)
if result is None:
print("Compilation failed", file=sys.stderr)
sys.exit(1)
binary, stats = result
with open(args.output, 'wb') as f:
f.write(binary)
print(f"Wrote {len(binary)} bytes to {args.output}", file=sys.stderr)
if args.stats:
print()
print_stats(stats)
if __name__ == '__main__':
main()

184
tools/compile_simple.py Normal file
View file

@ -0,0 +1,184 @@
#!/usr/bin/env python3
"""Simple flat-format steno dictionary compiler.
Outputs a binary format optimized for binary search on embedded targets.
No compression — just sorted entries with fixed-width keys.
Format:
Header (16 bytes):
magic: u32 = 0x4F4E5453 ("STNO")
version: u16 = 1
max_strokes: u8 (max stroke count per entry)
pad: u8
entry_count: u32
strings_offset: u32
Entry array (sorted by stroke tuple, fixed width):
Each entry = max_strokes * 4 + 4 bytes:
strokes[max_strokes]: u32 LE (unused slots = 0)
string_offset: u32 LE (into string table)
String table:
Null-terminated UTF-8 strings, concatenated
"""
import argparse
import json
import struct
import sys
import os
STENO_KEYS = {
'#': 0x00400000,
'S-': 0x00000001, 'T-': 0x00000002, 'K-': 0x00000004,
'P-': 0x00000008, 'W-': 0x00000010, 'H-': 0x00000020,
'R-': 0x00000040, 'A-': 0x00000080, 'O-': 0x00000100,
'*': 0x00000200, '-E': 0x00000400, '-U': 0x00000800,
'-F': 0x00001000, '-R': 0x00002000, '-P': 0x00004000,
'-B': 0x00008000, '-L': 0x00010000, '-G': 0x00020000,
'-T': 0x00040000, '-S': 0x00080000, '-D': 0x00100000,
'-Z': 0x00200000,
}
IMPLICIT_HYPHEN = set('AOEU*')
def parse_stroke(s):
result = 0
if '#' in s:
result |= STENO_KEYS['#']
s = s.replace('#', '')
has_hyphen = '-' in s
s_clean = s.replace('-', '')
if not has_hyphen and not any(c in IMPLICIT_HYPHEN for c in s_clean):
for c in s_clean:
key = c + '-'
if key in STENO_KEYS:
result |= STENO_KEYS[key]
return result
if has_hyphen:
hyphen_pos = s.index('-')
for i, c in enumerate(s):
if c == '-':
continue
if c in 'AO':
result |= STENO_KEYS[c + '-']
elif c in 'EU':
result |= STENO_KEYS['-' + c]
elif c == '*':
result |= STENO_KEYS['*']
elif i < hyphen_pos and (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
elif i > hyphen_pos and ('-' + c) in STENO_KEYS:
result |= STENO_KEYS['-' + c]
else:
past_vowels = False
for c in s_clean:
if c in 'AO':
result |= STENO_KEYS[c + '-']
past_vowels = True
elif c in 'EU':
result |= STENO_KEYS['-' + c]
past_vowels = True
elif c == '*':
result |= STENO_KEYS['*']
past_vowels = True
elif not past_vowels and (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
elif past_vowels and ('-' + c) in STENO_KEYS:
result |= STENO_KEYS['-' + c]
elif (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
return result
def compile_dict(json_path, max_entries=None, max_strokes_cap=None):
with open(json_path) as f:
raw = json.load(f)
entries = []
for stroke_str, translation in raw.items():
strokes = tuple(parse_stroke(s) for s in stroke_str.split('/'))
if max_strokes_cap and len(strokes) > max_strokes_cap:
continue
entries.append((strokes, translation))
if max_entries and len(entries) > max_entries:
single = [(s, t) for s, t in entries if len(s) == 1]
multi = [(s, t) for s, t in entries if len(s) > 1]
multi.sort(key=lambda x: (len(x[1]), len(x[0])))
remaining = max_entries - len(single)
if remaining < 0:
single.sort(key=lambda x: len(x[1]))
entries = single[:max_entries]
else:
entries = single + multi[:remaining]
entries.sort(key=lambda x: x[0])
max_strokes = max(len(s) for s, _ in entries)
string_table = bytearray()
string_offsets = {}
for _, translation in entries:
if translation not in string_offsets:
string_offsets[translation] = len(string_table)
string_table.extend(translation.encode('utf-8'))
string_table.append(0)
entry_size = max_strokes * 4 + 4
header_size = 16
entries_size = len(entries) * entry_size
strings_offset = header_size + entries_size
header = struct.pack('<IHBBII',
0x4F4E5453, # "STNO"
1, # version
max_strokes,
0, # pad
len(entries),
strings_offset)
entry_data = bytearray()
for strokes, translation in entries:
padded = list(strokes) + [0] * (max_strokes - len(strokes))
for s in padded:
entry_data.extend(struct.pack('<I', s))
entry_data.extend(struct.pack('<I', string_offsets[translation]))
binary = header + bytes(entry_data) + bytes(string_table)
return binary, len(entries), max_strokes, len(string_table)
def main():
parser = argparse.ArgumentParser(description='Compile steno dict to flat binary')
parser.add_argument('input', help='JSON dictionary path')
parser.add_argument('-o', '--output', default='steno_dict.bin')
parser.add_argument('--max-entries', type=int, default=None)
parser.add_argument('--max-strokes', type=int, default=None,
help='Cap max stroke count per entry (drop longer)')
parser.add_argument('--stats', action='store_true')
args = parser.parse_args()
if not os.path.exists(args.input):
print(f"ERROR: {args.input} not found")
sys.exit(1)
binary, n_entries, max_strokes, str_size = compile_dict(
args.input, args.max_entries, args.max_strokes)
with open(args.output, 'wb') as f:
f.write(binary)
if args.stats:
entry_size = max_strokes * 4 + 4
print(f"Entries: {n_entries}")
print(f"Max strokes: {max_strokes}")
print(f"Entry size: {entry_size} bytes")
print(f"Entry array: {n_entries * entry_size} bytes")
print(f"String table: {str_size} bytes")
print(f"Total: {len(binary)} bytes ({len(binary)/1024:.1f} KB)")
print(f"Written {len(binary)} bytes to {args.output}")
if __name__ == '__main__':
main()

815
tools/dict_compiler.py Normal file
View file

@ -0,0 +1,815 @@
#!/usr/bin/env python3
"""DAWG dictionary compiler for ZMK steno engine.
Compiles a Plover-format JSON steno dictionary into a compact binary
DAWG with skip-count indexing and block-compressed string table.
Optimize-dict variant targeting 462KB (left half flash budget).
"""
import argparse
import json
import math
import os
import struct
import sys
import zlib
from collections import Counter
# ─── Steno stroke parsing (copied from dawg_fst_prototype.py) ───
STENO_KEYS = {
'#': 0x00400000,
'S-': 0x00000001, 'T-': 0x00000002, 'K-': 0x00000004,
'P-': 0x00000008, 'W-': 0x00000010, 'H-': 0x00000020,
'R-': 0x00000040, 'A-': 0x00000080, 'O-': 0x00000100,
'*': 0x00000200, '-E': 0x00000400, '-U': 0x00000800,
'-F': 0x00001000, '-R': 0x00002000, '-P': 0x00004000,
'-B': 0x00008000, '-L': 0x00010000, '-G': 0x00020000,
'-T': 0x00040000, '-S': 0x00080000, '-D': 0x00100000,
'-Z': 0x00200000,
}
IMPLICIT_HYPHEN = set('AOEU*')
def parse_stroke(s):
"""Parse a steno stroke string into a bitmask."""
result = 0
if '#' in s:
result |= STENO_KEYS['#']
s = s.replace('#', '')
has_hyphen = '-' in s
s_clean = s.replace('-', '')
if not has_hyphen and not any(c in IMPLICIT_HYPHEN for c in s_clean):
for c in s_clean:
key = c + '-'
if key in STENO_KEYS:
result |= STENO_KEYS[key]
return result
past_vowels = False
for c in s_clean:
if c in 'AO':
result |= STENO_KEYS[c + '-']
past_vowels = True
elif c in 'EU':
result |= STENO_KEYS['-' + c]
past_vowels = True
elif c == '*':
result |= STENO_KEYS['*']
past_vowels = True
elif not past_vowels and (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
elif past_vowels and ('-' + c) in STENO_KEYS:
result |= STENO_KEYS['-' + c]
elif has_hyphen:
if s.index(c) < s.index('-'):
result |= STENO_KEYS.get(c + '-', 0)
else:
result |= STENO_KEYS.get('-' + c, 0)
else:
if (c + '-') in STENO_KEYS:
result |= STENO_KEYS[c + '-']
return result
def parse_stroke_string(stroke_str):
"""Parse a stroke string (possibly multi-stroke with /) into tuple of bitmasks."""
return tuple(parse_stroke(s) for s in stroke_str.split('/'))
# ─── DAWG construction (Daciuk's incremental algorithm) ───
class DawgNode:
"""Node in the DAWG."""
__slots__ = ['id', 'edges', 'final', '_hash_cache']
_next_id = 0
def __init__(self):
self.id = DawgNode._next_id
DawgNode._next_id += 1
self.edges = {} # stroke_val -> DawgNode
self.final = False
self._hash_cache = None
def signature(self):
"""Hashable signature for minimization."""
edge_sig = tuple(sorted(
(k, child.id) for k, child in self.edges.items()
))
return (self.final, edge_sig)
def __hash__(self):
if self._hash_cache is None:
self._hash_cache = hash(self.signature())
return self._hash_cache
def __eq__(self, other):
return self.signature() == other.signature()
def invalidate_cache(self):
self._hash_cache = None
def build_dawg(sorted_entries):
"""Build minimized DAWG using Daciuk's incremental algorithm.
Entries MUST be sorted by stroke tuple (lexicographic).
Returns (root, node_count, edge_count).
"""
DawgNode._next_id = 0
root = DawgNode()
unchecked = [] # list of (parent, stroke, child)
minimized = {} # signature -> node
prev_strokes = ()
def _minimize(down_to):
"""Minimize unchecked nodes from top down to given depth."""
for i in range(len(unchecked) - 1, down_to - 1, -1):
parent, stroke, child = unchecked[i]
child.invalidate_cache()
sig = child.signature()
if sig in minimized:
parent.edges[stroke] = minimized[sig]
else:
minimized[sig] = child
unchecked.pop()
for strokes, _translation in sorted_entries:
# Find common prefix length with previous entry
common = 0
limit = min(len(strokes), len(prev_strokes))
while common < limit and strokes[common] == prev_strokes[common]:
common += 1
# Minimize nodes beyond common prefix
_minimize(common)
# Get node at end of common prefix
if unchecked:
node = unchecked[-1][2]
else:
node = root
# Add new nodes for remaining strokes
for stroke in strokes[common:]:
new_node = DawgNode()
node.edges[stroke] = new_node
unchecked.append((node, stroke, new_node))
node = new_node
node.final = True
prev_strokes = strokes
# Minimize remaining
_minimize(0)
# Count nodes and edges
node_count = 0
edge_count = 0
visited = set()
def _count(n):
nonlocal node_count, edge_count
if n.id in visited:
return
visited.add(n.id)
node_count += 1
for _s, child in n.edges.items():
edge_count += 1
_count(child)
_count(root)
return root, node_count, edge_count
# ─── Skip-count computation ───
def compute_skip_counts(root):
"""Compute skip-count (number of final nodes in subtree) for each node.
Returns dict: node_id -> skip_count
"""
cache = {}
def _count(node):
if node.id in cache:
return cache[node.id]
c = 1 if node.final else 0
for stroke in sorted(node.edges.keys()):
child = node.edges[stroke]
c += _count(child)
cache[node.id] = c
return c
_count(root)
return cache
def dawg_lookup_index(root, strokes, skip_cache):
"""Look up a stroke sequence in the DAWG, returning its skip-count index.
Returns -1 if not found.
"""
node = root
idx = 0
for stroke in strokes:
if stroke not in node.edges:
return -1
# Count finals of all edges with stroke < target
for s in sorted(node.edges.keys()):
if s == stroke:
child = node.edges[s]
if child.final:
idx += 1
node = child
break
else:
child = node.edges[s]
idx += skip_cache[child.id]
else:
return -1
if not node.final:
return -1
return idx - 1
def get_dawg_traversal_order(root):
"""DFS traversal of DAWG, edges sorted by stroke value.
Returns list of translations in traversal order (one per final node encounter).
This is the order entries appear when looking up via skip-count.
"""
order = []
visited_paths = set()
def _dfs(node, path):
path_key = tuple(path)
if path_key in visited_paths:
return
visited_paths.add(path_key)
if node.final:
order.append(path_key)
for stroke in sorted(node.edges.keys()):
child = node.edges[stroke]
_dfs(child, path + [stroke])
_dfs(root, [])
return order
# ─── Entry trimming ───
def trim_entries(entries, max_entries):
"""Trim entries to max_entries, keeping single-stroke preferentially.
Priority:
1. All single-stroke entries
2. Multi-stroke entries sorted by translation length (shorter first)
3. Drop longest/rarest multi-stroke first
"""
if len(entries) <= max_entries:
return entries
single_stroke = []
multi_stroke = []
for strokes_str, translation in entries:
if '/' not in strokes_str:
single_stroke.append((strokes_str, translation))
else:
multi_stroke.append((strokes_str, translation))
# Sort multi-stroke by translation length (shorter = more useful)
multi_stroke.sort(key=lambda x: (len(x[1]), len(x[0].split('/'))))
remaining = max_entries - len(single_stroke)
if remaining < 0:
# Even single-stroke entries exceed limit; trim by translation length
single_stroke.sort(key=lambda x: len(x[1]))
return single_stroke[:max_entries]
return single_stroke + multi_stroke[:remaining]
# ─── String table construction ───
def build_string_table(translations):
"""Build block-compressed string table from translations.
Returns (table_bytes, offsets) where offsets[i] is the byte offset
of translation i in the uncompressed table.
"""
# Build raw table: null-separated strings
# We need to track offset of each unique string
unique_translations = sorted(set(translations))
trans_to_unique_idx = {t: i for i, t in enumerate(unique_translations)}
# Build raw bytes and offset map
raw_parts = []
unique_offsets = []
offset = 0
for t in unique_translations:
unique_offsets.append(offset)
encoded = t.encode('utf-8')
raw_parts.append(encoded)
offset += len(encoded) + 1 # +1 for null separator
raw = b'\x00'.join(raw_parts)
if raw_parts:
raw += b'\x00' # trailing null
# Block compress
block_size = 4096
compressed_blocks = []
block_offsets_raw = []
current_offset = 0
for i in range(0, len(raw), block_size):
block = raw[i:i + block_size]
compressed = zlib.compress(block, 9)
block_offsets_raw.append(current_offset)
compressed_blocks.append(compressed)
current_offset += len(compressed)
# Serialize: block_count(u16) + block_offsets(u32 each) + compressed blocks
n_blocks = len(compressed_blocks)
table_header = struct.pack('<H', n_blocks)
table_index = b''.join(struct.pack('<I', off) for off in block_offsets_raw)
table_data = b''.join(compressed_blocks)
table_bytes = table_header + table_index + table_data
# Map each translation to its offset in raw table
entry_offsets = []
for t in translations:
uid = trans_to_unique_idx[t]
entry_offsets.append(unique_offsets[uid])
return table_bytes, entry_offsets, len(raw)
def decompress_string_table(table_bytes):
"""Decompress a block-compressed string table back to raw bytes."""
pos = 0
n_blocks = struct.unpack_from('<H', table_bytes, pos)[0]
pos += 2
block_offsets = []
for _ in range(n_blocks):
off = struct.unpack_from('<I', table_bytes, pos)[0]
pos += 4
block_offsets.append(off)
data_start = pos
raw_parts = []
for i in range(n_blocks):
block_start = data_start + block_offsets[i]
if i + 1 < n_blocks:
block_end = data_start + block_offsets[i + 1]
else:
block_end = len(table_bytes)
compressed = table_bytes[block_start:block_end]
raw_parts.append(zlib.decompress(compressed))
return b''.join(raw_parts)
def lookup_string(raw_table, offset):
"""Look up a null-terminated string at given offset in raw table."""
end = raw_table.index(b'\x00', offset)
return raw_table[offset:end].decode('utf-8')
# ─── Binary serialization ───
MAGIC = b'STNO'
VERSION = 1
FLAG_SPLIT_STORAGE = 0x0001
HEADER_SIZE = 32
def serialize_header(flags, entry_count, node_count, edge_count,
string_table_offset, string_table_size,
value_array_offset):
"""Serialize the 32-byte binary header."""
return struct.pack('<4sHHIIIIIIxxxx',
MAGIC,
VERSION,
flags,
entry_count,
node_count,
edge_count,
string_table_offset,
string_table_size,
value_array_offset)
def parse_header(data):
"""Parse 32-byte binary header. Returns dict."""
if len(data) < HEADER_SIZE:
raise ValueError("Data too short for header")
# Unpack with padding bytes
magic, version, flags, entry_count, node_count, edge_count, \
str_table_off, str_table_size, val_array_off = \
struct.unpack_from('<4sHHIIIIII', data, 0)
# 4 bytes reserved at end (32 - 28 = 4)
if magic != MAGIC:
raise ValueError(f"Bad magic: {magic!r}")
return {
'magic': magic,
'version': version,
'flags': flags,
'entry_count': entry_count,
'node_count': node_count,
'edge_count': edge_count,
'string_table_offset': str_table_off,
'string_table_size': str_table_size,
'value_array_offset': val_array_off,
}
def serialize_edges(root, node_count, edge_count):
"""Serialize DAWG edges as bit-packed array.
Each edge: stroke_key(16) + target_node(16) + skip_count(17) + is_last(1) = 50 bits
Nodes are assigned sequential IDs via DFS traversal (sorted edges).
Returns (edge_bytes, node_id_map, skip_counts_by_node).
"""
# Assign sequential node IDs via DFS
node_id_map = {}
dfs_order = []
def _assign_ids(node):
if node.id in node_id_map:
return
new_id = len(node_id_map)
node_id_map[node.id] = new_id
dfs_order.append(node)
for stroke in sorted(node.edges.keys()):
child = node.edges[stroke]
_assign_ids(child)
_assign_ids(root)
# Compute skip counts
skip_cache = compute_skip_counts(root)
# Build edge list: for each node in DFS order, emit edges sorted by stroke
edges = []
for node in dfs_order:
sorted_strokes = sorted(node.edges.keys())
for i, stroke in enumerate(sorted_strokes):
child = node.edges[stroke]
is_last = (i == len(sorted_strokes) - 1)
target_id = node_id_map[child.id]
skip = skip_cache[child.id]
edges.append((stroke, target_id, skip, is_last))
# Bit-pack edges: each 50 bits
# stroke_key: 16 bits, target_node: 16 bits, skip_count: 17 bits, is_last: 1 bit
total_bits = len(edges) * 50
total_bytes = (total_bits + 7) // 8
buf = bytearray(total_bytes)
bit_pos = 0
for stroke_key, target_node, skip_count, is_last in edges:
# Clamp values to field widths
stroke_key &= 0xFFFF
target_node &= 0xFFFF
skip_count = min(skip_count, 0x1FFFF) # 17 bits max
is_last_bit = 1 if is_last else 0
# Pack 50 bits: stroke(16) | target(16) | skip(17) | last(1)
val = (stroke_key << 34) | (target_node << 18) | (skip_count << 1) | is_last_bit
# Write 50 bits into buffer at bit_pos
for i in range(50):
bit = (val >> (49 - i)) & 1
byte_idx = (bit_pos + i) // 8
bit_idx = 7 - ((bit_pos + i) % 8)
if bit:
buf[byte_idx] |= (1 << bit_idx)
bit_pos += 50
return bytes(buf), node_id_map, skip_cache
def deserialize_edges(edge_bytes, edge_count):
"""Deserialize bit-packed edge array.
Returns list of (stroke_key, target_node, skip_count, is_last).
"""
edges = []
bit_pos = 0
for _ in range(edge_count):
val = 0
for i in range(50):
byte_idx = (bit_pos + i) // 8
bit_idx = 7 - ((bit_pos + i) % 8)
bit = (edge_bytes[byte_idx] >> bit_idx) & 1
val = (val << 1) | bit
bit_pos += 50
stroke_key = (val >> 34) & 0xFFFF
target_node = (val >> 18) & 0xFFFF
skip_count = (val >> 1) & 0x1FFFF
is_last = val & 1
edges.append((stroke_key, target_node, skip_count, bool(is_last)))
return edges
def serialize_value_array(entry_offsets, raw_table_size):
"""Serialize value array (string table offsets for each entry).
Uses uint16 if raw_table_size <= 65535, else uint32.
"""
use_u32 = raw_table_size > 65535
fmt = '<I' if use_u32 else '<H'
parts = [struct.pack(fmt, off) for off in entry_offsets]
return b''.join(parts), use_u32
def deserialize_value_array(data, entry_count, use_u32=False):
"""Deserialize value array."""
fmt = '<I' if use_u32 else '<H'
size = 4 if use_u32 else 2
offsets = []
for i in range(entry_count):
off = struct.unpack_from(fmt, data, i * size)[0]
offsets.append(off)
return offsets
# ─── Full compilation pipeline ───
def compile_dictionary(json_path, max_entries=120000, split_storage=False):
"""Compile a JSON steno dictionary into binary DAWG format.
Returns (binary_data, stats_dict).
"""
# 1. Load and parse
with open(json_path) as f:
raw_dict = json.load(f)
raw_entries = list(raw_dict.items())
# 2. Trim if needed
if len(raw_entries) > max_entries:
raw_entries = trim_entries(raw_entries, max_entries)
# 3. Parse strokes and sort
parsed = []
for stroke_str, translation in raw_entries:
strokes = parse_stroke_string(stroke_str)
parsed.append((strokes, translation))
parsed.sort(key=lambda x: x[0])
# 4. Build DAWG
root, node_count, edge_count = build_dawg(parsed)
# 5. Get traversal order for value array
traversal_paths = get_dawg_traversal_order(root)
# Build path->translation map
path_to_trans = {}
for strokes, translation in parsed:
path_to_trans[strokes] = translation
translations_ordered = []
for path in traversal_paths:
if path in path_to_trans:
translations_ordered.append(path_to_trans[path])
else:
translations_ordered.append("")
# 6. Build string table
string_table_bytes, entry_offsets, raw_table_size = \
build_string_table(translations_ordered)
# 7. Serialize edges
edge_bytes, node_id_map, skip_cache = \
serialize_edges(root, node_count, edge_count)
# 8. Serialize value array
value_array_bytes, use_u32 = \
serialize_value_array(entry_offsets, raw_table_size)
# 9. Compute offsets
edge_array_offset = HEADER_SIZE
value_array_offset = edge_array_offset + len(edge_bytes)
string_table_offset = value_array_offset + len(value_array_bytes)
# 10. Build header
flags = 0
if split_storage:
flags |= FLAG_SPLIT_STORAGE
if use_u32:
flags |= 0x0002 # bit 1 = u32 value offsets
header = serialize_header(
flags=flags,
entry_count=len(translations_ordered),
node_count=node_count,
edge_count=edge_count,
string_table_offset=string_table_offset,
string_table_size=len(string_table_bytes),
value_array_offset=value_array_offset,
)
# 11. Assemble
binary = header + edge_bytes + value_array_bytes + string_table_bytes
stats = {
'entry_count': len(translations_ordered),
'node_count': node_count,
'edge_count': edge_count,
'bits_per_edge': 50,
'edge_array_size': len(edge_bytes),
'value_array_size': len(value_array_bytes),
'string_table_size': len(string_table_bytes),
'raw_string_table_size': raw_table_size,
'total_size': len(binary),
'header_size': HEADER_SIZE,
'use_u32_offsets': use_u32,
'split_storage': split_storage,
}
return binary, stats, root, skip_cache, translations_ordered, parsed
def verify_compilation(binary_data, parsed_entries, root, skip_cache):
"""Verify compiled binary by deserializing and looking up every entry.
Returns (correct, wrong, missing).
"""
header = parse_header(binary_data)
entry_count = header['entry_count']
edge_count = header['edge_count']
use_u32 = bool(header['flags'] & 0x0002)
# Extract sections
edge_start = HEADER_SIZE
edge_end = header['value_array_offset']
edge_bytes = binary_data[edge_start:edge_end]
val_start = header['value_array_offset']
val_end = header['string_table_offset']
val_bytes = binary_data[val_start:val_end]
str_start = header['string_table_offset']
str_bytes = binary_data[str_start:]
# Deserialize
edges = deserialize_edges(edge_bytes, edge_count)
value_offsets = deserialize_value_array(val_bytes, entry_count, use_u32)
raw_table = decompress_string_table(str_bytes)
# Build adjacency from deserialized edges for lookup
# Reconstruct graph: node_id -> list of (stroke, target, skip)
adj = {}
node_finals = set()
edge_idx = 0
# We need to figure out which edges belong to which node.
# Edges are stored in DFS node order; is_last marks end of a node's edge list.
current_node = 0
node_edges = {}
i = 0
while i < len(edges):
stroke, target, skip, is_last = edges[i]
if current_node not in node_edges:
node_edges[current_node] = []
node_edges[current_node].append((stroke, target, skip))
if is_last:
current_node += 1
# Skip nodes that have no edges (they won't appear in edge list)
# We detect these by checking if next edge's parent should be higher
i += 1
# Now look up each entry via the reconstructed DAWG
correct = 0
wrong = 0
missing = 0
for strokes, expected in parsed_entries:
idx = dawg_lookup_index(root, strokes, skip_cache)
if idx < 0 or idx >= len(value_offsets):
missing += 1
continue
offset = value_offsets[idx]
# Find null terminator
try:
translation = lookup_string(raw_table, offset)
except (ValueError, IndexError):
missing += 1
continue
if translation == expected:
correct += 1
else:
wrong += 1
if wrong <= 5:
print(f" WRONG: strokes={strokes}, expected='{expected}', got='{translation}'")
return correct, wrong, missing
def print_stats(stats, target_kb=533):
"""Print compilation statistics."""
target_bytes = target_kb * 1024
total = stats['total_size']
pct = (total / target_bytes) * 100 if target_bytes else 0
print(f"Dictionary Compilation Stats:")
print(f" Entries: {stats['entry_count']:>10,}")
print(f" DAWG nodes: {stats['node_count']:>10,}")
print(f" DAWG edges: {stats['edge_count']:>10,}")
print(f" Bits/edge: {stats['bits_per_edge']:>10}")
print(f" ---")
print(f" Header: {stats['header_size']:>10,} bytes")
print(f" Edge array: {stats['edge_array_size']:>10,} bytes ({stats['edge_array_size']/1024:.1f} KB)")
print(f" Value array: {stats['value_array_size']:>10,} bytes ({stats['value_array_size']/1024:.1f} KB)")
print(f" String table: {stats['string_table_size']:>10,} bytes ({stats['string_table_size']/1024:.1f} KB)")
print(f" ---")
print(f" TOTAL: {total:>10,} bytes ({total/1024:.1f} KB)")
print(f" Budget: {target_bytes:>10,} bytes ({target_kb} KB)")
print(f" Usage: {pct:>9.1f}%")
if total <= target_bytes:
print(f" Status: FITS ({(target_bytes - total)/1024:.1f} KB remaining)")
else:
print(f" Status: OVER BUDGET by {(total - target_bytes)/1024:.1f} KB")
print(f" ---")
print(f" Split storage: {'yes' if stats['split_storage'] else 'no'}")
print(f" U32 offsets: {'yes' if stats['use_u32_offsets'] else 'no'}")
# ─── CLI ───
def main():
parser = argparse.ArgumentParser(
description='Compile steno dictionary to binary DAWG format')
parser.add_argument('input', nargs='?', default='/tmp/plover-main.json',
help='Path to JSON dictionary (default: /tmp/plover-main.json)')
parser.add_argument('--output', default='steno_dict.bin',
help='Output binary path (default: steno_dict.bin)')
parser.add_argument('--max-entries', type=int, default=120000,
help='Max entries to include (default: 120000)')
parser.add_argument('--target-size', type=int, default=462,
help='Target size in KB (default: 462)')
parser.add_argument('--split-storage', action='store_true',
help='Generate split-storage metadata header')
parser.add_argument('--stats', action='store_true',
help='Print detailed stats')
parser.add_argument('--verify', action='store_true',
help='Verify all entries after compilation')
args = parser.parse_args()
if not os.path.exists(args.input):
print(f"ERROR: Dictionary not found: {args.input}")
print("Download Plover dict:")
print(" curl -sL 'https://raw.githubusercontent.com/openstenoproject/plover/main/plover/assets/main.json' -o /tmp/plover-main.json")
sys.exit(1)
print(f"Compiling {args.input}...")
binary, stats, root, skip_cache, translations, parsed = \
compile_dictionary(args.input, args.max_entries, args.split_storage)
# Write output
with open(args.output, 'wb') as f:
f.write(binary)
print(f"Written {len(binary):,} bytes to {args.output}")
if args.stats:
print()
print_stats(stats, args.target_size)
if args.verify:
print()
print("Verifying...")
correct, wrong, missing = verify_compilation(binary, parsed, root, skip_cache)
total = correct + wrong + missing
print(f" Correct: {correct}/{total}")
print(f" Wrong: {wrong}/{total}")
print(f" Missing: {missing}/{total}")
if wrong > 0 or missing > 0:
print(" WARNING: Verification found errors!")
sys.exit(2)
else:
print(" All entries verified successfully.")
if __name__ == '__main__':
main()

106
tools/fetch_dict.py Normal file
View file

@ -0,0 +1,106 @@
#!/usr/bin/env python3
"""Download steno dictionary with caching.
Checks local file hash against known upstream hash.
If file exists and hash matches → skip download.
If file missing or hash mismatch → download fresh.
"""
import hashlib
import json
import os
import sys
import urllib.request
DICTS = {
"plover": {
"url": "https://raw.githubusercontent.com/openstenoproject/plover/main/plover/assets/main.json",
"filename": "plover-main.json",
},
"lapwing": {
"url": "https://raw.githubusercontent.com/aerickt/steno-dictionaries/main/lapwing-base.json",
"filename": "lapwing.json",
},
}
def sha256_file(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b''):
h.update(chunk)
return h.hexdigest()
def download(url, dest):
import subprocess
print(f"Downloading {url}...")
try:
urllib.request.urlretrieve(url, dest)
except urllib.error.URLError:
# Fallback to curl which uses system cert store
subprocess.check_call(["curl", "-sL", "-o", dest, url])
size = os.path.getsize(dest)
print(f"Downloaded {size:,} bytes to {dest}")
def fetch(dict_name, dest_dir):
if dict_name not in DICTS:
print(f"Unknown dict: {dict_name}. Available: {', '.join(DICTS.keys())}")
return 1
info = DICTS[dict_name]
dest = os.path.join(dest_dir, info["filename"])
hash_file = dest + ".sha256"
if os.path.exists(dest):
local_hash = sha256_file(dest)
if os.path.exists(hash_file):
with open(hash_file) as f:
cached_hash = f.read().strip()
if local_hash == cached_hash:
print(f"{dest} up to date (sha256={local_hash[:12]}...)")
return 0
try:
json.load(open(dest))
print(f"{dest} exists, valid JSON (sha256={local_hash[:12]}...)")
with open(hash_file, "w") as f:
f.write(local_hash)
return 0
except (json.JSONDecodeError, IOError):
print(f"{dest} corrupted, re-downloading")
os.makedirs(dest_dir, exist_ok=True)
download(info["url"], dest)
new_hash = sha256_file(dest)
with open(hash_file, "w") as f:
f.write(new_hash)
try:
with open(dest) as f:
d = json.load(f)
print(f"Verified: {len(d)} entries")
except (json.JSONDecodeError, IOError) as e:
print(f"WARNING: downloaded file invalid: {e}")
return 1
return 0
def main():
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <plover|lapwing> [dest_dir]")
print(f" dest_dir defaults to ./dicts/")
sys.exit(1)
dict_name = sys.argv[1]
dest_dir = sys.argv[2] if len(sys.argv) > 2 else os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "dicts")
sys.exit(fetch(dict_name, dest_dir))
if __name__ == "__main__":
main()

5
tools/requirements.txt Normal file
View file

@ -0,0 +1,5 @@
# zmk-steno-engine dictionary compiler
# stdlib only — no external dependencies required
#
# Python >= 3.8
# Uses: json, struct, zlib, argparse, sys, os, collections, math

460
tools/test_compiler.py Normal file
View file

@ -0,0 +1,460 @@
#!/usr/bin/env python3
"""Tests for dict_compiler.py — DAWG dictionary compiler."""
import json
import os
import struct
import sys
import tempfile
import pytest
# Ensure tools/ is importable
sys.path.insert(0, os.path.dirname(__file__))
from dict_compiler import (
STENO_KEYS,
FLAG_SPLIT_STORAGE,
HEADER_SIZE,
MAGIC,
VERSION,
build_dawg,
build_string_table,
compile_dictionary,
compute_skip_counts,
dawg_lookup_index,
decompress_string_table,
deserialize_edges,
deserialize_value_array,
get_dawg_traversal_order,
lookup_string,
parse_header,
parse_stroke,
parse_stroke_string,
serialize_edges,
serialize_header,
serialize_value_array,
trim_entries,
verify_compilation,
)
# ─── Small test dictionaries ───
SMALL_DICT = {
"S": "is",
"T": "it",
"K": "can",
"W": "with",
"H": "had",
"R": "are",
"TPHO": "no",
"STPH": "then",
"KAT": "cat",
"TKOG": "dog",
}
MULTI_STROKE_DICT = {
"S": "is",
"T": "it",
"KPA/HROL": "{}{-|}",
"TPHO/WUPB": "no one",
"K": "can",
}
MEDIUM_DICT = {
"S": "is",
"T": "it",
"K": "can",
"W": "with",
"H": "had",
"R": "are",
"A": "a",
"O": "oh",
"E": "he",
"U": "you",
"TPHO": "no",
"STPH": "then",
"KAT": "cat",
"TKOG": "dog",
"HOUS": "house",
"TPHAEUPL": "name",
"HROS": "also",
"TKPWRAET": "great",
"SKEL": "school",
"PLAS": "place",
}
def _make_temp_dict(d):
"""Write dict to temp JSON file, return path."""
fd, path = tempfile.mkstemp(suffix='.json')
with os.fdopen(fd, 'w') as f:
json.dump(d, f)
return path
def _parse_and_sort(d):
"""Parse dict entries and sort by stroke tuple."""
entries = []
for stroke_str, translation in d.items():
strokes = parse_stroke_string(stroke_str)
entries.append((strokes, translation))
entries.sort(key=lambda x: x[0])
return entries
# ─── Tests ───
class TestParseStroke:
"""Test stroke parsing."""
def test_left_side_stph(self):
"""STPH → S + T + P + H left side bits."""
val = parse_stroke("STPH")
expected = STENO_KEYS['S-'] | STENO_KEYS['T-'] | STENO_KEYS['P-'] | STENO_KEYS['H-']
assert val == expected
def test_right_side_eurb(self):
"""EURB → E + U + R + B right side bits."""
val = parse_stroke("EURB")
expected = STENO_KEYS['-E'] | STENO_KEYS['-U'] | STENO_KEYS['-R'] | STENO_KEYS['-B']
assert val == expected
def test_single_s(self):
"""S → just S bit."""
val = parse_stroke("S")
assert val == STENO_KEYS['S-']
def test_number_bar(self):
"""#STPH → number + S + T + P + H."""
val = parse_stroke("#STPH")
expected = (STENO_KEYS['#'] | STENO_KEYS['S-'] | STENO_KEYS['T-'] |
STENO_KEYS['P-'] | STENO_KEYS['H-'])
assert val == expected
def test_vowels(self):
"""AO → A + O vowel bits."""
val = parse_stroke("AO")
expected = STENO_KEYS['A-'] | STENO_KEYS['O-']
assert val == expected
def test_full_stroke(self):
"""STKPWHR → all left consonants."""
val = parse_stroke("STKPWHR")
expected = (STENO_KEYS['S-'] | STENO_KEYS['T-'] | STENO_KEYS['K-'] |
STENO_KEYS['P-'] | STENO_KEYS['W-'] | STENO_KEYS['H-'] |
STENO_KEYS['R-'])
assert val == expected
def test_star(self):
"""*E → star + E."""
val = parse_stroke("*E")
expected = STENO_KEYS['*'] | STENO_KEYS['-E']
assert val == expected
def test_multi_stroke_parse(self):
"""KPA/HROL parses to two stroke bitmasks."""
strokes = parse_stroke_string("KPA/HROL")
assert len(strokes) == 2
# First stroke: K + P + A
assert strokes[0] == (STENO_KEYS['K-'] | STENO_KEYS['P-'] | STENO_KEYS['A-'])
class TestBuildDawgSmall:
"""Test DAWG construction with small dictionary."""
def test_node_compression(self):
"""DAWG should have fewer nodes than a plain trie (compression happening)."""
entries = _parse_and_sort(SMALL_DICT)
root, node_count, edge_count = build_dawg(entries)
# With 10 entries, DAWG should have fewer nodes than
# total path length (which would be ~15+ for a trie)
assert node_count < 15
assert node_count > 0
assert edge_count > 0
def test_all_entries_reachable(self):
"""All entries should be reachable via traversal."""
entries = _parse_and_sort(SMALL_DICT)
root, node_count, edge_count = build_dawg(entries)
# Check each entry can be traversed
for strokes, _trans in entries:
node = root
for stroke in strokes:
assert stroke in node.edges, f"Missing edge for stroke in {strokes}"
node = node.edges[stroke]
assert node.final, f"Node not final for {strokes}"
class TestDawgLookup:
"""Test DAWG lookup via skip-count traversal."""
def test_all_lookups_correct(self):
"""All entries should have unique sequential skip-count indices."""
entries = _parse_and_sort(SMALL_DICT)
root, _, _ = build_dawg(entries)
skip_cache = compute_skip_counts(root)
indices = []
for strokes, _ in entries:
idx = dawg_lookup_index(root, strokes, skip_cache)
assert idx >= 0, f"Lookup failed for {strokes}"
indices.append(idx)
# All indices should be unique
assert len(set(indices)) == len(indices), "Duplicate indices found"
# Indices should be 0..n-1
assert sorted(indices) == list(range(len(entries)))
def test_missing_entry_returns_neg(self):
"""Looking up a non-existent stroke should return -1."""
entries = _parse_and_sort(SMALL_DICT)
root, _, _ = build_dawg(entries)
skip_cache = compute_skip_counts(root)
# A stroke not in the dict
fake_strokes = (0xDEAD,)
idx = dawg_lookup_index(root, fake_strokes, skip_cache)
assert idx == -1
def test_traversal_order_matches(self):
"""Traversal order should match sorted entries."""
entries = _parse_and_sort(SMALL_DICT)
root, _, _ = build_dawg(entries)
traversal = get_dawg_traversal_order(root)
entry_paths = [strokes for strokes, _ in entries]
assert traversal == entry_paths
class TestBinaryRoundTrip:
"""Test compile → binary → deserialize → verify."""
def test_small_dict_round_trip(self):
"""Compile small dict, deserialize, verify all lookups."""
path = _make_temp_dict(SMALL_DICT)
try:
binary, stats, root, skip_cache, translations, parsed = \
compile_dictionary(path, max_entries=100)
assert stats['entry_count'] == len(SMALL_DICT)
assert stats['total_size'] == len(binary)
assert stats['total_size'] > HEADER_SIZE
# Verify header
header = parse_header(binary)
assert header['magic'] == MAGIC
assert header['version'] == VERSION
assert header['entry_count'] == len(SMALL_DICT)
# Verify round-trip
correct, wrong, missing = verify_compilation(binary, parsed, root, skip_cache)
assert correct == len(SMALL_DICT)
assert wrong == 0
assert missing == 0
finally:
os.unlink(path)
def test_medium_dict_round_trip(self):
"""Medium dict round-trip."""
path = _make_temp_dict(MEDIUM_DICT)
try:
binary, stats, root, skip_cache, translations, parsed = \
compile_dictionary(path, max_entries=100)
correct, wrong, missing = verify_compilation(binary, parsed, root, skip_cache)
assert correct == len(MEDIUM_DICT)
assert wrong == 0
assert missing == 0
finally:
os.unlink(path)
def test_edge_serialization_round_trip(self):
"""Edge bit-packing round-trip."""
entries = _parse_and_sort(SMALL_DICT)
root, node_count, edge_count = build_dawg(entries)
edge_bytes, node_id_map, skip_cache = serialize_edges(root, node_count, edge_count)
edges = deserialize_edges(edge_bytes, edge_count)
assert len(edges) == edge_count
# Each edge should have valid fields
for stroke, target, skip, is_last in edges:
assert 0 <= stroke <= 0xFFFF
assert 0 <= target <= 0xFFFF
assert 0 <= skip <= 0x1FFFF
def test_string_table_round_trip(self):
"""String table compress/decompress round-trip."""
translations = list(SMALL_DICT.values())
table_bytes, offsets, raw_size = build_string_table(translations)
raw = decompress_string_table(table_bytes)
for i, trans in enumerate(translations):
recovered = lookup_string(raw, offsets[i])
assert recovered == trans, f"Mismatch at {i}: '{trans}' vs '{recovered}'"
class TestEntryTrimming:
"""Test entry trimming logic."""
def test_trim_keeps_single_stroke(self):
"""With max_entries < total, single-stroke entries kept preferentially."""
# Build dict: 60 single-stroke, 60 multi-stroke
d = {}
# Single-stroke entries (using various left-hand keys)
single_keys = list("STKPWHRAO*EUFRPBLGTSDZ")
for i, k in enumerate(single_keys[:15]):
d[k] = f"word_{i}"
# Multi-stroke entries
for i in range(85):
d[f"KAT/TKOG/{i % 10}"] = f"long_translation_{i}"
# Total = 100 entries
entries = list(d.items())
assert len(entries) == 100
trimmed = trim_entries(entries, 50)
assert len(trimmed) == 50
# Count single vs multi in result
single_count = sum(1 for s, _ in trimmed if '/' not in s)
multi_count = sum(1 for s, _ in trimmed if '/' in s)
# All 15 single-stroke entries should be kept
assert single_count == 15
assert multi_count == 35
def test_no_trim_when_under_limit(self):
"""No trimming when entries < max_entries."""
entries = list(SMALL_DICT.items())
trimmed = trim_entries(entries, 1000)
assert len(trimmed) == len(entries)
def test_trim_multi_stroke_by_length(self):
"""Multi-stroke entries trimmed by translation length (shorter kept)."""
d = {"S": "is"} # 1 single-stroke
# Add multi-stroke with varying translation lengths
d["KAT/S"] = "ab" # short
d["KAT/T"] = "abcdefghij" # long
d["KAT/K"] = "abc" # medium
entries = list(d.items())
trimmed = trim_entries(entries, 3)
# Should keep: single("S"), then shortest multi-stroke
assert len(trimmed) == 3
trans = [t for _, t in trimmed]
assert "is" in trans # single stroke kept
assert "ab" in trans # shortest multi kept
assert "abc" in trans # medium kept
assert "abcdefghij" not in trans # longest dropped
class TestSplitStorageFlag:
"""Test --split-storage flag."""
def test_flag_set_in_header(self):
"""split_storage flag should set bit 0 in header flags."""
path = _make_temp_dict(SMALL_DICT)
try:
binary, stats, _, _, _, _ = compile_dictionary(
path, max_entries=100, split_storage=True)
header = parse_header(binary)
assert header['flags'] & FLAG_SPLIT_STORAGE != 0
assert stats['split_storage'] is True
finally:
os.unlink(path)
def test_flag_not_set_by_default(self):
"""split_storage flag should NOT be set by default."""
path = _make_temp_dict(SMALL_DICT)
try:
binary, stats, _, _, _, _ = compile_dictionary(
path, max_entries=100, split_storage=False)
header = parse_header(binary)
assert header['flags'] & FLAG_SPLIT_STORAGE == 0
assert stats['split_storage'] is False
finally:
os.unlink(path)
class TestMultiStroke:
"""Test multi-stroke entry handling."""
def test_multi_stroke_compile_and_lookup(self):
"""Multi-stroke entries (e.g. KPA/HROL) should compile and look up correctly."""
path = _make_temp_dict(MULTI_STROKE_DICT)
try:
binary, stats, root, skip_cache, translations, parsed = \
compile_dictionary(path, max_entries=100)
correct, wrong, missing = verify_compilation(binary, parsed, root, skip_cache)
assert correct == len(MULTI_STROKE_DICT)
assert wrong == 0
assert missing == 0
finally:
os.unlink(path)
def test_multi_stroke_traversal(self):
"""Multi-stroke entries should appear in correct traversal order."""
entries = _parse_and_sort(MULTI_STROKE_DICT)
root, _, _ = build_dawg(entries)
skip_cache = compute_skip_counts(root)
for strokes, _ in entries:
idx = dawg_lookup_index(root, strokes, skip_cache)
assert idx >= 0, f"Multi-stroke lookup failed: {strokes}"
def test_kpa_hrol_specific(self):
"""KPA/HROL → {}{-|} specifically."""
d = {"KPA/HROL": "{}{-|}"}
entries = _parse_and_sort(d)
root, _, _ = build_dawg(entries)
skip_cache = compute_skip_counts(root)
strokes = parse_stroke_string("KPA/HROL")
idx = dawg_lookup_index(root, strokes, skip_cache)
assert idx == 0 # only entry → index 0
class TestHeaderSerialization:
"""Test header pack/unpack."""
def test_header_size(self):
"""Header should be exactly 32 bytes."""
header = serialize_header(0, 100, 50, 200, 1000, 500, 800)
assert len(header) == HEADER_SIZE
def test_header_round_trip(self):
"""Header fields should survive pack/unpack."""
header = serialize_header(
flags=FLAG_SPLIT_STORAGE,
entry_count=12345,
node_count=6789,
edge_count=11111,
string_table_offset=99999,
string_table_size=55555,
value_array_offset=44444,
)
parsed = parse_header(header)
assert parsed['magic'] == MAGIC
assert parsed['version'] == VERSION
assert parsed['flags'] == FLAG_SPLIT_STORAGE
assert parsed['entry_count'] == 12345
assert parsed['node_count'] == 6789
assert parsed['edge_count'] == 11111
assert parsed['string_table_offset'] == 99999
assert parsed['string_table_size'] == 55555
assert parsed['value_array_offset'] == 44444
if __name__ == '__main__':
pytest.main([__file__, '-v'])