diff options
| -rw-r--r-- | .gitignore | 4 | ||||
| -rw-r--r-- | README.md | 2 | ||||
| -rw-r--r-- | Scripts/README.md | 84 | ||||
| -rw-r--r-- | Scripts/carla_to_CAN.py | 257 | ||||
| -rw-r--r-- | Scripts/record_playback.py | 141 | ||||
| -rw-r--r-- | Scripts/requirements.txt | 4 | ||||
| -rwxr-xr-x | Scripts/vcan.sh | 5 | ||||
| -rw-r--r-- | Widgets/ICPage.py | 47 | ||||
| -rw-r--r-- | extras/config.ini | 1 | ||||
| -rw-r--r-- | extras/config.py | 3 |
10 files changed, 530 insertions, 18 deletions
@@ -3,4 +3,6 @@ Widgets/.vssclient_history map.html test/ control-panel/ -res_rc.py
\ No newline at end of file +res_rc.py +Scripts/can_messages.txt +Scripts/agl-vcar.dbc @@ -20,6 +20,8 @@ A PyQt6 application to simulate CAN Bus signals using Kuksa.val for the AGL Demo $ source control-panel/bin/activate $ pip3 install -r requirements.txt $ pyside6-rcc assets/res.qrc -o res_rc.py + # (OR) + $ /usr/lib64/qt6/libexec/rcc -g python assets/res.qrc | sed '0,/PySide6/s//PyQt6/' > res_rc.py ``` ## # Usage diff --git a/Scripts/README.md b/Scripts/README.md new file mode 100644 index 0000000..e3ca6df --- /dev/null +++ b/Scripts/README.md @@ -0,0 +1,84 @@ +## Setting up CARLA + +You can follow the steps provided in the [CARLA documentation](https://carla.readthedocs.io/en/latest/start_quickstart/#carla-installation) for installing CARLA. + +We recommend using the [latest release](https://github.com/carla-simulator/carla/releases/), and using the supported Python version to run the `carla_to_CAN.py` Script. + +1. Running the CARLA Server + + ```bash + # Move to the installation directory + $ cd /path/to/CARLA_<version> + + # Start the CARLA Server + $ ./CarlaUE4.sh + + # To run using minimum resources + $ ./CarlaUE4.sh -quality-level=Low -prefernvidia + ``` + + You may also add the `-RenderOffScreen` flag to start CARLA in off-screen mode. Refer to the various [rendering options](https://carla.readthedocs.io/en/latest/adv_rendering_options/#no-rendering-mode) for more details. + + Another way of running the CARLA server without a display is by using [CARLA in Docker](https://carla.readthedocs.io/en/latest/build_docker/). + +2. Starting a Manual Simulation + + ```bash + # Navigate to directory containing the demo python scripts + # + $ cd /path/to/CARLA_<version>/PythonAPI/examples + ``` + + Create a Python virtual environment and resolve dependencies + ```bash + $ python3 -m venv carlavenv + $ source carlavenv/bin/activate + $ pip3 install -r requirements.txt + + # Start the manual_control.py script + $ python3 manual_control.py + ``` + +## Converting CARLA data into CAN + +The `carla_to_CAN.py` script can be run run alongside an existing CARLA simulation to fetch data and convert it into CAN messages based on the [agl-vcar.dbc](https://git.automotivelinux.org/src/agl-dbc/plain/agl-vcar.dbc) file. + +While the `record_playback.py` script is responsible for recording amd playing back the CAN data for later sessions. + +_NOTE_: This does **not** require the CARLA server to be running. + +To access these scripts, clone the [AGL Demo Control Panel](https://gerrit.automotivelinux.org/gerrit/admin/repos/src/agl-demo-control-panel,general) project. + +```bash +# Move to the Scripts directory +$ cd /path/to//agl-demo-control-panel/Scripts + +# Fetch the agl-vcar.dbc file +$ wget -nd -c "https://git.automotivelinux.org/src/agl-dbc/plain/agl-vcar.dbc" +``` + +Create a Python virtual environment and resolve dependencies +```bash +$ python3 -m venv carlavenv +$ source carlavenv/bin/activate +$ pip3 install -r requirements.txt + +# Optionally, set up the vcan0 interface +$ ./vcan.sh +``` + +1. Converting CARLA Data into CAN + + ```bash + $ python -u carla_to_CAN.py + # OR + $ python -u carla_to_CAN.py --host <carla_server_ip> --port <carla_server_port> + ``` + +2. Recording and Playback of CAN messages + + ```bash + $ python -u record_playback.py + # OR + $ python -u record_playback.py --interface (or) -i can0 # default vcan0 + ```
\ No newline at end of file diff --git a/Scripts/carla_to_CAN.py b/Scripts/carla_to_CAN.py new file mode 100644 index 0000000..951d29b --- /dev/null +++ b/Scripts/carla_to_CAN.py @@ -0,0 +1,257 @@ +# Copyright (C) 2024 Suchinton Chakravarty +# +# SPDX-License-Identifier: Apache-2.0 + +import carla +import math +import can +import cantools +import argparse + +# ============================================================================== +# -- CAN ----------------------------------------------------------------------- +# ============================================================================== + + +class CAN(object): + """ + Represents a Controller Area Network (CAN) interface for sending messages to a vehicle. + + Attributes: + db (cantools.database.can.Database): The CAN database. + can_bus (can.interface.Bus): The CAN bus interface. + speed_message (cantools.database.can.Message): The CAN message for vehicle speed. + gear_message (cantools.database.can.Message): The CAN message for transmission gear. + Vehicle_Status_2_message (cantools.database.can.Message): The CAN message for engine speed. + throttle_message (cantools.database.can.Message): The CAN message for throttle position. + Vehicle_Status_3_message (cantools.database.can.Message): The CAN message for indicator lights. + speed_cache (int): The cached vehicle speed. + throttle_cache (int): The cached throttle position. + engine_speed_cache (int): The cached engine RPM. + gear_cache (str): The cached transmission gear. + lights_cache (str): The cached indicator lights state. + """ + + def __init__(self): + self.db = cantools.database.load_file('agl-vcar.dbc') + self.can_bus = can.interface.Bus('vcan0', interface='socketcan') + + self.speed_message = self.db.get_message_by_name('Vehicle_Status_1') + self.gear_message = self.db.get_message_by_name('Transmission') + + self.Vehicle_Status_2_message = self.db.get_message_by_name( + 'Vehicle_Status_2') + self.throttle_message = self.db.get_message_by_name('Engine') + + self.Vehicle_Status_3_message = self.db.get_message_by_name( + 'Vehicle_Status_3') + + self.speed_cache = 0 + self.throttle_cache = 0 + self.engine_speed_cache = 0 + self.gear_cache = "P" + self.lights_cache = None + + def send_car_speed(self, speed): + """ + Sends the vehicle speed to the CAN bus. + + Args: + speed (int): The vehicle speed in km/h. + """ + if speed != 0: + self.send_gear("D") + else: + self.send_gear("P") + data = self.speed_message.encode({'PT_VehicleAvgSpeed': speed}) + message = can.Message( + arbitration_id=self.speed_message.frame_id, data=data) + self.can_bus.send(message) + + def send_gear(self, gear): + """ + Sends the transmission gear to the CAN bus. + + Args: + gear (str): The transmission gear ('P', 'R', 'N', 'D'). + Where, 0 = Neutral, 1/2/.. = Forward gears, -1/-2/.. = Reverse gears, 126 = Park, 127 = Drive + """ + if gear == "P": + data = self.gear_message.encode({'Gear': 126}) + elif gear == "R": + data = self.gear_message.encode({'Gear': -1, 'Gear': -2}) + elif gear == "N": + data = self.gear_message.encode({'Gear': 0}) + elif gear == "D": + data = self.gear_message.encode({'Gear': 127}) + message = can.Message( + arbitration_id=self.gear_message.frame_id, data=data) + self.can_bus.send(message) + + def send_engine_speed(self, engine_speed): + """ + Sends the engine speed to the CAN bus. + + Args: + engine_speed (int): The engine speed in RPM. + """ + data = self.Vehicle_Status_2_message.encode({'PT_FuelLevelPct': 100, + 'PT_EngineSpeed': engine_speed, + 'PT_FuelLevelLow': 0}) + + message = can.Message( + arbitration_id=self.Vehicle_Status_2_message.frame_id, data=data) + self.can_bus.send(message) + + def send_throttle(self, throttle): + """ + Sends the throttle position to the CAN bus. + + Args: + throttle (int): The throttle position in percentage. + """ + data = self.throttle_message.encode({'ThrottlePosition': throttle}) + message = can.Message( + arbitration_id=self.throttle_message.frame_id, data=data) + self.can_bus.send(message) + + def send_indicator(self, indicator): + """ + Sends the indicator lights state to the CAN bus. + + Args: + indicator (str): The indicator lights state ('LeftBlinker', 'RightBlinker', 'HazardLights'). + """ + # Mapping indicator names to signal values + indicators_mapping = { + 'LeftBlinker': {'PT_LeftTurnOn': 1}, + 'RightBlinker': {'PT_RightTurnOn': 1}, + 'HazardLights': {'PT_HazardOn': 1} + } + + # Default signal values + signals = {'PT_HazardOn': 0, 'PT_LeftTurnOn': 0, 'PT_RightTurnOn': 0} + + # Update signals based on the indicator argument + signals.update(indicators_mapping.get(indicator, {})) + + # Encode and send the CAN message + data = self.Vehicle_Status_3_message.encode(signals) + message = can.Message( + arbitration_id=self.Vehicle_Status_3_message.frame_id, data=data) + self.can_bus.send(message) + + def send_can_message(self, speed=0, rpm=0, throttle=0, gear="P", lights=None): + """ + Sends a complete set of CAN messages for vehicle control. + + Args: + speed (int): The vehicle speed in km/h. + rpm (int): The engine speed in RPM. + throttle (int): The throttle position in percentage. + gear (str): The transmission gear ('P', 'R', 'N', 'D'). + """ + if speed != self.speed_cache: + self.send_car_speed(speed) + self.speed_cache = speed + + if throttle != self.throttle_cache: + self.send_throttle(throttle) + self.throttle_cache = throttle + + if gear != self.gear_cache: + if gear == 1: + self.send_gear("D") + if gear == -1: + self.send_gear("R") + if gear == 0: + self.send_gear("N") + self.gear_cache = gear + + if rpm != self.engine_speed_cache: + self.send_engine_speed(rpm) + self.engine_speed_cache = rpm + + if lights is not None and lights != self.lights_cache: + self.send_indicator(lights) + self.lights_cache = lights + + +def main(host='127.0.0.1', port=2000): + parser = argparse.ArgumentParser(description='Carla to CAN Converter') + parser.add_argument('--host', default='127.0.0.1', help='IP of the host server') + parser.add_argument('--port', default=2000, type=int, help='TCP port to listen to') + args = parser.parse_args() + + client = carla.Client(args.host, args.port) + client.set_timeout(2.0) + + world = client.get_world() + + can = CAN() + + player_vehicle = None + + for actor in world.get_actors(): + if 'vehicle' in actor.type_id and actor.attributes['role_name'] == 'hero': + player_vehicle = actor + break + + if player_vehicle is None: + print("Player vehicle not found.") + return + + try: + + speed_kmh_cache = None + engine_rpm_cache = None + throttle_cache = None + gear_cache = None + lights_cache = None + + while True: + control = player_vehicle.get_control() + physics_control = player_vehicle.get_physics_control() + velocity = player_vehicle.get_velocity() + gear = player_vehicle.get_control().gear + speed_kmh = 3.6 * \ + math.sqrt(velocity.x**2 + velocity.y**2 + velocity.z**2) + + engine_rpm = physics_control.max_rpm * control.throttle + + if gear > 0: + gear_ratio = physics_control.forward_gears[min( + gear, len(physics_control.forward_gears)-1)].ratio + engine_rpm = physics_control.max_rpm * control.throttle / gear_ratio + else: + engine_rpm = physics_control.max_rpm * control.throttle + + lights = player_vehicle.get_light_state() + + # if any values have changed, try to send the CAN message + if (speed_kmh != speed_kmh_cache or + engine_rpm != engine_rpm_cache or + control.throttle != throttle_cache or + gear != gear_cache or + lights != lights_cache): + + speed_kmh_cache = speed_kmh + engine_rpm_cache = engine_rpm + throttle_cache = control.throttle + gear_cache = gear + lights_cache = lights + + can.send_can_message(speed_kmh, engine_rpm, + control.throttle, gear, lights) + + except Exception as e: + print( + f"An error occurred: {e}. The CARLA simulation might have stopped.") + finally: + if can.can_bus is not None: + can.can_bus.shutdown() + print("CAN bus properly shut down.") + + +if __name__ == "__main__": + main() diff --git a/Scripts/record_playback.py b/Scripts/record_playback.py new file mode 100644 index 0000000..e518356 --- /dev/null +++ b/Scripts/record_playback.py @@ -0,0 +1,141 @@ +# Copyright (C) 2024 Suchinton Chakravarty +# +# SPDX-License-Identifier: Apache-2.0 + +import can +import time +from rich.console import Console +import os +import argparse + +class CAN_playback: + def __init__(self, interface='vcan0'): + """ + Initialize the CAN Tool with the specified interface. + + Args: + interface (str): The CAN interface name (default: 'vcan0') + """ + self.console_mode = False + self.interface = interface + self.bus = can.interface.Bus(interface='socketcan', channel=self.interface, bitrate=500000) + self.output_file = os.path.join(os.path.dirname(__file__), 'can_messages.txt') + + def write_to_file(self, messages): + """ + Write captured CAN messages to a file. + + Args: + messages (list): List of can.Message objects to write + """ + with open(self.output_file, 'w') as file: + for msg in messages: + file.write(f"{msg.timestamp},{msg.arbitration_id},{msg.dlc},{','.join(map(lambda b: f'0x{b:02x}', msg.data))}\n") + + def playback_messages(self): + if os.path.exists(self.output_file): + #console.print("Replaying captured messages...") + messages = [] + with open(self.output_file, 'r') as file: + for line in file: + parts = line.strip().split(',', 3) # Split into at most 4 parts + timestamp, arbitration_id, dlc, data_str = parts + # Extract the data bytes, removing '0x' and splitting by ',' + data_bytes = [int(byte, 16) for byte in data_str.split(',') if byte] + + msg = can.Message( + timestamp=float(timestamp), + arbitration_id=int(arbitration_id, 0), + dlc=int(dlc), + data=data_bytes + ) + messages.append(msg) + self.replay_messages(messages) + + def stop(self): + self._running = False + if self.bus is not None: + self.bus.shutdown() + + def replay_messages(self, messages): + """ + Replay CAN messages on the specified bus. + + Args: + messages (list): List of can.Message objects to replay + """ + self._running = True + start_time = messages[0].timestamp + for msg in messages: + delay = msg.timestamp - start_time + self.bus.send(msg) + time.sleep(delay) + start_time = msg.timestamp + if self._running == False: return + + def capture_can_messages(self): + """ + Capture CAN messages from the specified bus. + + Returns: + list: List of captured can.Message objects + """ + messages = [] + + if self.console_mode: + console = Console() + console.print(f"Capturing CAN messages on {self.interface}. Press Ctrl+C to stop.") + + + try: + while True: + message = self.bus.recv() + if message is not None: + messages.append(message) + except KeyboardInterrupt: + console.print("Capture stopped.") + + return messages + +def main(): + from rich.console import Console + from rich.prompt import Prompt + + parser = argparse.ArgumentParser(description='CAN Message Capture and Replay Tool') + parser.add_argument('--interface', '-i', type=str, default='vcan0', help='Specify the CAN interface (default: vcan0)') + args = parser.parse_args() + + # Initialize the CAN Tool with the specified interface + can_tool = CAN_playback(interface=args.interface) + can_tool.console_mode = True + + console = Console() + while True: + console.print("\n[bold]CAN Message Capture and Replay[/bold]") + console.print("1. Capture CAN messages") + console.print("2. Replay captured messages") + console.print("3. Exit") + + choice = Prompt.ask("Enter your choice", choices=['1', '2', '3']) + + if choice == '1': + messages = can_tool.capture_can_messages() + console.print(f"Captured {len(messages)} messages.") + can_tool.write_to_file(messages) + console.print(f"CAN messages written to {can_tool.output_file}") + elif choice == '2': + if os.path.exists(can_tool.output_file): + console.print("Replaying captured messages...") + can_tool.playback_messages() + console.print("Replay completed.") + else: + console.print(f"No captured messages found in {can_tool.output_file}") + + + + else: + console.print("Exiting...") + break + +if __name__ == "__main__": + main() diff --git a/Scripts/requirements.txt b/Scripts/requirements.txt new file mode 100644 index 0000000..e6125b6 --- /dev/null +++ b/Scripts/requirements.txt @@ -0,0 +1,4 @@ +cantools +carla +python-can +rich
\ No newline at end of file diff --git a/Scripts/vcan.sh b/Scripts/vcan.sh new file mode 100755 index 0000000..dd85765 --- /dev/null +++ b/Scripts/vcan.sh @@ -0,0 +1,5 @@ +#!/bin/bash +sudo modprobe vcan +sudo ip link add dev vcan0 type vcan +sudo ip link set up vcan0 +echo Virtual CAN Bus has been opened!
\ No newline at end of file diff --git a/Widgets/ICPage.py b/Widgets/ICPage.py index f2e41a7..213e74c 100644 --- a/Widgets/ICPage.py +++ b/Widgets/ICPage.py @@ -11,6 +11,7 @@ from PyQt6.QtWidgets import QApplication from PyQt6.QtGui import QIcon, QPixmap, QPainter from PyQt6.QtCore import QObject, pyqtSignal from PyQt6.QtWidgets import QWidget +import threading current_dir = os.path.dirname(os.path.abspath(__file__)) @@ -23,8 +24,10 @@ Form, Base = uic.loadUiType(os.path.join(current_dir, "../ui/IC.ui")) # ======================================== +import extras.config as config from extras.KuksaClient import KuksaClient from extras.VehicleSimulator import VehicleSimulator +from Scripts.record_playback import CAN_playback import res_rc from Widgets.animatedToggle import AnimatedToggle @@ -263,24 +266,34 @@ class ICWidget(Base, Form): self.acceleration_timer.start(100) def handle_Script_toggle(self): - if self.Script_toggle.isChecked(): - self.Speed_slider.setEnabled(False) - self.RPM_slider.setEnabled(False) - self.accelerationBtn.setEnabled(False) - for button in self.driveGroupBtns.buttons(): - button.setEnabled(False) - self.set_Vehicle_RPM(1000) - self.set_Vehicle_Speed(0) - self.simulator_running = True - self.simulator.start() + if config.file_playback_enabled(): + can_tool = CAN_playback() + if self.Script_toggle.isChecked(): + can_tool_thread = threading.Thread( + target=can_tool.playback_messages) + can_tool_thread.start() + else: + can_tool.stop() + else: - self.simulator.stop() - self.simulator_running = False - self.Speed_slider.setEnabled(True) - self.RPM_slider.setEnabled(True) - self.accelerationBtn.setEnabled(True) - for button in self.driveGroupBtns.buttons(): - button.setEnabled(True) + if self.Script_toggle.isChecked(): + self.Speed_slider.setEnabled(False) + self.RPM_slider.setEnabled(False) + self.accelerationBtn.setEnabled(False) + for button in self.driveGroupBtns.buttons(): + button.setEnabled(False) + self.set_Vehicle_RPM(1000) + self.set_Vehicle_Speed(0) + self.simulator_running = True + self.simulator.start() + else: + self.simulator.stop() + self.simulator_running = False + self.Speed_slider.setEnabled(True) + self.RPM_slider.setEnabled(True) + self.accelerationBtn.setEnabled(True) + for button in self.driveGroupBtns.buttons(): + button.setEnabled(True) def updateSpeedAndEngineRpm(self, action, acceleration=(60/5)): if action == "Accelerate": diff --git a/extras/config.ini b/extras/config.ini index ec28639..a533578 100644 --- a/extras/config.ini +++ b/extras/config.ini @@ -2,6 +2,7 @@ fullscreen-mode = true hvac-enabled = true steering-wheel-enabled = true +file-playback-enabled = true [vss-server] ip = localhost diff --git a/extras/config.py b/extras/config.py index b1b1d7d..a0c60fd 100644 --- a/extras/config.py +++ b/extras/config.py @@ -151,6 +151,9 @@ def hvac_enabled(): def steering_wheel_enabled(): return config.getboolean('default', 'steering-wheel-enabled', fallback=True) +def file_playback_enabled(): + return config.getboolean('default', 'file-playback-enabled', fallback=True) + if not config.has_section('vss-server'): config.add_section('vss-server') |