Terminal enhancements

sandbox/life_game/run.py

-import sys
 import collections
-from sandbox.life_game.simulation import CellBornBehaviour, CellDieBehaviour, Cell, Empty
+from sandbox.life_game.simulation import Cell
+from sandbox.life_game.simulation import Empty
+from sandbox.life_game.simulation import CellDieEvent
+from sandbox.life_game.simulation import CellBornEvent
 
 from sandbox.life_game.utils import get_subjects_from_str_representation
 from synergine2.core import Core
 from synergine2.cycle import CycleManager
 from synergine2.simulation import Simulation
+from synergine2.simulation import Event
 from synergine2.terminals import Terminal
 from synergine2.terminals import TerminalPackage
 from synergine2.terminals import TerminalManager
 
 
 class SimplePrintTerminal(Terminal):
+    subscribed_events = [
+        CellDieEvent,
+        CellBornEvent,
+    ]
+
     def __init__(self):
         super().__init__()
-        self.subjects = None
+        self._cycle_born_count = 0
+        self._cycle_die_count = 0
+        self.register_event_handler(CellDieEvent, self.record_die)
+        self.register_event_handler(CellBornEvent, self.record_born)
 
-    def receive(self, value):
-        self.update_with_package(value)
-        self.print_str_representation()
+    def record_die(self, event: Event):
+        self._cycle_die_count += 1
 
-    def update_with_package(self, package: TerminalPackage):
-        self.subjects = package.subjects if package.subjects else self.subjects
-        for subject_id, actions in package.actions.items():
-            for action, value in actions.items():
-                if action == CellBornBehaviour:
-                    # Remove Empty subject
-                    self.subjects = [s for s in self.subjects[:] if s.id != subject_id]
-                    # Add born subject
-                    self.subjects.append(value)
-                if action == CellDieBehaviour:
-                    # Remove Cell subject
-                    self.subjects = [s for s in self.subjects[:] if s.id != subject_id]
-                    # Add Empty subject
-                    self.subjects.append(value)
+    def record_born(self, event: Event):
+        self._cycle_born_count += 1
+
+    def receive(self, package: TerminalPackage):
+        self._cycle_born_count = 0
+        self._cycle_die_count = 0
+        super().receive(package)
+        self.print_str_representation()
 
     def print_str_representation(self):
         items_positions = collections.defaultdict(list)
-        for subject in self.subjects:
+        for subject in self.subjects.values():
             if type(subject) == Cell:
                 items_positions['1'].append(subject.position)
             if type(subject) == Empty:
         print(get_str_representation_from_positions(
             items_positions,
             separator=' ',
-            #force_items_as=(('0', ' '),),
+            force_items_as=(('0', ' '),),
         ))
+
+        # Display current cycle events
+        print('This cycle: {0} born, {1} dead'.format(
+            self._cycle_born_count,
+            self._cycle_die_count,
+        ))
+
         print()
 
 

sandbox/life_game/simulation.py

 from synergine2.simulation import Subject
+from synergine2.simulation import Event
 from synergine2.simulation import Behaviour
 from synergine2.xyz import ProximityMechanism
 from synergine2.xyz import XYZSubjectMixin
 COLLECTION_CELL = 'COLLECTION_CELL'  # Collections of Cell type
 
 
+class CellDieEvent(Event):
+    def __init__(self, subject_id, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.subject_id = subject_id
+
+
+class CellBornEvent(Event):
+    def __init__(self, subject_id, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.subject_id = subject_id
+
+
 class CellProximityMechanism(ProximityMechanism):
     distance = 1.41  # distance when on angle
     feel_collections = [COLLECTION_CELL]
         )
         self.simulation.subjects.remove(self.subject)
         self.simulation.subjects.append(new_empty)
-        return new_empty
+        return [CellDieEvent(self.subject.id)]
 
 
 class CellBornBehaviour(Behaviour):
         )
         self.simulation.subjects.remove(self.subject)
         self.simulation.subjects.append(new_cell)
-        return new_cell
+        return [CellBornEvent(new_cell.id)]
 
 
 class Cell(XYZSubjectMixin, Subject):

synergine2/core.py

 
             while True:
                 # TODO: receive from terminals
-                actions = self.cycle_manager.next()
-                cycle_package = TerminalPackage(actions=actions)
+                events = self.cycle_manager.next()
+                cycle_package = TerminalPackage(
+                    events=events,
+                    add_subjects=self.simulation.subjects.adds,
+                    remove_subjects=self.simulation.subjects.removes,
+                )
                 self.terminal_manager.send(cycle_package)
+
+                # Reinitialize these list for next cycle
+                self.simulation.subjects.adds = []
+                self.simulation.subjects.removes = []
+
                 import time
                 time.sleep(1)  # TODO: tick control
         except KeyboardInterrupt:

synergine2/cycle.py

 import multiprocessing
-import collections
 
 from synergine2.processing import ProcessManager
-from synergine2.simulation import Subject, Behaviour, Mechanism
+from synergine2.simulation import Subject
+from synergine2.simulation import Behaviour
+from synergine2.simulation import Mechanism
+from synergine2.simulation import Event
+from synergine2.simulation import Subjects
 from synergine2.utils import ChunkManager
 
 
 class CycleManager(object):
     def __init__(
             self,
-            subjects: list,
+            subjects: Subjects,
             process_manager: ProcessManager=None,
     ):
         if process_manager is None:
         self.subjects = subjects
         self.process_manager = process_manager
         self.current_cycle = 0
+        self.first_cycle = True
+
+    def next(self) -> [Event]:
+        if self.first_cycle:
+            # To dispatch subjects add/removes, enable track on them
+            self.subjects.track_changes = True
+            self.first_cycle = False
 
-    def next(self):
         results = {}
         results_by_processes = self.process_manager.execute_jobs(self.subjects)
-        actions = collections.defaultdict(dict)
+        events = []
         for process_results in results_by_processes:
             results.update(process_results)
         for subject in self.subjects[:]:  # Duplicate list to prevent conflicts with behaviours subjects manipulations
             for behaviour_class in results[subject.id]:
                 # TODO: Ajouter une etape de selection des actions a faire (genre neuronnal)
-                # TODO: les behaviour_class ont le même uniqueid apres le process ?
-                action_result = subject.behaviours[behaviour_class].action(results[subject.id][behaviour_class])
-                actions[subject.id][behaviour_class] = action_result
+                # (genre se cacher et fuir son pas compatibles)
+                actions_events = subject.behaviours[behaviour_class].action(results[subject.id][behaviour_class])
+                events.extend(actions_events)
 
-        return actions
+        return events
 
     def computing(self, subjects):
         # compute mechanisms (prepare check to not compute slienced or not used mechanisms)

synergine2/simulation.py

 
 
 class Subjects(list):
+    """
+    TODO: Manage other list methods
+    """
     def __init__(self, *args, **kwargs):
         self.simulation = kwargs.pop('simulation')
+        self.removes = []
+        self.adds = []
+        self.track_changes = False
         super().__init__(*args, **kwargs)
 
     def remove(self, value: Subject):
+        # Remove from subjects list
         super().remove(value)
+        # Remove from collections
         for collection_name in value.collections:
             self.simulation.collections[collection_name].remove(value)
+        # Add to removed listing
+        if self.track_changes:
+            self.removes.append(value)
+
+    def append(self, p_object):
+        # Add to subjects list
+        super().append(p_object)
+        # Add to adds list
+        if self.track_changes:
+            self.adds.append(p_object)
 
 
 class Mechanism(object):
         raise NotImplementedError()
 
 
+class Event(object):
+    def __init__(self, *args, **kwargs):
+        pass
+
+
 class Behaviour(object):
     def __init__(
             self,
         """
         raise NotImplementedError()
 
-    def action(self, data) -> object:
+    def action(self, data) -> [Event]:
         """
         Method called in main process
         Return value will be give to terminals

synergine2/terminals.py

+import collections
+from copy import copy
 from multiprocessing import Queue
 
 from multiprocessing import Process
 from queue import Empty
 
 import time
-from synergine2.simulation import Simulation, Subject
+from synergine2.simulation import Subject
+from synergine2.simulation import Event
 
 STOP_SIGNAL = '__STOP_SIGNAL__'
 
     def __init__(
             self,
             subjects: [Subject]=None,
-            actions: ['TODO']=None,
+            add_subjects: [Subject]=None,
+            remove_subjects: [Subject]=None,
+            events: [Event]=None,
+            *args,
+            **kwargs
     ):
         self.subjects = subjects
-        self.actions = actions or {}
+        self.add_subjects = add_subjects or []
+        self.remove_subjects = remove_subjects or []
+        self.events = events or []
 
 
 class Terminal(object):
-    """Default behaviour is to do nothing.
-    DEFAULT_SLEEP is sleep time between each queue read"""
-    DEFAULT_SLEEP = 1
+    # Default behaviour is to do nothing.
+    # DEFAULT_SLEEP is sleep time between each queue read
+    default_sleep = 1
+    # List of subscribed Event classes. Terminal will not receive events
+    # who are not instance of listed classes
+    subscribed_events = [Event]
 
     def __init__(self):
         self._input_queue = None
         self._output_queue = None
         self._stop_required = False
+        self.subjects = {}
+        self.cycle_events = []
+        self.event_handlers = collections.defaultdict(list)
+
+    def accept_event(self, event: Event) -> bool:
+        for event_class in self.subscribed_events:
+            if isinstance(event, event_class):
+                return True
+        return False
 
     def start(self, input_queue: Queue, output_queue: Queue) -> None:
         self._input_queue = input_queue
         """
         try:
             while self.read():
-                time.sleep(self.DEFAULT_SLEEP)
+                time.sleep(self.default_sleep)
         except KeyboardInterrupt:
             pass
 
             except Empty:
                 return True  # Finished to read Queue
 
-    def receive(self, value):
-        raise NotImplementedError()
+    def receive(self, package: TerminalPackage):
+        self.update_with_package(package)
+
+    def send(self, package: TerminalPackage):
+        self._output_queue.put(package)
+
+    def register_event_handler(self, event_class, func):
+        self.event_handlers[event_class].append(func)
+
+    def update_with_package(self, package: TerminalPackage):
+        if package.subjects:
+            self.subjects = {s.id: s for s in package.subjects}
+
+        for new_subject in package.add_subjects:
+            self.subjects[new_subject.id] = new_subject
 
-    def send(self, value):
-        self._output_queue.put(value)
+        for deleted_subject in package.remove_subjects:
+            del self.subjects[deleted_subject.id]
+
+        self.cycle_events = package.events
+        self.execute_event_handlers(package.events)
+
+    def execute_event_handlers(self, events: [Event]):
+        for event in events:
+            for event_class, handlers in self.event_handlers.items():
+                if isinstance(event, event_class):
+                    for handler in handlers:
+                        handler(event)
 
 
 class TerminalManager(object):
     def __init__(self, terminals: [Terminal]):
-        self._terminals = terminals
-        self._outputs_queues = []
-        self._inputs_queues = []
+        self.terminals = terminals
+        self.outputs_queues = {}
+        self.inputs_queues = {}
 
     def start(self) -> None:
-        for terminal in self._terminals:
+        for terminal in self.terminals:
             output_queue = Queue()
-            self._outputs_queues.append(output_queue)
+            self.outputs_queues[terminal] = output_queue
 
             input_queue = Queue()
-            self._inputs_queues.append(input_queue)
+            self.inputs_queues[terminal] = input_queue
 
             process = Process(target=terminal.start, kwargs=dict(
                 input_queue=output_queue,
             process.start()
 
     def stop(self):
-        for output_queue in self._outputs_queues:
+        for output_queue in self.outputs_queues.values():
             output_queue.put(STOP_SIGNAL)
 
-    def send(self, value):
-        for output_queue in self._outputs_queues:
-            output_queue.put(value)
+    def send(self, package: TerminalPackage):
+        for terminal, output_queue in self.outputs_queues.items():
+            # Terminal maybe don't want all events, so reduce list of event
+            # Thirst make a copy to personalize this package
+            terminal_adapted_package = copy(package)
+            # Duplicate events list to personalize it
+            terminal_adapted_package.events = terminal_adapted_package.events[:]
+
+            for package_event in terminal_adapted_package.events[:]:
+                if not terminal.accept_event(package_event):
+                    terminal_adapted_package.events.remove(package_event)
+
+            output_queue.put(terminal_adapted_package)
 
     def receive(self) -> []:
-        values = []
-        for input_queue in self._inputs_queues:
+        packages = []
+        for input_queue in self.inputs_queues.values():
             try:
                 while True:
-                    values.append(input_queue.get(block=False, timeout=None))
+                    packages.append(input_queue.get(block=False, timeout=None))
             except Empty:
                 pass  # Queue is empty
 
-        return values
+        return packages

tests/test_terminals.py

 import time
 
+from synergine2.simulation import Event
 from synergine2.terminals import Terminal
+from synergine2.terminals import TerminalPackage
 from synergine2.terminals import TerminalManager
 from tests import BaseTest
 
 
+class ValueTerminalPackage(TerminalPackage):
+    def __init__(self, value, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.value = value
+
+
 class MultiplyTerminal(Terminal):
-    def receive(self, value):
-        self.send(value * 2)
-        self.send(value * 4)
+    def receive(self, package: ValueTerminalPackage):
+        self.send(ValueTerminalPackage(value=package.value * 2))
+        self.send(ValueTerminalPackage(value=package.value * 4))
 
 
 class DivideTerminal(Terminal):
-    def receive(self, value):
-        self.send(value / 2)
-        self.send(value / 4)
+    def receive(self, package: ValueTerminalPackage):
+        self.send(ValueTerminalPackage(value=package.value / 2))
+        self.send(ValueTerminalPackage(value=package.value / 4))
+
+
+class AnEvent(Event):
+    pass
+
+
+class AnOtherEvent(Event):
+    pass
+
+
+class SendBackTerminal(Terminal):
+    def receive(self, package: ValueTerminalPackage):
+        self.send(package)
 
 
 class TestTerminals(BaseTest):
             ]
         )
         terminals_manager.start()
-        terminals_manager.send(42)
+        terminals_manager.send(ValueTerminalPackage(value=42))
 
         # We wait max 2s (see time.sleep) to consider
         # process have finished. If not, it will fail
-        values = []
+        packages = []
         for i in range(200):
-            values.extend(terminals_manager.receive())
-            if len(values) == 2:
+            packages.extend(terminals_manager.receive())
+            if len(packages) == 2:
                 break
             time.sleep(0.01)
 
-        assert 2 == len(values)
-        values = [v for v in values]
+        assert 2 == len(packages)
+        values = [p.value for p in packages]
         assert 84 in values
         assert 168 in values
 
             ]
         )
         terminals_manager.start()
-        terminals_manager.send(42)
+        terminals_manager.send(ValueTerminalPackage(value=42))
 
         # We wait max 2s (see time.sleep) to consider
         # process have finished. If not, it will fail
-        values = []
+        packages = []
         for i in range(200):
-            values.extend(terminals_manager.receive())
-            if len(values) == 4:
+            packages.extend(terminals_manager.receive())
+            if len(packages) == 4:
                 break
             time.sleep(0.01)
 
-        assert 4 == len(values)
-        values = [v for v in values]
+        assert 4 == len(packages)
+        values = [p.value for p in packages]
         assert 84 in values
         assert 168 in values
         assert 21 in values
         assert 10.5 in values
 
-        terminals_manager.stop()  # pytest must execute this if have fail
+        terminals_manager.stop()  # TODO pytest must execute this if have fail
+
+    def test_event_listen_everything(self):
+        class ListenEverythingTerminal(SendBackTerminal):
+            pass
+
+        terminals_manager = TerminalManager(
+            terminals=[ListenEverythingTerminal()]
+        )
+        terminals_manager.start()
+        terminals_manager.send(ValueTerminalPackage(value=42))
+        an_event = AnEvent(84)
+        terminals_manager.send(TerminalPackage(events=[an_event]))
+
+        # We wait max 2s (see time.sleep) to consider
+        # process have finished. If not, it will fail
+        packages = []
+        for i in range(200):
+            packages.extend(terminals_manager.receive())
+            if len(packages) == 2:
+                break
+            time.sleep(0.01)
+
+        assert 2 == len(packages)
+        assert 42 == packages[0].value
+        assert AnEvent == type(packages[1].events[0])
+
+        terminals_manager.stop()  # TODO pytest must execute this if have fail
+
+    def test_event_listen_specified(self):
+        class ListenAnEventTerminal(SendBackTerminal):
+            subscribed_events = [AnOtherEvent]
+
+        terminals_manager = TerminalManager(
+            terminals=[ListenAnEventTerminal()]
+        )
+        terminals_manager.start()
+        terminals_manager.send(ValueTerminalPackage(value=42))
+        an_event = AnEvent(84)
+        an_other_event = AnOtherEvent(168)
+        terminals_manager.send(TerminalPackage(events=[an_event, an_other_event]))
+
+        # We wait max 2s (see time.sleep) to consider
+        # process have finished. If not, it will fail
+        packages = []
+        for i in range(200):
+            packages.extend(terminals_manager.receive())
+            if len(packages) == 1:
+                break
+            time.sleep(0.01)
+
+        assert 2 == len(packages)
+        assert AnOtherEvent == type(packages[1].events[0])
+
+        terminals_manager.stop()  # TODO pytest must execute this if have fail