Tracksight Refactor (#1257)

### Changelist 
<!-- Give a list of the changes covered in this PR. This will help both
you and the reviewer keep this PR within scope. -->

### Testing Done
<!-- Outline the testing that was done to demonstrate the changes are
solid. This could be unit tests, integration tests, testing on the car,
etc. Include relevant code snippets, screenshots, etc as needed. -->

### Resolved Tickets
<!-- Link any tickets that this PR resolves. -->

---------

Co-authored-by: Ashli <[email protected]>
Co-authored-by: Ashli Forbes <[email protected]>

Pipfile

@@ -21,6 +21,9 @@ python-can = "*"
 rich = "*"
 PyGithub = "*"
 black = "*"
+flask = "*"
+Definitions = "*"
+flask_cors = "*"
 
 [dev-packages]
 setuptools = "*"

software/hardware_tests/data_rate_test_rx.py

@@ -1,21 +1,23 @@
 import serial
 
+
 def receive_data():
-    ser = serial.Serial('/dev/ttyUSB0', baudrate=57600, stopbits=1, timeout=100) 
+    ser = serial.Serial("/dev/ttyUSB0", baudrate=57600, stopbits=1, timeout=100)
 
     tick_count = 0
 
     try:
         while True:
-            line = ser.readline(12).decode().strip() 
+            line = ser.readline(12).decode().strip()
             print(line)
-            tick_count+=1
+            tick_count += 1
 
     except KeyboardInterrupt:
         # Close the serial port on Ctrl+C
         ser.close()
         print(tick_count)
 
+
 if __name__ == "__main__":
 
-    receive_data()
+    receive_data()

software/hardware_tests/data_rate_test_rx_log.py

@@ -3,31 +3,36 @@
 from datetime import datetime
 from pathlib import Path
 
+
 def timestamp():
     return datetime.now().strftime("%Y%m%d%H%M%S")
 
+
 # csv_file_path = f'rx_data_{timestamp()}.csv'
 
-csv_file_path = Path("rx_data_"+ timestamp() + ".csv")
+csv_file_path = Path("rx_data_" + timestamp() + ".csv")
+
 
 def line_to_csv(data, csv_file_path):
-    with open(csv_file_path, 'a', newline='') as csv_file:
+    with open(csv_file_path, "a", newline="") as csv_file:
         csv_writer = csv.writer(csv_file)
         csv_writer.writerow([data])
 
+
 def receive_data():
-    ser = serial.Serial('/dev/ttyUSB0', baudrate=57600, stopbits=1, timeout=100) 
+    ser = serial.Serial("/dev/ttyUSB0", baudrate=57600, stopbits=1, timeout=100)
 
     try:
         while True:
             line = ser.readline().decode().strip()
-            line_to_csv(line,csv_file_path)
+            line_to_csv(line, csv_file_path)
             print(line)
 
     except KeyboardInterrupt:
         # Close the serial port on Ctrl+C
         ser.close()
 
+
 if __name__ == "__main__":
 
-    receive_data()
+    receive_data()

software/hardware_tests/data_rate_test_tx.py

@@ -1,29 +1,31 @@
 import serial
 import time
 
+
 def generate_data_rate(target_data_rate_kbps):
-    ser = serial.Serial('/dev/ttyUSB0', baudrate=57600, stopbits=1, timeout=100) 
+    ser = serial.Serial("/dev/ttyUSB0", baudrate=57600, stopbits=1, timeout=100)
 
     target_data_rate_bps = target_data_rate_kbps * 1000
     delay = 80 / target_data_rate_bps
     tick_count = 0
 
     try:
         while True:
-            msg = str(int(tick_count)) +'\n'
-            ser.write(msg) #10 characters at 8 bits each
+            msg = str(int(tick_count)) + "\n"
+            ser.write(msg)  # 10 characters at 8 bits each
             # 80 bits
 
             time.sleep(delay)
-            tick_count+=1
+            tick_count += 1
 
     except KeyboardInterrupt:
         # Close the serial port on Ctrl+C
         ser.close()
         print(tick_count)
 
+
 if __name__ == "__main__":
 
     target_data_rate_kbps = 10
-    
-    generate_data_rate(target_data_rate_kbps)
+
+    generate_data_rate(target_data_rate_kbps)

software/hardware_tests/proto_test_rx.py

@@ -5,7 +5,7 @@
 if __name__ == "__main__":
 
     # Try to open the serial port with the default baud rate.
-    ser = serial.Serial("/dev/ttyUSB0",baudrate = 57600, timeout=1)
+    ser = serial.Serial("/dev/ttyUSB0", baudrate=57600, timeout=1)
     ser.reset_input_buffer()
     ser.reset_output_buffer()
     s = set()
@@ -16,16 +16,18 @@
     # try:
     while True:
 
-                packet_size = int.from_bytes(ser.read(DEBUG_SIZE_MSG_BUF_SIZE), byteorder="little")
-                if packet_size in s:
-                        continue
-                reply = sample_pb2.TelemMessage()
-                bytes = ser.read(packet_size)
-                reply.ParseFromString(bytes)
+        packet_size = int.from_bytes(
+            ser.read(DEBUG_SIZE_MSG_BUF_SIZE), byteorder="little"
+        )
+        if packet_size in s:
+            continue
+        reply = sample_pb2.TelemMessage()
+        bytes = ser.read(packet_size)
+        reply.ParseFromString(bytes)
 
-                print(reply.can_id, reply.data, reply.time_stamp)
+        print(reply.can_id, reply.data, reply.time_stamp)
 
-                # print(ser.read(1))
+        # print(ser.read(1))
 
     # except KeyboardInterrupt:
-    #        ser.close()
+    #        ser.close()

software/tracksight/SocketTest.py

@@ -2,38 +2,46 @@
 import json
 import sys
 import os
+
 sys.path.append(os.path.dirname(os.path.dirname(__file__)))
-from Telemetry import app, socketio  # Replace 'your_app_file' with the actual filename of your Flask app
+from Telemetry import (
+    app,
+    socketio,
+)  # Replace 'your_app_file' with the actual filename of your Flask app
+
 
 @pytest.fixture
 def client():
-    app.config['TESTING'] = True
+    app.config["TESTING"] = True
     client = app.test_client()
     socketio_client = socketio.test_client(app)
     return client, socketio_client
 
+
 def test_hello_world(client):
     client, _ = client
-    response = client.get('/')
+    response = client.get("/")
     assert response.status_code == 200
     assert b"Wireless 2.0" in response.data
 
+
 def test_socketio_connect(client):
     _, socketio_client = client
     received_events = socketio_client.get_received()
     assert len(received_events) == 1
-    assert received_events[0]['name'] == 'message'
+    assert received_events[0]["name"] == "message"
+
 
 def test_socketio_message(client):
     _, socketio_client = client
-    socketio_client.emit('data', 'Test message')
+    socketio_client.emit("data", "Test message")
     received_events = socketio_client.get_received()
 
+
 def test_get_signals(client):
     _, socket_client = client
     req = '{"ids": ["Signal1"]}'
-    socket_client.emit('signals', json.loads(req))
+    socket_client.emit("signals", json.loads(req))
     events = socket_client.get_received()
     print(events)
     assert False
-

software/tracksight/backend/.gitignore

@@ -0,0 +1 @@
+*.log

software/tracksight/backend/app/process/__init__.py

@@ -0,0 +1,8 @@
+"""
+This module is responsible for the process of the application. It is the main module of the application.
+"""
+
+# from http_app import app as http_app
+
+
+__all__ = ["signal_util", "influx_handler", "definitions"]

software/tracksight/backend/app/process/flask_apps/.gitignore

@@ -0,0 +1 @@
+dashboards.json

software/tracksight/backend/app/process/flask_apps/__init__.py

@@ -0,0 +1,9 @@
+"""
+This module contains the Flask applications that are used to handle requests from the client.
+"""
+
+__all__ = [
+    "database_app",
+    "http_app",
+    "socket_app",
+]

software/tracksight/backend/app/process/flask_apps/database_app.py

@@ -0,0 +1,86 @@
+"""
+This module contains the flask app for the database.
+"""
+
+import json
+import os
+
+from flask import request, jsonify, Blueprint
+
+DB_JSON_PATH = "./dashboards.json"
+
+
+# HELPERS
+def read_json_file():
+    """
+
+    :return:
+    """
+    if not os.path.exists(DB_JSON_PATH):
+        return {}  # TODO respond with some error
+    with open(DB_JSON_PATH) as file:
+        return json.load(file)
+
+
+def write_json_file(data):
+    """
+
+    :param data:
+    """
+    with open(DB_JSON_PATH, "w") as file:
+        json.dump(data, file, indent=4)
+
+
+app = Blueprint("database_app", __name__)
+
+
+@app.route("/get-data", methods=["GET"])
+def get_data():
+    """
+
+    :return:
+    """
+    path = request.args.get("path")
+    data = read_json_file()
+    path_data = data.get(path, "No data found")
+    return jsonify(path_data)
+
+
+@app.route("/save-data", methods=["POST"])
+def save_data():
+    """
+
+    :return:
+    """
+    content = request.json
+    data_to_save = content["data"]
+    data = read_json_file()
+    # may want to add error handling when there is duplicate name?
+    name = data_to_save["dbname"]
+    data["dashboards"][name] = data_to_save
+    write_json_file(data)
+    return jsonify({"message": "Data saved successfully"})
+
+
+@app.route("/delete-data", methods=["POST"])
+def delete_data():
+    """
+
+    :return:
+    """
+    content = request.json
+    path = content["path"]
+    print(path)
+    data = read_json_file()
+
+    path_parts = path.split("/")
+    if (
+        len(path_parts) == 2
+        and path_parts[0] == "dashboards"
+        and path_parts[1] in data.get("dashboards", {})
+    ):
+        del data["dashboards"][path_parts[1]]
+        write_json_file(data)
+        return jsonify({"message": "Data deleted successfully"})
+    else:
+        return jsonify({"error": "Data not found"}), 404