Share lossless voice prefixes across companion announcements

scripts/generate_button_voice.py

@@ -3,7 +3,7 @@
 import argparse
 from pathlib import Path
 
-from generate_gps_voice import encode, read
+from generate_gps_voice import encode, read, shared_prefix, emit_array
 
 PHRASES = (
     ('ready', 'Ready'),
@@ -22,6 +22,64 @@ PHRASES = (
     ('alertsSilent', 'Silent'),
 )
 
+# Only meaningful byte-identical prefixes are shared. These retain the exact
+# approved phrase prosody; adaptive compressed tails are not standalone words.
+PREFIX_GROUPS = (
+    ('advertPrefix', 'advertQueued', 'advertFailed'),
+    ('soundPrefix', 'soundOff', 'alertsSoundOnly'),
+)
+
+def split_catalog(encoded):
+    tails = dict(encoded)
+    prefixes = {}
+    references = {}
+    for prefix_name, first, second in PREFIX_GROUPS:
+        prefix, one, two = shared_prefix(encoded[first], encoded[second])
+        # Tiny shared silence does not justify additional metadata/storage.
+        if len(prefix) < 64:
+            continue
+        prefixes[prefix_name] = prefix
+        tails[first], tails[second] = one, two
+        references[first] = references[second] = prefix_name
+    return prefixes, tails, references
+
+def emit_clip(lines, name, data, count, rate=8000, *, prefix=None, codec=None):
+    emit_array(lines, name, data)
+    fields = f'  {name}Data, sizeof({name}Data), {count}, {rate}'
+    if codec:
+        fields += f', VoiceCodec::{codec}'
+    if prefix:
+        fields += f', {prefix}Data, sizeof({prefix}Data)'
+    lines.extend([f'static constexpr VoiceClip {name}Clip = {{', fields, '};'])
+
+def emit_catalog(lines, samples, encoded, *, codec=None):
+    prefixes, tails, references = split_catalog(encoded)
+    for prefix_name, data in prefixes.items():
+        if prefix_name == 'soundPrefix':
+            lines.append('#ifdef HAS_DRV2605')
+        lines.append('// Shared encoded prefix; keep adaptive state at the seam.')
+        emit_array(lines, prefix_name, data)
+        if prefix_name == 'soundPrefix':
+            lines.append('#endif // HAS_DRV2605')
+    for name, phrase in PHRASES:
+        if name == 'alertsSoundVibration':
+            lines.append('#ifdef HAS_DRV2605')
+        lines.append(f'// "{phrase}"')
+        if name == 'soundOff' and name in references:
+            # Non-vibration boards have no second consumer for this prefix.
+            # Keep their original contiguous phrase, without extra span data.
+            lines.append('#ifdef HAS_DRV2605')
+            emit_clip(lines, name, tails[name], len(samples[name]),
+                      prefix=references[name], codec=codec)
+            lines.append('#else')
+            emit_clip(lines, name, encoded[name], len(samples[name]), codec=codec)
+            lines.append('#endif // HAS_DRV2605')
+        else:
+            emit_clip(lines, name, tails[name], len(samples[name]),
+                      prefix=references.get(name), codec=codec)
+    lines.append('#endif // HAS_DRV2605')
+    return sum(map(len, prefixes.values()))
+
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
@@ -39,27 +97,21 @@ def main():
     ]
     if args.voice is not None:
         lines.insert(2, f'// Voice: {args.voice}')
-    total = 0
+    samples_by_name = {}
+    encoded_by_name = {}
     for name, phrase in PHRASES:
         samples = read(args.directory / (name + '.wav'))
         # Shutdown drains audio nonblockingly for at most five seconds.
         if not samples or len(samples) > 32000:
             raise ValueError(f'{phrase}: expected a nonempty clip under four seconds')
         encoded = encode(samples)
-        total += len(encoded)
-        if name == 'alertsSoundVibration':
-            # Only buzzer+vibration devices need the additional mode catalog.
-            lines.append('#ifdef HAS_DRV2605')
-        lines.extend([f'// "{phrase}"', f'static const uint8_t {name}Data[] = {{'])
-        for offset in range(0, len(encoded), 20):
-            lines.append('  ' + ','.join(f'0x{b:02x}' for b in encoded[offset:offset + 20]) + ',')
-        lines.extend(['};', f'static constexpr VoiceClip {name}Clip = {{',
-                      f'  {name}Data, sizeof({name}Data), {len(samples)}, 8000', '};'])
+        samples_by_name[name] = samples
+        encoded_by_name[name] = encoded
         print(f'{phrase}: {len(samples) / 8000:.3f}s, {len(encoded)} bytes')
-    lines.append('#endif // HAS_DRV2605')
+    saved = emit_catalog(lines, samples_by_name, encoded_by_name)
     lines.append('}} // namespace mesh::audio')
     args.output.write_text('\n'.join(lines) + '\n', encoding='utf-8')
-    print(f'Button confirmations: {total} audio bytes')
+    print(f'Button confirmations: {sum(map(len, encoded_by_name.values())) - saved} audio bytes')
 
 
 if __name__ == '__main__':

scripts/generate_g726_voice.py

@@ -10,8 +10,8 @@ import struct
 import subprocess
 from pathlib import Path
 
-from generate_button_voice import PHRASES
-from generate_gps_voice import read
+from generate_button_voice import PHRASES, emit_catalog, emit_clip
+from generate_gps_voice import read, shared_prefix, emit_array
 
 
 def encode(samples, bitrate, ffmpeg="ffmpeg"):
@@ -59,22 +59,21 @@ def main():
     ]
     for bitrate in (16, 24):
         lines.append(f"#if MESH_GPS_VOICE_G726_BITRATE == {bitrate}")
-        total = 0
-        for name, phrase, pcm in samples:
-            mode = name.startswith("alerts")
-            if mode:
-                lines.append("#ifdef HAS_DRV2605")
-            data = encode(pcm, bitrate, args.ffmpeg)
-            total += len(data)
-            lines += [f'// "{phrase}"', f"static const uint8_t {name}Data[] = {{"]
-            for offset in range(0, len(data), 20):
-                lines.append("  " + ",".join(f"0x{b:02x}" for b in data[offset:offset + 20]) + ",")
-            lines += ["};", f"static constexpr VoiceClip {name}Clip = {{",
-                      f"  {name}Data, sizeof({name}Data), {len(pcm)}, 8000, VoiceCodec::G726{bitrate}", "};"]
-            if mode:
-                lines.append("#endif // HAS_DRV2605")
+        pcm_by_name = {name: pcm for name, _, pcm in samples}
+        encoded = {name: encode(pcm, bitrate, args.ffmpeg) for name, _, pcm in samples}
+        saved = emit_catalog(lines, pcm_by_name, encoded, codec=f'G726{bitrate}')
+        prefix, on, off = shared_prefix(encoded['gpsOn'], encoded['gpsOff'])
+        if len(prefix) < 64:
+            prefix, on, off = b'', encoded['gpsOn'], encoded['gpsOff']
+        if prefix:
+            lines.append('// GPS spans form one continuous stream, including cross-byte codes.')
+            emit_array(lines, 'gps', prefix)
+        for name, data in (('gpsOn', on), ('gpsOff', off)):
+            emit_clip(lines, name, data, len(pcm_by_name[name]),
+                      prefix='gps' if prefix else None, codec=f'G726{bitrate}')
         lines.append(f"#endif // {bitrate} kbps")
-        print(f"G.726-{bitrate}: {total} bytes across all 16 clips")
+        total = sum(map(len, encoded.values())) - saved - len(prefix)
+        print(f"G.726-{bitrate}: {total} bytes across all 16 phrases")
     lines.append("}} // namespace mesh::audio")
     args.output.write_text("\n".join(lines) + "\n", encoding="utf-8")
 

scripts/generate_gps_voice.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Encode 8/16 kHz mono 16-bit WAV speech as independent IMA ADPCM clips."""
+"""Encode GPS phrases, sharing their identical encoded prefix without loss."""
 import argparse
 from pathlib import Path
 import struct
@@ -41,6 +41,25 @@ def read(path, rate=8000):
         data = wav.readframes(wav.getnframes())
     return struct.unpack('<' + 'h' * (len(data)//2),data)
 
+def shared_prefix(on, off):
+    """Return three spans whose concatenations exactly restore both streams.
+
+    A split is byte-based, not a word/codec boundary. Playback must preserve
+    adaptive codec state and bit position at the seam. Leave a nonempty tail
+    for both phrases, including identical or very short fixture recordings.
+    """
+    limit = min(len(on), len(off)) - 1
+    split = 0
+    while split < limit and on[split] == off[split]:
+        split += 1
+    return on[:split], on[split:], off[split:]
+
+def emit_array(lines, name, data):
+    lines.append(f'static const uint8_t {name}Data[] = {{')
+    for i in range(0,len(data),20):
+        lines.append('  '+','.join(f'0x{b:02x}' for b in data[i:i+20])+',')
+    lines.append('};')
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--on',type=Path,required=True)
@@ -57,15 +76,25 @@ def main():
         'namespace mesh { namespace audio {']
     if args.voice is not None:
         lines.insert(2, f'// Voice: {args.voice}')
-    for name,path in (('On',args.on),('Off',args.off)):
-        samples = read(path,args.rate)
-        encoded = encode(samples)
-        lines.append(f'static const uint8_t gps{name}Data[] = {{')
-        for i in range(0,len(encoded),20):
-            lines.append('  '+','.join(f'0x{b:02x}' for b in encoded[i:i+20])+',')
-        lines.extend(['};',f'static constexpr VoiceClip gps{name}Clip = {{',
-            f'  gps{name}Data, sizeof(gps{name}Data), {len(samples)}, {args.rate}', '};'])
-        print(f'GPS {name.lower()}: {len(samples)/args.rate:.3f}s, {len(encoded)} bytes')
+    phrases = [read(path,args.rate) for path in (args.on,args.off)]
+    if any(not samples or len(samples) > args.rate * 4 for samples in phrases):
+        raise ValueError('GPS phrases must be nonempty clips under four seconds')
+    encoded = [encode(samples) for samples in phrases]
+    prefix, on, off = shared_prefix(*encoded)
+    if len(prefix) < 64:
+        prefix, on, off = b'', *encoded
+    if prefix:
+        lines.append('// Three stored spans; preserve decoder state across the shared prefix.')
+        emit_array(lines, 'gps', prefix)
+    for name,samples,data in zip(('On','Off'),phrases,(on,off)):
+        emit_array(lines, 'gps'+name, data)
+        fields = f'  gps{name}Data, sizeof(gps{name}Data), {len(samples)}, {args.rate}'
+        if prefix:
+            fields += ', gpsData, sizeof(gpsData)'
+        lines.extend([f'static constexpr VoiceClip gps{name}Clip = {{',
+                      fields, '};'])
+        print(f'GPS {name.lower()}: {len(samples)/args.rate:.3f}s, {len(data)} tail bytes')
+    print(f'GPS shared prefix: {len(prefix)} bytes; audio saved: {len(prefix)} bytes')
     lines.append('}} // namespace mesh::audio')
     args.output.write_text('\n'.join(lines)+'\n',encoding='utf-8')