timofej/neuropercolation
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

second try

Browse Source
This commit is contained in:
Richard Feistenauer 2019-01-31 13:38:48 +01:00
parent a8d17f6fe9
commit be57f89e78
8 changed files with 169 additions and 221 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -2,4 +2,5 @@
*.pyc__pycache__
__pycache__
.coverage
htmlcov/
htmlcov/
*.orig

60
scripts/main_ui.py
View File

@ -1,63 +1,51 @@
#!/usr/bin/env python3
import random
from cellular_automaton.cellular_automaton import CellularAutomaton, CellularAutomatonProcessor
from cellular_automaton.ca_rule import Rule
from cellular_automaton.ca_neighborhood import MooreNeighborhood, EdgeRule
from cellular_automaton.ca_display import PyGameFor2D
from cellular_automaton.ca_cell_state import CellState
from cellular_automaton.ca_grid import Grid
from cellular_automaton.ca_rule import Rule
class TestRule(Rule):
def evolve_cell(self, cell, iteration_index):
if cell.state is None:
return self._init_state(cell)
else:
return self._evolve_state(cell, iteration_index)
@staticmethod
def _evolve_state(cell, iteration_index):
def evolve_cell(last_cell_state, last_neighbour_states):
try:
left_neighbour_state = cell.neighbours[0].state.get_status_of_iteration(iteration_index - 1)
active = cell.state.set_status_of_iteration(left_neighbour_state, iteration_index)
if active:
cell.is_set_for_redraw = True
return active
return last_neighbour_states[0]
except IndexError:
return False
pass
return False
@staticmethod
def _init_state(cell):
class MyState(CellState):
def __init__(self):
rand = random.randrange(0, 101, 1)
if rand <= 99:
cell.state = MyStatus(0)
return False
else:
cell.state = MyStatus(1)
cell.is_set_for_redraw = True
return True
init = 0
if rand > 99:
init = 1
class MyStatus(CellState):
def __init__(self, initial_state):
super().__init__([initial_state])
super().__init__(init, draw_first_state=False)
def get_state_draw_color(self, iteration):
red = 0
if self._state_slots[iteration % 2][0]:
if self.get_state_of_last_iteration(iteration)[0]:
red = 255
return [red, 0, 0]
if __name__ == "__main__":
import random
from multiprocessing import freeze_support
from cellular_automaton.cellular_automaton import CellularAutomaton, CellularAutomatonProcessor
from cellular_automaton.ca_neighborhood import MooreNeighborhood, EdgeRule
from cellular_automaton.ca_display import PyGameFor2D
from cellular_automaton.ca_grid import Grid
freeze_support()
random.seed(1000)
rule = TestRule()
grid = Grid(dimension=[400, 400],
neighborhood=MooreNeighborhood(EdgeRule.FIRST_AND_LAST_CELL_OF_DIMENSION_ARE_NEIGHBORS))
neighborhood=MooreNeighborhood(EdgeRule.FIRST_AND_LAST_CELL_OF_DIMENSION_ARE_NEIGHBORS),
state_class=MyState)
ca = CellularAutomaton(grid, rule)
ca_window = PyGameFor2D(window_size=[1000, 800], cellular_automaton=ca)
ca_processor = CellularAutomatonProcessor(process_count=8)
ca_processor = CellularAutomatonProcessor(process_count=2, cellular_automaton=ca)
ca_window.main_loop(cellular_automaton_processor=ca_processor,
ca_iterations_per_draw=5)

66
src/cellular_automaton/ca_cell.py
View File

@ -1,7 +1,61 @@
import multiprocessing
from cellular_automaton.ca_cell_state import CellState
class Cell:
def __init__(self, name, coordinate: list):
self.name = name
self.coordinate = coordinate
self.neighbours = []
self.state = None
self.is_set_for_redraw = False
def __init__(self, name, state_class: CellState.__class__, coordinate: list):
self._name = name
self._coordinate = coordinate
self._state = state_class()
self._neighbours = []
self._active = multiprocessing.Value('i', 1)
self._age = multiprocessing.Value('i', 0)
def set_neighbours(self, neighbours):
self._neighbours = neighbours
def get_coordinate(self):
return self._coordinate
def evolve_if_ready(self, rule):
if self._neighbours_are_younger():
if self._is_active():
new_state = rule(self.get_current_state(), self.get_neighbour_states())
self.set_new_state_and_activate(new_state)
self.increase_age()
def _neighbours_are_younger(self):
for n in self._neighbours:
if n.get_age() < self.get_age():
return False
def get_age(self):
return self._age.value
def _is_active(self):
return self._active.value > self._age.value
def get_current_state(self):
return self._state.get_state_of_iteration(self._age.value)
def get_neighbour_states(self):
return [n.get_state_from_iteration(self._age.value) for n in self._neighbours]
def set_new_state_and_activate(self, new_state: CellState):
changed = self._state.set_current_state(new_state, self._age.value + 1)
if changed:
self._set_active()
def _set_active(self):
self.set_active_for_next_iteration(self._age.value)
for n in self._neighbours:
n.set_active_for_next_iteration(self._age.value)
def set_active_for_next_iteration(self, iteration):
self._active.value = max(self._active.value, iteration + 1)
def increase_age(self):
with self._age.get_lock():
self._age += 1

59
src/cellular_automaton/ca_cell_state.py
View File

@ -1,30 +1,65 @@
from multiprocessing import Array, Value
class CellState:
"""
This is the base class for all cell states.
When using the cellular automaton display, inherit this class and implement get_state_draw_color.
"""
def __init__(self, initial_state, state_save_slot_count=2):
def __init__(self, initial_state=(0., ), state_save_slot_count=2, draw_first_state=True):
self._state_save_slot_count = state_save_slot_count
self._state_slots = [initial_state] * state_save_slot_count
self._state_slots = [Array('d', initial_state)] * state_save_slot_count
if draw_first_state:
self._dirty = Value('i', 1)
else:
self._dirty = Value('i', 0)
def set_status_of_iteration(self, new_status, iteration):
""" Will set the new status for the iteration modulo number of saved states.
:param new_status: The new status to set.
def is_set_for_redraw(self):
return self._dirty != 1
def get_state_changes(self):
return self._dirty
def set_for_redraw(self):
self._dirty = 1
def was_redrawn(self):
self._dirty = 0
def set_current_state(self, new_state, current_iteration_index):
return self.set_state_of_iteration(new_state, current_iteration_index)
def get_state_of_last_iteration(self, current_iteration_index):
return self.get_state_of_iteration(current_iteration_index - 1)
def set_state_of_iteration(self, new_state, iteration):
""" Will set the new state for the iteration modulo number of saved states.
:param new_state: The new state to set.
:param iteration: Uses the iteration index, to differ between concurrent states.
:return True if status has changed.
:return True if state has changed.
"""
slot_count = self._state_save_slot_count
states = self._state_slots
states[iteration % slot_count] = new_status
current_state = states[iteration % slot_count]
return states[(iteration - 1) % slot_count] \
!= states[iteration % slot_count]
changed = False
for i in range(len(current_state)):
try:
if current_state[i] != new_state[i]:
changed = True
def get_status_of_iteration(self, iteration):
""" Will return the status for the iteration modulo number of saved states.
current_state[i] = new_state[i]
except IndexError:
raise IndexError("New State length or type is invalid!")
self._dirty |= changed
return changed
def get_state_of_iteration(self, iteration):
""" Will return the state for the iteration modulo number of saved states.
:param iteration: Uses the iteration index, to differ between concurrent states.
:return The status for this iteration.
:return The state for this iteration.
"""
return self._state_slots[iteration % self._state_save_slot_count]

8
src/cellular_automaton/ca_display.py
View File

@ -59,25 +59,27 @@ class PyGameFor2D:
while running:
time_ca_start = time.time()
cellular_automaton_processor.evolve_x_times(self._cellular_automaton, ca_iterations_per_draw)
time_ca_end = time.time()
self.ca_display._redraw_cellular_automaton()
time_ds_end = time.time()
self._print_process_duration(time_ca_end, time_ca_start, time_ds_end)
time.sleep(0.5)
for event in pygame.event.get():
if event.type == pygame.QUIT:
cellular_automaton_processor.stop()
running = False
def _cell_redraw_rectangles(cells, evolution_index, display_info):
for cell in cells:
if cell.is_set_for_redraw:
if cell.state.is_set_for_redraw():
cell_color = cell.state.get_state_draw_color(evolution_index)
cell_pos = _calculate_cell_position(display_info.cell_size, cell)
surface_pos = list(map(operator.add, cell_pos, display_info.grid_pos))
yield display_info.screen.fill(cell_color, (surface_pos, display_info.cell_size))
cell.is_set_for_redraw = False
cell.state.was_redrawn()
def _calculate_cell_position(cell_size, cell):

9
src/cellular_automaton/ca_grid.py
View File

@ -3,10 +3,11 @@ from cellular_automaton.ca_neighborhood import Neighborhood
class Grid:
def __init__(self, dimension: list, neighborhood: Neighborhood):
def __init__(self, dimension: list, neighborhood: Neighborhood, state_class):
self._dimension = dimension
self._cells = {}
self._active_cells = {}
self._state_class = state_class
self._init_cells(neighborhood)
@ -58,7 +59,7 @@ class Grid:
self._recursive_step_down_dimensions(coordinate, dimension_index, self._create_cells)
except IndexError:
coordinate_string = _join_coordinate(coordinate)
self._cells[coordinate_string] = Cell(coordinate_string, coordinate)
self._cells[coordinate_string] = Cell(coordinate_string, self._state_class, coordinate)
def _recursive_step_down_dimensions(self, coordinate, dimension_index, recursion_method):
""" For the range of the current dimension, recalls the recursion method.
@ -72,9 +73,9 @@ class Grid:
def _set_cell_neighbours(self, neighborhood):
for cell in self._cells.values():
neighbours_coordinates = neighborhood.calculate_cell_neighbor_coordinates(cell.coordinate,
neighbours_coordinates = neighborhood.calculate_cell_neighbor_coordinates(cell.get_coordinate(),
self._dimension)
cell.neighbours = list(map(self._get_cell_by_coordinate, neighbours_coordinates))
cell.set_neighbours(list(map(self._get_cell_by_coordinate, neighbours_coordinates)))
def _get_cell_by_coordinate(self, coordinate):
return self._cells[_join_coordinate(coordinate)]

8
src/cellular_automaton/ca_rule.py
View File

@ -1,4 +1,3 @@
from cellular_automaton.ca_cell import Cell
from abc import abstractmethod
@ -6,11 +5,12 @@ class Rule:
def __init__(self):
pass
@staticmethod
@abstractmethod
def evolve_cell(self, cell: Cell, iteration_index: int):
def evolve_cell(last_cell_state, last_neighbour_states):
""" Calculates and sets new state of 'cell'.
:param cell: The cell to calculate new state for.
:param iteration_index: The current iteration index, to choose the correct state.
:param last_cell_state: The cells current state to calculate new state for.
:param last_neighbour_states: The cells neighbours current states.
:return: True if state changed, False if not.
A cells evolution will only be called if it or at least one of its neighbours has changed last iteration cycle.
"""

177
src/cellular_automaton/cellular_automaton.py
View File

@ -1,9 +1,7 @@
import multiprocessing
from cellular_automaton.ca_grid import Grid
from cellular_automaton.ca_rule import Rule
from cellular_automaton.ca_cell_state import CellState
from multiprocessing import Process, Pipe, Array, Value
import multiprocessing
class CellularAutomaton:
@ -13,160 +11,29 @@ class CellularAutomaton:
self.evolution_iteration_index = 0
class _EvolutionProcess:
def __init__(self, process: Process, pipe: Pipe):
self.process = process
self.pipe = pipe
self.cell = None
class CellularAutomatonProcessor:
def __init__(self, process_count: int = 1):
self._processes = list(_create_processes(process_count))
def __init__(self, cellular_automaton, process_count: int = 1):
self.active = multiprocessing.Value('i', 1)
cells = list(cellular_automaton.grid.get_cells().values())
chunk_size = int(len(cells) / process_count)
self._processes = [multiprocessing.Process(target=_process_routine,
name=str(i),
args=(cells[i*chunk_size:i*chunk_size + chunk_size],
cellular_automaton.evolution_rule,
self.active))
for i in range(process_count)]
for p in self._processes:
p.start()
self.__cellular_automaton = None
def evolve_x_times(self, cellular_automaton: CellularAutomaton, evolution_steps: int):
""" Evolve all cells for x time steps.
:param cellular_automaton: The cellular automaton to evolve.
:param evolution_steps: The count of evolutions done.
:return: True if all cells are inactive
"""
for evo in range(evolution_steps):
finished = self.evolve(cellular_automaton)
if finished:
return True
return False
def evolve(self, cellular_automaton: CellularAutomaton):
""" Evolves all active cells for one time step.
:param cellular_automaton: The cellular automaton to evolve.
:return: True if all cells are inactive.
"""
self.__cellular_automaton = cellular_automaton
if self._is_evolution_finished():
print("finished")
return True
else:
cellular_automaton.evolution_iteration_index += 1
self._evolve_all_active_cells()
return False
def _is_evolution_finished(self):
return len(self.__cellular_automaton.grid.get_active_cell_names()) == 0
def _evolve_all_active_cells(self):
active_cells = self.__cellular_automaton.grid.get_active_cells()
self.__cellular_automaton.grid.clear_active_cells()
self._evolve_cells(active_cells.values())
print(len(self.__cellular_automaton.grid.get_active_cells()))
def _evolve_cells(self, cells):
cellular_automaton = self.__cellular_automaton
processes = self._processes
process_count = len(processes)
for i, cell in enumerate(cells):
evolution_process = processes[i % process_count]
if evolution_process.cell:
response = evolution_process.pipe.recv()
evolved_cell = evolution_process.cell
evolved_cell.state = response[0]
evolved_cell.is_set_for_redraw |= response[1]
if evolved_cell.is_set_for_redraw:
cellular_automaton.grid.set_cells_active([evolved_cell] + evolved_cell.neighbours)
evolution_process.cell = None
cell_info = self.read_cell_info(cell, cellular_automaton.evolution_iteration_index)
evolution_process.pipe.send(cell_info)
evolution_process.cell = cell
for evolution_process in processes:
response = evolution_process.pipe.recv()
cell = evolution_process.cell
cell.state = response[0]
cell.is_set_for_redraw |= response[1]
if cell.is_set_for_redraw:
cellular_automaton.grid.set_cells_active([cell] + cell.neighbours)
evolution_process.cell = None
# if cellular_automaton.evolution_rule.evolve_cell(cell.state, cellular_automaton.evolution_iteration_index):
# cellular_automaton.grid.set_cells_active([cell] + cell.neighbours)
@staticmethod
def read_cell_info(cell, iteration):
coordinate = cell.coordinate
return [coordinate, cell.state, [n.state for n in cell.neighbours], iteration]
# if cell.state is not None:
# cell_state = cell.state.get_status_of_iteration(iteration - 1)
# else:
# cell_state = None
#
# neighbor_states = []
# for neighbor in cell.neighbours:
# if neighbor.state is not None:
# neighbor_state = neighbor.state.get_status_of_iteration(iteration - 1)
# else:
# neighbor_state = None
# neighbor_states.append(neighbor_state)
#
# return [coordinate, cell_state, neighbor_states]
def stop(self):
self.active.value = 0
for p in self._processes:
p.join()
def _create_processes(count):
for i in range(count):
parent_pipe_connection, child_pipe_connection = Pipe()
p = Process(target=process_routine, args=(child_pipe_connection, ))
p.start()
yield _EvolutionProcess(p, parent_pipe_connection)
def _process_routine(cells, rule, active):
while active.value == 1:
for cell in cells:
cell.evolve_if_ready(rule)
def process_routine(pipe_conn: Pipe):
while True:
info = pipe_conn.recv()
pipe_conn.send(evolve_cell(*info))
def evolve_cell(coordinate, cell_state, neighbor_states, index):
if cell_state is None:
return _init_state()
else:
new_state = _evolve_state(cell_state, neighbor_states, index)
if new_state is None:
print(",".join([str(x) for x in neighbor_states]))
return [cell_state, False]
else:
changed = cell_state.set_status_of_iteration(new_state, index)
return [cell_state, changed]
def _evolve_state(cell_state, neighbor_states, index):
try:
left_neighbour_state = neighbor_states[0].get_status_of_iteration(index - 1)
return left_neighbour_state
except (IndexError, AttributeError):
return None
import random
def _init_state():
rand = random.randrange(0, 101, 1)
if rand <= 99:
return [MyStatus(0), False]
else:
return [MyStatus(1), True]
class MyStatus(CellState):
def __init__(self, initial_state):
super().__init__(initial_state)
def get_state_draw_color(self, iteration):
red = 0
if self._state_slots[iteration % 2][0]:
red = 255
return [red, 0, 0]
def __str__(self):
return super().__str__()