Feature/http sink

config/datasource/assemblyline/d1-goods-delivery.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "GoodsDelivery"
 spec:
   name: "GoodsDelivery"
-  group: "factory"
+  group: "GoodsDelivery"
   selection: driftingProbability
   startDistribution: [0.1, 0.7, 0.2]
   endDistribution: [0.1, 0.2, 0.7]

config/datasource/assemblyline/d2-material-preparation.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "MaterialPreparation"
 spec:
   name: "MaterialPreparation"
-  group: "factory"
+  group: "MaterialPreparation"
   selection: genericProbability
   distribution: [ 0.25, 0.7, 0.05 ]
   eventData:

config/datasource/assemblyline/d3-assembly-line-setup.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "AssemblyLineSetup"
 spec:
   name: "AssemblyLineSetup"
-  group: "factory"
+  group: "AssemblyLineSetup"
   selection: genericProbability
   distribution: [ 0.3, 0.4, 0.15, 0.05, 0.1 ]
   eventData:

config/datasource/assemblyline/d4-assembling.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "Assembling"
 spec:
   name: "Assembling"
-  group: "factory"
+  group: "Assembling"
   selection: genericProbability
   distribution: [ 0.8, 0.1, 0.08, 0.02 ]
   eventData:

config/datasource/assemblyline/d5-quality-control.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "QualityControl"
 spec:
   name: "QualityControl"
-  group: "factory"
+  group: "QualityControl"
   selection: genericProbability
   distribution: [ 0.3, 0.1, 0.6 ]
   eventData:

config/datasource/assemblyline/d6-packaging.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "Packaging"
 spec:
   name: "Packaging"
-  group: "factory"
+  group: "Packaging"
   selection: genericProbability
   distribution: [ 1.0 ]
   eventData:

config/datasource/assemblyline/d7-shipping.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "Shipping"
 spec:
   name: "Shipping"
-  group: "factory"
+  group: "Shipping"
   selection: genericProbability
   distribution: [ 0.8, 0.2 ]
   eventData:

config/datasource/concept-drift/cd-a.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "A"
 spec:
   name: "A"
-  group: "concept-drift"
+  group: "A"
   selection: driftingProbability
   startDistribution: [ 0.0, 1.0 ]
   endDistribution: [ 1.0, 0.0 ]

config/datasource/concept-drift/cd-b.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "B"
 spec:
   name: "B"
-  group: "concept-drift"
+  group: "B"
   selection: genericProbability
   distribution: [1.0]
   eventData:

config/datasource/concept-drift/cd-c.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "C"
 spec:
   name: "C"
-  group: "concept-drift"
+  group: "C"
   selection: genericProbability
   distribution: [1.0]
   eventData:

config/datasource/concept-drift/cd-d.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "D"
 spec:
   name: "D"
-  group: "concept-drift"
+  group: "D"
   selection: genericProbability
   distribution: [1.0]
   eventData:

config/datasource/concept-drift/cd-e.yaml

@@ -2,7 +2,7 @@ kind: datasource
 name: "E"
 spec:
   name: "E"
-  group: "concept-drift"
+  group: "E"
   selection: genericProbability
   distribution: [1.0]
   eventData:

config/datasource/image/image-a.yaml

@@ -0,0 +1,13 @@
+kind: datasource
+name: "A"
+spec:
+  name: "A"
+  group: "A"
+  selection: genericProbability
+  distribution: [1.0]
+  eventData:
+    - activity:
+        type: image
+        activity: "n02085620-Chihuahua/n02085620_7"
+      transition: "<end>"
+      duration: 1

config/datasource/image/start.yaml

@@ -0,0 +1,11 @@
+kind: datasource
+name: "<start>"
+spec:
+  name: "<start>"
+  group: "A"
+  selection: genericProbability
+  distribution: [1.0]
+  eventData:
+    - activity: ""
+      transition: "A"
+      duration: 0

config/simulation/stream.yaml

@@ -6,4 +6,4 @@ spec:
   maxConcurrentCases: 5
   load:
     loadBehavior: constant
-    load: 10
+    load: 1

config/sink/group/http-sink-assembling.yaml

@@ -0,0 +1,8 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8090/event"
+  dataSourceRefs:
+    - "Assembling"

config/sink/group/http-sink-assemblyline.yaml

@@ -0,0 +1,8 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8091/event"
+  dataSourceRefs:
+    - "AssemblyLineSetup"

config/sink/group/http-sink-goodsdelivery.yaml

@@ -0,0 +1,8 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8092/event"
+  dataSourceRefs:
+    - "GoodsDelivery"

config/sink/group/http-sink-materialpreparation.yaml

@@ -0,0 +1,8 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8093/event"
+  dataSourceRefs:
+    - "MaterialPreparation"

config/sink/group/http-sink-packaging.yaml

@@ -0,0 +1,8 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8094/event"
+  dataSourceRefs:
+    - "Packaging"

config/sink/group/http-sink-qualitycontrol.yaml

@@ -0,0 +1,8 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8095/event"
+  dataSourceRefs:
+    - "QualityControl"

config/sink/group/http-sink-shipping.yaml

@@ -0,0 +1,8 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8096/event"
+  dataSourceRefs:
+    - "Shipping"

config/sink/http-sink.yaml

@@ -0,0 +1,14 @@
+kind: sink
+name: http-sink
+spec:
+  type: http
+  id: "Sensor"
+  url: "http://localhost:8082/event"
+  dataSourceRefs:
+    - "GoodsDelivery"
+    - "MaterialPreparation"
+    - "AssemblyLineSetup"
+    - "Assembling"
+    - "QualityControl"
+    - "Packaging"
+    - "Shipping"

config/sink/traffic-fine.yaml

@@ -0,0 +1,17 @@
+kind: sink
+name: console-sink
+spec:
+  type: console
+  id: "Sensor"
+  dataSourceRefs:
+    - "Create Fine"
+    - "Send Fine"
+    - "Appeal to Judge"
+    - "Insert Fine Notification"
+    - "Send Appeal to Prefecture"
+    - "Insert Date Appeal to Prefecture"
+    - "Receive Result Appeal from Prefecture"
+    - "Payment"
+    - "Notify Result Appeal to Offender"
+    - "Send for Credit Collection"
+    - "Add penalty"

distributed_event_factory/core/abstract_datasource.py

@@ -8,10 +8,6 @@
 
 class DataSource(ABC):
 
-    @abstractmethod
-    def emit_event(self, case, activity_name, timestamp) -> None:
-        pass
-
     @abstractmethod
     def get_event_data(self) -> EventData:
         pass

distributed_event_factory/core/datasource.py

@@ -27,16 +27,5 @@ def get_event_provider(self):
     def get_event_data(self):
         return self.event_provider.get_event_data()
 
-    def emit_event(self, case, activity_name, timestamp) -> Event:
-        event = Event(
-            timestamp=timestamp.strftime("%Y-%m-%d %H:%M:%S"),
-            sensor_value=activity_name,
-            case_id=case,
-            sensor_name=self.sensor_id.get_name(),
-            group_id=self.group_id
-        )
-        self.event_log.append(event)
-        return event
-
     def get_event_log(self) -> List[Event]:
         return self.event_log

distributed_event_factory/core/datasource_id.py

@@ -14,3 +14,4 @@ def get_name(self):
 
 START_SENSOR_ID = DataSourceId("<start>")
 END_DATA_SOURCE_ID = DataSourceId("<end>")
+END_DATA_SOURCE = "<end>"

distributed_event_factory/event_factory.py

@@ -9,13 +9,15 @@
 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
+from distributed_event_factory.simulation.process_simulation import DefProcessSimulator
 
 
 class EventFactory:
     def __init__(self):
         self.sinks = dict()
         self.simulations = dict()
         self.datasources = dict()
+        self.process_simulator = None
         self.datasources["<end>"] = EndDataSource()
         self.parser = ParserRegistry()
 
@@ -43,6 +45,10 @@ def add_selection_parser(self, key: str, parser: SinkParser):
         self.parser.probability_selection_parser.add_dependency(key, parser)
         return self
 
+    def add_process_simulator(self, process_simulator):
+        self.process_simulator = process_simulator
+        return self
+
     def get_datasource(self, datasource_key):
         return self.datasources[datasource_key]
 
@@ -79,8 +85,11 @@ def add_file(self, filename):
                 self.add_datasource(name, parsed_object)
             elif kind == "sink":
                 self.add_sink(name, parsed_object)
+            elif kind == "processSimulator":
+                self.add_process_simulator(parsed_object)
+
         return self
 
     def run(self, hook=lambda: None):
         for simulation in self.simulations:
-            self.simulations[simulation].run_simulation(self.datasources, self.sinks, hook)
+            self.simulations[simulation].run_simulation(self.process_simulator, self.datasources, self.sinks, hook)

distributed_event_factory/event_factory_main.py

@@ -3,6 +3,13 @@
 
 from dotenv import load_dotenv
 from distributed_event_factory.event_factory import EventFactory
+from distributed_event_factory.provider.data.case_provider import CaseIdProvider
+from distributed_event_factory.provider.data.constant_count_provider import ConstantCountProvider
+from distributed_event_factory.provider.data.count_provider import CountProvider
+from distributed_event_factory.provider.data.increasing_case import IncreasingCaseIdProvider
+from distributed_event_factory.simulation.process_simulation import DefProcessSimulator
+from distributed_event_factory.simulation.xes_process_simulator import XesProcessSimulator
+from drift_conformance_checking_sink import DriftConformanceCheckingSink
 
 if __name__ == '__main__':
     if "local" in sys.argv:
@@ -22,5 +29,23 @@
     (event_factory
      .add_directory(f"{content_root}/config/datasource/{datasource}")
      .add_file(f"{content_root}/config/simulation/{simulation}")
-     .add_file(f"{content_root}/config/sink/{sink}")
-     ).run()
+     #.add_file(f"{content_root}/config/sink/{sink}")
+     .add_process_simulator(
+        DefProcessSimulator(
+              case_id_provider=IncreasingCaseIdProvider(),
+              data_sources=dict(),
+              max_concurrent_cases=ConstantCountProvider(1)
+            )
+    ).add_sink("Hi", DriftConformanceCheckingSink(["A","B","C","D","E"]))
+     .run()
+     #   XesProcessSimulator(
+     #       "../config/Road_Traffic_Fine_Management_Process.xes"
+     #   )
+        #DefProcessSimulator(
+        #   case_id_provider=IncreasingCaseIdProvider(),
+        #   data_sources=dict(),
+        #   max_concurrent_cases=ConstantCountProvider(1)
+        #)
+     )
+     #.add_sink("hello_algorithm", SayHelloAlgorithm(["GoodsDelivery"]))
+     #).run()

distributed_event_factory/parser/datasource/event/activity/activity_parser.py

@@ -12,6 +12,6 @@ def add_dependency(self, key: str, dependency):
         return self
 
     def parse(self, config):
-        if isinstance(config, str):
-            return ConstantActivityProvider(config)
-        return self.dependencies[config["type"]].parse(config)
+        if "type" in config:
+            return self.dependencies[config["type"]].parse(config)
+        return ConstantActivityProvider(config)

distributed_event_factory/parser/datasource/event/activity/constant_activitiy_parser.py

@@ -0,0 +1,11 @@
+from distributed_event_factory.parser.parser import Parser
+from distributed_event_factory.provider.activity.activity_provider import ConstantActivityProvider
+
+
+class ConstantActivityParser(Parser):
+
+    def add_dependency(self, key: str, dependency):
+        pass
+
+    def parse(self, config):
+        return ConstantActivityProvider(config)

distributed_event_factory/parser/datasource/event/activity/image_activity_parser.py

@@ -0,0 +1,11 @@
+from distributed_event_factory.parser.parser import Parser
+from distributed_event_factory.provider.activity.image_activity_provider import ImageActivityProvider
+
+
+class ImageActivityParser(Parser):
+
+    def add_dependency(self, key: str, dependency):
+        pass
+
+    def parse(self, config):
+        return ImageActivityProvider(activity=config["activity"])