Readme and Telemetry Packet processing improvements

fone2val/README.md

@@ -10,7 +10,7 @@ https://github.com/fraunhofer-iem/f1-telemetry-dashboard
 The custom [VSS File](./VSS/vss.json) contains specification points for further Application use.\
 The [`carTelemetry_feeder.ini`](./config/carTelemetry_feeder.ini)  contains `kuksa.val`, `listenerIPAddr` and `PS5_UDPPort` configuration.
 
-Before starting the [F1 feeder](./carTelemetry_feeder.py), you need to start the `kuksa.val databroker` docker container by running the following command in the main project folder:
+Before starting the [F1 feeder](./carTelemetry_feeder.py), you need to start the `kuksa.val databroker` docker container by running the following command in the main project folder (sucessfully tested with version 0.4.3):
 ```
 docker run -it -v ./VSS:/VSS --rm --net=host -p 127.0.0.1:8090:8090 -e LOG_LEVEL=ALL ghcr.io/eclipse/kuksa.val/databroker:master --insecure --vss /VSS/vss.json
 ```
@@ -23,22 +23,14 @@ General Information: This Project was run on an Ubuntu VM and created in coopera
 ```
 a. The F1 telemetry port/IP number for communication has to be updated in the ./config/carTelemetry_feeder.ini file.
 
-	> IP address of the Host/VM for example 192.168.178.154
-	> Same with the Port: fore example 20778
+	> IP address of the Host/VM for example 192.168.178.154 [listenerIPAddr] host
+	> Same with the Port: fore example 20778 [PS5_UDPort] port
 
-b. The listenerIPAddr of the host/VM a also needs to be updated in the ./config/carTelemetry_feeder.ini file.
-
-	> It has to match with the given IP in step a.
-
-c. The PS5_UDPPort of the host/VM a also needs to be updated in the ./config/carTelemetry_feeder.ini file.
-
-	> It has to match with the given Port in step a.
-
-d. kuksa.val IP for the VSSClient has to be updated in the ./config/carTelemetry_feeder.ini file.
+b. kuksa.val IP for the VSSClient has to be updated in the ./config/carTelemetry_feeder.ini file.
 
 	> Normally set to 127.0.0.1.
 
-e. kuksa.val port for the VSSClient has to be updated in the ./config/carTelemetry_feeder.ini file.
+c. kuksa.val port for the VSSClient has to be updated in the ./config/carTelemetry_feeder.ini file.
 
 	> Normaly set to 55555.
 ```

fone2val/carTelemetry_feeder.py

@@ -19,45 +19,67 @@
 from kuksa_client.grpc import VSSClient
 from kuksa_client.grpc import Datapoint
 from telemetry_f1_2021.listener import TelemetryListener
+from threading import Lock
 
 scriptDir = os.path.dirname(os.path.realpath(__file__))
 sys.path.append(os.path.join(scriptDir, "../../"))
 
+# telemetry packet ids
+TelemetryPacketID_Engine = 6
+TelemetryPacketID_CarStatus = 7
+TelemetryPacketID_CarDamage = 10
+TelemetryPacketID_LapTime = 2
+
 
 class Kuksa_Client():
     # Constructor
     def __init__(self, config):
         print("Init kuksa client...")
+        self.config = config
         if "kuksa_val" not in config:
             print("kuksa_val section missing from configuration, exiting")
             sys.exit(-1)
+        kuksaConfig = config['kuksa_val']
+        self.host = kuksaConfig.get('host')
+        self.port = kuksaConfig.getint('port')
 
     def shutdown(self):
         self.client.stop()
 
-# Christophers approach on sending Data to Kuksa Server
-    def setTelemetryData(self, teleData):
-        dataDictionary = {}
-
-        kuksaConfig = config['kuksa_val']
-        with VSSClient(kuksaConfig.get('host'), kuksaConfig.getint('port')) as client:
-            for x, y in teleData.items():
-                dataDictionary.update({
-                    str(x): Datapoint(y)
-                })
-            client.set_current_values(dataDictionary)
+    # Christophers approach on sending Data to Kuksa Server
+    def setTelemetryData(self, telemetryData):
+        with VSSClient(self.host, self.port) as client:
+            client.set_current_values(telemetryData)
 
 
 class carTelemetry_Client():
 
     def __init__(self, config, consumer):
         print("Init carTelemetry client...")
+        self.consumer = consumer
         if "listenerIPAddr" not in config:
             print("listenerIPAddr section missing from configuration, exiting")
             sys.exit(-1)
         if "PS5_UDPPort" not in config:
             print("PS5_UDPPort section missing from configuration, exiting")
             sys.exit(-1)
+        # init thread variables
+        self.datastructure_lock = Lock()
+        self.list_for_Ids = []
+        self.id_To_LastPacket = {}
+        self.id_To_ProcessingFunction = {}
+        self.id_To_ProcessingFunction[TelemetryPacketID_Engine] = (
+            self.processTelemetryPacket_Engine
+        )
+        self.id_To_ProcessingFunction[TelemetryPacketID_CarDamage] = (
+            self.processTelemetryPacket_CarDamage
+        )
+        self.id_To_ProcessingFunction[TelemetryPacketID_CarStatus] = (
+            self.processTelemetryPacket_CarStatus
+        )
+        self.id_To_ProcessingFunction[TelemetryPacketID_LapTime] = (
+            self.processTelemetryPacket_LapTime
+        )
         # extract carTelemetry Data
         print("Connecting to extract CarTelemetry Data")
 
@@ -67,73 +89,121 @@ def __init__(self, config, consumer):
         self.thread = threading.Thread(target=self.loop, args=())
         self.thread.start()
 
+    def get_next_packet(self):
+        print("start next packet thread")
+        while self.running:
+            try:
+                # listen to the data via UDP channel
+                packet = self.listener.get()
+                packetID = packet.m_header.m_packet_id
+                if packetID in [
+                    TelemetryPacketID_Engine,
+                    TelemetryPacketID_CarDamage,
+                    TelemetryPacketID_CarStatus,
+                    TelemetryPacketID_LapTime,
+                ]:
+                    with self.datastructure_lock:
+                        if packetID not in self.list_for_Ids:
+                            self.list_for_Ids.append(packetID)
+                        self.id_To_LastPacket[packetID] = packet
+            except Exception:
+                continue
+
     def loop(self):
         print("Car Telemetry data loop started")
 
+        # extract config
         config_ipAddr = config['listenerIPAddr']
         config_UDPport = config['PS5_UDPPort']
-
         listener_ip = config_ipAddr['host']
         udp_port = config_UDPport['port']
 
         print(f"listener_ip:{listener_ip}")
         print(f"udp_port:{udp_port}")
 
-        listener = TelemetryListener(port=int(udp_port), host=listener_ip)
-
+        # init threads to process telemetry data
+        self.listener = TelemetryListener(port=int(udp_port), host=listener_ip)
+        thread_get_next_packet = threading.Thread(target=self.get_next_packet)
+        thread_initPacketProcessing = threading.Thread(target=self.initPacketProcessing)
+        thread_get_next_packet.start()
+        thread_initPacketProcessing.start()
+        thread_get_next_packet.join()
+        thread_initPacketProcessing.join()
+
+    def processTelemetryPacket_Engine(self, telemetryPacket):
+        # Get data
+        carIndex = telemetryPacket.m_header.m_player_car_index
+        Speed = telemetryPacket.m_car_telemetry_data[carIndex].m_speed
+        EngineRPM = telemetryPacket.m_car_telemetry_data[carIndex].m_engine_rpm
+        # Store data
+        carTelemetry = {}
+        carTelemetry['Vehicle.Speed'] = Datapoint(Speed)
+        carTelemetry['Vehicle.RPM'] = Datapoint(EngineRPM)
+        return carTelemetry
+
+    def processTelemetryPacket_CarDamage(self, telemetryPacket):
+        # Get data
+        carIndex = telemetryPacket.m_header.m_player_car_index
+        leftWingDamage = telemetryPacket.m_car_damage_data[carIndex].m_front_left_wing_damage
+        rightWingDamage = telemetryPacket.m_car_damage_data[carIndex].m_front_right_wing_damage
+        # Extract nested data
+        tyreWear_1 = telemetryPacket.m_car_damage_data[carIndex].m_tyres_wear[0]
+        tyreWear_2 = telemetryPacket.m_car_damage_data[carIndex].m_tyres_wear[1]
+        tyreWear_3 = telemetryPacket.m_car_damage_data[carIndex].m_tyres_wear[2]
+        tyreWear_4 = telemetryPacket.m_car_damage_data[carIndex].m_tyres_wear[3]
+        # Store data
+        carTelemetry = {}
+        carTelemetry['Vehicle.FrontLeftWingDamage'] = Datapoint(leftWingDamage)
+        carTelemetry['Vehicle.FrontRightWingDamage'] = Datapoint(rightWingDamage)
+        carTelemetry['Vehicle.Tire.RearLeftWear'] = Datapoint(tyreWear_1)
+        carTelemetry['Vehicle.Tire.RearRightWear'] = Datapoint(tyreWear_2)
+        carTelemetry['Vehicle.Tire.FrontLeftWear'] = Datapoint(tyreWear_3)
+        carTelemetry['Vehicle.Tire.FrontRightWear'] = Datapoint(tyreWear_4)
+        return carTelemetry
+
+    def processTelemetryPacket_LapTime(self, telemetryPacket):
+        # Get data
+        carIndex = telemetryPacket.m_header.m_player_car_index
+        lastLapTime_in_ms = telemetryPacket.m_lap_data[carIndex].m_last_lap_time_in_ms
+        # Preprocessing
+        lastLapTime_in_s = lastLapTime_in_ms / 1000
+        # Store data
+        carTelemetry = {}
+        carTelemetry['Vehicle.LastLapTime'] = Datapoint(lastLapTime_in_s)
+        return carTelemetry
+
+    def processTelemetryPacket_CarStatus(self, telemetryPacket):
+        # Get data
+        carIndex = telemetryPacket.m_header.m_player_car_index
+        fuelInTank = telemetryPacket.m_car_status_data[carIndex].m_fuel_in_tank
+        fuelCapacity = telemetryPacket.m_car_status_data[carIndex].m_fuel_capacity
+        # Preprocessing
+        fuelInPercent = int((fuelInTank / fuelCapacity) * 100)
+        # Store data
+        carTelemetry = {}
+        carTelemetry['Vehicle.FuelLevel'] = Datapoint(fuelInPercent)
+        return carTelemetry
+
+    def initPacketProcessing(self):
         while self.running:
             try:
-                # listen to the data via UDP channel
-                packet = listener.get()
-
-                # Update packet ID
-                packetID = packet.m_header.m_packet_id
-                # player carIndex
-                carIndex = packet.m_header.m_player_car_index
-                # Check for telemetry data - packet ID 6.
-                if (packetID == 6):
-
-                    EngineRPM = packet.m_car_telemetry_data[carIndex].m_engine_rpm
-                    Speed = packet.m_car_telemetry_data[carIndex].m_speed
-
-                    self.carTelemetry['Vehicle.Speed'] = Speed
-                    self.carTelemetry['Vehicle.RPM'] = EngineRPM
-
-                    # Set the data to the KUKSA_VAL
-                    self.consumer.setTelemetryData(self.carTelemetry)
-
-                if (packetID == 7):  # car status data packet
-                    fuelInTank = packet.m_car_status_data[carIndex].m_fuel_in_tank
-                    fuelCapacity = packet.m_car_status_data[carIndex].m_fuel_capacity
-                    fuelInPercent = fuelInTank/fuelCapacity
-
-                    self.carTelemetry['Vehicle.FuelLevel'] = int(fuelInPercent*100)
-                    self.consumer.setTelemetryData(self.carTelemetry)
-
-                if (packetID == 10):  # car dmg packet
-
-                    leftWingDamage = packet.m_car_damage_data[carIndex].m_front_left_wing_damage
-                    rightWingDamage = packet.m_car_damage_data[carIndex].m_front_right_wing_damage
-
-                    tyreWear_1 = packet.m_car_damage_data[carIndex].m_tyres_wear[0]
-                    tyreWear_2 = packet.m_car_damage_data[carIndex].m_tyres_wear[1]
-                    tyreWear_3 = packet.m_car_damage_data[carIndex].m_tyres_wear[2]
-                    tyreWear_4 = packet.m_car_damage_data[carIndex].m_tyres_wear[3]
-
-                    self.carTelemetry['Vehicle.FrontLeftWingDamage'] = leftWingDamage
-                    self.carTelemetry['Vehicle.FrontRightWingDamage'] = rightWingDamage
-                    self.carTelemetry['Vehicle.Tire.RearLeftWear'] = tyreWear_1
-                    self.carTelemetry['Vehicle.Tire.RearRightWear'] = tyreWear_2
-                    self.carTelemetry['Vehicle.Tire.FrontLeftWear'] = tyreWear_3
-                    self.carTelemetry['Vehicle.Tire.FrontRightWear'] = tyreWear_4
-
-                    self.consumer.setTelemetryData(self.carTelemetry)
-                if (packetID == 2):
-                    lastLapTime = packet.m_lap_data[carIndex].m_last_lap_time_in_ms
-
-                    self.carTelemetry['Vehicle.LastLapTime'] = lastLapTime/1000
-
-                    self.consumer.setTelemetryData(self.carTelemetry)
+                # Initializing variables
+                packetID = None
+                packet = None
+                # Lock datastructures
+                with self.datastructure_lock:
+                    # Update packet ID and get dependend packet
+                    if len(self.list_for_Ids) > 0:
+                        packetID = self.list_for_Ids.pop()
+                        packet = self.id_To_LastPacket[packetID]
+                    else:
+                        packetID = None
+                        packet = None
+                if packet is not None:
+                    # Process telemetry packet
+                    carTelemetry = self.id_To_ProcessingFunction[packetID](packet)
+                    # Forward data to KUKSA_VAL
+                    self.consumer.setTelemetryData(carTelemetry)
             except Exception:
                 continue