Add calculated signals to your data lake

Parquet Data Lake Custom Messages

In this section we explain how you can customize your Amazon/Google/Azure cloud function to create custom messages - and output the results as Parquet files in your data lake.

Add calculated signals via Grafana and/or SQL

If you are using Grafana, you can handle 90%+ of custom signal calculations via Grafana’s powerful transformations and/or by customizing your SQL queries.

Note

We recommend to review if your custom signals can be handled in the frontend before proceeding - feel free to contact us for sparring

Add calculated signals via cloud function

If your custom signals cannot be handled in the frontend, you can instead create new Parquet data lake tables that contain your calculated signals, moving the processing to your backend.

To create a custom signal table via your cloud function, follow below steps:

  1. Download our example custom-messages.json below and modify it as per your needs
  2. Once ready, upload it to your input bucket root
  3. If required, download your cloud function zip, modify the custom_message_functions.py and deploy your customized cloud function

custom-messages.json (J1939 DM1 example)

Note

The default cloud function includes some built-in functions that do not require customizing the cloud function (see examples below)

How it works

Adding the custom-messages.json to your input bucket will add the following cloud function behavior:

  1. The function will download and validate your custom-messages.json file
  2. For each entry, it loads the message(s) in messages_filtered_list into a data frame
  3. It determines if/when the event occurs in the file based on specified signal thresholds
  4. It optionally resamples the data frame (required for cross-message calculations)
  5. The data frame is parsed to the custom_message_functions.py in the cloud function
  6. The matching ‘function’ code is applied to create the custom message
  7. The new custom Parquet file is uploaded to the output bucket

Note

Make sure to use valid custom message/signal names[1]

JSON syntax

To illustrate the JSON syntax, consider the example below:

[
{
   "messages_filtered_list": [["CAN9_GnssSpeed", "CAN9_GnssDistance"]],
   "messages_match_type": "equals",
   "raster": "0.2s",
   "custom_message_name": "CAN9_CALC_GnssDeltaDistance",
   "prefix": false,
   "function": "delta_distance"
}
]

For each entry you specify a custom_message_name. The messages_filtered_list reflects the CAN messages that will be parsed to your custom function. If messages_match_type equals "exact", the exact list of CAN messages specified is evaluated. If it equals "contains", the messages_filtered_list should equal a string (not a list) and the event then includes all CAN messages which contain this string in their name. You can set a raster to resample the messages before applying the custom calculations.

The prefix lets you control whether you wish to update the signal names to include the message name as a prefix (this may be required for uniqueness). The function lets you specify what function should be applied to the data within the custom_message_functions.py script.

Python syntax

The custom_message_functions.py can be modified to create practically any customized logic you prefer. Below is a built-in example related to the JSON entry above:

# Example 3: Add trip distance delta signals
if function == "delta_distance":
   df_messages['DeltaDistance'] = df_messages['DistanceTrip'].diff()
   df_messages["DeltaDistanceHighSpeed"] = df_messages['DeltaDistance'].where((df_messages['Speed'] > 20) & (df_messages['SpeedValid'] == 1), None)
   df_messages["DeltaDistanceLowSpeed"] = df_messages['DeltaDistance'].where((df_messages['Speed'] <= 20) & (df_messages['SpeedValid'] == 1), None)
   signals_to_include = ["DeltaDistance","DeltaDistanceHighSpeed","DeltaDistanceLowSpeed"]
   df_messages = df_messages[signals_to_include]
   df_messages = df_messages[(df_messages.notnull()).any(axis=1)]

Examples

Example 1: Combine J1939 DM1 messages

Visualizing the J1939 Diagnostic Message 1 (DM1) in e.g. Grafana is normally complex as explained in our J1939-73 intro and our J1939 protocol guidance. This is because the DM1 data is split across multiple tables (per source addresses and DTC). To solve this, the default cloud function supports the custom function combine_dtcs that combines all DM1 messages into a single table per CAN channel. To add this custom table to your data lake, simply add the ready-to-use custom-messages.json from above to your input bucket (no cloud function customization required).

[
{
   "messages_filtered_list": "CAN1_DM01_DTC_",
   "messages_match_type": "contains",
   "raster": "",
   "custom_message_name": "CAN1_CALC_DM01",
   "prefix": false,
   "function": "combine_dtcs"
}
]

Example 2: Add custom geofences

The built-in custom_geofences function can be used to evaluate when your CANedge enters/exits custom geofence areas. To leverage this, you first upload a geofences.json file with syntax as below to your input bucket/container:

[
{
   "id": 1,
   "name": "Home Base",
   "latitude": 56.072706,
   "longitude": 10.103398,
   "radius": 0.2
},
{
   "id": 2,
   "name": "Service Center",
   "latitude": 56.116626,
   "longitude": 10.154564,
   "radius": 0.3
}
]

Here the syntax is as follows:

  • id: Numeric identifier for the geofence (used in the GeofenceId output signal)
  • name: Descriptive name for the geofence
  • latitude: Geographic latitude in decimal degrees (-90 to 90)
  • longitude: Geographic longitude in decimal degrees (-180 to 180)
  • radius: Radius of the geofence in kilometers (must be greater than 0)

Next, you can add the below to your custom-messages.json file. The script will now create a new Parquet table with the geofence status.

[
{
   "messages_filtered_list": [["CAN9_GnssPos"]],
   "messages_match_type": "equals",
   "raster": "0.2s",
   "custom_message_name": "CAN9_CALC_CustomGeofences",
   "prefix": false,
   "function": "custom_geofences"
}
]

Example 3: Combine select messages

You can create a single table that combines all signals across multiple messages with a shared timestamp. This can be done by using the below example JSON entry. The "resample" function is supported by the default cloud function.

[
{
   "messages_filtered_list": [["CAN1_Message1", "CAN1_Message2", "CAN1_Message3"]],
   "messages_match_type": "exact",
   "raster": "0.5s",
   "custom_message_name": "CAN1_CALC_Combined123",
   "prefix": false,
   "function": "resample"
}
]

Example 4: Combine all messages

As a special case, you can use a variation of the above to create a single message that contains all your signals across all your messages. In this case, the prefix should typically be set to true to ensure column uniqueness.

[
{
   "messages_filtered_list": [[]],
   "messages_match_type": "all_messages",
   "raster": "1s",
   "custom_message_name": "ALL_DATA_RESAMPLED",
   "prefix": true,
   "function": "resample"
}
]
[1]See e.g. the Athena naming guidelines for tables (aka messages) and columns (aka signals)