Added custom signal calculation example

Author: MatinF

examples/data-processing/custom_signal.py

@@ -0,0 +1,53 @@
+import mdf_iter
+import canedge_browser
+import can_decoder
+
+import pandas as pd
+
+from datetime import datetime, timezone
+from utils import setup_fs, custom_sig
+
+
+# specify which devices to process (from local folder or S3 bucket)
+devices = ["LOG/958D2219"]
+
+# specify which time period to fetch log files for
+start = datetime(year=2020, month=1, day=13, hour=0, minute=0, tzinfo=timezone.utc)
+stop = datetime(year=2099, month=1, day=1, tzinfo=timezone.utc)
+
+# specify DBC path
+dbc_path = r"CSS-Electronics-SAE-J1939-DEMO.dbc"
+
+# ---------------------------------------------------
+# initialize DBC converter and file loader
+db = can_decoder.load_dbc(dbc_path)
+df_decoder = can_decoder.DataFrameDecoder(db)
+
+fs = setup_fs()
+
+# List log files based on inputs and select first log file
+log_files = canedge_browser.get_log_files(fs, devices, start_date=start, stop_date=stop)
+log_file = log_files[0]
+
+with fs.open(log_file, "rb") as handle:
+    mdf_file = mdf_iter.MdfFile(handle)
+    df_raw = mdf_file.get_data_frame()
+
+# extract all DBC decodable signals and print dataframe
+df_phys = df_decoder.decode_frame(df_raw)
+
+
+# define a function for calculating the custom signal
+def ratio(s1, s2):
+    if s2 != 0:
+        return 100 * s1 / s2
+    else:
+        return np.nan
+
+
+# calculate custom signal and append to main dataframe, then save as CSV
+df_ratio = custom_sig(df_phys, "WheelBasedVehicleSpeed", "EngineSpeed", ratio, "Ratio")
+
+df_phys = df_phys.append(df_ratio)
+df_phys.to_csv("data_incl_custom_signal.csv")
+print("Saved data incl. custom signal to CSV")

examples/data-processing/process_data.py

@@ -49,11 +49,12 @@
     path = device_id + log_file.split(device_id)[1].replace("MF4", "csv").replace(
         "/", "_"
     )
-    df_phys.to_csv(path)
 
     if df_phys.empty:
         continue
 
+    df_phys.to_csv(path)
+
     # create a list of the individual DBC decoded dataframes:
     df_concat.append(df_phys)
 

examples/data-processing/utils.py

@@ -8,7 +8,7 @@ def setup_fs_s3():
         secret="<secret>",
         client_kwargs={
             "endpoint_url": "<endpoint>",
-            "verify": "path\\to\\public_certificate.crt",  # for TLS enabled MinIO servers
+            # "verify": "path\\to\\public_certificate.crt",  # for TLS enabled MinIO servers
         },
     )
 
@@ -26,3 +26,33 @@ def setup_fs():
     fs = LocalFileSystem()
 
     return fs
+
+
+def custom_sig(df, signal1, signal2, function, new_signal):
+    """Helper function for calculating a new signal based on two signals and a function.
+    Returns a dataframe with the new signal name and physical values
+    """
+    import pandas as pd
+
+    try:
+        s1 = df[df["Signal"] == signal1]["Physical Value"].rename(signal1)
+        s2 = df[df["Signal"] == signal2]["Physical Value"].rename(signal2)
+
+        df_new_sig = pd.merge_ordered(
+            s1, s2, on="TimeStamp", fill_method="ffill",
+        ).set_index("TimeStamp")
+
+        df_new_sig = (
+            df_new_sig.apply(lambda x: function(x[0], x[1]), axis=1)
+            .dropna()
+            .rename("Physical Value")
+            .to_frame()
+        )
+
+        df_new_sig["Signal"] = new_signal
+
+        return df_new_sig
+
+    except:
+        print(f"Warning: Custom signal {new_signal} not created\n")
+        return pd.DataFrame()