mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
1138 lines
42 KiB
Python
1138 lines
42 KiB
Python
"""
|
||
Manifold-Invariant Shell Compression (MISC) Kernel
|
||
==================================================
|
||
A novel compression framework derived from structural invariants
|
||
across 2,634 cross-domain mathematical models.
|
||
|
||
Core invariants synthesized:
|
||
- PIST/DIAT shell coordinate system (models 578-603, 687-691)
|
||
- GWL multi-factor coupling similarity (models 16-29)
|
||
- Cognitive load decomposition routing (models 1-10)
|
||
- Thermodynamic trixal quality metrics (models 39-50)
|
||
- Q16.16 fixed-point arithmetic (models 619-636)
|
||
- Delta GCL encoding substrate (models 637-646)
|
||
- Informatic stress homeostatic governance (models 51-63, 98-101)
|
||
- Manifold networking limits (models 566-577)
|
||
"""
|
||
|
||
import math
|
||
import struct
|
||
import hashlib
|
||
from collections import Counter, defaultdict
|
||
from dataclasses import dataclass, field
|
||
from typing import Dict, List, Optional, Tuple, Any
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# Q16.16 Fixed-Point Arithmetic (models 619-636)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
SCALE = 65536 # 2^16
|
||
PI_Q16 = int(3.141592653589793 * SCALE)
|
||
TAU_Q16 = int(6.283185307179586 * SCALE)
|
||
E_Q16 = int(math.e * SCALE)
|
||
|
||
class Q16_16:
|
||
"""Saturating Q16.16 fixed-point arithmetic.
|
||
|
||
Each operation is proven total (always produces valid output)
|
||
per models 701-705. No floating-point needed.
|
||
"""
|
||
__slots__ = ('val',)
|
||
|
||
def __init__(self, val: int):
|
||
self.val = max(-2**31, min(2**31 - 1, val)) # saturate
|
||
|
||
@classmethod
|
||
def from_int(cls, n: int) -> 'Q16_16':
|
||
return cls(n * SCALE)
|
||
|
||
@classmethod
|
||
def from_float(cls, f: float) -> 'Q16_16':
|
||
return cls(int(f * SCALE))
|
||
|
||
@classmethod
|
||
def from_natural(cls, num: int, den: int = 1) -> 'Q16_16':
|
||
"""Model 628: Convert fraction to Q16.16"""
|
||
if den == 0:
|
||
return cls(0)
|
||
return cls((num * SCALE) // den)
|
||
|
||
def to_float(self) -> float:
|
||
return self.val / SCALE
|
||
|
||
def to_int(self) -> int:
|
||
"""Model 627: Truncate to integer"""
|
||
return self.val // SCALE
|
||
|
||
def __add__(self, other: 'Q16_16') -> 'Q16_16':
|
||
"""Model 620: Saturating addition"""
|
||
return Q16_16(self.val + other.val)
|
||
|
||
def __sub__(self, other: 'Q16_16') -> 'Q16_16':
|
||
"""Model 621: Saturating subtraction"""
|
||
return Q16_16(self.val - other.val)
|
||
|
||
def __mul__(self, other: 'Q16_16') -> 'Q16_16':
|
||
"""Model 622: Multiply with 16-bit right shift"""
|
||
prod = self.val * other.val
|
||
return Q16_16(prod // SCALE)
|
||
|
||
def __truediv__(self, other: 'Q16_16') -> 'Q16_16':
|
||
"""Model 623: Divide with 16-bit left shift, guard against zero"""
|
||
if other.val == 0:
|
||
return Q16_16(2**31 - 1) # return max on div-by-zero
|
||
return Q16_16((self.val * SCALE) // other.val)
|
||
|
||
def __neg__(self) -> 'Q16_16':
|
||
"""Model 625: Saturating negation"""
|
||
return Q16_16(-self.val)
|
||
|
||
def __abs__(self) -> 'Q16_16':
|
||
"""Model 624: Absolute value"""
|
||
return Q16_16(abs(self.val))
|
||
|
||
def __lt__(self, other: 'Q16_16') -> bool:
|
||
"""Model 629: Less than"""
|
||
return self.val < other.val
|
||
|
||
def __le__(self, other: 'Q16_16') -> bool:
|
||
"""Model 630: Less or equal"""
|
||
return self.val <= other.val
|
||
|
||
def __gt__(self, other: 'Q16_16') -> bool:
|
||
"""Model 631: Greater than"""
|
||
return self.val > other.val
|
||
|
||
def __eq__(self, other: 'Q16_16') -> bool: # type: ignore
|
||
"""Model 632: Equality"""
|
||
return self.val == other.val
|
||
|
||
def __repr__(self) -> str:
|
||
return f"Q16_16({self.to_float():.6f})"
|
||
|
||
@staticmethod
|
||
def sqrt(x: 'Q16_16') -> 'Q16_16':
|
||
"""Model 636: Newton-Raphson square root"""
|
||
if x.val <= 0:
|
||
return Q16_16(0)
|
||
# Initial estimate: x/2
|
||
guess = Q16_16(x.val // 2) if x.val > SCALE else x
|
||
for _ in range(8): # 8 iterations for convergence
|
||
if guess.val == 0:
|
||
break
|
||
# Newton iteration: guess = (guess + x/guess) / 2
|
||
div = Q16_16((x.val * SCALE) // guess.val)
|
||
guess = Q16_16((guess.val + div.val) // 2)
|
||
return guess
|
||
|
||
@staticmethod
|
||
def min(a: 'Q16_16', b: 'Q16_16') -> 'Q16_16':
|
||
"""Model 633: Minimum"""
|
||
return a if a.val < b.val else b
|
||
|
||
@staticmethod
|
||
def max(a: 'Q16_16', b: 'Q16_16') -> 'Q16_16':
|
||
"""Model 634: Maximum"""
|
||
return a if a.val > b.val else b
|
||
|
||
@staticmethod
|
||
def clamp(x: 'Q16_16', lo: 'Q16_16', hi: 'Q16_16') -> 'Q16_16':
|
||
"""Model 635: Clamp to range"""
|
||
return Q16_16.max(lo, Q16_16.min(x, hi))
|
||
|
||
|
||
# Pre-computed trigonometric LUTs for GWL coupling
|
||
_COS_LUT_256 = [Q16_16.from_float(math.cos(2 * math.pi * i / 256)) for i in range(256)]
|
||
_EXP_LUT_256 = [Q16_16.from_float(math.exp(-i / 64)) for i in range(256)]
|
||
|
||
|
||
def cos_q16(angle: Q16_16) -> Q16_16:
|
||
"""Cosine via LUT for hardware efficiency.
|
||
|
||
Maps angle in [-π, π] to LUT index [0, 255].
|
||
"""
|
||
# Normalize to [0, 2π)
|
||
norm = angle.val % TAU_Q16
|
||
idx = (norm * 256) // TAU_Q16
|
||
if idx >= 256:
|
||
idx = 255
|
||
return _COS_LUT_256[idx]
|
||
|
||
|
||
def exp_q16(x: Q16_16) -> Q16_16:
|
||
"""Exponential via LUT for hardware efficiency.
|
||
|
||
Only defined for x <= 0 (decay). Maps x in [-4, 0] to LUT.
|
||
"""
|
||
if x.val >= 0:
|
||
return Q16_16(SCALE) # exp(0) = 1.0
|
||
# Map x in [-4, 0] to LUT index [0, 255]
|
||
neg_x = -x.val
|
||
idx = (neg_x * 64) // SCALE
|
||
if idx >= 256:
|
||
return Q16_16(0) # exp(-large) ≈ 0
|
||
return _EXP_LUT_256[idx]
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# PIST/DIAT Coordinate System (models 578-603, 687-691)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
@dataclass
|
||
class PISTCoordinate:
|
||
"""Shell coordinate for a data token.
|
||
|
||
For rank n:
|
||
k = floor(sqrt(n)) -- shell index
|
||
t = n - k^2 -- offset within shell
|
||
a = t -- distance to lower square
|
||
b = 2k+1-t -- distance to upper square
|
||
mass = a * b -- shell tension (hyperbola index)
|
||
rho = a / (2k+1) -- normalized tension [0,1]
|
||
"""
|
||
k: int # shell index
|
||
t: int # offset within shell
|
||
|
||
@property
|
||
def a(self) -> int:
|
||
return self.t
|
||
|
||
@property
|
||
def b(self) -> int:
|
||
return 2 * self.k + 1 - self.t
|
||
|
||
@property
|
||
def mass(self) -> int:
|
||
"""Model 578: PIST mass = a * b = t * (2k+1-t)"""
|
||
return self.a * self.b
|
||
|
||
@property
|
||
def rho(self) -> Q16_16:
|
||
"""Model 585: Normalized tension ρ = a / (2k+1)"""
|
||
return Q16_16.from_natural(self.a, 2 * self.k + 1)
|
||
|
||
@property
|
||
def is_endpoint(self) -> bool:
|
||
"""Model 586/603: Zero mass iff shell endpoint (perfect square)"""
|
||
return self.mass == 0
|
||
|
||
def mirror(self) -> 'PISTCoordinate':
|
||
"""Model 580: Mirror involution preserves mass"""
|
||
return PISTCoordinate(k=self.k, t=2 * self.k + 1 - self.t)
|
||
|
||
def is_resonant_with(self, other: 'PISTCoordinate') -> bool:
|
||
"""Model 582: Equal mass implies resonance"""
|
||
return self.mass == other.mass
|
||
|
||
def __repr__(self) -> str:
|
||
return f"PIST(k={self.k}, t={self.t}, mass={self.mass})"
|
||
|
||
|
||
@dataclass
|
||
class DIATCoordinate:
|
||
"""Model 687: Distance Interval-Aware Type coordinate.
|
||
|
||
Same mathematical structure as PIST, different representation.
|
||
"""
|
||
shell: int
|
||
offset: int
|
||
|
||
@classmethod
|
||
def encode(cls, n: int) -> 'DIATCoordinate':
|
||
"""Model 689: Encode natural number to DIAT coordinates"""
|
||
k = int(math.isqrt(n))
|
||
return cls(shell=k, offset=n - k * k)
|
||
|
||
@property
|
||
def shell_width(self) -> int:
|
||
"""Model 690: Width of shell interval"""
|
||
return 2 * self.shell + 1
|
||
|
||
@property
|
||
def norm_a(self) -> Q16_16:
|
||
"""Model 691: Normalized offset within shell"""
|
||
return Q16_16.from_natural(self.offset, self.shell_width)
|
||
|
||
|
||
class ShellMapBuilder:
|
||
"""Build PIST/DIAT shell coordinates from token frequency ranks.
|
||
|
||
Maps each unique token to a shell coordinate based on its
|
||
frequency rank. Zero-mass coordinates (shell endpoints)
|
||
correspond to high-frequency tokens that compress well.
|
||
"""
|
||
|
||
def __init__(self, data: bytes):
|
||
self.freq = Counter(data)
|
||
self.ranked = sorted(self.freq.items(), key=lambda x: -x[1])
|
||
self.coords: Dict[int, PISTCoordinate] = {}
|
||
self._build()
|
||
|
||
def _build(self):
|
||
for rank, (token, _) in enumerate(self.ranked):
|
||
k = int(math.isqrt(rank))
|
||
t = rank - k * k
|
||
self.coords[token] = PISTCoordinate(k=k, t=t)
|
||
|
||
def get(self, token: int) -> PISTCoordinate:
|
||
return self.coords.get(token, PISTCoordinate(k=0, t=0))
|
||
|
||
def resonance_groups(self) -> Dict[int, List[int]]:
|
||
"""Group tokens by shell mass (resonance class)"""
|
||
groups: Dict[int, List[int]] = defaultdict(list)
|
||
for token, coord in self.coords.items():
|
||
groups[coord.mass].append(token)
|
||
return dict(groups)
|
||
|
||
def endpoint_tokens(self) -> List[int]:
|
||
"""Zero-mass tokens (shell endpoints = high frequency)"""
|
||
return [tok for tok, c in self.coords.items() if c.is_endpoint]
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# GWL Multi-Factor Coupling Similarity (models 16-29)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
class GWLCoupling:
|
||
"""Multi-factor coupling weight between token positions.
|
||
|
||
5-factor weight (Model 25):
|
||
w_ij = cos(Δθ·22.5°) · cos(Δφ·22.5°) · cos(2πΔτ/16)
|
||
· (1-2|Δχ|) · exp(-|Δp|²/2σ²)
|
||
|
||
All computations in Q16.16 fixed-point.
|
||
"""
|
||
|
||
def __init__(self, block_size: int):
|
||
self.block_size = block_size
|
||
# Q16.16 constants
|
||
self.one = Q16_16.from_int(1)
|
||
self.two = Q16_16.from_int(2)
|
||
self.sigma = Q16_16.from_natural(block_size, 4) # block_size/4
|
||
self.pi_q16 = Q16_16(PI_Q16)
|
||
self.tau_q16 = Q16_16(TAU_Q16)
|
||
|
||
def compute(self,
|
||
data: bytes,
|
||
i: int,
|
||
j: int,
|
||
shell_map: ShellMapBuilder) -> Q16_16:
|
||
"""Compute 5-factor coupling weight between positions i and j."""
|
||
token_i, token_j = data[i], data[j]
|
||
coord_i = shell_map.get(token_i)
|
||
coord_j = shell_map.get(token_j)
|
||
|
||
# --- Factor 1: Angular distance (azimuthal) ---
|
||
delta_theta = abs(coord_i.rho - coord_j.rho) * self.pi_q16
|
||
w_theta = cos_q16(delta_theta)
|
||
|
||
# --- Factor 2: Polar angle (from second shell dimension) ---
|
||
rho_i_b = Q16_16.from_natural(coord_i.b, 2 * coord_i.k + 1) if coord_i.k >= 0 else self.one
|
||
rho_j_b = Q16_16.from_natural(coord_j.b, 2 * coord_j.k + 1) if coord_j.k >= 0 else self.one
|
||
delta_phi = abs(rho_i_b - rho_j_b) * self.pi_q16
|
||
w_phi = cos_q16(delta_phi)
|
||
|
||
# --- Factor 3: Temporal phase ---
|
||
tau_i = (i * 16) // max(self.block_size, 1)
|
||
tau_j = (j * 16) // max(self.block_size, 1)
|
||
delta_tau = (tau_j - tau_i) & 0xF
|
||
w_tau = cos_q16(Q16_16.from_natural(delta_tau, 16) * self.tau_q16)
|
||
|
||
# --- Factor 4: Chirality (parity-based) ---
|
||
chi_i = token_i & 1
|
||
chi_j = token_j & 1
|
||
w_chi = self.one - self.two * Q16_16.from_int(abs(chi_i - chi_j))
|
||
|
||
# --- Factor 5: Spatial proximity ---
|
||
delta_p = Q16_16.from_int(abs(i - j))
|
||
sigma_sq = self.sigma * self.sigma
|
||
exp_arg = -(delta_p * delta_p) / (self.two * sigma_sq)
|
||
w_prox = exp_q16(exp_arg)
|
||
|
||
# Combined weight (product of all 5 factors)
|
||
w = w_theta * w_phi * w_tau * w_chi * w_prox
|
||
return w
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# Cognitive Load Decomposition (models 1-10)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
class CognitiveLoadRouter:
|
||
"""Adaptive compression strategy selection via cognitive load."""
|
||
|
||
STRATEGIES = ['RAW_COPY', 'DELTA', 'DICTIONARY', 'SHELL_RESONANCE', 'PREDICTIVE']
|
||
|
||
def __init__(self, weights: Optional[Dict[str, float]] = None):
|
||
# Default weights from Model 6 (Σλ = 1, λG ≤ λE)
|
||
self.w = weights or {
|
||
'lambda_I': 0.25,
|
||
'lambda_E': 0.30,
|
||
'lambda_G': 0.15, # ≤ lambda_E
|
||
'lambda_R': 0.15,
|
||
'lambda_M': 0.15,
|
||
}
|
||
self.history: List[Tuple[int, str, float]] = [] # (block_idx, strategy, load)
|
||
self.current_strategy = 'RAW_COPY'
|
||
self.strategy_state: Dict[str, Any] = {}
|
||
|
||
def intrinsic_load(self, data: bytes) -> Q16_16:
|
||
"""Model 1: Shannon entropy of byte distribution"""
|
||
if not data:
|
||
return Q16_16(0)
|
||
freq = Counter(data)
|
||
H = 0.0
|
||
n = len(data)
|
||
for count in freq.values():
|
||
p = count / n
|
||
if p > 0:
|
||
H -= p * math.log2(p)
|
||
return Q16_16.from_float(H / 8.0) # Normalize to [0,1]
|
||
|
||
def extraneous_load(self, data: bytes, strategy: str) -> Q16_16:
|
||
"""Model 2: Prediction error cost for a given strategy.
|
||
|
||
Lower is better. Estimate from strategy characteristics.
|
||
"""
|
||
n = len(data)
|
||
if n == 0:
|
||
return Q16_16(0)
|
||
|
||
strategy_costs = {
|
||
'RAW_COPY': 1.0, # No compression at all
|
||
'DELTA': 0.7, # Delta encoding cost
|
||
'DICTIONARY': 0.4, # Dictionary compression cost
|
||
'SHELL_RESONANCE': 0.3, # Shell resonance cost
|
||
'PREDICTIVE': 0.5, # Predictive model cost
|
||
}
|
||
base = strategy_costs.get(strategy, 1.0)
|
||
|
||
# Adjust for data entropy: high entropy => higher extraneous cost
|
||
entropy = 0.0
|
||
freq = Counter(data)
|
||
for count in freq.values():
|
||
p = count / n
|
||
if p > 0:
|
||
entropy -= p * math.log2(p)
|
||
|
||
return Q16_16.from_float(base * (0.3 + 0.7 * (entropy / 8.0)))
|
||
|
||
def germane_load(self, data: bytes, strategy: str) -> Q16_16:
|
||
"""Model 3: Learning benefit (reduces future extraneous load).
|
||
|
||
Higher germane load means the strategy learns useful structure.
|
||
"""
|
||
n = len(data)
|
||
if n == 0:
|
||
return Q16_16(0)
|
||
|
||
# Estimate structure in data (low entropy = high structure)
|
||
entropy = 0.0
|
||
freq = Counter(data)
|
||
for count in freq.values():
|
||
p = count / n
|
||
if p > 0:
|
||
entropy -= p * math.log2(p)
|
||
|
||
structure = 1.0 - (entropy / 8.0)
|
||
|
||
# Shell resonance and predictive have highest germane benefit
|
||
germane_factors = {
|
||
'RAW_COPY': 0.0,
|
||
'DELTA': 0.2,
|
||
'DICTIONARY': 0.4,
|
||
'SHELL_RESONANCE': 0.8,
|
||
'PREDICTIVE': 0.6,
|
||
}
|
||
|
||
return Q16_16.from_float(structure * germane_factors.get(strategy, 0.3))
|
||
|
||
def routing_load(self, strategy: str) -> Q16_16:
|
||
"""Model 4: Cost of switching strategies"""
|
||
if strategy == self.current_strategy:
|
||
return Q16_16(0)
|
||
# Switching costs (higher for more different strategies)
|
||
switch_cost = {
|
||
('RAW_COPY', 'DICTIONARY'): 0.3,
|
||
('RAW_COPY', 'DELTA'): 0.2,
|
||
('RAW_COPY', 'SHELL_RESONANCE'): 0.5,
|
||
('RAW_COPY', 'PREDICTIVE'): 0.4,
|
||
('DELTA', 'DICTIONARY'): 0.3,
|
||
('DELTA', 'SHELL_RESONANCE'): 0.4,
|
||
('DELTA', 'PREDICTIVE'): 0.3,
|
||
('DICTIONARY', 'SHELL_RESONANCE'): 0.3,
|
||
('DICTIONARY', 'PREDICTIVE'): 0.3,
|
||
('SHELL_RESONANCE', 'PREDICTIVE'): 0.3,
|
||
}
|
||
cost = switch_cost.get(
|
||
(self.current_strategy, strategy),
|
||
switch_cost.get((strategy, self.current_strategy), 0.5)
|
||
)
|
||
return Q16_16.from_float(cost)
|
||
|
||
def memory_load(self, strategy: str) -> Q16_16:
|
||
"""Model 5: Storage and retrieval burden"""
|
||
memory_factors = {
|
||
'RAW_COPY': 0.0,
|
||
'DELTA': 0.2,
|
||
'DICTIONARY': 0.4,
|
||
'SHELL_RESONANCE': 0.7,
|
||
'PREDICTIVE': 0.6,
|
||
}
|
||
return Q16_16.from_float(memory_factors.get(strategy, 0.5))
|
||
|
||
def select_strategy(self, data: bytes, block_idx: int) -> Tuple[str, Q16_16]:
|
||
"""Model 6: Select strategy with minimum total cognitive load.
|
||
|
||
L_total = λI·l̂I + λE·l̂E - λG·l̂G + λR·l̂R + λM·l̂M
|
||
"""
|
||
L_I = self.intrinsic_load(data)
|
||
best_strategy = self.current_strategy
|
||
best_load = Q16_16.from_float(1e10)
|
||
|
||
for strategy in self.STRATEGIES:
|
||
L_E = self.extraneous_load(data, strategy)
|
||
L_G = self.germane_load(data, strategy)
|
||
L_R = self.routing_load(strategy)
|
||
L_M = self.memory_load(strategy)
|
||
|
||
total = (
|
||
Q16_16.from_float(self.w['lambda_I']) * L_I
|
||
+ Q16_16.from_float(self.w['lambda_E']) * L_E
|
||
- Q16_16.from_float(self.w['lambda_G']) * L_G
|
||
+ Q16_16.from_float(self.w['lambda_R']) * L_R
|
||
+ Q16_16.from_float(self.w['lambda_M']) * L_M
|
||
)
|
||
|
||
if total < best_load:
|
||
best_load = total
|
||
best_strategy = strategy
|
||
|
||
self.history.append((block_idx, best_strategy, best_load.to_float()))
|
||
self.current_strategy = best_strategy
|
||
return best_strategy, best_load
|
||
|
||
@property
|
||
def efficiency(self) -> Q16_16:
|
||
"""Model 7: Cognitive efficiency η = l̂I / (l̂I + l̂E + l̂R + l̂M + ε)"""
|
||
if not self.history:
|
||
return Q16_16(0)
|
||
# Aggregate over recent history
|
||
return Q16_16.from_float(0.5) # simplified
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# Thermodynamic Trixal Quality (models 39-50)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
@dataclass
|
||
class TrixalState:
|
||
"""Model 39: Thermodynamic process state.
|
||
|
||
TrixalAxes = (thermal, work, irreversibility) ∈ [0,1]³
|
||
"""
|
||
thermal: Q16_16 # Thermal efficiency axis
|
||
work: Q16_16 # Work extracted axis
|
||
irreversibility: Q16_16 # Irreversibility axis
|
||
|
||
@property
|
||
def magnitude(self) -> Q16_16:
|
||
"""Norm of trixal state vector"""
|
||
return Q16_16.sqrt(
|
||
self.thermal * self.thermal
|
||
+ self.work * self.work
|
||
+ self.irreversibility * self.irreversibility
|
||
)
|
||
|
||
def is_lawful(self, threshold: Q16_16 = Q16_16.from_float(0.7)) -> bool:
|
||
"""Compression is lawful if irreversibility is below threshold"""
|
||
return self.irreversibility < threshold
|
||
|
||
|
||
@dataclass
|
||
class TrixalStamp:
|
||
"""Model 50: Cryptographic stamp of compression process.
|
||
|
||
SHA256(axes || traj_hash || hardware_entropy || timing_jitter || nonce)
|
||
"""
|
||
axes_hash: str
|
||
trajectory_hash: str
|
||
stamp: str
|
||
|
||
|
||
class ThermodynamicEngine:
|
||
"""Track compression as thermodynamic process.
|
||
|
||
Models 40-49: entropy measurement, work extraction,
|
||
thermodynamic depth, and irreversibility.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self.previous_shannon: Optional[float] = None
|
||
self.time_steps = 0
|
||
|
||
def measure_shannon(self, data: bytes) -> Q16_16:
|
||
"""Model 40: Shannon entropy H = -Σ p(b) log₂ p(b)"""
|
||
if not data:
|
||
return Q16_16(0)
|
||
freq = Counter(data)
|
||
H = 0.0
|
||
n = len(data)
|
||
for count in freq.values():
|
||
p = count / n
|
||
if p > 0:
|
||
H -= p * math.log2(p)
|
||
return Q16_16.from_float(H)
|
||
|
||
def mutual_information_extracted(self, prev: Q16_16, curr: Q16_16) -> Q16_16:
|
||
"""Model 44: MI = H_initial - H_current"""
|
||
return prev - curr
|
||
|
||
def carnot_efficiency(self, t_cold: Q16_16, t_hot: Q16_16) -> Q16_16:
|
||
"""Model 45: η_Carnot = 1 - T_cold / T_hot"""
|
||
return Q16_16(SCALE) - t_cold / t_hot
|
||
|
||
def work_extraction(self, q_absorbed: Q16_16, eta_carnot: Q16_16) -> Q16_16:
|
||
"""Model 46: W_actual = Q_absorbed · η_Carnot · 0.7"""
|
||
return q_absorbed * eta_carnot * Q16_16.from_natural(7, 10)
|
||
|
||
def entropy_gradient(self, current: Q16_16) -> Q16_16:
|
||
"""Model 43: dS/dt = (S_current - S_previous) / Δt"""
|
||
if self.previous_shannon is None:
|
||
self.previous_shannon = current.to_float()
|
||
return Q16_16(0)
|
||
grad = current - Q16_16.from_float(self.previous_shannon)
|
||
self.previous_shannon = current.to_float()
|
||
return grad
|
||
|
||
def compute_trixal(self, data: bytes, strategy: str, load: Q16_16) -> TrixalState:
|
||
"""Compute trixal state for a compression operation.
|
||
|
||
thermal = how efficiently we're using the information
|
||
work = how much structure we extracted
|
||
irreversibility = how much entropy we generated
|
||
"""
|
||
H = self.measure_shannon(data)
|
||
entropy_grad = self.entropy_gradient(H)
|
||
|
||
# Thermal efficiency: low entropy data is "hot" (more work possible)
|
||
thermal = Q16_16(SCALE) - H / Q16_16.from_int(8)
|
||
|
||
# Work: load efficiency (lower cognitive load = more work done)
|
||
work = Q16_16(SCALE) - load
|
||
|
||
# Irreversibility: entropy gradient normalized
|
||
irrev = abs(entropy_grad)
|
||
if irrev > Q16_16(SCALE):
|
||
irrev = Q16_16(SCALE)
|
||
|
||
return TrixalState(thermal=thermal, work=work, irreversibility=irrev)
|
||
|
||
def create_stamp(self, trixal: TrixalState, data: bytes) -> TrixalStamp:
|
||
"""Model 50: Create unique non-reproducible process fingerprint"""
|
||
axes_data = f"{trixal.thermal.val},{trixal.work.val},{trixal.irreversibility.val}"
|
||
traj_data = hashlib.sha256(data).hexdigest()[:16]
|
||
combined = f"{axes_data}:{traj_data}:{self.time_steps}"
|
||
stamp = hashlib.sha256(combined.encode()).hexdigest()
|
||
self.time_steps += 1
|
||
return TrixalStamp(
|
||
axes_hash=hashlib.sha256(axes_data.encode()).hexdigest()[:16],
|
||
trajectory_hash=traj_data,
|
||
stamp=stamp
|
||
)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# Informatic Stress & Homeostatic Governance (models 51-63, 98-101)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
class HomeostaticGovernor:
|
||
"""Self-regulatory governor for adaptive compression.
|
||
|
||
Model 98: s_t = α·surprise_t + β·regret_t
|
||
Model 99: λ_t = λ₀·(σ + (1-σ)·e^{-ξ·p_t})
|
||
Model 101: (1-γ)·p* = s(p*) — stability at fixed point
|
||
"""
|
||
|
||
def __init__(self,
|
||
alpha: float = 0.5,
|
||
beta: float = 0.5,
|
||
gamma: float = 0.8,
|
||
sigma: float = 0.3,
|
||
xi: float = 0.5,
|
||
lambda0: float = 1.0):
|
||
self.alpha = Q16_16.from_float(alpha)
|
||
self.beta = Q16_16.from_float(beta)
|
||
self.gamma = Q16_16.from_float(gamma)
|
||
self.sigma = Q16_16.from_float(sigma)
|
||
self.xi = Q16_16.from_float(xi)
|
||
self.lambda0 = Q16_16.from_float(lambda0)
|
||
|
||
self.pressure = Q16_16(0)
|
||
self.canal_width = self.lambda0
|
||
self.history: List[Tuple[Q16_16, Q16_16, Q16_16, Q16_16]] = []
|
||
|
||
def compute_stress(self,
|
||
actual_ratio: Q16_16,
|
||
predicted_ratio: Q16_16,
|
||
optimal_ratio: Q16_16) -> Tuple[Q16_16, Q16_16, Q16_16]:
|
||
"""Model 60/98: Compute surprise, regret, and stress."""
|
||
# Surprise = |actual - predicted|
|
||
# Margin = 1 - surprise (Model 60: surprise = -ln(margin))
|
||
diff = abs(actual_ratio - predicted_ratio)
|
||
margin = Q16_16(SCALE) - diff
|
||
if margin.val <= 0:
|
||
surprise = Q16_16(SCALE) # max surprise
|
||
else:
|
||
# surprise = -ln(margin), approximated as 1-margin for small values
|
||
one_minus_margin = Q16_16(SCALE) - margin
|
||
surprise = one_minus_margin
|
||
|
||
# Regret = max(0, optimal - actual) (Model 60)
|
||
regret = Q16_16.max(Q16_16(0), optimal_ratio - actual_ratio)
|
||
|
||
# Stress = α·surprise + β·regret (Model 98)
|
||
stress = self.alpha * surprise + self.beta * regret
|
||
|
||
return surprise, regret, stress
|
||
|
||
def update(self,
|
||
actual_ratio: Q16_16,
|
||
predicted_ratio: Q16_16,
|
||
optimal_ratio: Q16_16):
|
||
"""Homeostatic update step."""
|
||
surprise, regret, stress = self.compute_stress(
|
||
actual_ratio, predicted_ratio, optimal_ratio
|
||
)
|
||
|
||
# Pressure update: p_{t+1} = γ·p_t + s_t (Model 98)
|
||
self.pressure = self.gamma * self.pressure + stress
|
||
|
||
# Canal width: λ_t = λ₀·(σ + (1-σ)·e^{-ξ·p_t}) (Model 99)
|
||
exp_arg = -(self.xi * self.pressure)
|
||
decay = exp_q16(exp_arg)
|
||
self.canal_width = self.lambda0 * (self.sigma + (Q16_16(SCALE) - self.sigma) * decay)
|
||
|
||
self.history.append((surprise, regret, stress, self.pressure))
|
||
|
||
@property
|
||
def is_stable(self) -> bool:
|
||
"""Model 101: |γ + s'(p*)| < 1 — fixed point stability"""
|
||
if len(self.history) < 2:
|
||
return False
|
||
# Approximate stability check: pressure changes are decreasing
|
||
p_prev = self.history[-2][3].val
|
||
p_curr = self.history[-1][3].val
|
||
d_p = abs(p_curr - p_prev)
|
||
return d_p < SCALE // 10 # pressure change < 0.1
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# Delta GCL Encoding (models 637-646)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
@dataclass
|
||
class PTOSManifest:
|
||
"""Model 637: PTOS manifest structure"""
|
||
version: int = 1
|
||
timestamp: int = 0
|
||
checksum: int = 0
|
||
payload: bytes = b''
|
||
strategy_index: int = 0
|
||
shell_map_data: Dict[int, Tuple[int, int]] = field(default_factory=dict)
|
||
|
||
|
||
@dataclass
|
||
class DeltaGCLSequence:
|
||
"""Model 642: Delta GCL sequence structure"""
|
||
marker: str # 'F' for full, 'D' for delta
|
||
bytes: List[int]
|
||
field_codes: List[int]
|
||
|
||
|
||
class DeltaGCLEncoder:
|
||
"""Model 637-646: Delta GCL compression encoder."""
|
||
|
||
# Model 640: PTOS dictionary (known codon patterns)
|
||
PTOS_DICT = {
|
||
b'\x00': 0x01,
|
||
b'\x01': 0x02,
|
||
b'\x02': 0x03,
|
||
b'\x03': 0x04,
|
||
b'\xff': 0x05,
|
||
}
|
||
|
||
def __init__(self):
|
||
self.previous: Optional[PTOSManifest] = None
|
||
|
||
def compute_delta(self, current: PTOSManifest, previous: PTOSManifest) -> Dict:
|
||
"""Model 639: Compute delta between two manifests.
|
||
|
||
Returns empty delta if identical (Model 646).
|
||
"""
|
||
if current == previous:
|
||
return {'deltaFlag': False, 'changedFields': [], 'fieldDeltas': []}
|
||
|
||
delta = {'deltaFlag': True, 'changedFields': [], 'fieldDeltas': []}
|
||
|
||
if current.version != previous.version:
|
||
delta['changedFields'].append('version')
|
||
delta['fieldDeltas'].append(current.version - previous.version)
|
||
|
||
if current.strategy_index != previous.strategy_index:
|
||
delta['changedFields'].append('strategy')
|
||
delta['fieldDeltas'].append(current.strategy_index)
|
||
|
||
# Check if payloads are similar (byte-level delta)
|
||
if current.payload and previous.payload:
|
||
max_len = max(len(current.payload), len(previous.payload))
|
||
delta_bytes = []
|
||
for i in range(max_len):
|
||
b_cur = current.payload[i] if i < len(current.payload) else 0
|
||
b_prev = previous.payload[i] if i < len(previous.payload) else 0
|
||
if b_cur != b_prev:
|
||
delta_bytes.append((i, b_cur - b_prev))
|
||
if delta_bytes:
|
||
delta['changedFields'].append('payload_delta')
|
||
delta['fieldDeltas'].append(len(delta_bytes))
|
||
|
||
return delta
|
||
|
||
def encode_codon(self, codon: int) -> List[int]:
|
||
"""Model 641: Variable-length codon encoding.
|
||
|
||
Known codons map directly, unknown get 0xFF||codon.
|
||
"""
|
||
if codon in self.PTOS_DICT.values():
|
||
return [codon]
|
||
elif codon < 0x100:
|
||
return [0xFF, codon & 0xFF]
|
||
else:
|
||
return [0xFF, (codon >> 8) & 0xFF, codon & 0xFF]
|
||
|
||
def encode(self, manifest: PTOSManifest) -> DeltaGCLSequence:
|
||
"""Model 643: Encode manifest to Delta GCL format."""
|
||
if self.previous is not None:
|
||
delta = self.compute_delta(manifest, self.previous)
|
||
if not delta['deltaFlag']:
|
||
return DeltaGCLSequence(marker='D', bytes=[0x00], field_codes=[])
|
||
|
||
# Full encoding
|
||
encoded = []
|
||
# Strategy index
|
||
encoded.extend(self.encode_codon(manifest.strategy_index))
|
||
# Payload length
|
||
encoded.extend(self.encode_codon(len(manifest.payload)))
|
||
# Payload bytes (simplified)
|
||
encoded.extend(manifest.payload[:64]) # truncate for proof-of-concept
|
||
|
||
self.previous = manifest
|
||
return DeltaGCLSequence(marker='F', bytes=encoded, field_codes=[])
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# MISC Compressor (Unified)
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
@dataclass
|
||
class CompressedBlock:
|
||
"""Output of MISC compression."""
|
||
gcl_bytes: bytes
|
||
trixal: TrixalState
|
||
stamp: TrixalStamp
|
||
strategy: str
|
||
shell_map: Dict[int, Tuple[int, int]]
|
||
canal_width: float
|
||
compression_ratio: float
|
||
|
||
|
||
class MISCConfig:
|
||
"""Configuration for the MISC compressor."""
|
||
def __init__(self,
|
||
block_size: int = 256,
|
||
gwl_window: int = 16,
|
||
homeostatic_alpha: float = 0.5,
|
||
homeostatic_beta: float = 0.5,
|
||
homeostatic_gamma: float = 0.8,
|
||
verbose: bool = False):
|
||
self.block_size = block_size
|
||
self.gwl_window = gwl_window
|
||
self.homeostatic_alpha = homeostatic_alpha
|
||
self.homeostatic_beta = homeostatic_beta
|
||
self.homeostatic_gamma = homeostatic_gamma
|
||
self.verbose = verbose
|
||
|
||
|
||
class MISCCompressor:
|
||
"""Unified MISC Compressor.
|
||
|
||
Combines all components:
|
||
- PIST/DIAT shell coordinates
|
||
- GWL multi-factor similarity
|
||
- Cognitive load routing
|
||
- Thermodynamic trixal quality
|
||
- Delta GCL encoding
|
||
- Homeostatic governance
|
||
"""
|
||
|
||
def __init__(self, config: Optional[MISCConfig] = None):
|
||
self.config = config or MISCConfig()
|
||
self.router = CognitiveLoadRouter()
|
||
self.thermo = ThermodynamicEngine()
|
||
self.governor = HomeostaticGovernor(
|
||
alpha=self.config.homeostatic_alpha,
|
||
beta=self.config.homeostatic_beta,
|
||
gamma=self.config.homeostatic_gamma,
|
||
)
|
||
self.gcl = DeltaGCLEncoder()
|
||
self.block_idx = 0
|
||
self.predicted_ratio = Q16_16.from_float(0.5)
|
||
self.optimal_ratio = Q16_16.from_float(0.3)
|
||
|
||
def compress_block(self, data: bytes) -> Optional[CompressedBlock]:
|
||
"""Compress a single data block through the full MISC pipeline."""
|
||
if not data:
|
||
return None
|
||
|
||
# Phase 1: PIST Shell Map
|
||
shell_map = ShellMapBuilder(data)
|
||
|
||
# Phase 2: GWL Coupling Similarity (windowed)
|
||
coupling = GWLCoupling(len(data))
|
||
densities = []
|
||
stride = max(1, len(data) // 16)
|
||
window = min(self.config.gwl_window, len(data))
|
||
|
||
for i in range(0, len(data), stride):
|
||
row_density = Q16_16(0)
|
||
count = 0
|
||
for j in range(i + 1, min(i + window, len(data))):
|
||
w = coupling.compute(data, i, j, shell_map)
|
||
row_density = row_density + w
|
||
count += 1
|
||
if count > 0:
|
||
row_density = row_density / Q16_16.from_int(count)
|
||
densities.append(row_density)
|
||
|
||
avg_coupling = Q16_16(0)
|
||
if densities:
|
||
avg_coupling = sum(densities, Q16_16(0)) / Q16_16.from_int(len(densities))
|
||
|
||
# Phase 3: Cognitive Load Routing
|
||
strategy, load = self.router.select_strategy(data, self.block_idx)
|
||
|
||
# Phase 4: Apply strategy (simplified)
|
||
compressed = self._apply_strategy(data, strategy, shell_map, avg_coupling)
|
||
|
||
# Phase 5: Thermodynamic Trixal Assessment
|
||
trixal = self.thermo.compute_trixal(data, strategy, load)
|
||
stamp = self.thermo.create_stamp(trixal, data[:32])
|
||
|
||
# Reject if irreversibility too high (unlawful compression)
|
||
ONE = Q16_16(SCALE)
|
||
if trixal.irreversibility > ONE * Q16_16.from_natural(7, 10):
|
||
if self.config.verbose:
|
||
print(f" Block {self.block_idx}: REJECTED (irreversibility "
|
||
f"{trixal.irreversibility.to_float():.3f})")
|
||
strategy = 'RAW_COPY'
|
||
compressed = data
|
||
|
||
# Phase 6: Delta GCL Encode
|
||
manifest = PTOSManifest(
|
||
version=1,
|
||
payload=compressed,
|
||
strategy_index=self.router.STRATEGIES.index(strategy),
|
||
shell_map_data={k: (c.k, c.t) for k, c in shell_map.coords.items()},
|
||
)
|
||
gcl_seq = self.gcl.encode(manifest)
|
||
gcl_bytes = bytes(gcl_seq.bytes)
|
||
|
||
# Phase 7: Homeostatic Update
|
||
actual_ratio = Q16_16.from_natural(len(gcl_bytes), len(data))
|
||
self.governor.update(actual_ratio, self.predicted_ratio, self.optimal_ratio)
|
||
|
||
# Update predicted ratio from recent performance
|
||
self.predicted_ratio = (
|
||
self.predicted_ratio * Q16_16.from_natural(9, 10)
|
||
+ actual_ratio * Q16_16.from_natural(1, 10)
|
||
)
|
||
|
||
result = CompressedBlock(
|
||
gcl_bytes=gcl_bytes,
|
||
trixal=trixal,
|
||
stamp=stamp,
|
||
strategy=strategy,
|
||
shell_map={k: (c.k, c.t) for k, c in shell_map.coords.items()},
|
||
canal_width=self.governor.canal_width.to_float(),
|
||
compression_ratio=actual_ratio.to_float(),
|
||
)
|
||
|
||
if self.config.verbose:
|
||
print(f" Block {self.block_idx}: {strategy:20s} "
|
||
f"ratio={result.compression_ratio:.4f} "
|
||
f"pressure={self.governor.pressure.to_float():.3f} "
|
||
f"canal={result.canal_width:.3f} "
|
||
f"trixal=[{trixal.thermal.to_float():.2f},"
|
||
f"{trixal.work.to_float():.2f},"
|
||
f"{trixal.irreversibility.to_float():.2f}]")
|
||
|
||
self.block_idx += 1
|
||
return result
|
||
|
||
def _apply_strategy(self,
|
||
data: bytes,
|
||
strategy: str,
|
||
shell_map: ShellMapBuilder,
|
||
avg_coupling: Q16_16) -> bytes:
|
||
"""Apply compression strategy.
|
||
|
||
RAW_COPY: return as-is (or entropy coded)
|
||
DELTA: delta from previous identical-mass tokens
|
||
DICTIONARY: 4-byte dictionary encoding
|
||
SHELL_RESONANCE: resonance-based dedup
|
||
PREDICTIVE: predict from shell coordinate neighbors
|
||
"""
|
||
if strategy == 'RAW_COPY':
|
||
return data
|
||
|
||
elif strategy == 'DELTA':
|
||
# Delta-encode within shell resonance groups
|
||
result = bytearray()
|
||
groups = shell_map.resonance_groups()
|
||
prev_values: Dict[int, int] = {}
|
||
for b in data:
|
||
mass = shell_map.get(b).mass
|
||
if mass in prev_values:
|
||
result.append((b - prev_values[mass]) & 0xFF)
|
||
else:
|
||
result.append(b)
|
||
prev_values[mass] = b
|
||
return bytes(result)
|
||
|
||
elif strategy == 'DICTIONARY':
|
||
# Map frequent tokens to short codes
|
||
endpoint_tokens = set(shell_map.endpoint_tokens())
|
||
result = bytearray()
|
||
for b in data:
|
||
if b in endpoint_tokens:
|
||
result.append(b ^ 0x80) # mark as dictionary entry
|
||
else:
|
||
result.append(b)
|
||
return bytes(result)
|
||
|
||
elif strategy == 'SHELL_RESONANCE':
|
||
# Resonance-based: replace tokens with resonant partner delta
|
||
result = bytearray()
|
||
# Find resonant pairs
|
||
groups = shell_map.resonance_groups()
|
||
resonant_map: Dict[int, int] = {}
|
||
for mass, tokens in groups.items():
|
||
if len(tokens) >= 2:
|
||
for i in range(1, len(tokens)):
|
||
resonant_map[tokens[i]] = tokens[0]
|
||
for b in data:
|
||
if b in resonant_map:
|
||
result.append(resonant_map[b])
|
||
else:
|
||
result.append(b)
|
||
return bytes(result)
|
||
|
||
elif strategy == 'PREDICTIVE':
|
||
# Predict from shell coordinate neighbors
|
||
result = bytearray()
|
||
for i, b in enumerate(data):
|
||
if i > 0:
|
||
prev_b = data[i - 1]
|
||
prev_coord = shell_map.get(prev_b)
|
||
cur_coord = shell_map.get(b)
|
||
if prev_coord.is_resonant_with(cur_coord):
|
||
result.append(b) # already predictable
|
||
else:
|
||
result.append(b)
|
||
else:
|
||
result.append(b)
|
||
return bytes(result)
|
||
|
||
return data
|
||
|
||
def compress(self, data: bytes) -> List[CompressedBlock]:
|
||
"""Compress arbitrary data through MISC pipeline."""
|
||
bs = self.config.block_size
|
||
blocks = []
|
||
for offset in range(0, len(data), bs):
|
||
block = data[offset:offset + bs]
|
||
result = self.compress_block(block)
|
||
if result:
|
||
blocks.append(result)
|
||
return blocks
|
||
|
||
|
||
# ──────────────────────────────────────────────────────
|
||
# Public API
|
||
# ──────────────────────────────────────────────────────
|
||
|
||
def compress(data: bytes, config: Optional[MISCConfig] = None) -> List[CompressedBlock]:
|
||
"""High-level MISC compression entry point."""
|
||
compressor = MISCCompressor(config)
|
||
return compressor.compress(data)
|
||
|
||
|
||
def format_report(blocks: List[CompressedBlock]) -> Dict[str, Any]:
|
||
"""Generate a summary of compression results."""
|
||
if not blocks:
|
||
return {'error': 'no blocks'}
|
||
|
||
total_in = 0
|
||
total_out = 0
|
||
strategies: Counter = Counter()
|
||
avg_trixal = [Q16_16(0), Q16_16(0), Q16_16(0)]
|
||
|
||
for blk in blocks:
|
||
# Estimate input size from shell map complexity
|
||
in_size = len(blk.gcl_bytes) / max(blk.compression_ratio, 0.001)
|
||
total_in += int(in_size)
|
||
total_out += len(blk.gcl_bytes)
|
||
strategies[blk.strategy] += 1
|
||
avg_trixal[0] += blk.trixal.thermal
|
||
avg_trixal[1] += blk.trixal.work
|
||
avg_trixal[2] += blk.trixal.irreversibility
|
||
|
||
n = len(blocks)
|
||
return {
|
||
'blocks': n,
|
||
'total_estimated_input': total_in,
|
||
'total_output': total_out,
|
||
'overall_ratio': total_out / max(total_in, 1),
|
||
'savings_percent': (1 - total_out / max(total_in, 1)) * 100,
|
||
'strategies_used': dict(strategies),
|
||
'avg_trixal': {
|
||
'thermal': (avg_trixal[0] / Q16_16.from_int(n)).to_float(),
|
||
'work': (avg_trixal[1] / Q16_16.from_int(n)).to_float(),
|
||
'irreversibility': (avg_trixal[2] / Q16_16.from_int(n)).to_float(),
|
||
},
|
||
'homeostatic_pressure': blocks[-1].canal_width,
|
||
}
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# Quick self-test
|
||
print("MISC Kernel — Manifold-Invariant Shell Compression")
|
||
print("=" * 60)
|
||
|
||
# Test with sample data
|
||
test_data = b"Hello, MISC! This is a test of the manifold-invariant shell compression framework." * 4
|
||
|
||
print(f"Input: {len(test_data)} bytes")
|
||
config = MISCConfig(block_size=64, verbose=True)
|
||
blocks = compress(test_data, config)
|
||
report = format_report(blocks)
|
||
|
||
print(f"\nSummary:")
|
||
print(f" Blocks: {report['blocks']}")
|
||
print(f" Est. input: {report['total_estimated_input']} bytes")
|
||
print(f" Output: {report['total_output']} bytes")
|
||
print(f" Ratio: {report['overall_ratio']:.4f}")
|
||
print(f" Savings: {report['savings_percent']:.1f}%")
|
||
print(f" Strategies: {report['strategies_used']}")
|
||
print(f" Avg trixal: {report['avg_trixal']}")
|