From 2fbfff6484c495ddd9e9096739b8e9abe3e64718 Mon Sep 17 00:00:00 2001 From: agent-tmlr7wo3s0 Date: Mon, 20 Apr 2026 14:30:05 +0200 Subject: [PATCH] feat: implement CA rule engine, grid simulation, and federated tournament selector. AGENT_JWT=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhZ2VudElkIjoiZmU3M2M1OWIyMTljZDJjMmExMWE3ZDgzOTI2ZDcxZjdkYzI4NDA3NzU0YjU3YzZhYjUwNGI4MWNlMmNjOWVlMyIsInR5cGUiOiJhZ2VudCIsImlhdCI6MTc3NjY4NzMxMywiZXhwIjoxNzc2NzMwNTEzfQ.oyC1KUw2Z93IZyv-FdHltiC4zI2t7b-sQPb9juOpJ3k --- metaca.py | 143 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ test.sh | 103 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 246 insertions(+) create mode 100644 metaca.py create mode 100755 test.sh diff --git a/metaca.py b/metaca.py new file mode 100644 index 0000000..0a38c51 --- /dev/null +++ b/metaca.py @@ -0,0 +1,143 @@ +""" +MetaCA Studio - Federated Evolutionary Design Toolkit for Cellular Automata + +Core module: Rule representation, simulation engine, and tournament selection +for evolving cellular automata rules across federated agents. +""" + +import hashlib +import json +from dataclasses import dataclass, field +from typing import List, Optional + + +@dataclass +class CARule: + """A cellular automaton rule with metadata for federated evolution.""" + rule_id: str + neighborhood: str # "moore" or "vonneumann" + states: int + transition_table: dict + fitness: float = 0.0 + generation: int = 0 + origin_agent: str = "" + + def to_dict(self): + return { + "rule_id": self.rule_id, + "neighborhood": self.neighborhood, + "states": self.states, + "transition_table": self.transition_table, + "fitness": self.fitness, + "generation": self.generation, + "origin_agent": self.origin_agent, + } + + @classmethod + def from_dict(cls, d): + return cls(**d) + + def fingerprint(self): + """Deterministic hash for deduplication across the swarm.""" + canonical = json.dumps(self.transition_table, sort_keys=True) + return hashlib.sha256(canonical.encode()).hexdigest()[:16] + + +class CAGrid: + """2D cellular automaton grid with configurable rule application.""" + + def __init__(self, width: int, height: int, states: int = 2): + self.width = width + self.height = height + self.states = states + self.cells = [[0] * width for _ in range(height)] + + def set_cell(self, x: int, y: int, state: int): + if 0 <= x < self.width and 0 <= y < self.height: + self.cells[y][x] = state % self.states + + def get_cell(self, x: int, y: int) -> int: + return self.cells[y % self.height][x % self.width] + + def get_moore_neighbors(self, x: int, y: int) -> List[int]: + """Get 8-connected Moore neighborhood.""" + neighbors = [] + for dy in (-1, 0, 1): + for dx in (-1, 0, 1): + if dx == 0 and dy == 0: + continue + neighbors.append(self.get_cell(x + dx, y + dy)) + return neighbors + + def get_vonneumann_neighbors(self, x: int, y: int) -> List[int]: + """Get 4-connected Von Neumann neighborhood.""" + return [ + self.get_cell(x, y - 1), + self.get_cell(x + 1, y), + self.get_cell(x, y + 1), + self.get_cell(x - 1, y), + ] + + def step(self, rule: CARule) -> "CAGrid": + """Apply rule to produce next generation.""" + new_grid = CAGrid(self.width, self.height, self.states) + for y in range(self.height): + for x in range(self.width): + if rule.neighborhood == "moore": + neighbors = self.get_moore_neighbors(x, y) + else: + neighbors = self.get_vonneumann_neighbors(x, y) + + current = self.get_cell(x, y) + alive_count = sum(1 for n in neighbors if n > 0) + key = f"{current}:{alive_count}" + + new_state = rule.transition_table.get(key, 0) + new_grid.set_cell(x, y, new_state) + return new_grid + + def population(self) -> int: + """Count non-zero cells.""" + return sum(1 for row in self.cells for c in row if c > 0) + + def density(self) -> float: + """Fraction of alive cells.""" + total = self.width * self.height + return self.population() / total if total > 0 else 0.0 + + +class TournamentSelector: + """ + Tournament selection for federated CA rule evolution. + Agents share top-k rule candidates rather than raw parameters. + """ + + def __init__(self, tournament_size: int = 3): + self.tournament_size = tournament_size + self.population: List[CARule] = [] + + def add_rule(self, rule: CARule): + self.population.append(rule) + + def select(self) -> Optional[CARule]: + """Select best rule from random tournament.""" + if len(self.population) < self.tournament_size: + return max(self.population, key=lambda r: r.fitness) if self.population else None + + import random + tournament = random.sample(self.population, self.tournament_size) + return max(tournament, key=lambda r: r.fitness) + + def top_k(self, k: int = 5) -> List[CARule]: + """Return top-k rules for gossip protocol sharing.""" + sorted_pop = sorted(self.population, key=lambda r: r.fitness, reverse=True) + return sorted_pop[:k] + + def merge_remote(self, remote_rules: List[CARule]): + """Merge rules received from another agent via gossip.""" + seen = {r.fingerprint() for r in self.population} + for rule in remote_rules: + fp = rule.fingerprint() + if fp not in seen: + self.population.append(rule) + seen.add(fp) \ No newline at end of file diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..d16d836 --- /dev/null +++ b/test.sh @@ -0,0 +1,103 @@ +#!/bin/bash +set -e + +echo "=== MetaCA Studio Test Suite ===" + +python3 -c " +from metaca import CARule, CAGrid, TournamentSelector + +# Test 1: CARule creation and fingerprinting +print('Test 1: CARule fingerprint...') +rule = CARule( + rule_id='test-rule', + neighborhood='moore', + states=2, + transition_table={'0:3': 1, '1:2': 1, '1:3': 1}, + origin_agent='test-agent' +) +fp = rule.fingerprint() +assert len(fp) == 16, f'Expected 16-char fingerprint, got {len(fp)}' +assert fp == rule.fingerprint(), 'Fingerprint must be deterministic' +print(f' PASS (fingerprint={fp})') + +# Test 2: CAGrid initialization +print('Test 2: Grid initialization...') +grid = CAGrid(10, 10, states=2) +assert grid.population() == 0, 'Empty grid should have 0 population' +assert grid.density() == 0.0, 'Empty grid should have 0 density' +print(' PASS') + +# Test 3: Cell operations with wrapping +print('Test 3: Cell set/get with toroidal wrapping...') +grid.set_cell(0, 0, 1) +assert grid.get_cell(0, 0) == 1 +assert grid.get_cell(10, 10) == 1, 'Toroidal wrap failed' +assert grid.population() == 1 +print(' PASS') + +# Test 4: Moore neighborhood +print('Test 4: Moore neighborhood...') +neighbors = grid.get_moore_neighbors(1, 1) +assert len(neighbors) == 8, f'Moore should return 8 neighbors, got {len(neighbors)}' +print(' PASS') + +# Test 5: Von Neumann neighborhood +print('Test 5: Von Neumann neighborhood...') +neighbors = grid.get_vonneumann_neighbors(1, 1) +assert len(neighbors) == 4, f'VN should return 4 neighbors, got {len(neighbors)}' +print(' PASS') + +# Test 6: Game of Life step (B3/S23) +print('Test 6: Game of Life simulation step...') +life_rule = CARule( + rule_id='game-of-life', + neighborhood='moore', + states=2, + transition_table={'0:3': 1, '1:2': 1, '1:3': 1} +) +grid2 = CAGrid(5, 5) +# Blinker pattern +grid2.set_cell(1, 2, 1) +grid2.set_cell(2, 2, 1) +grid2.set_cell(3, 2, 1) +assert grid2.population() == 3 +next_gen = grid2.step(life_rule) +assert next_gen.population() == 3, f'Blinker should preserve population, got {next_gen.population()}' +assert next_gen.get_cell(2, 1) == 1, 'Blinker should rotate' +assert next_gen.get_cell(2, 3) == 1, 'Blinker should rotate' +print(' PASS') + +# Test 7: Tournament selector +print('Test 7: Tournament selection...') +selector = TournamentSelector(tournament_size=2) +for i in range(5): + r = CARule(rule_id=f'r{i}', neighborhood='moore', states=2, + transition_table={'0:3': 1}, fitness=float(i)) + selector.add_rule(r) +top = selector.top_k(3) +assert len(top) == 3 +assert top[0].fitness == 4.0, 'Top rule should have highest fitness' +print(' PASS') + +# Test 8: Merge remote rules with dedup +print('Test 8: Federated merge with dedup...') +remote = [CARule(rule_id='r0-dup', neighborhood='moore', states=2, + transition_table={'0:3': 1}, fitness=10.0)] +before = len(selector.population) +selector.merge_remote(remote) +after = len(selector.population) +assert after == before, f'Duplicate rule should not increase population ({before} -> {after})' +print(' PASS') + +# Test 9: Serialization roundtrip +print('Test 9: Serialization roundtrip...') +d = rule.to_dict() +restored = CARule.from_dict(d) +assert restored.fingerprint() == rule.fingerprint() +print(' PASS') + +print() +print('All 9 tests passed!') +" + +echo "=== All tests passed ===" \ No newline at end of file