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
$ 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
--- /dev/null
+## 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
--- /dev/null
+# 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()
--- /dev/null
+# 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()
--- /dev/null
+cantools
+carla
+python-can
+rich
\ No newline at end of file
--- /dev/null
+#!/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
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__))
# ========================================
+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
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":
fullscreen-mode = true
hvac-enabled = true
steering-wheel-enabled = true
+file-playback-enabled = true
[vss-server]
ip = localhost
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')
Code Review / src / agl-demo-control-panel.git / commitdiff