Implements a Von Neumann cellular automaton as well as two new entropy coding data compression methods.

Author: Marioman2023

cellular_automata/Von_Neumann_CA.py

@@ -0,0 +1,94 @@
+"""
+Von Neumann CA with multi-state fading "heatmap" effect in the terminal.
+
+Requirements: numpy
+
+Rules:
+ - Uses Von Neumann neighborhood (4 neighbors).
+ - Alive cells follow BIRTH / SURVIVE rules.
+ - Dead cells fade out gradually through colored stages.
+"""
+
+import numpy as np
+import os
+import time
+
+# ---------- Configuration ----------
+GRID_SIZE = (20, 40)     # (rows, cols)
+PROB_ALIVE = 0.25        # initial alive probability
+WRAP = True              # wrap edges (toroidal)
+BIRTH = {3}              # birth neighbor counts
+SURVIVE = {1, 2}         # survival neighbor counts
+SLEEP_TIME = 0.1         # seconds between frames
+MAX_AGE = 5              # how many steps before a cell fully disappears (0 = dead)
+COLORS = {               # Characters & colors for different ages
+    0: " ",                          # dead
+    1: "\033[92m█\033[0m",           # bright green (newborn)
+    2: "\033[93m█\033[0m",           # yellow
+    3: "\033[91m█\033[0m",           # red
+    4: "\033[31m░\033[0m",           # dim red fading
+    5: "\033[90m·\033[0m",           # grey dust
+}
+# -----------------------------------
+
+def make_initial_grid(shape, prob_alive, seed=None):
+    rng = np.random.default_rng(seed)
+    alive = (rng.random(shape) < prob_alive).astype(np.uint8)
+    return alive.astype(np.uint8) # age 1 for alive, 0 for dead
+
+def count_von_neumann_neighbors(alive_mask, wrap=True):
+    """Count Von Neumann neighbors (4 directions)"""
+    up     = np.roll(alive_mask, -1, axis=0)
+    down   = np.roll(alive_mask,  1, axis=0)
+    left   = np.roll(alive_mask, -1, axis=1)
+    right  = np.roll(alive_mask,  1, axis=1)
+    counts = up + down + left + right
+
+    if not wrap:
+        counts[ 0,  :] -= alive_mask[-1,  :]
+        counts[-1,  :] -= alive_mask[ 0,  :]
+        counts[ :,  0] -= alive_mask[ :, -1]
+        counts[ :, -1] -= alive_mask[ :,  0]
+        counts = np.clip(counts, 0, 4)
+
+    return counts
+
+def step(age_grid, birth=BIRTH, survive=SURVIVE, wrap=WRAP, max_age=MAX_AGE):
+    alive_mask = age_grid > 0
+    neighbor_counts = count_von_neumann_neighbors(alive_mask.astype(np.uint8), wrap=wrap)
+
+    born_mask = (~alive_mask) & np.isin(neighbor_counts, list(birth))
+    survive_mask = alive_mask & np.isin(neighbor_counts, list(survive))
+
+    new_age_grid = age_grid.copy()
+
+    # Alive cells that survive get age reset to 1 if born, else age increment
+    new_age_grid[born_mask] = 1
+    new_age_grid[survive_mask & alive_mask] = 1 # reset alive age for fresh color
+
+    # Fade out dead cells
+    fade_mask = (~born_mask) & (~survive_mask)
+    new_age_grid[fade_mask & (new_age_grid > 0)] += 1
+    new_age_grid[new_age_grid > max_age] = 0 # fully dead
+
+    return new_age_grid
+
+def display(age_grid):
+    """Render grid with colors for each age stage"""
+    os.system("cls" if os.name == "nt" else "clear")
+    for row in age_grid:
+        print("".join(COLORS.get(age, COLORS[MAX_AGE]) for age in row))
+
+def main():
+    grid = make_initial_grid(GRID_SIZE, prob_alive=PROB_ALIVE)
+
+    try:
+        while True:
+            display(grid)
+            grid = step(grid, birth=BIRTH, survive=SURVIVE, wrap=WRAP, max_age=MAX_AGE)
+            time.sleep(SLEEP_TIME)
+    except KeyboardInterrupt:
+        print("\nStopped.")
+
+if __name__ == "__main__":
+    main()

data_compression/Arithmetic_Coding.py

@@ -0,0 +1,69 @@
+from collections_extended import bag
+from decimal import Decimal, getcontext
+
+# Set high precision for decimal calculations
+getcontext().prec = 50
+
+def build_probability_table(data):
+    """Returns a dictionary int the form (symbol: probability)"""
+    freq = bag(data) # A bag is like a multiset
+    return {char: Decimal(freq.count(char)) / Decimal(len(data)) for char in set(freq)}
+
+def arithmetic_encode(data, prob_table):
+    """Preforms arithmetic coding compression"""
+    symbols = sorted(prob_table.keys())
+    cumulative = {}
+    cumulative_sum = Decimal('0.0')
+    for sym in symbols:
+        cumulative[sym] = cumulative_sum
+        cumulative_sum += prob_table[sym]
+
+    low, high = Decimal('0.0'), Decimal('1.0')
+    for symbol in data:
+        range_ = high - low
+        high   = low  + range_ * (cumulative[symbol] + prob_table[symbol])
+        low    = low  + range_ *  cumulative[symbol]
+
+    return (low + high) / 2, len(data)
+
+def arithmetic_decode(encoded_value, length, prob_table):
+    """Decodes an arithmetic-coded value"""
+    symbols = sorted(prob_table.keys())
+    cumulative = {}
+    cumulative_sum = Decimal('0.0')
+    for sym in symbols:
+        cumulative[sym] = cumulative_sum
+        cumulative_sum += prob_table[sym]
+
+    result = []
+    low, high = Decimal('0.0'), Decimal('1.0')
+    value = Decimal(str(encoded_value))
+
+    for _ in range(length):
+        range_ = high - low
+        for sym in symbols:
+            sym_low  =     low + range_ * cumulative[sym]
+            sym_high = sym_low + range_ * prob_table[sym]
+            if sym_low <= value < sym_high:
+                result.append(sym)
+                low, high = sym_low, sym_high
+                break
+
+    return ''.join(result)
+
+if __name__ == "__main__":
+    text = "this is text used for testing"
+    print(f"Original: {text}")
+
+    prob_table = build_probability_table(text)
+    encoded_value, length = arithmetic_encode(text, prob_table)
+    print(f"Encoded value: {encoded_value}")
+
+    decoded_text = arithmetic_decode(encoded_value, length, prob_table)
+    print(f"Decoded: {decoded_text}")
+    
+    # Show compression ratio
+    import sys
+    original_size = sys.getsizeof(text)
+    encoded_size = sys.getsizeof(str(encoded_value))
+    print(f"Compression ratio: {original_size / encoded_size:.2f}")

data_compression/CABAC.py

@@ -0,0 +1,98 @@
+class CABAC:
+    def __init__(self):
+        self.low     = 0
+        self.high    = (1 << 32) - 1
+        self.context = [0.5] * 256 # probability model for 256 contexts
+
+    def _update_context(self, ctx, bit):
+        # Simple adaptation: move probability toward observed bit
+        alpha = 0.05
+        self.context[ctx] = (1 - alpha) * self.context[ctx] + alpha * bit
+
+    def encode_bit(self, bit, ctx, output):
+        prob   = self.context[ctx]
+        range_ = self.high - self.low + 1
+        split  = self.low  + int(range_ * prob)
+
+        if bit == 0:
+            self.high = split
+        else:
+            self.low  = split + 1
+
+        while (self.high ^ self.low) < (1 << 24):
+            output.append((self.high >> 24) & 0xFF)
+            self.low  =  (self.low   << 8)  & 0xFFFFFFFF
+            self.high =  ((self.high << 8)  & 0xFFFFFFFF) | 0xFF
+
+        self._update_context(ctx, bit)
+
+    def finish_encoding(self, output):
+        for _ in range(4):
+            output.append((self.low >> 24) & 0xFF)
+            self.low =    (self.low << 8)  & 0xFFFFFFFF
+
+    def decode_bit(self, ctx, input_bits):
+        prob   = self.context[ctx]
+        range_ = self.high - self.low + 1
+        split  = self.low  + int(range_ * prob)
+
+        if self.code <= split:
+            self.high = split
+            bit = 0
+        else:
+            self.low = split + 1
+            bit = 1
+
+        while (self.high ^ self.low) < (1 << 24):
+            self.code = ((self.code << 8) & 0xFFFFFFFF) | next(input_bits)
+            self.low  = (self.low   << 8) & 0xFFFFFFFF
+            self.high = ((self.high << 8) & 0xFFFFFFFF) | 0xFF
+
+        self._update_context(ctx, bit)
+        return bit
+
+    def start_decoding(self, encoded_bytes):
+        self.low = 0
+        self.high = (1 << 32) - 1
+        self.code = 0
+        input_bits = iter(encoded_bytes)
+        for _ in range(4):
+            self.code = (self.code << 8) | next(input_bits)
+        return input_bits
+
+
+def string_to_bits(s):
+    return [(byte >> i) & 1 for byte in s.encode('utf-8') for i in range(7, -1, -1)]
+
+def bits_to_string(bits):
+    b = bytearray()
+    for i in range(0, len(bits), 8):
+        byte = 0
+        for bit in bits[i:i+8]:
+            byte = (byte << 1) | bit
+        b.append(byte)
+    return b.decode('utf-8')
+
+def main(text: str):
+    encoder = CABAC()
+    output_bytes = []
+    bits = string_to_bits(text)
+
+    for i, bit in enumerate(bits):
+        ctx = i % 256 # simple positional context
+        encoder.encode_bit(bit, ctx, output_bytes)
+    encoder.finish_encoding(output_bytes)
+
+    # Decode
+    decoder = CABAC()
+    bitstream = decoder.start_decoding(iter(output_bytes))
+    decoded_bits = [decoder.decode_bit(i % 256, bitstream) for i in range(len(bits))]
+    decoded_text = bits_to_string(decoded_bits)
+
+    print("Original:", text)
+    print("Decoded :", decoded_text)
+    print("Compressed size (bytes):", len(output_bytes))
+
+if __name__ == "__main__":
+    # Example usage
+    main("Hello CABAC!")