Skip to content

Instantly share code, notes, and snippets.

@douxxtech

douxxtech/gol.py

Last active November 6, 2025 23:02
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save douxxtech/3979e1551f45dd109b93acdf22ef83ac to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save douxxtech/3979e1551f45dd109b93acdf22ef83ac to your computer and use it in GitHub Desktop.
Download ZIP
Raw
gol.py
# My own Game Of Life
# Blog post: https://douxx.blog/?p=4-my-own-game-of-life
# Made by douxx.tech / github.com/douxx.tech
# side note: there are some hardcoded values, like
# the 'hardcore' triggers values around lines 190.
import sys
import socket
import struct
import numpy as np
from PIL import Image
import argparse
import time
from pysstv.color import Robot36
from piwave import PiWave
import requests
import json
class Log:
COLORS = {
'reset': '\033[0m',
'bright_red': '\033[91m',
'bright_green': '\033[92m',
'bright_yellow': '\033[93m',
'bright_cyan': '\033[96m',
'magenta': '\033[35m',
'cyan': '\033[36m',
'yellow': '\033[33m',
'green': '\033[32m',
'blue': '\033[34m',
'bright_blue': '\033[94m',
}
ICONS = {
'success': 'OK',
'error': 'ERR',
'warning': 'WARN',
'info': 'INFO',
'rtl': 'RTL',
'gol': 'GOL',
'signal': 'SIG',
'image': 'IMG',
'gen': 'GEN',
'sstv': 'SSTV',
'fm': 'FM',
'stats': 'STATS',
}
@classmethod
def print(cls, message: str, style: str = '', icon: str = '', end: str = '\n'):
color = cls.COLORS.get(style, '')
icon_char = cls.ICONS.get(icon, '')
if icon_char:
if color:
print(f"{color}[{icon_char}]\033[0m {message}", end=end)
else:
print(f"[{icon_char}] {message}", end=end)
else:
if color:
print(f"{color}{message}\033[0m", end=end)
else:
print(f"{message}", end=end)
sys.stdout.flush()
@classmethod
def header(cls, text: str):
cls.print(text, 'bright_blue', end='\n\n')
sys.stdout.flush()
@classmethod
def success(cls, message: str):
cls.print(message, 'bright_green', 'success')
@classmethod
def info(cls, message: str):
cls.print(message, 'bright_cyan', 'info')
@classmethod
def rtl_message(cls, message: str):
cls.print(message, 'magenta', 'rtl')
@classmethod
def gol_message(cls, message: str):
cls.print(message, 'cyan', 'gol')
@classmethod
def signal_message(cls, message: str):
cls.print(message, 'yellow', 'signal')
@classmethod
def image_message(cls, message: str):
cls.print(message, 'green', 'image')
@classmethod
def gen_message(cls, message: str):
cls.print(message, 'blue', 'gen')
@classmethod
def sstv_message(cls, message: str):
cls.print(message, 'bright_yellow', 'sstv')
@classmethod
def fm_message(cls, message: str):
cls.print(message, 'bright_green', 'fm')
@classmethod
def stats_message(cls, message: str):
cls.print(message, 'bright_cyan', 'stats')
class RTLGameOfLife:
def __init__(self, host, port, rtl_frequency, fm_frequency, mutation_rate, output_prefix, set_back_freq, sample_time, hardcore_mode, discord):
self.host = host
self.port = port
self.rtl_frequency = rtl_frequency
self.fm_frequency = fm_frequency
self.mutation_rate = mutation_rate
self.output_prefix = output_prefix
self.set_back_freq = set_back_freq
self.sample_time = sample_time
self.hardcore = hardcore_mode
self.hardcore_active = False
self.hardcore_pending = False
self.discord = discord
self.grid_size = 60
self.grid = np.zeros((self.grid_size, self.grid_size), dtype=bool)
self.generation = 0
self.prev_alive = 0
self.prev_grid = None
Log.header("RTL-SDR Game of Life -> SSTV -> FM Broadcast")
Log.info(f"Grid: {self.grid_size}x{self.grid_size}")
Log.info(f"RTL-TCP: {host}:{port} @ {rtl_frequency} MHz")
Log.info(f"FM Broadcast: {fm_frequency} MHz")
if self.discord:
Log.info(f"Discord webhook: enabled")
print()
def count_neighbors(self, x, y):
count = 0
for i in range(-1, 2):
for j in range(-1, 2):
if i == 0 and j == 0:
continue
nx = (x + i) % self.grid_size
ny = (y + j) % self.grid_size
if self.grid[nx][ny]:
count += 1
return count
def apply_gol_rules(self):
# save previous grid for change tracking
self.prev_grid = self.grid.copy()
new_grid = np.zeros_like(self.grid)
if self.hardcore_pending:
self.hardcore_active = not self.hardcore_active
self.hardcore_pending = False
if self.hardcore_active:
Log.gol_message("HARDCORE mode ACTIVATED (rules apply this generation)")
else:
Log.gol_message("HARDCORE mode DEACTIVATED (rules apply this generation)")
if self.hardcore_active:
Log.gol_message("Applying HARDCORE mode rules")
births = 0
deaths = 0
for i in range(self.grid_size):
for j in range(self.grid_size):
neighbors = self.count_neighbors(i, j)
was_alive = self.grid[i][j]
if self.hardcore_active:
if was_alive:
new_grid[i][j] = neighbors == 2
else:
new_grid[i][j] = neighbors == 4
else:
if was_alive:
new_grid[i][j] = neighbors in [2, 3]
else:
new_grid[i][j] = neighbors == 3
# track births and deaths
if new_grid[i][j] and not was_alive:
births += 1
elif not new_grid[i][j] and was_alive:
deaths += 1
self.grid = new_grid
return births, deaths
def check_pandemic_trigger(self):
border = (
np.sum(self.grid[0,:]) +
np.sum(self.grid[-1,:]) +
np.sum(self.grid[:,0]) +
np.sum(self.grid[:,-1])
)
maxcells = self.grid_size * 4
p = border / maxcells
if self.hardcore and not self.hardcore_active and not self.hardcore_pending and p > 0.35:
Log.gol_message(f"HARDCORE trigger detected: border density={p*100:.1f}% (will activate NEXT generation)")
self.hardcore_pending = True
if self.hardcore and self.hardcore_active and not self.hardcore_pending and p < 0.15:
Log.gol_message(f"HARDCORE deactivation trigger: border density={p*100:.1f}% (will deactivate NEXT generation)")
self.hardcore_pending = True
def connect_and_sample(self, target_duration):
Log.rtl_message(f"Connecting to {self.host}:{self.port}...")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
sock.connect((self.host, self.port))
freq_hz = int(self.rtl_frequency * 1_000_000)
freq_cmd = struct.pack('>BI', 1, freq_hz)
sock.send(freq_cmd)
Log.rtl_message(f"Set frequency to {self.rtl_frequency} MHz")
Log.signal_message(f"Sampling for {target_duration:.1f} seconds...")
samples = []
start_time = time.time()
bytes_received = 0
while time.time() - start_time < target_duration:
data = sock.recv(16384)
if not data:
break
samples.extend(data)
bytes_received += len(data)
elapsed = time.time() - start_time
if self.set_back_freq:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
sock.connect((self.host, self.port))
freq_hz = int(self.set_back_freq * 1_000_000)
freq_cmd = struct.pack('>BI', 1, freq_hz)
sock.send(freq_cmd)
Log.rtl_message(f"Set RTL back to {self.set_back_freq} MHz")
except Exception as e:
Log.rtl_message(f"Failed to set-back frequency: {e}")
sock.close()
Log.signal_message(f"Collected {len(samples)} samples in {elapsed:.2f}s ({bytes_received/1024:.1f} KB)")
return samples
def mutate_from_signal(self, samples):
"""
Direct signal-to-grid mapping using hash-based deterministic selection.
Returns mutation statistics for logging and reporting.
"""
samples_array = np.frombuffer(bytes(samples), dtype=np.uint8).astype(np.float32)
# Ensure even number of samples for I/Q pairs
if len(samples_array) % 2 != 0:
samples_array = samples_array[:-1]
n = self.grid_size * self.grid_size
# Instantaneous power (I^2 + Q^2 approx.)
i_samples = samples_array[0::2]
q_samples = samples_array[1::2]
# Center around 127.5 (DC offset)
i_centered = i_samples - 127.5
q_centered = q_samples - 127.5
# Instantaneous power
power = i_centered**2 + q_centered**2
# Downsample power to grid cells
step = max(1, len(power) // n)
cell_powers = np.array([power[i:i+step].max() for i in range(0, len(power), step)])[:n]
min_p = cell_powers.min()
max_p = cell_powers.max()
range_p = max_p - min_p
if range_p > 0:
normalized = ((cell_powers - min_p) / range_p * 255).astype(np.uint8)
else:
normalized = (cell_powers / (cell_powers.mean() + 1) * 255).astype(np.uint8)
modulo_threshold = int(256 / (10 * self.mutation_rate + 1))
modulo_threshold = max(1, min(128, modulo_threshold))
mutation_mask = (normalized % modulo_threshold == 0).reshape(self.grid_size, self.grid_size)
# Flood protection
current_alive = np.sum(self.grid)
current_density = current_alive / n
flood_protected = False
if current_density > 0.6:
potential_new_alive = np.sum(mutation_mask & ~self.grid)
potential_new_dead = np.sum(mutation_mask & self.grid)
if current_alive + potential_new_alive - potential_new_dead > n * 0.75:
mutation_mask = mutation_mask & self.grid
flood_protected = True
Log.gol_message(f"Flood protection: limiting mutations to only kill cells (density: {current_density*100:.1f}%)")
# Track what mutations do
cells_killed = np.sum(mutation_mask & self.grid)
cells_born = np.sum(mutation_mask & ~self.grid)
self.grid ^= mutation_mask
flipped = int(np.sum(mutation_mask))
pct = (flipped / n) * 100
new_alive = np.sum(self.grid)
new_density = new_alive / n
Log.gol_message(f"Signal mutation: {flipped} cells ({pct:.2f}%) - killed: {cells_killed}, born: {cells_born}")
Log.gol_message(f"Density: {current_density*100:.2f}% → {new_density*100:.2f}% (mod threshold: {modulo_threshold})")
return {
"flipped": flipped,
"cells_killed": cells_killed,
"cells_born": cells_born,
"density_before": current_density,
"density_after": new_density,
"threshold": modulo_threshold,
"flood_protected": flood_protected,
}
def create_image(self):
img = Image.new('RGB', (320, 240), color=(0, 0, 0))
pixels = img.load()
scale_x = 4
scale_y = 4
offset_x = (320 - 60 * scale_x) // 2
offset_y = 0
for i in range(self.grid_size):
for j in range(self.grid_size):
color = (255, 255, 255) if self.grid[i][j] else (0, 0, 0)
for dy in range(scale_y):
for dx in range(scale_x):
pixels[offset_x + j * scale_x + dx, offset_y + i * scale_y + dy] = color
alive_count = np.sum(self.grid)
return img, alive_count
def print_detailed_stats(self, mutation_stats, gol_stats, alive_count):
"""Print detailed generation statistics"""
total_cells = self.grid_size * self.grid_size
density = alive_count / total_cells
births, deaths = gol_stats
delta = alive_count - self.prev_alive
# Calculate changes from previous grid if available
if self.prev_grid is not None:
cells_changed = np.sum(self.grid != self.prev_grid)
change_pct = (cells_changed / total_cells) * 100
else:
cells_changed = 0
change_pct = 0.0
Log.stats_message("═" * 60)
Log.stats_message(f"Generation {self.generation} Summary:")
Log.stats_message(f" Population: {alive_count}/{total_cells} ({density*100:.2f}%)")
Log.stats_message(f" Change: {delta:+d} cells ({(delta/total_cells)*100:+.2f}%)")
Log.stats_message(f" Conway's rules: {births} births, {deaths} deaths")
Log.stats_message(f" Grid changes: {cells_changed} cells ({change_pct:.2f}%)")
Log.stats_message(f" Mutations: {mutation_stats['flipped']} ({mutation_stats['cells_born']} born, {mutation_stats['cells_killed']} killed)")
if mutation_stats['flood_protected']:
Log.stats_message(f" Flood protection: ACTIVE")
if self.hardcore_active:
Log.stats_message(f" Hardcore mode: 🔥 ACTIVE")
elif self.hardcore_pending:
Log.stats_message(f" Hardcore mode: ⏳ PENDING (next gen)")
Log.stats_message("═" * 60)
def run(self):
Log.info("Initializing PiWave...")
pw = PiWave(
frequency=self.fm_frequency,
ps="SSTV-GOL",
rt="Game of Life SSTV",
debug=True
)
Log.success("PiWave initialized")
print()
Log.gol_message("Initializing empty grid...")
self.grid = np.zeros((self.grid_size, self.grid_size), dtype=bool)
Log.success("Grid initialized - waiting for radio signals...")
print()
while True:
Log.gen_message(f"Generation {self.generation}")
samples = self.connect_and_sample(self.sample_time)
mutation_stats = self.mutate_from_signal(samples)
Log.gol_message("Applying Conway's rules...")
births, deaths = self.apply_gol_rules()
self.check_pandemic_trigger()
self.generation += 1
img, alive_count = self.create_image()
png_filename = f"{self.output_prefix}.png"
img.save(png_filename)
Log.image_message(f"Saved {png_filename}")
self.print_detailed_stats(mutation_stats, (births, deaths), alive_count)
if self.discord:
delta = alive_count - self.prev_alive
total_cells = self.grid_size * self.grid_size
density = alive_count / total_cells
pct_change = (delta / total_cells) * 100
hardcore_status = "**ACTIVE**" if self.hardcore_active else ("__PENDING__" if self.hardcore_pending else "inactive")
payload = {
"username": "GOL-Bot",
"content": (
f"**Generation {self.generation}**\n"
f"Population: {alive_count}/{total_cells} ({density*100:.2f}%)\n"
f"Change: {delta:+d} ({pct_change:+.2f}%)\n"
f"Conway: {births} births, {deaths} deaths\n"
f"Mutations: {mutation_stats['flipped']} ({mutation_stats['cells_born']}↑ {mutation_stats['cells_killed']}↓)\n"
f"RTL freq: {self.rtl_frequency} MHz\n"
f"Hardcore: {hardcore_status}"
)
}
try:
with open(png_filename, "rb") as f:
files = {"file": (f"generation_{self.generation}.png", f, "image/png")}
r = requests.post(self.discord, data={"payload_json": json.dumps(payload)}, files=files)
if r.status_code >= 300:
Log.print(f"Discord webhook failed: {r.status_code}", 'bright_red', 'error')
else:
Log.success("Sent to Discord")
except Exception as e:
Log.print(f"Discord error: {e}", 'bright_red', 'error')
self.prev_alive = alive_count
wav_filename = f"{self.output_prefix}.wav"
Log.sstv_message(f"Encoding to SSTV (Robot 36)...")
sstv = Robot36(img, 48000, 16)
sstv.write_wav(wav_filename)
Log.sstv_message(f"Saved {wav_filename}")
time.sleep(0.5)
samples = None
Log.fm_message(f"Broadcasting on {self.fm_frequency} MHz...")
pw.play(wav_filename)
print()
time.sleep(38)
parser = argparse.ArgumentParser(description='RTL-SDR Game of Life -> SSTV -> FM Broadcaster')
parser.add_argument('-H', '--host', default='localhost', help='RTL-TCP host')
parser.add_argument('-P', '--port', type=int, default=1234, help='RTL-TCP port')
parser.add_argument('-r', '--rtl-freq', type=float, required=True, help='RTL frequency in MHz')
parser.add_argument('-f', '--fm-freq', type=float, required=True, help='FM broadcast frequency in MHz')
parser.add_argument('-o', '--output', default='gol', help='Output filename name')
parser.add_argument('-t', '--sample-time', type=float, default=2, help="Time (in seconds) of sampling.")
parser.add_argument('-m', '--mutation-rate', type=float, default=2, help="Mutation rate control. Higher values = sparser mutations (0.1-2.0 recommended)")
parser.add_argument('-s', '--set-back', type=float, help='Optional frequency to set RTL back to after sampling (MHz)')
parser.add_argument('-p', '--hardcore', action='store_true', help='enable border pandemic hardcore mode')
parser.add_argument('-d', '--discord', help='Discord webhook URL')
args = parser.parse_args()
gol = RTLGameOfLife(
host=args.host,
port=args.port,
rtl_frequency=args.rtl_freq,
fm_frequency=args.fm_freq,
mutation_rate=args.mutation_rate,
output_prefix=args.output,
set_back_freq=args.set_back,
sample_time=args.sample_time,
hardcore_mode=args.hardcore,
discord=args.discord
)
gol.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment