Merge pull request #1 from CSS-Electronics/develop

Restructuring of script to enable increased modularity

Author: MatinF

.gitignore

@@ -1,4 +1,3 @@
 __pycache__
-inputs_test.py
 *.whl
-temp-variables.py
+lambda_handler.py

README.md

@@ -24,27 +24,35 @@ To use the script, install dependencies via the `requirements.txt`:
 ---
 ## Deployment
 
-### 1: Test script with sample data 
-1. Clone the script folder incl. the J1939 sample data and demo DBC
+### 1: Test script with sample data
+1. Download this repository incl. the J1939 data and demo DBC
 2. In `inputs.py` add your InfluxDB details, then run `python main.py` via the command line
 
 *Note: If you use a free InfluxDB Cloud user, the sample data will be removed after a period (as it is >30 days old).*
 
-### 2: Modify script with your own details 
-1. Local data: Add your own data next to the scripts as per the SD structure:
+### 2: Modify script with your own details
+1. Local disk: Add your own data next to the scripts as per the SD structure:
    `LOG/<device_ID>/<session>/<split>.MF4`
-2. S3 server: Add your S3 server details in `inputs.py` and set `use_s3 = True`
-3. In `inputs.py` update the DBC path and the device list to match yours
-4. In `last_run.txt`, specify from when you wish to load log files (e.g. `2020-01-13 00:00:00`)
+2. S3 server: Add your S3 server details in `inputs.py` and set `s3 = True`
+3. In `inputs.py` update the DBC path list and the device list to match yours
+4. In `last_run.txt`, specify when you wish to load log files from (e.g. `2020-01-13 00:00:00`)
 5. Optionally modify the signal filters or resampling frequency
 
 
 ### 3: Enable dynamic start time
-1. In `inputs.py` set `dynamic = True` 
+1. In `inputs.py` set `dynamic = True`
 2. Follow the CANedge Intro guide for setting up e.g. Windows Task Scheduler
 
 ---
-## Deleting data from InfluxDB 
-If you need to delete data in InfluxDB that you e.g. uploaded as part of a test, you can use the `delete_influx(name)` function from `utils.py`. Call it by parsing the name of the 'measurement' to delete (i.e. the device serial number):
+## Other practical information
 
-``delete_influx("958D2219")``
+### Change verbosity
+By default, summary information is printed as part of the processing. You can parse `verbose=False` as an input argument in `list_log_files`, `SetupInflux` and `DataWriter` to avoid this.
+
+### Delete data from InfluxDB
+If you need to delete data in InfluxDB that you e.g. uploaded as part of a test, you can use the `delete_influx(name)` function from the `SetupInflux` class. Call it by parsing the name of the 'measurement' to delete (i.e. the device serial number):
+
+``influx.delete_influx("958D2219")``
+
+### Multiple channels
+If your log files contain data from two CAN channels, you may need to adjust the script in case you have duplicate signal names across both channels. For example, if you're extracting the signal `EngineSpeed` from both channels. 

dashboard-writer/inputs.py

@@ -1,5 +1,3 @@
-from datetime import datetime, timezone
-
 # -----------------------------------------------
 # specify your InfluxDB details
 influx_bucket = "influx_bucket_name"
@@ -9,28 +7,27 @@
 
 # -----------------------------------------------
 # specify devices to process (from a local folder or S3 bucket)
-# If local, ensure files are organized as on SD (folder/device_id/session/split.MF4)
-devices = ["LOG/958D2219"]  # local: ["folder/device_id"] | S3: ["bucket/device_id"]
+# If local, ensure logs are organized correctly: folder/device_id/session/split.MF4
+# Syntax: Local: ["folder/device_id"] | S3: ["bucket/device_id"]
+devices = ["LOG/958D2219"]
+
 
 # -----------------------------------------------
-# specify DBC path and the list of signals to process ([]: process all signals)
-dbc_path = "CSS-Electronics-SAE-J1939-DEMO.dbc"
+# specify DBC paths and a list of signals to process ([]: include all signals)
+dbc_paths = ["CSS-Electronics-SAE-J1939-DEMO.dbc"]
 signals = []
 
-# optionally modify resampling frequency ("": disable resampling)
+# specify resampling frequency ("": no resampling)
 res = "1S"
 
 # -----------------------------------------------
-# specify your S3 details (if relevant)
-use_s3 = False
+# specify whether to load data from S3 (and add server details if relevant)
+s3 = True
 key = "s3_key"
 secret = "s3_secret"
 endpoint = "s3_endpoint"
-# cert = "path/to/cert.crt"  # if MinIO + TLS, add path to cert and update utils.py/setup_fs_s3 to verify
-
+# cert = "path/to/cert.crt"  # if MinIO + TLS, add path to cert and update utils.py/setup_fs to verify
 
+# -----------------------------------------------
 # toggle whether to update the last execution datetime on each script execution
-use_dynamic = False
-
-# set stop date to a point in the future to load all log files after start date
-stop = datetime(year=2099, month=1, day=1, tzinfo=timezone.utc)
+dynamic = False

dashboard-writer/main.py

@@ -1,65 +1,14 @@
-import mdf_iter, canedge_browser, can_decoder
-from utils import setup_fs, setup_fs_s3, write_influx, load_last_run, set_last_run, print_summary
+from utils import setup_fs, load_dbc_files, list_log_files, SetupInflux, DataWriter
 import inputs
-from pathlib import Path
 
+# setup filesystem (local or S3), load DBC files and list log files for processing
+fs = setup_fs(inputs.s3, inputs.key, inputs.secret, inputs.endpoint)
+db_list = load_dbc_files(inputs.dbc_paths)
+log_files = list_log_files(fs, inputs.devices, inputs.dynamic)
 
-# function for loading raw CAN data from S3, DBC converting it and writing it to InfluxDB
-def process_data_and_write():
+# initialize connection to InfluxDB
+influx = SetupInflux(influx_url=inputs.influx_url, token=inputs.token, org_id=inputs.org_id, influx_bucket=inputs.influx_bucket)
 
-    # initialize DBC converter, file loader and start date
-    db = can_decoder.load_dbc(Path(__file__).parent / inputs.dbc_path)
-    df_decoder = can_decoder.DataFrameDecoder(db)
-
-    if inputs.use_s3:
-        fs = setup_fs_s3()
-    else:
-        fs = setup_fs()
-
-    start = load_last_run()
-
-    if inputs.use_dynamic:
-        set_last_run()
-
-    log_files = canedge_browser.get_log_files(fs, inputs.devices, start_date=start, stop_date=inputs.stop)
-    print(f"Found {len(log_files)} log files")
-
-    for log_file in log_files:
-        # open log file, get device_id and extract dataframe with raw CAN data
-        with fs.open(log_file, "rb") as handle:
-            mdf_file = mdf_iter.MdfFile(handle)
-            device_id = mdf_file.get_metadata()["HDComment.Device Information.serial number"]["value_raw"]
-            df_raw = mdf_file.get_data_frame()
-
-        # DBC convert the raw CAN dataframe
-        df_phys = df_decoder.decode_frame(df_raw)
-        if df_phys.empty:
-            continue
-
-        print_summary(device_id, log_file, df_phys)
-
-        # group the data to enable a signal-by-signal loop
-        df_phys_grouped = df_phys.groupby("Signal")["Physical Value"]
-
-        # for each signal in your list, resample the data and write to InfluxDB
-        for signal, group in df_phys_grouped:
-            if signal in inputs.signals or len(inputs.signals) == 0:
-                df_signal = group.to_frame().rename(columns={"Physical Value": signal})
-                print(f"Signal: {signal} (mean: {round(df_signal[signal].mean(),2)})")
-
-                if inputs.res != "":
-                    cnt = len(df_signal)
-                    df_signal = df_signal.resample(inputs.res).pad().dropna()
-                    print(f"- Resampling to {inputs.res} ({cnt} --> {len(df_signal)} records)")
-
-                # print(df_signal)
-                write_influx(device_id, df_signal)
-
-    return
-
-
-# execute the script
-if __name__ == "__main__":
-
-    process_data_and_write()
-    pass
+# process the log files and write extracted signals to InfluxDB
+writer = DataWriter(fs=fs, db_list=db_list, signals=inputs.signals, res=inputs.res, db_func=influx.write_influx)
+writer.decode_log_files(log_files)