Added multi-threaded RNG sample generation and results analysis

Results are provided and are very promising!

Author: DefinetlyNotAI

.idea/dictionaries/project.xml

@@ -0,0 +1,40 @@
+NOTE: Manually created file to store results of RNG tests. You can run your own test.
+
+PS > python .\rng_test.py
+----------------------------------------
+Starting RNG tests...
+----------------------------------------
+
+Read 10000000 bytes from rng_output.bin
+
+----------------------------------------
+Shannon Entropy:
+    7.9995 bits per byte
+----------------------------------------
+Frequency test:
+    Total bits: 80000000
+    Zeros: 40082720
+    Ones: 40024692
+----------------------------------------
+Runs test:
+    Runs: 39989904
+    Avg run length: 2.00
+    Max run length: 21
+    Min run length: 1
+----------------------------------------
+Autocorrelation test (lag=1):
+    Autocorrelation: 0.5001 (expected ~0.5)
+----------------------------------------
+Autocorrelation test (lag=8):
+    Autocorrelation: 0.4997 (expected ~0.5)
+----------------------------------------
+Bit position frequency test:
+    Bit position 0: set bit frequency = 0.5007 (expected ~0.5)
+    Bit position 1: set bit frequency = 0.4987 (expected ~0.5)
+    Bit position 2: set bit frequency = 0.5008 (expected ~0.5)
+    Bit position 3: set bit frequency = 0.5004 (expected ~0.5)
+    Bit position 4: set bit frequency = 0.4994 (expected ~0.5)
+    Bit position 5: set bit frequency = 0.5018 (expected ~0.5)
+    Bit position 6: set bit frequency = 0.5000 (expected ~0.5)
+    Bit position 7: set bit frequency = 0.5007 (expected ~0.5)
+----------------------------------------

example/rng_tests/Results.txt

@@ -0,0 +1,52 @@
+import os
+import threading
+
+from tqdm import tqdm
+
+import sys
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
+from pyCTools.hwrng import MaxRNG
+
+# Initialize the MaxRNG instance
+rng = MaxRNG()
+rng.dll.maxrng_init()
+
+
+def worker(size, index, results):
+    try:
+        data = rng.maxrng_threadsafe(size)
+        results[index] = data
+    except RuntimeError as e:
+        results[index] = e
+
+
+def save_random_samples(total_bytes=10_000_000, chunk_size=1024):
+    bytes_generated = 0
+
+    with open("rng_output.bin", "wb") as f, tqdm(total=total_bytes, unit="B", unit_scale=True,
+                                                 desc="Saving random samples") as pbar:
+        while bytes_generated < total_bytes:
+            num_cores = os.cpu_count() or 1
+            num_threads_ = min(num_cores, (total_bytes - bytes_generated + chunk_size - 1) // chunk_size)
+
+            results = [None] * num_threads_
+            threads_ = []
+            for i_ in range(num_threads_):
+                size = min(chunk_size, total_bytes - bytes_generated - i_ * chunk_size)
+                t_ = threading.Thread(target=worker, args=(size, i_, results))
+                threads_.append(t_)
+                t_.start()
+            for t_ in threads_:
+                t_.join()
+            for i_, result_ in enumerate(results):
+                if isinstance(result_, Exception):
+                    raise result_
+                f.write(result_)
+                bytes_generated += len(result_)
+                pbar.update(len(result_))
+    print(f"Saved {bytes_generated} bytes to rng_output.bin")
+
+
+save_random_samples()
+print("Random samples saved to rng_output.bin")

example/rng_tests/generate_bin.py

@@ -0,0 +1,112 @@
+import math
+
+
+def shannon_entropy(data: bytes):
+    if not data:
+        return None
+    freq = {}
+    for b in data:
+        freq[b] = freq.get(b, 0) + 1
+    entropy = 0.0
+    length = len(data)
+    for count in freq.values():
+        p = count / length
+        entropy -= p * math.log2(p)
+    print(f"Shannon Entropy:")
+    print(f"    {entropy:.4f} bits per byte")
+    return None
+
+
+def frequency_test(data: bytes):
+    zeros = sum(bin(b).count("0") for b in data)
+    ones = sum(bin(b).count("1") for b in data)
+    print("Frequency test:")
+    print(f"    Total bits: {len(data) * 8}")
+    print(f"    Zeros: {zeros}")
+    print(f"    Ones: {ones}")
+
+
+def runs_test(data: bytes):
+    bits = ''.join(f'{b:08b}' for b in data)
+    runs = 1
+    prev_bit = bits[0]
+
+    run_lengths = []
+    current_run_length = 1
+
+    for bit in bits[1:]:
+        if bit == prev_bit:
+            current_run_length += 1
+        else:
+            run_lengths.append(current_run_length)
+            current_run_length = 1
+            runs += 1
+        prev_bit = bit
+    run_lengths.append(current_run_length)
+
+    avg_run_length = sum(run_lengths) / len(run_lengths)
+    max_run_length = max(run_lengths)
+    min_run_length = min(run_lengths)
+
+    print("Runs test:")
+    print(f"    Runs: {runs}")
+    print(f"    Avg run length: {avg_run_length:.2f}")
+    print(f"    Max run length: {max_run_length}")
+    print(f"    Min run length: {min_run_length}")
+
+
+def autocorrelation_test(data: bytes, lag=1):
+    bits = ''.join(f'{b:08b}' for b in data)
+    n = len(bits)
+    matches = 0
+    for i in range(n - lag):
+        if bits[i] == bits[i + lag]:
+            matches += 1
+    autocorr = matches / (n - lag)
+    print(f"Autocorrelation test (lag={lag}):")
+    print(f"    Autocorrelation: {autocorr:.4f} (expected ~0.5)")
+
+
+def bit_position_frequency(data: bytes):
+    print("Bit position frequency test:")
+    counts = [0] * 8  # count of set bits per position
+    total_bytes = len(data)
+
+    for b in data:
+        for i in range(8):
+            if b & (1 << i):
+                counts[i] += 1
+
+    for i, c in enumerate(counts):
+        freq = c / total_bytes
+        print(f"    Bit position {i}: set bit frequency = {freq:.4f} (expected ~0.5)")
+
+
+def main():
+    print("-" * 40)
+    print("Starting RNG tests...")
+    print("-" * 40 + "\n")
+    filename = "rng_output.bin"
+    with open(filename, "rb") as f:
+        data = f.read()
+
+    print(f"Read {len(data)} bytes from {filename}\n")
+
+    print("-" * 40)
+    shannon_entropy(data)
+    print("-" * 40)
+    frequency_test(data)
+    print("-" * 40)
+    runs_test(data)
+    print("-" * 40)
+    autocorrelation_test(data, lag=1)
+    print("-" * 40)
+    autocorrelation_test(data, lag=8)
+    print("-" * 40)
+    bit_position_frequency(data)
+    print("-" * 40)
+    print("\nAll tests completed.")
+
+
+if __name__ == "__main__":
+    main()