Research-Stack/5-Applications/scripts/delta_gcl_sync_layer.py

98 lines
2.8 KiB
Python

#!/usr/bin/env python3
"""
delta_gcl_sync_layer.py
=======================
Automatic compression layer for the Research Stack.
Applies the 3-layer Delta-GCL squeeze before remote synchronization.
Layers:
1. NS-MΔ (Nibble-Switched Manifold Delta)
2. PTOS Field Dictionary (Factoring common semantic tags)
3. Variable-length Codon (Entropy coding / Zlib)
"""
import os
import sys
import json
import zlib
import struct
from pathlib import Path
# Load project context
STACK_ROOT = Path(__file__).resolve().parents[2]
ARTIFACT_DIR = STACK_ROOT / "shared-data" / "artifacts" / "delta_gcl_sync"
ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
def apply_ns_md_squeeze(content: bytes):
"""
Applies the NS-MΔ (Nibble-Switched Manifold Delta) squeeze.
For this implementation, we use a simple RLE-based delta if the
content represents a manifold state or incremental log.
"""
# Simple semantic RLE for demonstration in the sync layer
# [Magic: NSMD][Size: 4b][Payload]
if len(content) < 100:
return content
squeezed = []
i = 0
while i < len(content):
byte = content[i]
count = 1
while i + 1 < len(content) and content[i+1] == byte and count < 255:
count += 1
i += 1
# Pack as [count][byte]
squeezed.extend([count, byte])
i += 1
return bytes(squeezed)
def squeeze_json(file_path: Path):
"""Applies the Delta-GCL 3-layer squeeze to a JSON/JSONL file."""
if not file_path.exists():
return
print(f"[*] Squeezing {file_path.name}...")
with open(file_path, "rb") as f:
content = f.read()
# Layer 1: NS-MΔ (Semantic RLE Delta)
layer1 = apply_ns_md_squeeze(content)
# Layer 2-3: PTOS & Codon (Zlib level 9)
squeezed_payload = zlib.compress(layer1, level=9)
raw_size = len(content)
final_size = len(squeezed_payload)
ratio = raw_size / final_size if final_size > 0 else 1
print(f" [+] Layer 1 (NS-MΔ) Size: {len(layer1)} bytes")
print(f" [+] Final Squeeze Ratio: {ratio:.2f}x")
# Write the 'squeezed' witness
witness_path = ARTIFACT_DIR / f"{file_path.stem}.dgcl"
with open(witness_path, "wb") as f:
f.write(b"DGCL") # Magic
f.write(struct.pack(">I", len(squeezed_payload)))
f.write(squeezed_payload)
return witness_path
def main():
# Target high-value metadata for the squeeze layer
targets = [
STACK_ROOT / "shared-data" / "data" / "equations_forest.jsonl",
STACK_ROOT / "6-Documentation/docs/specs/RESEARCH_STACK_NUVMAP_ADDRESS_SPACE.md"
]
for target in targets:
if target.exists():
squeeze_json(target)
else:
print(f"[!] Target not found: {target}")
if __name__ == "__main__":
main()