Rudimentary integration of signal analysis with Seed Source

Author: pjkundert

slip39/gui/main.py

@@ -13,8 +13,8 @@
 import PySimpleGUI as sg
 
 from ..api		import Account, create, group_parser, random_secret, cryptopaths_parser, paper_wallet_available
-from ..recovery		import recover, recover_bip39, produce_bip39, analyze_entropy
-from ..util		import log_level, log_cfg, ordinal, chunker, hue_shift
+from ..recovery		import recover, recover_bip39, produce_bip39, scan_entropy, display_entropy
+from ..util		import log_level, log_cfg, ordinal, chunker, hue_shift, rate_dB
 from ..layout		import write_pdfs, printers_available
 from ..defaults		import (
     GROUPS, GROUP_THRESHOLD_RATIO, MNEM_PREFIX, CRYPTO_PATHS, BITS, BITS_DEFAULT,
@@ -34,6 +34,9 @@
                                   " earth warn lunar olympic clothes piece campus alpha short endless"
 
 
+SD_SEED_FRAME			= 'Seed Source: Input or Create your Seed Entropy here'
+
+
 def theme_color( thing, theme=None ):
     """Get the currency configured PySimpleGUI Theme color for thing == eg. "TEXT", "BACKGROUND.
     """
@@ -211,7 +214,7 @@ def groups_layout(
     ] + [
         [
             # SLIP-39 only available in Recovery; SLIP-39 Passphrase only in Pro; BIP-39 and Fixed Hex only in Pro
-            sg.Frame( 'Seed Source: Input or Create your Seed Entropy here', [
+            sg.Frame( SD_SEED_FRAME, [
                 [
                     sg.Text( "Random:" if not LO_BAK else "Source:",            visible=LO_CRE, **T_hue( T_kwds, 0/20 )),
                     sg.Radio( "128-bit",          "SD", key='-SD-128-RND-',     default=LO_CRE,
@@ -458,6 +461,7 @@ def update_seed_data( event, window, values ):
     restores our last-known radio button and data.  Since we cannot know if/when our main window is
     going to disappear and be replaced, we constantly save the current state.
 
+    Reports the quality of the Seed Data in the frame label.
     """
     SD_CONTROLS			= [
         '-SD-128-RND-',
@@ -615,14 +619,18 @@ def update_seed_data( event, window, values ):
 
     # Analyze the seed for Signal harmonic or Shannon entropy failures, if we're in a __TIMEOUT__
     # (between keystrokes or after a major controls change).  Otherwise, if the seed's changed,
-    # request a __TIMEOUT__; when it, perform the entropy analysis.
-    values['-SD-SIG-']		= ''
+    # request a __TIMEOUT__; when it invokes, perform the entropy analysis.
+    values['-SD-SIGS-']		= ''
+
     if status is None and event == '__TIMEOUT__':
         seed_bytes		= codecs.decode( seed, 'hex_codec' )
-        analysis		= analyze_entropy( seed_bytes, what=f"{len(seed_bytes)*8}-bit Seed Source", show_details=False )
+        signals, shannons	= scan_entropy( seed_bytes, show_details=True )
+        analysis		= display_entropy( signals, shannons, what=f"{len(seed_bytes)*8}-bit Seed Source" )
         if analysis:
             values['-SD-SIG-']	= analysis
             status		= analysis.split( '\n' )[0]
+        window['-SD-SEED-F-'].update(
+            f"{SD_SEED_FRAME}; {rate_dB( max( signals ).dB, what='Harmonics')}, {rate_dB( max( signals ).dB, what='Shannon')}, " )
     elif changed:
         log.info( f"Seed Data requests __TIMEOUT__ w/ current source: {update_seed_data.src!r}" )
         values['__TIMEOUT__']	= .5
@@ -1001,7 +1009,7 @@ def app(
         # specific instructional .txt we can load.  Only if the current instructions is empty will
         # we go all the way back to load the generic SLIP-39.txt.  If the event corresponds to an
         # object with text/backround_color, use it in the instructional text.
-        txt_segs		= ( event or '' ).strip('-').split( '-' )
+        txt_segs		= ( event or '' ).strip( '-' ).split( '-' )
         for txt_i in range( len( txt_segs ), 0 if instructions else -1, -1 ):
             txt_name		= '-'.join( [ 'SLIP', '39' ] + txt_segs[:txt_i] ) + '.txt'
             txt_path		= os.path.join( os.path.dirname( __file__ ), txt_name )

slip39/recovery/__init__.py

@@ -25,7 +25,9 @@
 
 from ..util		import ordinal
 
-from .entropy		import shannon_entropy, signal_entropy, analyze_entropy  # noqa F401
+from .entropy		import (  # noqa F401
+    shannon_entropy, signal_entropy, analyze_entropy, scan_entropy, display_entropy
+)
 
 log				= logging.getLogger( __package__ )
 

slip39/recovery/entropy.py

@@ -1019,6 +1019,30 @@ def scan_entropy(
     return signals, shannons
 
 
+def display_entropy( signals, shannons, what=None ):
+    result			= None
+    if signals or shannons:
+        report			= ''
+        dBs			= sorted( list( s.dB for s in signals ) + list( s.dB for s in shannons) )
+        dBr			= dBs[:1] + dBs[max( 1, len(dBs)-1):]
+        report		       += f"Entropy analysis {('of ' + what) if what else ''}: {len(signals)+len(shannons)}x"
+        report		       += f" {'-'.join( f'{dB:.1f}' for dB in dBr )}dB non-random energy patterns in"
+        report		       += f" {commas( sorted( set( s.stride for s in signals )), final_and=True)}-bit symbols\n"
+        for s,summary in sorted(
+                [
+                    (s, f"{s.dB:5.1f}dB Signal harmonic feature at offset {s.offset} in {s.symbols}x {s.stride}-bit symbols")
+                    for s in signals
+                ] + [
+                    (s, f"{s.dB:5.1f}dB Shannon entropy reduced at offset {s.offset} in {s.symbols}x {s.stride}-bit symbols")
+                    for s in shannons
+                ], reverse=True ):
+            report	       += f"-{summary}{': ' if s.details else ''}\n"
+            if s.details:
+                report	       += f"{s.details}\n"
+        result			= report
+    return result
+
+
 def analyze_entropy(
     entropy: bytes,
     strides: Optional[Union[int,Tuple[int,int]]] = None,  # If only a specific stride/s makes sense, eg. for ASCII symbols
@@ -1052,27 +1076,9 @@ def analyze_entropy(
     0.25% failure on each individual test.
 
     """
-    signals, shannons		= scan_entropy(
-        entropy, strides, overlap, ignore_dc=ignore_dc, show_details=show_details,
-        signal_threshold=signal_threshold, shannon_threshold=shannon_threshold )
-    result			= None
-    if signals or shannons:
-        report			= ''
-        dBs			= sorted( list( s.dB for s in signals ) + list( s.dB for s in shannons) )
-        dBr			= dBs[:1] + dBs[max( 1, len(dBs)-1):]
-        report		       += f"Entropy analysis {('of ' + what) if what else ''}: {len(signals)+len(shannons)}x"
-        report		       += f" {'-'.join( f'{dB:.1f}' for dB in dBr )}dB non-random energy patterns in"
-        report		       += f" {commas( sorted( set( s.stride for s in signals )), final_and=True)}-bit symbols\n"
-        for s,summary in sorted(
-                [
-                    (s, f"{s.dB:5.1f}dB Signal harmonic feature at offset {s.offset} in {s.symbols}x {s.stride}-bit symbols")
-                    for s in signals
-                ] + [
-                    (s, f"{s.dB:5.1f}dB Shannon entropy reduced at offset {s.offset} in {s.symbols}x {s.stride}-bit symbols")
-                    for s in shannons
-                ], reverse=True ):
-            report	       += f"-{summary}{': ' if s.details else ''}\n"
-            if s.details:
-                report	       += f"{s.details}\n"
-        result			= report
-    return result
+    return display_entropy(
+        *scan_entropy(
+            entropy, strides, overlap, ignore_dc=ignore_dc, show_details=show_details,
+            signal_threshold=signal_threshold, shannon_threshold=shannon_threshold ),
+        what=what
+    )

slip39/recovery_test.py

@@ -21,7 +21,7 @@
 from .recovery		import recover, recover_bip39, shannon_entropy, signal_entropy, analyze_entropy
 from .recovery.entropy	import fft, ifft, pfft, dft, dft_on_real, dft_to_rms_mags, entropy_bin_dfts, denoise_mags, signal_draw, signal_recover_real, scan_entropy
 from .dependency_test	import substitute, nonrandom_bytes, SEED_XMAS, SEED_ONES
-from .util		import avg, rms, ordinal, commas
+from .util		import avg, rms, ordinal, commas, round_onto
 
 log				= logging.getLogger( __package__ )
 
@@ -330,6 +330,21 @@ def test_util():
     assert commas( [1,3,5], final_and=True ) == '1, 3 and 5'
     assert commas( [1,2,5], final_and=True ) == '1, 2 and 5'
 
+    assert round_onto( -.1, [-5, -1, 0, 1, 5], keep_sign=False ) == 0
+    assert round_onto( -.9, [-5, -1, 0, 1, 5], keep_sign=False ) == -1
+    assert round_onto( -10, [0, -1, -5, 5, 1], keep_sign=False ) == -5
+    assert round_onto( 100, [-5, -1, 0, 1, 5], keep_sign=False ) == +5
+    assert round_onto( 100, [0, -1, -5, 5, 1], keep_sign=False ) == +5
+    assert round_onto( 2.9, [-5, -1, 0, 1, 5], keep_sign=False ) == +1
+    assert round_onto( 3.1, [-5, -1, 0, 1, 5], keep_sign=False ) == +5
+    assert round_onto( .01, [-5, -1, 0, 1, 5], keep_sign=False ) == 0
+    assert round_onto( -.1, [-5, -1, 0, 1, 5], keep_sign=False ) == 0
+    assert round_onto( -.1, [-5, -1, 0, 1, 5], keep_sign=True  ) == -1
+    assert round_onto( -.1, [-5, -1,    1, 5], keep_sign=True  ) == -1
+    assert round_onto( -.1, [-5,        1, 5], keep_sign=True  ) == -5
+    assert round_onto( +.1, [-5, -1,    1, 5], keep_sign=True  ) == +1
+    assert round_onto( +.1, [-5, -1,       5], keep_sign=True  ) == +5
+
 
 def test_dft_smoke():
     """Test some basic assumptions on DFTs"""

slip39/util.py

@@ -83,6 +83,39 @@ def int_seq( seq ):
     return ', '.join( map( str, seq ))
 
 
+def round_onto( value, keys, keep_sign=True ):
+    """Find the numeric key closest to value, maintaining sign if possible."""
+    keys			= sorted( keys )
+    near, near_i		= min(
+        (abs( k - value), i)
+        for i, k in enumerate( keys )
+    )
+    print( f"round_onto: Rounding {value} onto {keys[near_i]}; {ordinal(near_i+1)} key in: {keys} ({keep_sign=})" )
+    if keep_sign and (( value < 0 ) != ( keys[near_i] < 0 )):
+        # The sign differs; if value -'ve but closest was +'ve, and there is a lower key available
+        if value < 0:
+            print( f"round_onto: sign of {value} != {keys[near_i]}: shift down in {keys[:near_i+1]}..." )
+            if near_i > 0:
+                near_i	       -= 1
+        else:
+            print( f"round_onto: sign of {value} != {keys[near_i]}: shift up in ...{keys[near_i:]}" )
+            if near_i + 1 < len( keys ):
+                near_i	       += 1
+    return keys[near_i]
+
+
+def rate_dB( dB, what=None ):
+    strength			= {
+        2: "very bad",
+        1: "bad",
+        0: "poor",
+        -1: "ok",
+        -2: "strong",
+    }
+    rating			= strength[round_onto( dB, strength.keys(), keep_sign=True )]
+    return f"{what or ''}{': ' if what else ''}{dB:.1f}dB ({rating})"
+
+
 def input_secure( prompt, secret=True, file=None ):
     """When getting secure (optionally secret) input from standard input, we don't want to use getpass, which
     attempts to read from /dev/tty.