Updated code to have an integration with the orchestration blueprint

.gitignore

@@ -27,5 +27,7 @@ data/*
 config/
 *.env
 
+.sampledata
+
 # Python
 __pycache__/

README.md

@@ -39,6 +39,6 @@ The telematics platform (representing a vehicle OEM) stores all insurance events
 
 The code is organized in three parts:
 
-- [Edge](./src/edge/README.md) contains examples for risk event detectors.
+- [Edge](./edge/README.md) contains examples for risk event detectors and integration with the Eclipse Orchestration Blueprint
 - *Telematics Platform* contains examples to receive vehicle data, store it and process it as a data product.
 - [Insurance Platform](./src/edc/README.md) showcases consumption of data using *Eclipse Dataspaces*

edge/README.md

@@ -22,16 +22,23 @@ SPDX-License-Identifier: Apache-2.0
 
 This code implements risk event detectors that categorize driving style.
 
-![Edge Overview](../../docs/images/edge-overview.png)
+![Edge Overview](/docs/images/edge-overview.svg)
 
-# Risk Events Detector
+# Structure of the directory
+
+The directory has the follopwing structure
+
+- *applications / insurance_event_detector* contains the python code for the main application
+- *digital-twin-providers / vehicle_properties_provider* contains the python code that registers the signals and simulates the vehicle, described as 2b in the diagram
+- *digital-twin-model* contains a DTDL representation of the COVESA signal specification used in the project.
+- *proto-build* contains all Python gRPC interfaces to the different components (service_discovery, invehicle_digital_twin, module)
+
+## Risk Events Detector
 
 Each Risk event detector monitors vehicle signals everytime that they change, at a given refresh rate. The algorithm of each event detector varies - in some cases it requires analyzing previous data, or it can be a simple ramp up or ramp down threshold detection.
 
 Once an event is triggered, the event detector will capture additional signals before and after the event.
 
-## Threshold detectors
-
 The following events react on a flank up or flank down
 
 | Event             | Monitored Signals             |   Additional Captured Signals    |
@@ -44,4 +51,44 @@ The following events react on a flank up or flank down
 | harsh_acceleration | Vehicle_Acceleration_Lateral | Vehicle Acceleration Longitudinal / Lateral, Speed, Accelerator Pedal Position, Steering Wheel Angle, ADAS ABS Error, ADAS ABS Is Engaged, ADAS TCS IsEngaged
 | harsh_cornering | Vehicle_Acceleration_Longitudinal | Vehicle Acceleration Longitudinal / Lateral, Speed, Brake Pressure, Accelerator Pedal Position, Steering Wheel Angle, Steering Wheel Angle Sign, Left Turn Light Indicator, Right Turn Light Indicator, ADAS TCS IsEngaged. Vehicle Speed Wheel Front Left / Front Right / Rear Left / Rear Right
 
+## Integration with Orchestration Blueprint
+
+The Insurance Event Detector uses charriot to discover the digital twin service. It will collect all necessary signals and use the digital twin service to detemrine if they exist. Once the signals are discovered, it will use the managed subscribed (through the digital twin service) to read the metadata and connect.
+
+![Edge Overview](/docs/images/eclipse-orchestration-integration.svg)
+
+To provide simulation, a vehicle provider will register signals (described in a DTDL file) and provide updates using a sample CSV file. The CSV file has the following structure:
+
+``` csv
+source_id, signal, timestamp, value
+"b7ed5744-8715-4b3d-a322-7c7e0c399f69","Vehicle_Speed_Wheel_RearLeft","142.785732","53.715"
+"b7ed5744-8715-4b3d-a322-7c7e0c399f69","Vehicle_Speed_Wheel_FrontRight","142.785732","54.3975"
+"b7ed5744-8715-4b3d-a322-7c7e0c399f69","Vehicle_Speed_Wheel_FrontLeft","142.785732","53.685"
+"b7ed5744-8715-4b3d-a322-7c7e0c399f69","Vehicle_Speed_Wheel_RearRight","142.785732","54.225"
+```
+
+# Running the application
+
+As a prerequisite to run he application, it is necessary to run the base modules of the orchestration blueprint. Make sure that the network is the same (e.g. host).
+
+The following modules should be running - for example, when running using Eclipse Ankaios using docker and the local network from the Eclupse Orchestration project (asumming the code is checked out in ~/repos/maestro-challenge).
+
+``` bash
+docker run -it --privileged --name custom_ankaios_dev -v ~/repos/maestro-challenge/eclipse-ankaios:/workspaces/app -v ~/repos/maestro-challenge/in-vehicle-stack:/workspaces/app/in-vehicle-stack --network host -p 25551:25551 --workdir /workspaces/app custom-ankaios-dev:0.1  /bin/bash
+```
+
+follow the instructions to run the base workloads using the run_maestro.sh script. listing the workloads
+
+```bash
+/workspaces/app# ank get workloads
+ WORKLOAD NAME              AGENT     RUNTIME   EXECUTION STATE
+ digital_twin_cloud_sync    agent_A   podman    Running
+ digital_twin_vehicle       agent_A   podman    Running
+ dynamic_topic_management   agent_A   podman    Running
+ mqtt_broker                agent_A   podman    Running
+ service_discovery          agent_A   podman    Running
+```
+
+Once this is running it is possible to run the vehicle_provider and the insurance_event_detector applications
+
 

edge/applications/insurance_event_detector/main.py

@@ -1,9 +1,11 @@
 import argparse
 import csv
+import time
 
-from event_detector import risk_event_detector
-import event_definitions
+from applications.insurance_event_detector.event_detector import risk_event_detector
+from applications.insurance_event_detector import event_definitions
 
+from proto_build import consumer
 
 # This script is a very basic simulation of the seuence of events to detect risk events in a vehicle.
 # 
@@ -68,6 +70,11 @@ def risk_event_callback(riskEvent):
 def process_signal(data):
     return Signal(data[1], float(data[3]), float(data[2]))
 
+def process_mqtt_signal(data):
+    topic = data.topic.replace("/", "_")
+    return Signal(topic, float(data.payload), float(time.time() * 1000))
+
+
 # Each time that a signal change is posted in the in-vehicle digital twin, the risk event detectors will be notified.
 # Each risk event detector has individual logic that decides if it should be triggered
 # In case the risk event detects a problem, it will post the notification in the callback
@@ -95,6 +102,36 @@ def process_sample_file(filename):
 
 
 
+
+def on_message(client, userdata, msg):
+    #print(f"Received message {msg.payload} on topic {msg.topic}")
+
+    hist_signals = 60
+    signal = process_mqtt_signal(msg)
+
+    if signal.name in signal_dict:
+        update_signal_value(signal_dict, signal, hist_signals)
+        risk_event_detector(event_dict, timeout_dict, signal, signal_dict, risk_event_callback)
+
+
+def process_vehicle_integration():
+
+    # Make the ids compatible with DTDL
+    collectedSignals = [ ("dtmi:" + element.replace("_", ":") + ";1") for element in signal_dict.keys()]
+
+    print(f"{collectedSignals}")
+
+    consumer.start(collectedSignals)
+
+    consumer.mqttClient.on_message = on_message
+
+    consumer.mqttClient.loop_forever()
+
+
+event_dict = {}
+timeout_dict = {}
+signal_dict = {}
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Starts the sample process")
     parser.add_argument("-f", "--file", dest="file", help="Path to the file containing the recording.")
@@ -115,5 +152,11 @@ def process_sample_file(filename):
         "harsh_cornering": event_definitions.harsh_cornering,
         }
 
+    timeout_dict = setup_timeout_dict(event_dict.values())
+    signal_dict = setup_signal_dict(event_dict)
+
     if(args.file):
-        process_sample_file(args.file)
+        process_sample_file(args.file)
+    else:
+        process_vehicle_integration()
+