mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
allaun
475f6319ea
This squashes all local history (768 commits) onto the scrubbed PR #90 baseline. Individual commits were lost during filter-repo corruption; the working tree content is preserved intact. Build: N/A (working tree state only)
881 lines
28 KiB
Markdown
881 lines
28 KiB
Markdown
# PIST-GCL Compression Algorithm Validation Plan
|
||
|
||
**Version:** 2.0
|
||
**Date:** 2026-05-31
|
||
**File Under Validation:** `/home/allaun/Research Stack/5-Applications/pist-scripts/pist_gcl_compression.py`
|
||
|
||
---
|
||
|
||
## Executive Summary
|
||
|
||
This validation plan documents comprehensive testing procedures for the PIST-GCL (PIST Geometry + Cognitive Load) manifold compression algorithm. The algorithm implements a 4-layer architecture combining number-theoretic geometry (PIST), cognitive load routing, variable-length delta encoding with PTOS dictionary, and thermodynamic verification.
|
||
|
||
**Key Claim:** The algorithm achieves lossless roundtrip compression with thermodynamic compliance (`dS/dt ≤ 0`) across 12 diverse test cases.
|
||
|
||
---
|
||
|
||
## 1. Core Component Analysis
|
||
|
||
### 1.1 PIST Encoding Functions
|
||
|
||
**Location:** Lines 38-64
|
||
|
||
| Function | Purpose | Formula |
|
||
|----------|---------|---------|
|
||
| `pist_encode(n: int) -> tuple` | Encode byte → (shell, offset) | `k = ⌊√n⌋, t = n - k²` |
|
||
| `pist_decode(k: int, t: int) -> int` | Decode coordinates back to byte | `n = k² + t` |
|
||
| `pist_mass(k: int, t: int) -> int` | Compute PIST mass | `mass = t·(2k+1-t)` |
|
||
| `pist_normalized_tension(k, t) -> float` | Normalized phase indicator | `ρ = t/(2k+1)` |
|
||
| `pist_phase(k, t) -> str` | Phase classification | `grounded` if mass=0, else `seismic` |
|
||
|
||
**Validation Requirements:**
|
||
1. Verify formula `mass = t·(2k+1-t)` for all 256 byte values (0-255)
|
||
2. Confirm grounded bytes (perfect squares: 0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225) have mass=0
|
||
3. Verify maximum mass occurs at shell midpoint (t ≈ k+0.5)
|
||
4. Confirm `pist_mirror(k, t) = (k, 2k+1-t)` preserves mass
|
||
|
||
---
|
||
|
||
### 1.2 Cognitive Load Functions
|
||
|
||
**Location:** Lines 73-131
|
||
|
||
| Function | Purpose |
|
||
|----------|---------|
|
||
| `intrinsic_load(data: bytes) -> float` | Shannon entropy: `L_I = -Σ p·log₂(p)` |
|
||
| `extraneous_load_bpb(data: bytes) -> float` | BPB penalty: `L_E = max(0, BPB - optimal)` |
|
||
| `germane_load(L_E, trust=0.5, S=10) -> float` | Learning from experience |
|
||
| `surprise_metric(mi_actual, mi_predicted) -> float` | Triggers learning |
|
||
|
||
**HomeostaticCanal Class:**
|
||
- Adaptive threshold: `threshold = 0.3 + 2.0·pressure` (0.3 to 20.3)
|
||
- Canal width: `λ_t = λ₀·(σ + (1-σ)·e^{-ξ·p_t})`
|
||
- Pressure update: `p_{t+1} = 0.95·p_t + (0.6·surprise + 0.4·regret)`
|
||
|
||
---
|
||
|
||
### 1.3 Dictionary and Entropy Coding
|
||
|
||
**Location:** Lines 139-248
|
||
|
||
**PTOSDictionary Class:**
|
||
- 8-entry initial dictionary + learned entries (up to 256)
|
||
- 4-byte code format: `0x41534D00 | i`
|
||
- Methods: `lookup(quad)`, `learn(quad)`
|
||
|
||
**HuffmanCoder Class:**
|
||
- Static Huffman tree from byte frequencies
|
||
- Methods: `build(data)`, `encode(data)`, `decode(data)`
|
||
- Stores padding count in last byte
|
||
|
||
---
|
||
|
||
### 1.4 Compression Pipeline
|
||
|
||
**Location:** Lines 362-835
|
||
|
||
**PISTGCLCompressor Class Architecture:**
|
||
|
||
```
|
||
Layer 0: PIST Remap
|
||
└─ bytes → (shell, offset, mass, tension, phase)
|
||
|
||
Layer 1: Cognitive Route
|
||
├─ Compute L_I (intrinsic load)
|
||
├─ Compute L_E (extraneous load via BPB)
|
||
├─ Update canal pressure
|
||
└─ Route/drop based on mass/tension
|
||
|
||
Layer 2a: Variable-Length Delta Encoding
|
||
├─ RLE for consecutive identical (shell, mass) pairs
|
||
├─ 1-byte: coarse deltas (5-bit shell + 3-bit mass)
|
||
├─ 2-byte: medium deltas (8-bit shell + 12-bit mass)
|
||
└─ 3-byte: wide deltas (16-bit shell + 16-bit mass)
|
||
|
||
Layer 2b: PTOS + VLE + Huffman
|
||
├─ PTOS dictionary lookup on 4-byte quads
|
||
├─ VLE encoding (0xFE=multi, 0xFF=escape)
|
||
└─ Huffman entropy coding
|
||
|
||
Layer 3: Thermodynamic Verify
|
||
├─ Compute dS = S_out - S_in
|
||
├─ Verify dS ≤ threshold (relaxed for small blocks)
|
||
└─ Compute Landauer bound
|
||
```
|
||
|
||
**Compression Header Format:**
|
||
```
|
||
[orig_len:4 bytes][routed_count:2 bytes][dropped:2 bytes]
|
||
```
|
||
|
||
**RLE Format:**
|
||
```
|
||
0xFD + shell(1) + mass(1) + count(1)
|
||
```
|
||
|
||
---
|
||
|
||
## 2. Component-Level Validation Tests
|
||
|
||
### 2.1 PIST Encoding Validation
|
||
|
||
**Test File:** `test_pist_encoding.py`
|
||
|
||
```python
|
||
def test_pist_mass_formula():
|
||
"""Verify mass = t·(2k+1-t) for all 256 byte values."""
|
||
failures = []
|
||
for n in range(256):
|
||
k, t = pist_encode(n)
|
||
expected_mass = t * (2 * k + 1 - t)
|
||
actual_mass = pist_mass(k, t)
|
||
if expected_mass != actual_mass:
|
||
failures.append((n, k, t, expected_mass, actual_mass))
|
||
|
||
assert len(failures) == 0, f"Mass formula failures: {failures}"
|
||
|
||
# Verify grounded bytes (perfect squares)
|
||
perfect_squares = [0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225]
|
||
for n in perfect_squares:
|
||
k, t = pist_encode(n)
|
||
assert pist_mass(k, t) == 0, f"Grounded byte {n} has non-zero mass"
|
||
|
||
# Verify mirror preserves mass
|
||
for n in range(256):
|
||
k, t = pist_encode(n)
|
||
m1 = pist_mass(k, t)
|
||
k2, t2 = pist_mirror(k, t)
|
||
m2 = pist_mass(k2, t2)
|
||
assert m1 == m2, f"Mirror fails for {n}: {m1} != {m2}"
|
||
```
|
||
|
||
**Expected Results:**
|
||
- All 256 bytes encode/decode correctly
|
||
- 16 grounded bytes (perfect squares) have mass=0
|
||
- Mirror involution preserves mass for all bytes
|
||
|
||
---
|
||
|
||
### 2.2 Delta Encoding Validation
|
||
|
||
**Test File:** `test_delta_encoding.py`
|
||
|
||
**RLE Validation:**
|
||
```python
|
||
def test_rle_two_byte_run_field():
|
||
"""Verify 2-byte run field handles runs ≥256 correctly."""
|
||
compressor = PISTGCLCompressor()
|
||
|
||
# Create data with run > 255
|
||
# Note: Current implementation uses 1-byte run field (max 255)
|
||
# This is a known limitation that should be documented
|
||
long_run_data = bytes([42] * 300)
|
||
result = compressor.compress(long_run_data)
|
||
|
||
# Verify compression works (may not hit RLE if routed bytes differ)
|
||
decompressed = compressor.decompress(result['compressed'])
|
||
assert decompressed == long_run_data, "RLE roundtrip failed"
|
||
```
|
||
|
||
**Note:** The current RLE implementation uses a 1-byte run field (max 255). For runs ≥256, the implementation will create multiple RLE markers.
|
||
|
||
---
|
||
|
||
### 2.3 Cognitive Routing Validation
|
||
|
||
**Test File:** `test_cognitive_routing.py`
|
||
|
||
```python
|
||
def test_cognitive_route_routed_count():
|
||
"""Verify routing logic and routed count."""
|
||
compressor = PISTGCLCompressor()
|
||
|
||
# High entropy data should route more bytes
|
||
random_data = bytes([i % 256 for i in range(256)])
|
||
coords = compressor._pist_remap_block(random_data)
|
||
routed, dropped, stats = compressor._cognitive_route(coords)
|
||
assert stats['routed'] > 0, "No bytes routed for random data"
|
||
assert stats['total'] == len(coords), "Total count mismatch"
|
||
|
||
# Low entropy data may drop more (grounded/low-tension)
|
||
zero_data = bytes(200)
|
||
coords = compressor._pist_remap_block(zero_data)
|
||
routed, dropped, stats = compressor._cognitive_route(coords)
|
||
# Grounded bytes (all zeros) should be dropped
|
||
assert dropped > 0, "No bytes dropped for zero data"
|
||
```
|
||
|
||
**Thermodynamic Validation:**
|
||
```python
|
||
def test_thermodynamic_compliance():
|
||
"""Verify dS/dt ≤ 0 for real compression."""
|
||
compressor = PISTGCLCompressor()
|
||
|
||
# High entropy data should compress well
|
||
random_data = bytes([i % 256 for i in range(1024)])
|
||
result = compressor.compress(random_data)
|
||
stats = result['stats']
|
||
|
||
# dS/dt should be ≤ threshold (relaxed for small blocks)
|
||
dS_dt = stats['dS/dt']
|
||
if len(random_data) < 100:
|
||
assert dS_dt <= 2.0, f"dS/dt exceeds threshold: {dS_dt}"
|
||
else:
|
||
assert dS_dt <= 1.0, f"dS/dt exceeds threshold: {dS_dt}"
|
||
|
||
# Landauer bound should be positive
|
||
assert stats['landauer_bound_J'] >= 0, "Negative Landauer bound"
|
||
```
|
||
|
||
---
|
||
|
||
### 2.4 PTOS Dictionary Validation
|
||
|
||
**Test File:** `test_ptos_dictionary.py`
|
||
|
||
```python
|
||
def test_ptos_dictionary_lookup():
|
||
"""Verify PTOS dictionary lookup and learning."""
|
||
ptos = PTOSDictionary()
|
||
|
||
# Initial entries should be found
|
||
initial_quad = b'\x00\x00\x00\x00'
|
||
code = ptos.lookup(initial_quad)
|
||
assert code != initial_quad, "Initial entry not found"
|
||
|
||
# Learned entries should be added
|
||
new_quad = b'\xAB\xCD\xEF\x12'
|
||
ptos.learn(new_quad)
|
||
code = ptos.lookup(new_quad)
|
||
assert code != new_quad, "Learned entry not found"
|
||
assert len(ptos.entries) <= 256, "Dictionary exceeds max size"
|
||
```
|
||
|
||
---
|
||
|
||
### 2.5 Huffman Coding Validation
|
||
|
||
**Test File:** `test_huffman_coding.py`
|
||
|
||
```python
|
||
def test_huffman_roundtrip():
|
||
"""Verify Huffman encode/decode roundtrip."""
|
||
coder = HuffmanCoder()
|
||
|
||
test_data = bytes([0, 1, 2, 3, 4, 5, 6, 7] * 16)
|
||
|
||
# Build and encode
|
||
coder.build(test_data)
|
||
encoded = coder.encode(test_data)
|
||
|
||
# Decode
|
||
decoded = coder.decode(encoded)
|
||
|
||
assert decoded == test_data, "Huffman roundtrip failed"
|
||
assert len(encoded) <= len(test_data), "Huffman expansion on compressible data"
|
||
```
|
||
|
||
---
|
||
|
||
## 3. Roundtrip Validation Tests (All 12 Test Cases)
|
||
|
||
**Test File:** `test_roundtrip_all_cases.py`
|
||
|
||
### 3.1 Test Case Specifications
|
||
|
||
| # | Test Case | Size | Pattern | Expected Behavior |
|
||
|---|-----------|------|---------|-------------------|
|
||
| 1 | English text | 356 bytes | ASCII text | Moderate compression, some routed bytes |
|
||
| 2 | Zero block | 200 bytes | All `\x00` | High compression, all grounded |
|
||
| 3 | Uniform random | 1024 bytes | 0-255 repeat | Low compression, high entropy |
|
||
| 4 | Periodic AB | 256 bytes | `AB`×128 | High compression, low entropy |
|
||
| 5 | Periodic ABC | 255 bytes | `ABC`×85 | High compression, low entropy |
|
||
| 6 | Rising sawtooth | 512 bytes | `i%256` | Moderate compression, sequential pattern |
|
||
| 7 | Low entropy ramp | 128 bytes | `0..127` | Moderate compression, monotonic |
|
||
| 8 | Binary bitmask | 256 bytes | `0xFF if i%3==0` | Moderate compression, sparse |
|
||
| 9 | Pattern 0x00-0x07 | 256 bytes | `i%8` | High compression, 3-bit pattern |
|
||
| 10 | All same byte | 200 bytes | `\x42`×200 | High compression, RLE ideal |
|
||
| 11 | Empty bytes | 0 bytes | Edge case | Minimal header |
|
||
| 12 | Single byte | 1 byte | Edge case | Minimal header + byte |
|
||
|
||
### 3.2 Roundtrip Test Implementation
|
||
|
||
```python
|
||
def test_roundtrip_all_cases():
|
||
"""Test roundtrip for all 12 cases."""
|
||
compressor = PISTGCLCompressor()
|
||
|
||
test_cases = [
|
||
("English text", b'Hello, World! This is a test of the PIST-GCL manifold compression algorithm. '
|
||
b'The algorithm uses PIST geometry (hyperbolic shell coordinates from number theory), '
|
||
b'cognitive load routing (BPB-aware adaptation via homeostatic control), '
|
||
b'and thermodynamic verification (Landauer limit, dS/dt <= 0). '
|
||
b'This demonstrates cross-domain compression on natural language.'),
|
||
("Zero block", b'\x00' * 200),
|
||
("Uniform random", bytes(range(256)) * 4),
|
||
("Periodic AB", b'AB' * 128),
|
||
("Periodic ABC", b'ABC' * 85),
|
||
("Rising sawtooth", bytes(i % 256 for i in range(512))),
|
||
("Low entropy ramp", bytes(i for i in range(128))),
|
||
("Binary bitmask", bytes(0xFF if i % 3 == 0 else 0x00 for i in range(256))),
|
||
("Pattern 0x00-0x07", bytes(i % 8 for i in range(256))),
|
||
("All same byte", b'\x42' * 200),
|
||
("Empty bytes", b''),
|
||
("Single byte", b'\x42'),
|
||
]
|
||
|
||
results = []
|
||
for name, data in test_cases:
|
||
result = compressor.compress(data)
|
||
decompressed = compressor.decompress(result['compressed'])
|
||
|
||
stats = result['stats']
|
||
|
||
# Verify roundtrip
|
||
roundtrip_ok = (decompressed == data)
|
||
|
||
# Verify compression ratio >= 1.0
|
||
ratio_ok = stats.get('ratio', 0) >= 1.0
|
||
|
||
# Verify thermodynamic compliance
|
||
dS_dt = stats.get('dS/dt', 0)
|
||
threshold = 2.0 if len(data) < 100 else 1.0
|
||
thermodynamic_ok = dS_dt <= threshold
|
||
|
||
# Verify routed count
|
||
routed_count_ok = stats.get('routed_count', 0) >= 0
|
||
|
||
# Verify verified flag
|
||
verified_ok = result['verified'] in [True, False]
|
||
|
||
results.append({
|
||
'name': name,
|
||
'original_size': len(data),
|
||
'compressed_size': stats.get('compressed_size', 0),
|
||
'ratio': stats.get('ratio', 0),
|
||
'dS_dt': dS_dt,
|
||
'routed_count': stats.get('routed_count', 0),
|
||
'dropped': stats.get('dropped', 0),
|
||
'roundtrip_ok': roundtrip_ok,
|
||
'ratio_ok': ratio_ok,
|
||
'thermodynamic_ok': thermodynamic_ok,
|
||
'routed_count_ok': routed_count_ok,
|
||
'verified_ok': verified_ok,
|
||
})
|
||
|
||
status = "PASS" if all([
|
||
roundtrip_ok, ratio_ok, thermodynamic_ok,
|
||
routed_count_ok, verified_ok
|
||
]) else "FAIL"
|
||
|
||
print(f"{status} {name:20s} orig={len(data):4d} cmp={stats.get('compressed_size', 0):4d} "
|
||
f"ratio={stats.get('ratio', 0):5.2f}x dS={dS_dt:+6.3f} routed={stats.get('routed_count', 0):4d}")
|
||
|
||
# Summary
|
||
all_pass = all(r['roundtrip_ok'] and r['ratio_ok'] for r in results)
|
||
print(f"\n{'ALL TESTS PASS' if all_pass else 'SOME TESTS FAILED'}")
|
||
|
||
return results
|
||
```
|
||
|
||
---
|
||
|
||
## 4. Concurrent Compression Validation
|
||
|
||
**Test File:** `test_concurrent_compression.py`
|
||
|
||
### 4.1 Thread-Safety Requirements
|
||
|
||
The `concurrent_compress()` method (not yet implemented in the file) must:
|
||
|
||
1. Split data into chunks
|
||
2. Create fresh `PTOSDictionary()` and `HuffmanCoder()` per chunk (via `copy.deepcopy(self)`)
|
||
3. Compress chunks in parallel
|
||
4. Merge results in order via `as_completed`
|
||
|
||
### 4.2 Implementation Template
|
||
|
||
```python
|
||
import copy
|
||
import concurrent.futures
|
||
|
||
def concurrent_compress(self, data: bytes, num_workers: int = 4) -> dict:
|
||
"""
|
||
Thread-safe concurrent compression.
|
||
Each chunk gets fresh PTOSDictionary and HuffmanCoder via deepcopy.
|
||
"""
|
||
if not data:
|
||
return {'compressed': b'\x00\x00\x00\x00', 'stats': {}, 'verified': True}
|
||
|
||
# Split data into chunks
|
||
chunk_size = max(1, len(data) // num_workers)
|
||
chunks = []
|
||
for i in range(0, len(data), chunk_size):
|
||
chunks.append(data[i:i + chunk_size])
|
||
|
||
# Compress chunks in parallel
|
||
results = []
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
|
||
futures = {}
|
||
for i, chunk in enumerate(chunks):
|
||
# Create fresh compressor instance per chunk
|
||
fresh_compressor = copy.deepcopy(self)
|
||
futures[executor.submit(fresh_compressor.compress, chunk)] = i
|
||
|
||
# Collect results in order
|
||
ordered_results = [None] * len(chunks)
|
||
for future in concurrent.futures.as_completed(futures):
|
||
idx = futures[future]
|
||
ordered_results[idx] = future.result()
|
||
|
||
# Merge compressed data
|
||
merged_compressed = b''.join(r['compressed'] for r in ordered_results)
|
||
|
||
# Merge stats (simplified: sum sizes, average ratios)
|
||
merged_stats = {
|
||
'original_size': len(data),
|
||
'compressed_size': len(merged_compressed),
|
||
'ratio': len(data) / max(1, len(merged_compressed)),
|
||
'chunks': len(chunks),
|
||
}
|
||
|
||
# Verify all chunks verified
|
||
verified = all(r['verified'] for r in ordered_results)
|
||
|
||
return {'compressed': merged_compressed, 'stats': merged_stats, 'verified': verified}
|
||
```
|
||
|
||
### 4.3 Thread-Safety Tests
|
||
|
||
```python
|
||
def test_concurrent_thread_safety():
|
||
"""Verify concurrent compression produces same result as sequential."""
|
||
original_data = bytes(range(256)) * 4 # 1024 bytes
|
||
num_workers = 4
|
||
|
||
# Sequential compression
|
||
seq_compressor = PISTGCLCompressor()
|
||
seq_result = seq_compressor.compress(original_data)
|
||
|
||
# Concurrent compression
|
||
conc_compressor = PISTGCLCompressor()
|
||
conc_result = conc_compressor.concurrent_compress(original_data, num_workers)
|
||
|
||
# Decompress and verify
|
||
seq_decompressed = seq_compressor.decompress(seq_result['compressed'])
|
||
conc_decompressed = conc_compressor.decompress(conc_result['compressed'])
|
||
|
||
assert seq_decompressed == original_data, "Sequential roundtrip failed"
|
||
assert conc_decompressed == original_data, "Concurrent roundtrip failed"
|
||
assert seq_decompressed == conc_decompressed, "Concurrent differs from sequential"
|
||
|
||
print("PASS: Concurrent compression thread-safe")
|
||
|
||
|
||
def test_concurrent_fresh_instances():
|
||
"""Verify each chunk gets fresh PTOSDictionary and HuffmanCoder."""
|
||
data = bytes(range(256)) * 2
|
||
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
|
||
results = list(executor.map(
|
||
lambda chunk: PISTGCLCompressor().compress(chunk),
|
||
[data[:256], data[256:]]
|
||
))
|
||
|
||
# Verify both chunks compress correctly
|
||
for i, result in enumerate(results):
|
||
compressor = PISTGCLCompressor()
|
||
decompressed = compressor.decompress(result['compressed'])
|
||
assert decompressed == data[i*256:(i+1)*256], f"Chunk {i} roundtrip failed"
|
||
|
||
print("PASS: Fresh instances per chunk")
|
||
```
|
||
|
||
---
|
||
|
||
## 5. Adaptive Precision Validation
|
||
|
||
**Test File:** `test_adaptive_precision.py`
|
||
|
||
### 5.1 Precision Modes
|
||
|
||
| Mode | Delta Encoding | RLE | Use Case |
|
||
|------|----------------|-----|----------|
|
||
| High | 2-3 bytes/coord | Conservative | High-entropy data |
|
||
| Medium | 1-2 bytes | Balanced | Mixed entropy |
|
||
| Low | Coarse deltas | Aggressive | Low-entropy data |
|
||
|
||
### 5.2 Auto Mode Implementation
|
||
|
||
```python
|
||
def compress_adaptive(self, data: bytes, mode: str = 'auto') -> dict:
|
||
"""
|
||
Adaptive precision compression.
|
||
|
||
Modes:
|
||
- 'high': Fine-grained deltas (2-3 bytes/coord)
|
||
- 'medium': Balanced (1-2 bytes)
|
||
- 'low': Coarse deltas + aggressive RLE
|
||
- 'auto': Measure distribution and pick best
|
||
"""
|
||
if mode == 'auto':
|
||
# Measure distribution
|
||
byte_freq = Counter(data)
|
||
entropy = intrinsic_load(data)
|
||
|
||
# Simple heuristic: low entropy → low precision
|
||
if entropy < 4.0:
|
||
mode = 'low'
|
||
elif entropy < 6.0:
|
||
mode = 'medium'
|
||
else:
|
||
mode = 'high'
|
||
|
||
# Store mode for delta encoding
|
||
self._precision_mode = mode
|
||
|
||
return self.compress(data)
|
||
```
|
||
|
||
### 5.3 Adaptive Tests
|
||
|
||
```python
|
||
def test_adaptive_precision_modes():
|
||
"""Test all precision modes."""
|
||
test_cases = [
|
||
("Low entropy ramp", bytes(i for i in range(128)), 'low'),
|
||
("Periodic AB", b'AB' * 128, 'low'),
|
||
("Uniform random", bytes(range(256)) * 4, 'high'),
|
||
("English text", b'Hello World! ' * 20, 'medium'),
|
||
]
|
||
|
||
for name, data, expected_mode in test_cases:
|
||
compressor = PISTGCLCompressor()
|
||
|
||
# Test specific mode
|
||
result = compressor.compress_adaptive(data, mode=expected_mode)
|
||
decompressed = compressor.decompress(result['compressed'])
|
||
|
||
assert decompressed == data, f"{name} {expected_mode} mode roundtrip failed"
|
||
|
||
# Verify compression ratio
|
||
ratio = result['stats']['ratio']
|
||
assert ratio >= 1.0, f"{name} {expected_mode} mode ratio < 1.0"
|
||
|
||
print(f"PASS: {name} {expected_mode} mode ratio={ratio:.2f}x")
|
||
|
||
|
||
def test_auto_precision_selection():
|
||
"""Test auto mode selects appropriate precision."""
|
||
test_cases = [
|
||
("Very low entropy", bytes([0, 1, 2, 3] * 100), 'low'),
|
||
("Low entropy", bytes(range(64)) * 4, 'low'),
|
||
("Medium entropy", bytes(range(128)) * 2, 'medium'),
|
||
("High entropy", bytes([i % 256 for i in range(1024)]), 'high'),
|
||
]
|
||
|
||
for name, data, expected_mode in test_cases:
|
||
compressor = PISTGCLCompressor()
|
||
|
||
# Auto mode
|
||
result = compressor.compress_adaptive(data, mode='auto')
|
||
|
||
# Verify compression works
|
||
decompressed = compressor.decompress(result['compressed'])
|
||
assert decompressed == data, f"{name} auto mode roundtrip failed"
|
||
|
||
print(f"PASS: {name} auto mode selected")
|
||
```
|
||
|
||
---
|
||
|
||
## 6. Edge Case Validation
|
||
|
||
**Test File:** `test_edge_cases.py`
|
||
|
||
### 6.1 Edge Cases
|
||
|
||
| Case | Input | Expected Behavior |
|
||
|------|-------|-------------------|
|
||
| Empty | `b''` | Header only, verified=True |
|
||
| Single byte | `b'\x42'` | Minimal header + data |
|
||
| All zeros | `b'\x00' * 1024` | All grounded, high compression |
|
||
| All same | `b'\xFF' * 1024` | Low entropy, good compression |
|
||
| Max values | `bytes(range(256))` | High entropy, poor compression |
|
||
|
||
### 6.2 Edge Case Tests
|
||
|
||
```python
|
||
def test_edge_cases():
|
||
"""Test edge cases."""
|
||
compressor = PISTGCLCompressor()
|
||
|
||
edge_cases = [
|
||
("Empty", b''),
|
||
("Single byte", b'\x42'),
|
||
("Two bytes", b'\x00\xFF'),
|
||
("All zeros 1KB", b'\x00' * 1024),
|
||
("All same 1KB", b'\xFF' * 1024),
|
||
("Max values", bytes(range(256))),
|
||
("Max values x4", bytes(range(256)) * 4),
|
||
]
|
||
|
||
for name, data in edge_cases:
|
||
result = compressor.compress(data)
|
||
decompressed = compressor.decompress(result['compressed'])
|
||
|
||
# Verify roundtrip
|
||
assert decompressed == data, f"{name} roundtrip failed"
|
||
|
||
# Verify stats
|
||
stats = result['stats']
|
||
assert stats['original_size'] == len(data), f"{name} size mismatch"
|
||
assert stats['ratio'] >= 1.0, f"{name} ratio < 1.0"
|
||
|
||
# Verify verified flag
|
||
assert result['verified'] in [True, False], f"{name} invalid verified flag"
|
||
|
||
print(f"PASS: {name} orig={len(data):4d} cmp={stats['compressed_size']:4d} ratio={stats['ratio']:.2f}x")
|
||
```
|
||
|
||
---
|
||
|
||
## 7. Performance and Scalability Validation
|
||
|
||
**Test File:** `test_performance.py`
|
||
|
||
### 7.1 Performance Metrics
|
||
|
||
| Metric | Target | Notes |
|
||
|--------|--------|-------|
|
||
| Roundtrip correctness | 100% | All data must reconstruct |
|
||
| Compression ratio | ≥1.0 | No expansion on compressible data |
|
||
| dS/dt ≤ threshold | 100% | Thermodynamic compliance |
|
||
| Memory usage | O(n) | Linear in input size |
|
||
| Thread safety | Verified | Concurrent produces same result |
|
||
|
||
### 7.2 Performance Tests
|
||
|
||
```python
|
||
def test_performance_scalability():
|
||
"""Test compression performance across sizes."""
|
||
import time
|
||
|
||
compressor = PISTGCLCompressor()
|
||
|
||
sizes = [16, 64, 256, 1024, 4096, 16384, 65536]
|
||
|
||
print(f"{'Size':>8} {'Time (ms)':>12} {'Ratio':>8} {'dS/dt':>8}")
|
||
print("-" * 40)
|
||
|
||
for size in sizes:
|
||
data = bytes([i % 256 for i in range(size)])
|
||
|
||
t0 = time.time()
|
||
result = compressor.compress(data)
|
||
t1 = time.time()
|
||
|
||
decompressed = compressor.decompress(result['compressed'])
|
||
t2 = time.time()
|
||
|
||
assert decompressed == data, f"Roundtrip failed for size {size}"
|
||
|
||
elapsed_compress = (t1 - t0) * 1000
|
||
elapsed_decompress = (t2 - t1) * 1000
|
||
ratio = result['stats']['ratio']
|
||
dS_dt = result['stats']['dS/dt']
|
||
|
||
print(f"{size:>8} {elapsed_compress:>10.2f} {ratio:>8.2f}x {dS_dt:>+7.3f}")
|
||
|
||
print("\nPASS: Performance scalability verified")
|
||
```
|
||
|
||
---
|
||
|
||
## 8. Validation Checklist
|
||
|
||
### 8.1 Component Validation
|
||
|
||
- [ ] PIST mass formula verified for all 256 bytes
|
||
- [ ] PIST encoding/decoding roundtrip verified
|
||
- [ ] Mirror involution preserves mass
|
||
- [ ] Grounded bytes (perfect squares) have mass=0
|
||
- [ ] Cognitive load functions compute correctly
|
||
- [ ] Homeostatic canal state updates properly
|
||
- [ ] PTOS dictionary lookup/learn works
|
||
- [ ] Huffman encode/decode roundtrip works
|
||
- [ ] Delta encoding handles all three precision modes
|
||
- [ ] RLE handles runs up to 255
|
||
- [ ] VLE encoding/decoding works
|
||
|
||
### 8.2 Roundtrip Validation
|
||
|
||
- [ ] English text (356 bytes) roundtrip
|
||
- [ ] Zero block (200 bytes) roundtrip
|
||
- [ ] Uniform random (1024 bytes) roundtrip
|
||
- [ ] Periodic AB (256 bytes) roundtrip
|
||
- [ ] Periodic ABC (255 bytes) roundtrip
|
||
- [ ] Rising sawtooth (512 bytes) roundtrip
|
||
- [ ] Low entropy ramp (128 bytes) roundtrip
|
||
- [ ] Binary bitmask (256 bytes) roundtrip
|
||
- [ ] Pattern 0x00-0x07 (256 bytes) roundtrip
|
||
- [ ] All same byte (200 bytes) roundtrip
|
||
- [ ] Empty bytes edge case
|
||
- [ ] Single byte edge case
|
||
|
||
### 8.3 Thermodynamic Validation
|
||
|
||
- [ ] dS/dt ≤ threshold for all test cases
|
||
- [ ] Landauer bound computed correctly
|
||
- [ ] Compression ratio ≥ 1.0 for all cases
|
||
|
||
### 8.4 Concurrent Validation
|
||
|
||
- [ ] Thread-safe concurrent compression
|
||
- [ ] Fresh PTOSDictionary per chunk
|
||
- [ ] Fresh HuffmanCoder per chunk
|
||
- [ ] Results merge in order via as_completed
|
||
|
||
### 8.5 Adaptive Precision Validation
|
||
|
||
- [ ] High precision mode works
|
||
- [ ] Medium precision mode works
|
||
- [ ] Low precision mode works
|
||
- [ ] Auto mode selects appropriate precision
|
||
|
||
### 8.6 Edge Case Validation
|
||
|
||
- [ ] Empty input handled
|
||
- [ ] Single byte handled
|
||
- [ ] All zeros handled
|
||
- [ ] All same byte handled
|
||
- [ ] Max values handled
|
||
|
||
---
|
||
|
||
## 9. Expected Results Summary
|
||
|
||
### 9.1 Compression Ratios (Expected)
|
||
|
||
| Test Case | Expected Ratio | Notes |
|
||
|-----------|----------------|-------|
|
||
| English text | 1.2-1.5x | Moderate compression |
|
||
| Zero block | 10-20x | All grounded bytes dropped |
|
||
| Uniform random | 0.8-1.0x | Near Shannon limit |
|
||
| Periodic AB | 5-10x | High pattern redundancy |
|
||
| Periodic ABC | 5-10x | High pattern redundancy |
|
||
| Rising sawtooth | 1.5-2.0x | Sequential pattern |
|
||
| Low entropy ramp | 2-3x | Monotonic pattern |
|
||
| Binary bitmask | 2-3x | Sparse pattern |
|
||
| Pattern 0x00-0x07 | 3-5x | 3-bit pattern |
|
||
| All same byte | 10-20x | Ideal RLE case |
|
||
| Empty | 1.0x | Edge case |
|
||
| Single byte | 1.0x | Overhead dominates |
|
||
|
||
### 9.2 Thermodynamic Compliance
|
||
|
||
All test cases should satisfy:
|
||
- `dS/dt ≤ 2.0` for blocks < 100 bytes
|
||
- `dS/dt ≤ 1.0` for blocks ≥ 100 bytes
|
||
- `Landauer_bound_J ≥ 0`
|
||
|
||
### 9.3 Verified Flag
|
||
|
||
All successful roundtrips should set `verified=True`. Small blocks may have relaxed thresholds.
|
||
|
||
---
|
||
|
||
## 10. Known Limitations
|
||
|
||
### 10.1 RLE Run Field
|
||
|
||
**Issue:** Current RLE uses 1-byte run field (max 255).
|
||
|
||
**Impact:** Runs ≥256 create multiple RLE markers.
|
||
|
||
**Mitigation:** Document limitation; consider 2-byte run field for future version.
|
||
|
||
### 10.2 Delta Decoding Approximation
|
||
|
||
**Issue:** Variable-length delta decoding may approximate values.
|
||
|
||
**Impact:** Small rounding errors possible for large deltas.
|
||
|
||
**Mitigation:** Verify roundtrip for all test cases; document if approximation occurs.
|
||
|
||
### 10.3 Concurrent Compression
|
||
|
||
**Issue:** `concurrent_compress()` method not yet implemented.
|
||
|
||
**Impact:** No parallel compression support.
|
||
|
||
**Mitigation:** Implement per Section 4.2; add thread-safety tests.
|
||
|
||
---
|
||
|
||
## 11. Validation Execution Plan
|
||
|
||
### Phase 1: Component Tests (1-2 hours)
|
||
- Run `test_pist_encoding.py`
|
||
- Run `test_delta_encoding.py`
|
||
- Run `test_cognitive_routing.py`
|
||
- Run `test_ptos_dictionary.py`
|
||
- Run `test_huffman_coding.py`
|
||
|
||
### Phase 2: Roundtrip Tests (30 minutes)
|
||
- Run `test_roundtrip_all_cases.py`
|
||
- Verify all 12 test cases pass
|
||
- Document any failures
|
||
|
||
### Phase 3: Thermodynamic Tests (30 minutes)
|
||
- Run thermodynamic compliance tests
|
||
- Verify dS/dt ≤ threshold
|
||
- Verify Landauer bounds
|
||
|
||
### Phase 4: Concurrent Tests (1 hour)
|
||
- Implement `concurrent_compress()` per Section 4.2
|
||
- Run thread-safety tests
|
||
- Verify ordered results
|
||
|
||
### Phase 5: Adaptive Tests (30 minutes)
|
||
- Run adaptive precision tests
|
||
- Verify auto mode selection
|
||
|
||
### Phase 6: Edge Case Tests (30 minutes)
|
||
- Run edge case tests
|
||
- Verify all edge cases handled
|
||
|
||
### Phase 7: Performance Tests (1 hour)
|
||
- Run scalability tests
|
||
- Measure compression/decompression speed
|
||
- Document performance characteristics
|
||
|
||
---
|
||
|
||
## 12. Success Criteria
|
||
|
||
The PIST-GCL compression algorithm is validated when:
|
||
|
||
1. **Roundtrip Correctness:** All 12 test cases pass roundtrip verification
|
||
2. **Compression Ratio:** All ratios ≥ 1.0 (no expansion on compressible data)
|
||
3. **Thermodynamic Compliance:** dS/dt ≤ threshold for 100% of test cases
|
||
4. **Thread Safety:** Concurrent compression produces identical results to sequential
|
||
5. **Edge Cases:** All edge cases (empty, single byte, max values) handled correctly
|
||
|
||
---
|
||
|
||
## 13. References
|
||
|
||
1. PIST Geometry: `mass = t·(2k+1-t)` — hyperbolic shell coordinate system
|
||
2. Cognitive Load: `L_I = -Σ p·log₂(p)`, `L_E = BPB - optimal`
|
||
3. Homeostatic Canal: `λ_t = λ₀·(σ + (1-σ)·e^{-ξ·p_t})`
|
||
4. Thermodynamic: `dS/dt ≤ 0`, Landauer bound `W ≥ N·k_B·T·ln(2)`
|
||
5. PTOS Dictionary: 4-byte codec with learning
|
||
6. Huffman Coding: Static entropy coding backend
|
||
|
||
---
|
||
|
||
**Document Version:** 1.0
|
||
**Last Updated:** 2026-05-31
|
||
**Prepared By:** Validation Planning System
|
||
**Status:** Ready for Execution
|