release 0.2.0

.github/workflows/docker-image.yml

@@ -15,6 +15,6 @@ jobs:
         username: ${{ secrets.DOCKERHUB_USERNAME }}
         password: ${{ secrets.DOCKERHUB_PASSWORD }}
     - name: Build image
-      run: docker build . --file Dockerfile --tag hendrikreiter/distributed_event_factory:0.2.0-SNAPSHOT
+      run: docker build . --file Dockerfile --tag hendrikreiter/distributed_event_factory:0.2.0
     - name: Push image
-      run: docker push hendrikreiter/distributed_event_factory:0.2.0-SNAPSHOT
+      run: docker push hendrikreiter/distributed_event_factory:0.2.0

config/simulation/stream.yaml

@@ -4,7 +4,5 @@ spec:
   type: stream
   caseId: increasing
   load:
-    loadBehavior: sinus
-    mean: 100
-    amplitude: 80
-    phase: 10
+    loadBehavior: constant
+    load: 10

distributed_event_factory/event_factory.py

@@ -8,6 +8,7 @@
 from distributed_event_factory.parser.simulation.case.case_id_parser import CaseIdParser
 from distributed_event_factory.parser.simulation.load.load_parser import LoadParser
 from distributed_event_factory.parser.sink.sink_parser import SinkParser
+from distributed_event_factory.provider.sink.sink_provider import Sink
 
 
 class EventFactory:
@@ -20,21 +21,27 @@ def __init__(self):
 
     def add_load_parser(self, key: str, parser: LoadParser):
         self.parser.load_parser.add_dependency(key, parser)
+        return self
 
     def add_case_id_parser(self, key: str, parser: CaseIdParser):
         self.parser.case_id_parser.add_dependency(key, parser)
+        return self
 
     def add_transition_parser(self, key: str, parser: TransitionParser):
         self.parser.transition_parser.add_dependency(key, parser)
+        return self
 
     def add_activity_parser(self, key: str, parser: ActivityParser):
         self.parser.activity_parser.add_dependency(key, parser)
+        return self
 
     def add_sink_parser(self, key: str, parser: SinkParser):
         self.parser.sink_parser.add_dependency(key, parser)
+        return self
 
     def add_selection_parser(self, key: str, parser: SinkParser):
         self.parser.probability_selection_parser.add_dependency(key, parser)
+        return self
 
     def get_datasource(self, datasource_key):
         return self.datasources[datasource_key]
@@ -47,20 +54,32 @@ def add_directory(self, directory):
             self.add_file(directory + "/" + filename)
         return self
 
+    def add_sink(self, name, sink: Sink):
+        self.sinks[name] = sink
+        return self
+
+    def add_datasource(self, name, datasource):
+        self.datasources[name] = datasource
+        return self
+
+    def add_simulation(self, name, simulation):
+        self.simulations[name] = simulation
+        return self
+
     def add_file(self, filename):
         with open(filename) as file:
             configuration = yaml.safe_load(file)
             kind = configuration['kind']
             name = configuration['name']
             parsed_object = self.parser.kind_parser.parse(configuration)
             if kind == "simulation":
-                self.simulations[name] = parsed_object
+                self.add_simulation(name, parsed_object)
             elif kind == "datasource":
-                self.datasources[name] = parsed_object
+                self.add_datasource(name, parsed_object)
             elif kind == "sink":
-                self.sinks[name] = parsed_object
+                self.add_sink(name, parsed_object)
         return self
 
-    def run(self):
+    def run(self, hook=lambda: None):
         for simulation in self.simulations:
-            self.simulations[simulation].run_simulation(self.datasources, self.sinks)
+            self.simulations[simulation].run_simulation(self.datasources, self.sinks, hook)

distributed_event_factory/parser/simulation/variant/stream_simulation_parser.py

@@ -1,5 +1,5 @@
 from distributed_event_factory.parser.parser import Parser
-from distributed_event_factory.simulation.stream_simulation import StreamSimulation
+from distributed_event_factory.simulation.stream import StreamSimulation
 
 
 class StreamSimulationParser(Parser):

distributed_event_factory/provider/sink/console/console_sink.py

@@ -6,8 +6,8 @@
 class PrintConsole(Sink):
 
     def __init__(self, id, data_source_ref):
+        super.__init__(data_source_ref)
         self.id = id
-        self.data_source_ref = data_source_ref
 
     def send(self, event: Event) -> None:
         print("Sensor " + event.node + ": " + str(event))

distributed_event_factory/simulation/abstract_simulation.py

@@ -10,11 +10,12 @@ def send_event(self, event):
             for sink in self.sinks[event.node]:
                 sink.send(event)
         else:
-            print("Skip event. No sink configured")
+            print(f"Skip event. No sink configured. Event: {event}")
 
     def setup_sinks(self, sinks):
         for sink in sinks:
-            for data_source in sinks[sink].data_source_ref:
-                if data_source not in self.sinks:
-                    self.sinks[data_source] = []
-                self.sinks[data_source].append(sinks[sink])
+            if sinks[sink].data_source_ref:
+                for data_source in sinks[sink].data_source_ref:
+                    if data_source not in self.sinks:
+                        self.sinks[data_source] = []
+                    self.sinks[data_source].append(sinks[sink])

distributed_event_factory/simulation/countbased.py

@@ -11,12 +11,15 @@ def __init__(self, simulation_steps: int, case_id_provider: CaseIdProvider):
         self.simulation_steps = simulation_steps
         self.sinks = dict()
 
-    def run_simulation(self, datasources, sinks):
+    def run_simulation(self, datasources, sinks, hook=lambda: None):
         super().setup_sinks(sinks)
         process_simulator = ProcessSimulator(
             case_id_provider=self.case_id_provider,
             data_sources=datasources
         )
 
         for i in range(self.simulation_steps):
-            self.send_event(process_simulator.simulate())
+            event = process_simulator.simulate()
+            print(event)
+            self.send_event(event)
+            hook()

distributed_event_factory/simulation/loadtest.py

@@ -33,7 +33,7 @@ def start_simulation(self, sinks):
             for s in sinks[sink]:
                 s.start()
 
-    def run_simulation(self, data_sources: Dict[str, DataSource], sinks: Dict[str, LoadTestHttpSink]):
+    def run_simulation(self, data_sources: Dict[str, DataSource], sinks: Dict[str, LoadTestHttpSink], hook = lambda: None):
         self.setup_sinks(sinks)
         process_simulator = ProcessSimulator(
             case_id_provider=self.case_id_provider,
@@ -46,5 +46,6 @@ def run_simulation(self, data_sources: Dict[str, DataSource], sinks: Dict[str, L
             self.start_timeframe(self.sinks)
             for _ in range(int(self.load_provider.get_load_value())):
                 self.send_event(process_simulator.simulate())
+                hook()
             self.end_timeframe(self.sinks)
             iteration = iteration + 1

distributed_event_factory/simulation/stream_simulation.py → distributed_event_factory/simulation/stream.py

@@ -17,7 +17,7 @@ def __init__(
         self.load_provider = load_provider
         self.sinks = dict()
 
-    def run_simulation(self, datasources, sinks):
+    def run_simulation(self, datasources, sinks, hook):
         self.setup_sinks(sinks)
         process_simulator = ProcessSimulator(
             case_id_provider=self.case_id_provider,
@@ -29,5 +29,6 @@ def run_simulation(self, datasources, sinks):
                     lambda: self.send_event(process_simulator.simulate()),
                     period=1/self.load_provider.get_load_value()
                 )
+                hook()
                 time.sleep(1)
                 scheduler.cancel()

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "distributed-event-factory"
-version = "0.2.0.dev6"
+version = "0.2.0"
 authors = [
   { name="Hendrik Reiter", email="[email protected]" },
 ]