This demo shows how disparate sensors that detect the same object could have their data feeds 'fused' together. This fusion process would enable a more accurate picture of real world objects to be mapped.
This demo achieves this by making some assumptions around minimum distance and direction of travel of any number of object tracks relative to each another. In a real world scenario, the following considerations would also need to be taken into account, but for simplicity, have been excluded from this demo:
- Order of precedence of sensor data e.g. AIS data potentially taking primacy over radar.
- Aggregation of specific data elements from original source sensors during the fusion process e.g. when combining tracks, you may want to include unique insight from each sensor such as MMSI from AIS, and distance from the radar for radar data.
- Differing rules for minimum distance and relative directions of movement based on the physical characteristics of the detected object e.g. the minimum distances between sailing boats is likely to be different to the minimum distance between container ships. Equally, ships in port will be closer to one another than ships at sea.
In the ./fusion folder, you'll find the following Python scripts:
calculatetrack.py
fusionservice.py
testdatagen.py
fusionservice.py is the one you need to run to see the demo.
The fusion service takes an input of JSON messages and continues to check for new messages based on the time period detailed within the poll_frequency variable.
For the demo, the JSON messages are generated by testdatagen.py, but this could be replaced by a repeated check of Kafka for new messages on a topic.
Within the get_messages function, a check is conducted of the previously seen messages (hash_list) to see if any of those messages are now considered 'stale' i.e. no further updates should be sent for the given ORIGINATORID. We assume that ORIGINATORID is a unique ID assigned by the source sensor, but could relate to the same object e.g. a radar could assign Ship A an ORIGINATORID of '0', with Ship A also being detected by a UAV reporting an ORIGINATORID of '1'. The fusion service aims to identify these cases by analysing the track of '0' and '1' to determine whether they are likely the same object, but detected through disparate sensors.
Once stale historic tracks have been removed via the check performed by the stale_message function, the new messages retrieved from Kafka/testdatagen.py are checked to ensure that none of those messages are stale. This could be due to messages arriving out of sync or delayed for some reason.
For messages that are considered valid, the get_messages function generates a unique hash value based on the ORIGINATORID and DETECTIONTIME. This hash is added to the JSON message, along with a further element called TRACK - this is where previous position locations will be appended for a given object.
The message, along with the hash_list is then passed to the track_updater function. This function looks to see if the ORIGINATORID has previously been seen. If it has been seen, then if the DETECTIONTIME on the message falls before the most recent DETECTIONTIME recorded for the message, the message is appended to the TRACK element of the existing record for that ORIGINATORID in hash_list. If the DETECTIONTIME falls after the most recent DETECTIONTIME, then the reverse happens.
Next, the calculate_track function is used (resides within calculatetrack.py), which based on the values given for the distance_threshold and direction_threshold, compares historic records in hash_list to determine if they have the same track, and therefore likely represent the same object. This is achieved by calculating the average distance between each position event held within the TRACK JSON element for each record in hash_list, then comparing this with other records in hash_list. If the distances are greater than the distance_threshold, then the tracks are considered different. If the distance is within the distance_threshold, then the next check that is performed is to check the direction of travel of the objects (check_same_direction). This function determines the dot product of the direction vectors of the TRACK records. A value of 1 means the direction has to be exactly the same, a value of -1 means the direction can be exactly opposite, and a value of 0 allows for perpendicular directions.
If the dot product of the direction vectors of two consecutive points is less than the direction_threshold, the track is not considered to be moving in the same direction. Otherwise, the tracks are considered to be the same, and are merged.
The new hash_list is returned to fusionservice.py, with the send_messages function being used to publish messaged to Kafka on a 'Fusion' topic. For now, the ORIGINATORID is retained as the top level ID, but in reality you may want to generate a unique ID from the fusion service which maps to any ORIGINATORID's that have been associated with that fused track.
e.g.
[
{
"ORIGINATORID": "0",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"HASH": "db983d39b5e12e9f7944e70b1600f327ef4bb7b21d18ea3c86831545fb3b0dc2",
"TRACK": [
{
"ORIGINATORID": "0",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"DISTANCE": 0.0,
"SPEED": 0.0
},
{
"ORIGINATORID": "1",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"DISTANCE": 0.0,
"SPEED": 0.0
}
]
}
]
Becomes
[
{
"FUSIONID: "A"
"ORIGINATORID": "0",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"HASH": "db983d39b5e12e9f7944e70b1600f327ef4bb7b21d18ea3c86831545fb3b0dc2",
"TRACK": [
{
"ORIGINATORID": "0",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"DISTANCE": 0.0,
"SPEED": 0.0
},
{
"ORIGINATORID": "1",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"DISTANCE": 0.0,
"SPEED": 0.0
}
]
}
]
The following section of the code is included to demonstrate the fusion process working, and should be removed for 'real' data:
# For demo purposes only!
n = n+1
if n < 12:
with open(f'./data/data-after-message-{n}.json', 'w') as f:
json.dump(hash_list, f, indent=4)
# End of demo addition ^^
This contains some json message exports from fusionservice.py to show how tracks are being combined and seperated as more data is obtained. data-sent.json is an example of the totality of the analysis of a batch of messages.
The DISTANCE field within each record (n) under TRACK is the haversine distance between n and n+1 in meters.
The SPEED field within each record (n) under TRACK is the DISTANCE / DETECTIONTIME(n+1) - DETECTIONTIME(n) in meters per second.
{
"ORIGINATORID": "1",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"DISTANCE": 0.0,
"SPEED": 0.0
},
{
"ORIGINATORID": "2",
"DETECTIONTIME": "2023-11-23T11:54:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.0005,
"DISTANCE": 55.59754011676646,
"SPEED": 0.9266256686127743
},
{
"ORIGINATORID": "0",
"DETECTIONTIME": "2023-11-23T11:53:55",
"LATITUDE": 0.0,
"LONGITUDE": 0.001,
"DISTANCE": 55.59754011676646,
"SPEED": 0.9266256686127743
}