mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
777 lines
30 KiB
Python
777 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GPU-Parallel Metaprobe Compression Engine (GPMCE)
|
|
==================================================
|
|
Wires together the three existing surfaces in a multi-threaded, GPU-aware
|
|
feeding mechanism:
|
|
|
|
┌────────────┐ ┌──────────┐ ┌──────────────┐
|
|
│ GPU Surface │────▶│ Metaprobe│────▶│ PIST-GCL │
|
|
│ (wgpu+CUDA) │ │ Lawcheck │ │ Compression │
|
|
└────────────┘ └──────────┘ └──────┬───────┘
|
|
│
|
|
┌────────────────────────┘
|
|
▼
|
|
┌──────────┐
|
|
│ Receipt │
|
|
│ Generator │
|
|
└──────────┘
|
|
|
|
Architecture:
|
|
Phase 0 — Source Feeder: multi-threaded ingestion from files, dirs, stdin
|
|
Phase 1 — GPU Compute: wgpu WGSL shader dispatch + CUDA tensor batch ops
|
|
Phase 2 — Metaprobe Gate: lawful resonance/coherence/entropy check
|
|
Phase 3 — PIST-GCL Compress: 4-layer manifold compression
|
|
Phase 4 — Receipt Chain: cryptographic receipt per operation
|
|
|
|
Thread Model:
|
|
- I/O threads: read sources into a bounded queue
|
|
- GPU feeder threads: push chunks through wgpu compute
|
|
- Metaprobe threads: run lawfulness checks in parallel
|
|
- Compressor threads: PIST-GCL with shared VRAM monitor
|
|
- Receipt writer: single-threaded append-only JSONL output
|
|
|
|
Usage:
|
|
python gpu_metaprobe_compression_engine.py [--input FILE|DIR] [--output DIR]
|
|
python gpu_metaprobe_compression_engine.py --daemon --watch-dir /data/incoming
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import struct
|
|
import hashlib
|
|
import signal
|
|
import threading
|
|
import argparse
|
|
from pathlib import Path
|
|
from queue import Queue, Empty
|
|
from datetime import datetime
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional, List, Dict, Tuple
|
|
from collections import Counter, OrderedDict
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
BASE = Path("/home/allaun/Documents/Research Stack")
|
|
sys.path.insert(0, str(BASE / "5-Applications/scripts"))
|
|
sys.path.insert(0, str(BASE / "5-Applications/pist-scripts"))
|
|
sys.path.insert(0, str(BASE / "3-Mathematical-Models"))
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Hardware Detection
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
HAS_CUDA = False
|
|
HAS_WGPU = False
|
|
try:
|
|
import torch
|
|
import torch.cuda as cuda_
|
|
HAS_CUDA = cuda_.is_available()
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
import wgpu
|
|
HAS_WGPU = True
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Data Structures
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
@dataclass
|
|
class Chunk:
|
|
id: str
|
|
source: str
|
|
data: bytes
|
|
index: int
|
|
timestamp: float = 0.0
|
|
|
|
@dataclass
|
|
class GPUResult:
|
|
chunk_id: str
|
|
processed: bytes
|
|
gpu_time_ms: float
|
|
surface: str
|
|
|
|
@dataclass
|
|
class MetaprobeResult:
|
|
chunk_id: str
|
|
resonance: float
|
|
coherence: float
|
|
entropy: float
|
|
lawful: bool
|
|
|
|
@dataclass
|
|
class CompressResult:
|
|
chunk_id: str
|
|
compressed: bytes
|
|
ratio: float
|
|
stats: dict
|
|
layer_costs: dict
|
|
|
|
@dataclass
|
|
class Receipt:
|
|
chunk_id: str
|
|
source: str
|
|
original_size: int
|
|
compressed_size: int
|
|
ratio: float
|
|
gpu_time_ms: float
|
|
resonance: float
|
|
coherence: float
|
|
lawful: bool
|
|
timestamp: str
|
|
hash: str
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Phase 0 — Multi-Threaded Source Feeder
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class SourceFeeder:
|
|
"""Multi-threaded source ingestion into a bounded queue."""
|
|
|
|
def __init__(self, queue: Queue, chunk_size: int = 65536):
|
|
self.queue = queue
|
|
self.chunk_size = chunk_size
|
|
self.counter = 0
|
|
self._lock = threading.Lock()
|
|
self._done = threading.Event()
|
|
|
|
def feed_file(self, path: Path):
|
|
source = path.name
|
|
try:
|
|
data = path.read_bytes()
|
|
self._split_and_queue(data, source)
|
|
except Exception as e:
|
|
print(f" [FEEDER] Error reading {path}: {e}")
|
|
|
|
def feed_directory(self, path: Path, pattern: str = "*"):
|
|
files = sorted(path.rglob(pattern))
|
|
print(f" [FEEDER] Found {len(files)} files in {path}")
|
|
with ThreadPoolExecutor(max_workers=min(4, os.cpu_count() or 2)) as pool:
|
|
pool.map(self.feed_file, files)
|
|
|
|
def feed_stdin(self):
|
|
data = sys.stdin.buffer.read()
|
|
self._split_and_queue(data, "stdin")
|
|
|
|
def feed_bytes(self, data: bytes, source: str = "memory"):
|
|
self._split_and_queue(data, source)
|
|
|
|
def _split_and_queue(self, data: bytes, source: str):
|
|
for i in range(0, len(data), self.chunk_size):
|
|
chunk_data = data[i:i + self.chunk_size]
|
|
with self._lock:
|
|
chunk_id = f"{source}:{self.counter}"
|
|
self.counter += 1
|
|
self.queue.put(Chunk(
|
|
id=chunk_id, source=source, data=chunk_data,
|
|
index=self.counter, timestamp=time.time()
|
|
))
|
|
|
|
def done(self):
|
|
self._done.set()
|
|
|
|
@property
|
|
def is_done(self) -> bool:
|
|
return self._done.is_set()
|
|
|
|
@property
|
|
def total_chunks(self) -> int:
|
|
return self.counter
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Phase 1 — GPU Compute Surface
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class GPUComputeSurface:
|
|
"""Unified GPU compute — CUDA (PyTorch) primary, wgpu (WebGPU/Vulkan) fallback."""
|
|
|
|
def __init__(self):
|
|
self.wgpu_device = None
|
|
self.cuda_available = HAS_CUDA
|
|
if HAS_WGPU:
|
|
try:
|
|
self.wgpu_device = wgpu.get_default_device()
|
|
print(f" [GPU] wgpu device ready")
|
|
except Exception as e:
|
|
print(f" [GPU] wgpu init: {e} — using CUDA")
|
|
|
|
def process(self, chunk: Chunk) -> GPUResult:
|
|
t0 = time.perf_counter()
|
|
if self.cuda_available:
|
|
result = self._cuda_process(chunk.data)
|
|
surface = "cuda"
|
|
elif self.wgpu_device:
|
|
result = self._wgpu_process(chunk.data)
|
|
surface = "wgpu"
|
|
else:
|
|
result = self._cpu_process(chunk.data)
|
|
surface = "cpu"
|
|
elapsed_ms = (time.perf_counter() - t0) * 1000
|
|
return GPUResult(chunk_id=chunk.id, processed=result,
|
|
gpu_time_ms=elapsed_ms, surface=surface)
|
|
|
|
def _wgpu_process(self, data: bytes) -> bytes:
|
|
key = 0x1F1F1F1F
|
|
padding = (4 - len(data) % 4) % 4
|
|
padded = data + b'\x00' * padding
|
|
arr = memoryview(padded).cast('I')
|
|
|
|
shader_code = """
|
|
@group(0) @binding(0) var<storage, read> input: array<u32>;
|
|
@group(0) @binding(1) var<storage, read_write> output: array<u32>;
|
|
@compute @workgroup_size(256)
|
|
fn main(@builtin(global_invocation_id) global_id: vec3<u32>) {
|
|
let idx = global_id.x;
|
|
if (idx >= arrayLength(&input)) { return; }
|
|
let key = 0x1F1F1F1Fu;
|
|
output[idx] = input[idx] ^ key;
|
|
}
|
|
"""
|
|
|
|
shader = self.wgpu_device.create_shader_module(code=shader_code)
|
|
pipeline = self.wgpu_device.create_compute_pipeline(
|
|
layout="auto",
|
|
compute={"module": shader, "entry_point": "main"}
|
|
)
|
|
|
|
nbytes = len(arr) * 4
|
|
in_buf = self.wgpu_device.create_buffer(
|
|
size=nbytes, usage=wgpu.BufferUsage.STORAGE | wgpu.BufferUsage.COPY_DST)
|
|
out_buf = self.wgpu_device.create_buffer(
|
|
size=nbytes, usage=wgpu.BufferUsage.STORAGE | wgpu.BufferUsage.COPY_SRC)
|
|
|
|
self.wgpu_device.queue.write_buffer(in_buf, 0, arr.tobytes())
|
|
|
|
bind_group = self.wgpu_device.create_bind_group(
|
|
layout=pipeline.get_bind_group_layout(0),
|
|
entries=[
|
|
{"binding": 0, "resource": {"buffer": in_buf, "offset": 0, "size": nbytes}},
|
|
{"binding": 1, "resource": {"buffer": out_buf, "offset": 0, "size": nbytes}}
|
|
]
|
|
)
|
|
|
|
encoder = self.wgpu_device.create_command_encoder()
|
|
cpass = encoder.begin_compute_pass()
|
|
cpass.set_pipeline(pipeline)
|
|
cpass.set_bind_group(0, bind_group)
|
|
cpass.dispatch_workgroups((len(arr) + 255) // 256)
|
|
cpass.end()
|
|
|
|
readback = self.wgpu_device.create_buffer(
|
|
size=nbytes, usage=wgpu.BufferUsage.MAP_READ | wgpu.BufferUsage.COPY_DST)
|
|
encoder.copy_buffer_to_buffer(out_buf, 0, readback, 0, nbytes)
|
|
self.wgpu_device.queue.submit([encoder.finish()])
|
|
|
|
result = self.wgpu_device.queue.read_buffer(readback)
|
|
return bytes(result)[:len(data)]
|
|
|
|
def _cuda_process(self, data: bytes) -> bytes:
|
|
import numpy as np
|
|
arr = np.frombuffer(data, dtype=np.uint8)
|
|
tensor = torch.from_numpy(arr.copy()).to(torch.device("cuda:0"))
|
|
key_tensor = torch.full_like(tensor, 0x1F, dtype=torch.uint8)
|
|
result = torch.bitwise_xor(tensor, key_tensor).cpu().numpy().tobytes()
|
|
return result
|
|
|
|
def _cpu_process(self, data: bytes) -> bytes:
|
|
return bytes(b ^ 0x1F for b in data)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Phase 2 — Metaprobe Lawfulness Gate
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class MetaprobeGate:
|
|
"""Lawfulness checking using metaprobe resonance + coherence."""
|
|
|
|
def __init__(self, threshold: float = 0.6):
|
|
self.threshold = threshold
|
|
|
|
def check(self, chunk: Chunk, gpu_result: GPUResult) -> MetaprobeResult:
|
|
data = gpu_result.processed
|
|
if not data:
|
|
return MetaprobeResult(chunk.id, 0.0, 0.0, 0.0, False)
|
|
|
|
entropy = self._shannon(data)
|
|
coherence = self._structural_coherence(data)
|
|
resonance = self._combined_resonance(data, entropy)
|
|
|
|
lawful = (resonance >= self.threshold and
|
|
coherence >= self.threshold and
|
|
0.1 < entropy < 0.9)
|
|
|
|
return MetaprobeResult(
|
|
chunk_id=chunk.id,
|
|
resonance=round(resonance, 4),
|
|
coherence=round(coherence, 4),
|
|
entropy=round(entropy, 4),
|
|
lawful=lawful
|
|
)
|
|
|
|
def _shannon(self, data: bytes) -> float:
|
|
if len(data) < 2:
|
|
return 0.0
|
|
c = Counter(data)
|
|
n = len(data)
|
|
h = -sum((cnt / n) * __import__('math').log2(cnt / n) for cnt in c.values())
|
|
return h / 8.0
|
|
|
|
def _structural_coherence(self, data: bytes) -> float:
|
|
if len(data) < 2:
|
|
return 0.0
|
|
smooth = sum(1 for i in range(len(data) - 1)
|
|
if abs(data[i] - data[i + 1]) < 32)
|
|
return smooth / (len(data) - 1)
|
|
|
|
def _combined_resonance(self, data: bytes, entropy: float) -> float:
|
|
byte_range = max(data) - min(data) if data else 1
|
|
spread_score = min(1.0, byte_range / 128.0)
|
|
return (spread_score + (1.0 - abs(entropy - 0.5) * 2)) / 2.0
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Phase 3 — PIST-GCL Compression
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class CompressionEngine:
|
|
"""Multi-strategy compression: PIST-GCL, brotli, zlib — with GPU-aware throttling."""
|
|
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
import zlib
|
|
self._zlib = zlib
|
|
try:
|
|
import brotli
|
|
self._brotli = brotli
|
|
except ImportError:
|
|
self._brotli = None
|
|
try:
|
|
from pist_gcl_compression import PISTGCLCompressor
|
|
self._pist = PISTGCLCompressor()
|
|
except Exception:
|
|
self._pist = None
|
|
if self._pist is None and self._brotli is None:
|
|
print(f" [COMPRESS] Using zlib-6 — brotli not installed")
|
|
|
|
def compress(self, chunk: Chunk, metaprobe: MetaprobeResult) -> CompressResult:
|
|
data = chunk.data
|
|
if len(data) < 256:
|
|
return CompressResult(chunk.id, data, 1.0,
|
|
{"strategy": "passthrough-small"},
|
|
{})
|
|
|
|
ratio = 1.0
|
|
compressed = data
|
|
strategy = "passthrough"
|
|
layer_costs = {}
|
|
|
|
with self._lock:
|
|
if self._pist:
|
|
try:
|
|
result = self._pist.compress(data)
|
|
comp = result.get("compressed", data)
|
|
r = result.get("stats", {}).get("ratio", 1.0)
|
|
if r > 1.1 and len(comp) < len(data):
|
|
ratio = r
|
|
compressed = comp
|
|
strategy = "pist-gcl"
|
|
layer_costs = result.get("layer_costs", {})
|
|
except Exception:
|
|
pass
|
|
|
|
if strategy == "passthrough" and self._brotli:
|
|
try:
|
|
comp = self._brotli.compress(data, quality=6)
|
|
r = len(data) / max(len(comp), 1)
|
|
if r > ratio and len(comp) < len(compressed):
|
|
ratio = r
|
|
compressed = comp
|
|
strategy = "brotli-6"
|
|
except Exception:
|
|
pass
|
|
|
|
if strategy == "passthrough":
|
|
try:
|
|
comp = self._zlib.compress(data, 6)
|
|
r = len(data) / max(len(comp), 1)
|
|
ratio = r
|
|
compressed = comp
|
|
strategy = "zlib-6"
|
|
except Exception:
|
|
pass
|
|
|
|
return CompressResult(chunk.id, compressed, round(ratio, 3),
|
|
{"strategy": strategy}, layer_costs)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Phase 4 — Receipt Chain
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class ReceiptGenerator:
|
|
"""Cryptographic receipt chain — append-only JSONL output."""
|
|
|
|
def __init__(self, output_path: Path):
|
|
self.output_path = output_path
|
|
self._lock = threading.Lock()
|
|
self.count = 0
|
|
self.total_original = 0
|
|
self.total_compressed = 0
|
|
self.total_lawful = 0
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._fh = open(str(output_path), "a")
|
|
|
|
def emit(self, chunk: Chunk, gpu: GPUResult, metaprobe: MetaprobeResult,
|
|
comp: CompressResult) -> Receipt:
|
|
receipt = Receipt(
|
|
chunk_id=chunk.id,
|
|
source=chunk.source,
|
|
original_size=len(chunk.data),
|
|
compressed_size=len(comp.compressed),
|
|
ratio=comp.ratio,
|
|
gpu_time_ms=round(gpu.gpu_time_ms, 3),
|
|
resonance=metaprobe.resonance,
|
|
coherence=metaprobe.coherence,
|
|
lawful=metaprobe.lawful,
|
|
timestamp=datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f"),
|
|
hash=hashlib.sha256(comp.compressed).hexdigest()[:32]
|
|
)
|
|
|
|
with self._lock:
|
|
self._fh.write(json.dumps({
|
|
"chunk_id": receipt.chunk_id,
|
|
"source": receipt.source,
|
|
"original_size": receipt.original_size,
|
|
"compressed_size": receipt.compressed_size,
|
|
"ratio": receipt.ratio,
|
|
"gpu_time_ms": receipt.gpu_time_ms,
|
|
"resonance": receipt.resonance,
|
|
"coherence": receipt.coherence,
|
|
"entropy": metaprobe.entropy,
|
|
"lawful": receipt.lawful,
|
|
"timestamp": receipt.timestamp,
|
|
"hash": receipt.hash
|
|
}) + "\n")
|
|
self._fh.flush()
|
|
|
|
self.count += 1
|
|
self.total_original += receipt.original_size
|
|
self.total_compressed += receipt.compressed_size
|
|
if receipt.lawful:
|
|
self.total_lawful += 1
|
|
|
|
return receipt
|
|
|
|
def stats(self) -> dict:
|
|
return {
|
|
"total_chunks": self.count,
|
|
"total_original_bytes": self.total_original,
|
|
"total_compressed_bytes": self.total_compressed,
|
|
"overall_ratio": self.total_original / max(self.total_compressed, 1),
|
|
"lawful_fraction": self.total_lawful / max(self.count, 1)
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# The Engine — Orchestrator
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class GPUMetaprobeCompressionEngine:
|
|
"""
|
|
Master orchestrator. Phase pipeline with thread-pool parallelism:
|
|
|
|
Each chunk flows through: GPU → Metaprobe → Compress → Receipt
|
|
Thread pools for GPU, metaprobe, and compression run concurrently.
|
|
"""
|
|
|
|
def __init__(self, output_dir: Path, workers: Optional[int] = None,
|
|
metaprobe_threshold: float = 0.6):
|
|
self.output_dir = output_dir
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
self.chunk_size = 65536
|
|
self.threshold = metaprobe_threshold
|
|
|
|
cpu = os.cpu_count() or 12
|
|
self.n_gpu_workers = min(2, workers or cpu)
|
|
self.n_meta_workers = min(4, workers or cpu)
|
|
self.n_comp_workers = min(2, workers or cpu)
|
|
|
|
self.gpu_queue = Queue(maxsize=256)
|
|
self.meta_queue = Queue(maxsize=256)
|
|
self.comp_queue = Queue(maxsize=256)
|
|
|
|
self.gpu_surface = GPUComputeSurface()
|
|
self.metaprobe_gate = MetaprobeGate(threshold=self.threshold)
|
|
self.compression = CompressionEngine()
|
|
self.receipts = ReceiptGenerator(output_dir / "receipts.jsonl")
|
|
|
|
self.running = False
|
|
self._start_time = 0.0
|
|
|
|
def start(self):
|
|
self.running = True
|
|
self._start_time = time.time()
|
|
pools = []
|
|
|
|
pools.append(ThreadPoolExecutor(self.n_gpu_workers, thread_name_prefix="gpu"))
|
|
pools.append(ThreadPoolExecutor(self.n_meta_workers, thread_name_prefix="meta"))
|
|
pools.append(ThreadPoolExecutor(self.n_comp_workers, thread_name_prefix="comp"))
|
|
|
|
pools[0].submit(self._gpu_worker, pools)
|
|
pools[1].submit(self._metaprobe_worker, pools)
|
|
pools[2].submit(self._compression_worker)
|
|
|
|
return pools
|
|
|
|
def stop(self, pools: list):
|
|
self.running = False
|
|
for pool in pools:
|
|
pool.shutdown(wait=True)
|
|
|
|
def feed(self, feeder: SourceFeeder):
|
|
feeder.done()
|
|
return feeder.total_chunks
|
|
|
|
def _gpu_worker(self, pools: list):
|
|
while self.running:
|
|
try:
|
|
chunk = self.gpu_queue.get(timeout=0.5)
|
|
except Empty:
|
|
continue
|
|
try:
|
|
result = self.gpu_surface.process(chunk)
|
|
self.meta_queue.put((chunk, result))
|
|
except Exception as e:
|
|
print(f" [GPU] Error on {chunk.id}: {e}")
|
|
finally:
|
|
self.gpu_queue.task_done()
|
|
|
|
def _metaprobe_worker(self, pools: list):
|
|
while self.running:
|
|
try:
|
|
chunk, gpu_result = self.meta_queue.get(timeout=0.5)
|
|
except Empty:
|
|
continue
|
|
try:
|
|
meta_result = self.metaprobe_gate.check(chunk, gpu_result)
|
|
self.comp_queue.put((chunk, gpu_result, meta_result))
|
|
except Exception as e:
|
|
print(f" [META] Error on {chunk.id}: {e}")
|
|
finally:
|
|
self.meta_queue.task_done()
|
|
|
|
def _compression_worker(self):
|
|
while self.running:
|
|
try:
|
|
chunk, gpu_result, meta_result = self.comp_queue.get(timeout=0.5)
|
|
except Empty:
|
|
continue
|
|
try:
|
|
comp_result = self.compression.compress(chunk, meta_result)
|
|
receipt = self.receipts.emit(chunk, gpu_result, meta_result, comp_result)
|
|
if self.receipts.count % 100 == 0:
|
|
self._print_progress(receipt)
|
|
except Exception as e:
|
|
print(f" [COMP] Error on {chunk.id}: {e}")
|
|
finally:
|
|
self.comp_queue.task_done()
|
|
|
|
def _print_progress(self, receipt: Receipt):
|
|
elapsed = time.time() - self._start_time
|
|
stats = self.receipts.stats()
|
|
throughput = stats["total_original_bytes"] / max(elapsed, 0.001) / (1024**2)
|
|
print(f"\n [ENGINE] {stats['total_chunks']} chunks | "
|
|
f"{stats['total_original_bytes']/1024:.1f}KB → "
|
|
f"{stats['total_compressed_bytes']/1024:.1f}KB | "
|
|
f"ratio={stats['overall_ratio']:.2f}x | "
|
|
f"lawful={stats['lawful_fraction']:.1%} | "
|
|
f"{throughput:.1f} MB/s")
|
|
|
|
def finalize(self, pools: list) -> dict:
|
|
self.stop(pools)
|
|
stats = self.receipts.stats()
|
|
elapsed = time.time() - self._start_time
|
|
stats["elapsed_seconds"] = round(elapsed, 1)
|
|
stats["throughput_mbps"] = round(
|
|
stats["total_original_bytes"] / max(elapsed, 0.001) / (1024**2), 1
|
|
)
|
|
stats["timestamp"] = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
|
|
stats["gpu_surface"] = "wgpu" if HAS_WGPU else ("cuda" if HAS_CUDA else "cpu")
|
|
|
|
report_path = self.output_dir / "compression_report.json"
|
|
with open(str(report_path), "w") as f:
|
|
json.dump(stats, f, indent=2)
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f" ENGINE COMPLETE")
|
|
print(f"{'='*60}")
|
|
for k, v in stats.items():
|
|
if isinstance(v, float):
|
|
print(f" {k}: {v:.3f}")
|
|
else:
|
|
print(f" {k}: {v}")
|
|
print(f" receipts: {self.output_dir}/receipts.jsonl")
|
|
print(f" report: {report_path}")
|
|
print(f"{'='*60}")
|
|
return stats
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Daemon Mode
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class Daemon:
|
|
"""File-watching daemon that feeds the engine continuously."""
|
|
|
|
def __init__(self, engine: GPUMetaprobeCompressionEngine, watch_dir: Path,
|
|
interval: float = 30.0):
|
|
self.engine = engine
|
|
self.watch_dir = Path(watch_dir)
|
|
self.interval = interval
|
|
self.seen_hashes = set()
|
|
|
|
def run(self):
|
|
print(f" [DAEMON] Watching {self.watch_dir} every {self.interval}s")
|
|
feeder = SourceFeeder(self.engine.gpu_queue, self.engine.chunk_size)
|
|
self.engine.gpu_queue = feeder.queue
|
|
|
|
while True:
|
|
files = sorted(self.watch_dir.rglob("*"))
|
|
new_files = []
|
|
for f in files:
|
|
if not f.is_file() or f.suffix in (".jsonl", ".json", ".tmp"):
|
|
continue
|
|
try:
|
|
h = hashlib.sha256(str(f).encode()).hexdigest()
|
|
if h not in self.seen_hashes:
|
|
self.seen_hashes.add(h)
|
|
new_files.append(f)
|
|
except Exception:
|
|
pass
|
|
|
|
if new_files:
|
|
print(f" [DAEMON] {len(new_files)} new files")
|
|
feeder = SourceFeeder(self.engine.gpu_queue, self.engine.chunk_size)
|
|
for f in new_files:
|
|
feeder.feed_file(f)
|
|
feeder.done()
|
|
|
|
time.sleep(self.interval)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# CLI Entry Point
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="GPU-Parallel Metaprobe Compression Engine")
|
|
parser.add_argument("--input", "-i", help="Input file or directory")
|
|
parser.add_argument("--output", "-o", default="5-Applications/out/gpmce",
|
|
help="Output directory")
|
|
parser.add_argument("--daemon", action="store_true", help="Run as file-watching daemon")
|
|
parser.add_argument("--watch-dir", default="~/Downloads",
|
|
help="Directory to watch in daemon mode")
|
|
parser.add_argument("--threshold", type=float, default=0.6,
|
|
help="Metaprobe lawfulness threshold (0-1)")
|
|
parser.add_argument("--benchmark", action="store_true",
|
|
help="Run self-benchmark with synthetic data")
|
|
parser.add_argument("--workers", type=int, default=None,
|
|
help="Worker thread count")
|
|
args = parser.parse_args()
|
|
|
|
output_dir = (BASE / args.output).resolve()
|
|
print(f"\n{'='*60}")
|
|
print(f" GPU-Parallel Metaprobe Compression Engine")
|
|
print(f" GPU: wgpu={'✓' if HAS_WGPU else '✗'} CUDA={'✓' if HAS_CUDA else '✗'}")
|
|
print(f" Output: {output_dir}")
|
|
print(f"{'='*60}")
|
|
|
|
engine = GPUMetaprobeCompressionEngine(
|
|
output_dir, workers=args.workers,
|
|
metaprobe_threshold=args.threshold
|
|
)
|
|
|
|
if args.benchmark:
|
|
_run_benchmark(engine)
|
|
return
|
|
|
|
pools = engine.start()
|
|
|
|
if args.daemon:
|
|
watch_dir = Path(args.watch_dir).expanduser()
|
|
daemon = Daemon(engine, watch_dir)
|
|
daemon.run()
|
|
return
|
|
|
|
feeder = SourceFeeder(engine.gpu_queue, engine.chunk_size)
|
|
engine.gpu_queue = feeder.queue
|
|
|
|
if args.input:
|
|
path = Path(args.input).expanduser()
|
|
if path.is_dir():
|
|
feeder.feed_directory(path)
|
|
elif path.is_file():
|
|
feeder.feed_file(path)
|
|
else:
|
|
print(f" [ERROR] Input not found: {path}")
|
|
sys.exit(1)
|
|
else:
|
|
feeder.feed_stdin()
|
|
|
|
feeder.done()
|
|
engine.finalize(pools)
|
|
|
|
|
|
def _run_benchmark(engine: GPUMetaprobeCompressionEngine):
|
|
"""Self-benchmark with structured compressible data across sizes."""
|
|
print(f"\n [BENCH] Running self-benchmark...")
|
|
|
|
import zlib as bench_zlib
|
|
patterns = [
|
|
b"The quick brown fox jumps over the lazy dog. " * 5000,
|
|
b"Lorem ipsum dolor sit amet consectetur adipiscing elit. " * 5000,
|
|
b"ABCDEFGHIJKLMNOPQRSTUVWXYZ" * 1000,
|
|
b"package main; import fmt; func main() { fmt.Println('hello'); } " * 2000,
|
|
]
|
|
json_lines = []
|
|
for i in range(2000):
|
|
json_lines.append(
|
|
b'{"id":' + str(i).encode() +
|
|
b',"name":"item-' + str(i).encode() +
|
|
b'","value":' + str(i * 1.5).encode() +
|
|
b',"tags":["alpha","beta"]}'
|
|
)
|
|
patterns.append(b"\n".join(json_lines))
|
|
text = b"".join(patterns)
|
|
ref_ratio = len(text) / max(len(bench_zlib.compress(text, 9)), 1)
|
|
print(f" [BENCH] Reference: zlib-9 ratio = {ref_ratio:.2f}x on {len(text)} bytes")
|
|
|
|
feeder = SourceFeeder(engine.gpu_queue, engine.chunk_size)
|
|
engine.gpu_queue = feeder.queue
|
|
pools = engine.start()
|
|
feeder.feed_bytes(text, "bench-structured")
|
|
feeder.done()
|
|
stats = engine.finalize(pools)
|
|
|
|
bench_path = engine.output_dir / "benchmark_results.json"
|
|
with open(str(bench_path), "w") as f:
|
|
json.dump({"benchmark": stats, "reference_zlib9_ratio": round(ref_ratio, 3),
|
|
"hardware": {"wgpu": HAS_WGPU, "cuda": HAS_CUDA,
|
|
"cpu_count": os.cpu_count(),
|
|
"surface": stats["gpu_surface"]}}, f, indent=2)
|
|
print(f"\n [BENCH] PIST-GCL ratio={stats['overall_ratio']:.2f}x vs zlib-9={ref_ratio:.2f}x")
|
|
print(f" [BENCH] Results: {bench_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|