Fix doc
This commit is contained in:
parent
6be663c914
commit
9d3aa54272
25 changed files with 405 additions and 26432 deletions
|
|
@ -1,86 +0,0 @@
|
|||
# CAN Communication Scripts
|
||||
|
||||
This repository contains Python scripts for testing and interacting with a CAN bus system. These scripts enable sending and receiving CAN messages to control a motor, set angles, and adjust velocities.
|
||||
|
||||
## Prerequisites
|
||||
|
||||
1. **Python 3.7+** installed on your system.
|
||||
2. **`python-can` library** installed. Install it via pip:
|
||||
```bash
|
||||
pip install python-can
|
||||
```
|
||||
3. **SocketCAN interface** properly configured on your Linux system. The default channel is `can0`.
|
||||
|
||||
## Usage
|
||||
|
||||
### 1. Receiving CAN Messages
|
||||
|
||||
The script `python_can.py` listens to the CAN bus and processes incoming messages.
|
||||
|
||||
#### Run:
|
||||
```bash
|
||||
python3 python_can.py
|
||||
```
|
||||
|
||||
#### Features:
|
||||
- Processes messages with data length 5.
|
||||
- Parses the first byte (`flag`) to determine the type:
|
||||
- `'A'`: Angle (float).
|
||||
- `'V'`: Velocity (float).
|
||||
- `'E'`: Enable/disable status (boolean).
|
||||
|
||||
### 2. Enabling or Disabling the Motor
|
||||
|
||||
The script `python_enable_motor.py` sends commands to enable or disable the motor.
|
||||
|
||||
#### Run:
|
||||
```bash
|
||||
python3 python_enable_motor.py <0|1>
|
||||
```
|
||||
|
||||
#### Arguments:
|
||||
- `0`: Disable the motor.
|
||||
- `1`: Enable the motor.
|
||||
|
||||
### 3. Sending Target Angle
|
||||
|
||||
The script `python_send_angle.py` sends a target angle to the CAN bus.
|
||||
|
||||
#### Run:
|
||||
```bash
|
||||
python3 python_send_angle.py
|
||||
```
|
||||
|
||||
#### Behavior:
|
||||
- Sends a message with a predefined target angle every second.
|
||||
- Adjust the target angle in the script (`target_angle` variable).
|
||||
|
||||
### 4. Sending Target Velocity
|
||||
|
||||
The script `python_send_velocity.py` sends a target velocity to the CAN bus.
|
||||
|
||||
#### Run:
|
||||
```bash
|
||||
python3 python_send_velocity.py
|
||||
```
|
||||
|
||||
#### Behavior:
|
||||
- Sends a message with a predefined target velocity every second.
|
||||
- Adjust the target velocity in the script (`target_speed` variable).
|
||||
|
||||
## Configuration
|
||||
|
||||
### CAN Interface
|
||||
The scripts use the following default CAN bus settings:
|
||||
- **Channel**: `can0`
|
||||
- **Bitrate**: `1 Mbps`
|
||||
|
||||
If your configuration differs, update the `Bus()` initialization in the scripts.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
1. **"Error initializing CAN bus"**:
|
||||
- Ensure your CAN interface is correctly configured and active:
|
||||
```bash
|
||||
sudo ip link set can0 up type can bitrate 1000000
|
||||
```
|
||||
141
controller/fw/bootloader/test/firmware_can.py
Normal file
141
controller/fw/bootloader/test/firmware_can.py
Normal file
|
|
@ -0,0 +1,141 @@
|
|||
import can
|
||||
import sys
|
||||
import time
|
||||
from intelhex import IntelHex
|
||||
# Конфигурация
|
||||
CAN_CHANNEL = 'socketcan'
|
||||
CAN_INTERFACE = 'can0'
|
||||
CAN_BITRATE = 1000000
|
||||
#ch =int(input("Введите id устройства:"))
|
||||
ch = int(sys.argv[2])
|
||||
BOOT_CAN_ID = (ch * 16) + 1
|
||||
DATA_CAN_ID = (ch * 16) + 3
|
||||
BOOT_CAN_END = (ch * 16) + 2
|
||||
ACK_CAN_ID = 0x05
|
||||
|
||||
#конфиг для crc16 ibm
|
||||
|
||||
|
||||
|
||||
def debug_print(msg):
|
||||
print(f"[DEBUG] {msg}")
|
||||
|
||||
def calculate_crc16(data: bytes) -> int:
|
||||
crc = 0xFFFF
|
||||
for byte in data:
|
||||
crc ^= byte
|
||||
for _ in range(8):
|
||||
if crc & 0x0001:
|
||||
crc = (crc >> 1) ^ 0xA001
|
||||
else:
|
||||
crc >>= 1
|
||||
return crc
|
||||
|
||||
def send_firmware(hex_file):
|
||||
try:
|
||||
debug_print("Инициализация CAN...")
|
||||
bus = can.interface.Bus(
|
||||
channel=CAN_INTERFACE,
|
||||
bustype=CAN_CHANNEL,
|
||||
bitrate=CAN_BITRATE
|
||||
)
|
||||
|
||||
debug_print("Чтение HEX-файла...")
|
||||
ih = IntelHex(hex_file)
|
||||
binary_data = ih.tobinstr() # Исправлено на tobinstr()
|
||||
fw_size = len(binary_data)
|
||||
debug_print(f"Размер прошивки: {fw_size} байт")
|
||||
|
||||
# Расчет CRC
|
||||
debug_print("Расчёт CRC...")
|
||||
# calculator = Calculator(Crc16.IBM)
|
||||
fw_crc = calculate_crc16(binary_data)
|
||||
debug_print(f"CRC: 0x{fw_crc:04X}")
|
||||
|
||||
# Отправка START
|
||||
start_data = bytearray([0x01])
|
||||
start_data += fw_size.to_bytes(4, 'little')
|
||||
start_data += fw_crc.to_bytes(2, 'little')
|
||||
|
||||
debug_print(f"START: {list(start_data)}")
|
||||
start_msg = can.Message(
|
||||
arbitration_id=BOOT_CAN_ID,
|
||||
data=bytes(start_data),
|
||||
is_extended_id=False
|
||||
)
|
||||
|
||||
try:
|
||||
bus.send(start_msg)
|
||||
except can.CanError as e:
|
||||
debug_print(f"Ошибка отправки START: {str(e)}")
|
||||
return
|
||||
|
||||
# Ожидание ACK
|
||||
debug_print("Ожидание ACK...")
|
||||
ack = wait_for_ack(bus)
|
||||
if not ack:
|
||||
debug_print("Таймаут ACK START")
|
||||
return
|
||||
debug_print(f"Получен ACK: {list(ack.data)}")
|
||||
|
||||
# Отправка данных
|
||||
packet_size = 8
|
||||
for i in range(0, len(binary_data), packet_size):
|
||||
chunk = binary_data[i:i+packet_size]
|
||||
# Дополнение до 8 байт
|
||||
if len(chunk) < 8:
|
||||
chunk += b'\xFF' * (8 - len(chunk))
|
||||
|
||||
debug_print(f"Пакет {i//8}: {list(chunk)}")
|
||||
data_msg = can.Message(
|
||||
arbitration_id=DATA_CAN_ID,
|
||||
data=chunk,
|
||||
is_extended_id=False
|
||||
)
|
||||
|
||||
try:
|
||||
bus.send(data_msg)
|
||||
except can.CanError as e:
|
||||
debug_print(f"Ошибка отправки данных: {str(e)}")
|
||||
return
|
||||
|
||||
ack = wait_for_ack(bus)
|
||||
if not ack:
|
||||
debug_print("Таймаут ACK DATA")
|
||||
return
|
||||
|
||||
# Финал
|
||||
debug_print("Отправка FINISH...")
|
||||
finish_msg = can.Message(
|
||||
arbitration_id=BOOT_CAN_END,
|
||||
data=bytes([0xAA]),
|
||||
is_extended_id=False
|
||||
)
|
||||
bus.send(finish_msg)
|
||||
|
||||
ack = wait_for_ack(bus, timeout=1.0)
|
||||
if ack and ack.data[0] == 0xAA:
|
||||
debug_print("Прошивка подтверждена!")
|
||||
else:
|
||||
debug_print("Ошибка верификации!")
|
||||
|
||||
except Exception as e:
|
||||
debug_print(f"Критическая ошибка: {str(e)}")
|
||||
finally:
|
||||
bus.shutdown()
|
||||
|
||||
def wait_for_ack(bus, timeout=1.0):
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < timeout:
|
||||
msg = bus.recv(timeout=0.1) # Неблокирующий режим
|
||||
if msg and msg.arbitration_id == ACK_CAN_ID:
|
||||
return msg
|
||||
return None
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
if len(sys.argv) != 3:
|
||||
print("Использование: sudo python3 can_flasher.py firmware.hex")
|
||||
sys.exit(1)
|
||||
|
||||
send_firmware(sys.argv[1])
|
||||
70
controller/fw/bootloader/test/firmware_update_flag.py
Normal file
70
controller/fw/bootloader/test/firmware_update_flag.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
import can
|
||||
import time
|
||||
import sys
|
||||
# Конфигурация
|
||||
CAN_INTERFACE = 'can0'
|
||||
OLD_DEVICE_ID = int(sys.argv[1]) # Текущий ID устройства (по умолчанию)
|
||||
REG_WRITE = 0x8 # Код команды чтения
|
||||
REG_ID = 0x55 # Адрес регистра с Firmware Update
|
||||
|
||||
def send_can_message(bus, can_id, data):
|
||||
"""Отправка CAN-сообщения"""
|
||||
try:
|
||||
msg = can.Message(
|
||||
arbitration_id=can_id,
|
||||
data=data,
|
||||
is_extended_id=False
|
||||
)
|
||||
bus.send(msg)
|
||||
print(f"[Отправка] CAN ID: 0x{can_id:03X}, Данные: {list(data)}")
|
||||
return True
|
||||
except can.CanError as e:
|
||||
print(f"Ошибка CAN: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def validate_crc16(data):
|
||||
"""Расчет CRC16 (MODBUS) для проверки целостности данных"""
|
||||
crc = 0xFFFF
|
||||
for byte in data:
|
||||
crc ^= byte
|
||||
for _ in range(8):
|
||||
if crc & 0x0001:
|
||||
crc = (crc >> 1) ^ 0xA001
|
||||
else:
|
||||
crc >>= 1
|
||||
return crc
|
||||
|
||||
# Инициализация CAN-интерфейса
|
||||
bus = can.interface.Bus(channel=CAN_INTERFACE, bustype='socketcan')
|
||||
|
||||
# ======= 1. Запрос текущего ID устройства =======
|
||||
|
||||
# Формируем CAN ID для чтения: (OLD_DEVICE_ID << 4) | REG_READ
|
||||
can_id_read = (OLD_DEVICE_ID << 4) | REG_WRITE
|
||||
|
||||
# Данные для запроса: [регистр, резервный байт]
|
||||
data_read = [REG_ID, 0x00]
|
||||
|
||||
# Формируем полные данные для расчета CRC:
|
||||
# - CAN ID разбивается на 2 байта (little-endian)
|
||||
# - Добавляем данные запроса
|
||||
full_data_for_crc = list(can_id_read.to_bytes(2, 'little')) + data_read
|
||||
|
||||
# Рассчитываем CRC и разбиваем на байты (little-endian)
|
||||
crc = validate_crc16(full_data_for_crc)
|
||||
crc_bytes = list(crc.to_bytes(2, 'little'))
|
||||
|
||||
# Собираем итоговый пакет: данные + CRC
|
||||
packet_read = data_read + crc_bytes
|
||||
|
||||
print("Переход в boot режим", packet_read)
|
||||
send_can_message(bus, can_id_read, packet_read)
|
||||
|
||||
bus.shutdown()
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
if len(sys.argv) != 2:
|
||||
print("Использование: python3 firmware_test.py address")
|
||||
sys.exit(1)
|
||||
|
|
@ -1,47 +0,0 @@
|
|||
import can
|
||||
import struct
|
||||
import time
|
||||
|
||||
def process_can_message(msg):
|
||||
if msg.dlc == 5: # Check the message length
|
||||
print(f"Received message with ID: {msg.arbitration_id}")
|
||||
print(f"Data: {msg.data}")
|
||||
|
||||
# The first byte determines the data type (flag)
|
||||
flag = chr(msg.data[0])
|
||||
|
||||
if flag == 'A': # Angle
|
||||
angle_bytes = msg.data[1:5]
|
||||
angle = struct.unpack('<f', bytes(angle_bytes))[0]
|
||||
print(f"Angle: {angle} degrees")
|
||||
elif flag == 'V': # Velocity
|
||||
velocity_bytes = msg.data[1:5]
|
||||
velocity = struct.unpack('<f', bytes(velocity_bytes))[0]
|
||||
print(f"Velocity: {velocity} rad/s")
|
||||
elif flag == 'E' and msg.dlc >= 2: # Enable/Disable
|
||||
enabled = msg.data[1] # Expecting 1 byte (0 or 1)
|
||||
print(f"Enabled: {bool(enabled)}")
|
||||
else:
|
||||
print(f"Unknown flag: {flag}")
|
||||
else:
|
||||
print(f"Received message with unexpected length: {msg.dlc}")
|
||||
|
||||
def receive_can_messages():
|
||||
try:
|
||||
# Connect to the CAN bus
|
||||
bus = can.interface.Bus(channel='can0', bustype='socketcan')
|
||||
|
||||
print("Waiting for messages on the CAN bus...")
|
||||
|
||||
while True:
|
||||
msg = bus.recv()
|
||||
if msg:
|
||||
process_can_message(msg)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\nExiting program...")
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
receive_can_messages()
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
import can
|
||||
import sys
|
||||
|
||||
# Function to send the motor enable/disable command
|
||||
def send_motor_enable(bus, enable):
|
||||
"""
|
||||
Sends a command to enable or disable the motor.
|
||||
|
||||
:param bus: The CAN bus
|
||||
:param enable: 1 to enable the motor, 0 to disable it
|
||||
"""
|
||||
msg = can.Message()
|
||||
msg.arbitration_id = 1 # Message ID
|
||||
msg.is_extended_id = False
|
||||
msg.dlc = 2 # Message length (flag + 1 byte of data)
|
||||
msg.data = [ord('E'), enable] # 'E' for the command, followed by 0 or 1
|
||||
|
||||
try:
|
||||
bus.send(msg)
|
||||
state = "enabled" if enable else "disabled"
|
||||
print(f"Sent message to {state} motor")
|
||||
print(f"Message data: {msg.data}")
|
||||
except can.CanError as e:
|
||||
print(f"Message failed to send: {e}")
|
||||
sys.exit(1) # Exit the program on failure
|
||||
|
||||
def main():
|
||||
# CAN interface setup
|
||||
bus = None # Define outside the try block for proper shutdown
|
||||
try:
|
||||
bus = can.interface.Bus(channel='can0', bustype='socketcan', bitrate=1000000) # Ensure the bitrate matches the microcontroller settings
|
||||
print("CAN bus initialized.")
|
||||
|
||||
# Ensure the state is passed via arguments
|
||||
if len(sys.argv) != 2 or sys.argv[1] not in ['0', '1']:
|
||||
print("Usage: python3 script_name.py <0|1>")
|
||||
print("0 - Disable motor, 1 - Enable motor")
|
||||
sys.exit(1)
|
||||
|
||||
enable = int(sys.argv[1])
|
||||
send_motor_enable(bus, enable)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error initializing CAN bus: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
finally:
|
||||
# Ensure the bus is properly shut down
|
||||
if bus is not None:
|
||||
bus.shutdown()
|
||||
print("CAN bus shut down.")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
|
@ -1,37 +0,0 @@
|
|||
import can
|
||||
import struct
|
||||
import time
|
||||
import argparse
|
||||
|
||||
# Function to send the target angle
|
||||
def send_target_angle(bus, target_angle):
|
||||
msg = can.Message()
|
||||
msg.arbitration_id = 1 # Message ID
|
||||
msg.is_extended_id = False
|
||||
msg.dlc = 5 # Message length
|
||||
msg.data = [ord('A')] + list(struct.pack('<f', target_angle)) # 'A' for the command identifier, followed by the angle in float format
|
||||
|
||||
try:
|
||||
bus.send(msg)
|
||||
print(f"Sent message with target angle: {target_angle} degrees")
|
||||
print(f"Message data: {msg.data}")
|
||||
except can.CanError:
|
||||
print("Message failed to send")
|
||||
|
||||
# Main function
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Send target angles over CAN bus.")
|
||||
parser.add_argument("--angle", type=float, required=True, help="Target angle to send over the CAN bus")
|
||||
args = parser.parse_args()
|
||||
|
||||
target_angle = args.angle
|
||||
|
||||
# CAN interface setup
|
||||
bus = can.interface.Bus(channel='can0', bustype='socketcan', bitrate=1000000) # Ensure the bitrate matches the microcontroller settings
|
||||
print("CAN bus initialized, sending target angles...")
|
||||
|
||||
# Loop to send messages
|
||||
send_target_angle(bus, target_angle)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
|
@ -1,76 +0,0 @@
|
|||
import can
|
||||
import struct
|
||||
import time
|
||||
import sys
|
||||
|
||||
# Function to send the target speed
|
||||
def send_target_speed(bus, target_speed):
|
||||
msg = can.Message()
|
||||
msg.arbitration_id = 1 # Message ID
|
||||
msg.is_extended_id = False
|
||||
msg.dlc = 5 # Message length
|
||||
msg.data = bytearray([ord('V')] + list(struct.pack('<f', target_speed))) # 'V' for the command identifier, followed by the speed in float format
|
||||
|
||||
try:
|
||||
bus.send(msg)
|
||||
print(f"Sent message with target speed: {target_speed} rad/s")
|
||||
except can.CanError:
|
||||
print("Message failed to send")
|
||||
|
||||
# Function to send the motor enable/disable command
|
||||
def send_motor_enable(bus, enable):
|
||||
"""
|
||||
Sends a command to enable or disable the motor.
|
||||
|
||||
:param bus: The CAN bus
|
||||
:param enable: 1 to enable the motor, 0 to disable it
|
||||
"""
|
||||
msg = can.Message()
|
||||
msg.arbitration_id = 1 # Message ID
|
||||
msg.is_extended_id = False
|
||||
msg.dlc = 2 # Message length (flag + 1 byte of data)
|
||||
msg.data = bytearray([ord('E'), enable]) # 'E' for the command, followed by 0 or 1
|
||||
|
||||
try:
|
||||
bus.send(msg)
|
||||
state = "enabled" if enable else "disabled"
|
||||
print(f"Sent message to {state} motor")
|
||||
except can.CanError as e:
|
||||
print(f"Message failed to send: {e}")
|
||||
sys.exit(1) # Exit the program on failure
|
||||
|
||||
send_target_speed(bus,0.0)
|
||||
|
||||
def main():
|
||||
# CAN interface setup
|
||||
bus = None # Define outside the try block for proper shutdown
|
||||
try:
|
||||
bus = can.interface.Bus(channel='COM4', bustype='slcan', bitrate=1000000) # Ensure the bitrate matches the microcontroller settings
|
||||
print("CAN bus initialized.")
|
||||
|
||||
while True:
|
||||
user_input = input("Enter target speed: ")
|
||||
if user_input.lower() == 'exit':
|
||||
print("Exiting...")
|
||||
break
|
||||
try:
|
||||
target_speed = float(user_input)
|
||||
send_target_speed(bus, target_speed)
|
||||
except ValueError:
|
||||
print("Invalid input. Please enter a valid number.")
|
||||
|
||||
# Disable motor before exiting
|
||||
send_motor_enable(bus, 0)
|
||||
print("Motor disabled.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error initializing1 CAN bus: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
finally:
|
||||
if bus is not None:
|
||||
bus.shutdown()
|
||||
print("CAN bus shut down.")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
|
@ -1,73 +0,0 @@
|
|||
import can
|
||||
import time
|
||||
|
||||
def send_write_read_requests():
|
||||
try:
|
||||
bus = can.interface.Bus(channel='can0', bustype='socketcan')
|
||||
|
||||
# Конфигурация сообщений (ЗАПОЛНИТЕ ВАШИ ЗНАЧЕНИЯ)
|
||||
write_msg = {
|
||||
'arbitration_id': 0x01, # CAN ID для записи
|
||||
'data': [0x27, 0xA0, 0xFF, 0x00], # Данные для записи (4 байта)
|
||||
'description': "Установка id устройства"
|
||||
}
|
||||
|
||||
read_msg = {
|
||||
'arbitration_id': 0x01, # CAN ID для чтения
|
||||
'data': [0xFF,0x99], # Адрес новый + команда запроса данных
|
||||
'description': "Запрос id устройства",
|
||||
'response_id': 0xFF, # Ожидаемый ID ответа
|
||||
'timeout': 1.0 # Таймаут ожидания ответа (сек)
|
||||
}
|
||||
|
||||
# 1. Отправка команды записи
|
||||
print("Отправка команды записи...")
|
||||
msg = can.Message(
|
||||
arbitration_id=write_msg['arbitration_id'],
|
||||
data=write_msg['data'],
|
||||
is_extended_id=False
|
||||
)
|
||||
bus.send(msg)
|
||||
print(f"Запись: ID={hex(msg.arbitration_id)}, Данные={list(msg.data)}")
|
||||
|
||||
# Ждем обработки команды устройством
|
||||
time.sleep(2.0)
|
||||
|
||||
# 2. Отправка запроса чтения и ожидание ответа
|
||||
print("\nОтправка запроса чтения...")
|
||||
msg = can.Message(
|
||||
arbitration_id=read_msg['arbitration_id'],
|
||||
data=read_msg['data'],
|
||||
is_extended_id=False
|
||||
)
|
||||
bus.send(msg)
|
||||
print(f"Чтение: ID={hex(msg.arbitration_id)}, Команда={list(msg.data)}")
|
||||
|
||||
# Ожидаем ответа
|
||||
start_time = time.time()
|
||||
response_received = False
|
||||
|
||||
print("\nОжидание ответа...")
|
||||
while (time.time() - start_time) < read_msg['timeout']:
|
||||
response = bus.recv(timeout=0.1)
|
||||
|
||||
if response and response.arbitration_id == read_msg['response_id']:
|
||||
print(f"\nПолучен ответ: ID={hex(response.arbitration_id)}")
|
||||
print(f"Данные: {list(response.data)}")
|
||||
print(f"Длина: {response.dlc} байт")
|
||||
response_received = True
|
||||
break
|
||||
|
||||
if not response_received:
|
||||
print("\nОшибка: ответ не получен в течение заданного времени")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\nПрерывание пользователем")
|
||||
except Exception as e:
|
||||
print(f"Ошибка: {str(e)}")
|
||||
finally:
|
||||
bus.shutdown()
|
||||
print("\nCAN соединение закрыто")
|
||||
|
||||
if __name__ == '__main__':
|
||||
send_write_read_requests()
|
||||
|
|
@ -1,35 +0,0 @@
|
|||
import can
|
||||
import struct
|
||||
import time
|
||||
|
||||
# Function to send the target speed
|
||||
def send_target_speed(bus, target_speed):
|
||||
msg = can.Message()
|
||||
msg.arbitration_id = 1 # Message ID
|
||||
msg.is_extended_id = False
|
||||
msg.dlc = 5 # Message length
|
||||
msg.data = [ord('V')] + list(struct.pack('<f', target_speed)) # 'V' for the command identifier, followed by the speed in float format
|
||||
|
||||
try:
|
||||
bus.send(msg)
|
||||
print(f"Sent message with target speed: {target_speed} m/s")
|
||||
print(f"Message data: {msg.data}")
|
||||
except can.CanError:
|
||||
print("Message failed to send")
|
||||
|
||||
# Main function
|
||||
def main():
|
||||
# CAN interface setup
|
||||
bus = can.interface.Bus(channel='can0', bustype='socketcan', bitrate=1000000) # Ensure the bitrate matches the microcontroller settings
|
||||
print("CAN bus initialized, sending target speed impulses...")
|
||||
|
||||
# Send impulses of target speed from -2 to 2 m/s
|
||||
target_speeds = [-1, 1]
|
||||
|
||||
while True:
|
||||
for speed in target_speeds:
|
||||
send_target_speed(bus, speed)
|
||||
time.sleep(1) # 1-second delay between messages
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
78
controller/fw/bootloader/test/st-link.py
Normal file
78
controller/fw/bootloader/test/st-link.py
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
import subprocess
|
||||
import os
|
||||
import sys
|
||||
|
||||
def flash_hex_with_stlink(hex_file_path):
|
||||
if not os.path.isfile(hex_file_path):
|
||||
print(f"❌ Файл не найден: {hex_file_path}")
|
||||
return False
|
||||
|
||||
command = [
|
||||
"st-flash",
|
||||
"--format", "ihex",
|
||||
"write",
|
||||
hex_file_path
|
||||
]
|
||||
|
||||
try:
|
||||
print(f"⚡️ Прошиваем {hex_file_path} через ST-Link...")
|
||||
result = subprocess.run(
|
||||
command,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
universal_newlines=True,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
print("▬▬▬ STDOUT ▬▬▬")
|
||||
print(result.stdout)
|
||||
|
||||
print("▬▬▬ STDERR ▬▬▬")
|
||||
print(result.stderr)
|
||||
|
||||
if result.returncode == 0:
|
||||
print("✅ Прошивка успешно завершена!")
|
||||
|
||||
# Добавленный блок сброса
|
||||
try:
|
||||
print("🔄 Выполняем сброс устройства...")
|
||||
reset_result = subprocess.run(
|
||||
["st-info", "--reset"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
universal_newlines=True,
|
||||
timeout=10
|
||||
)
|
||||
if reset_result.returncode == 0:
|
||||
print("♻️ Устройство успешно сброшено!")
|
||||
else:
|
||||
print(f"⚠️ Ошибка (код: {reset_result.returncode})")
|
||||
print("▬▬▬ STDERR сброса ▬▬▬")
|
||||
print(reset_result.stderr)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Ошибка при сбросе: {str(e)}")
|
||||
|
||||
return True
|
||||
else:
|
||||
print(f"❌ Ошибка прошивки (код: {result.returncode})")
|
||||
return False
|
||||
|
||||
except FileNotFoundError:
|
||||
print("❌ st-flash не найден! Установите stlink-tools.")
|
||||
return False
|
||||
except subprocess.TimeoutExpired:
|
||||
print("❌ Таймаут операции! Проверьте подключение ST-Link.")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ Неизвестная ошибка: {str(e)}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) != 2:
|
||||
print("Использование: python stlink_flash.py <firmware.hex>")
|
||||
sys.exit(1)
|
||||
|
||||
if flash_hex_with_stlink(sys.argv[1]):
|
||||
sys.exit(0)
|
||||
else:
|
||||
sys.exit(1)
|
||||
100
controller/fw/bootloader/test/st-link_full.py
Normal file
100
controller/fw/bootloader/test/st-link_full.py
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
import subprocess
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
def flash_hex_with_stlink(hex_file_path, component_name):
|
||||
if not os.path.isfile(hex_file_path):
|
||||
print(f"❌ Файл {component_name} не найден: {hex_file_path}")
|
||||
return False
|
||||
|
||||
command = [
|
||||
"st-flash",
|
||||
"--format", "ihex",
|
||||
"write",
|
||||
hex_file_path
|
||||
]
|
||||
|
||||
try:
|
||||
print(f"⚡️ Прошиваем {component_name} ({hex_file_path}) через ST-Link...")
|
||||
result = subprocess.run(
|
||||
command,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
universal_newlines=True,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
print("▬▬▬ STDOUT ▬▬▬")
|
||||
print(result.stdout)
|
||||
|
||||
print("▬▬▬ STDERR ▬▬▬")
|
||||
print(result.stderr)
|
||||
|
||||
if result.returncode == 0:
|
||||
print(f"✅ {component_name} успешно прошит!")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ Ошибка прошивки {component_name} (код: {result.returncode})")
|
||||
return False
|
||||
|
||||
except FileNotFoundError:
|
||||
print("❌ st-flash не найден! Установите stlink-tools.")
|
||||
return False
|
||||
except subprocess.TimeoutExpired:
|
||||
print(f"❌ Таймаут операции при прошивке {component_name}! Проверьте подключение ST-Link.")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ Неизвестная ошибка при прошивке {component_name}: {str(e)}")
|
||||
return False
|
||||
|
||||
def reset_device():
|
||||
try:
|
||||
print("🔄 Выполняем сброс(перезагрузку) устройства...")
|
||||
reset_result = subprocess.run(
|
||||
["st-info", "--reset"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
universal_newlines=True,
|
||||
timeout=10
|
||||
)
|
||||
if reset_result.returncode == 0:
|
||||
print("♻️ Устройство успешно сброшено!")
|
||||
return True
|
||||
else:
|
||||
print(f"⚠️ Ошибка при сбросе (код: {reset_result.returncode})")
|
||||
print("▬▬▬ STDERR сброса ▬▬▬")
|
||||
print(reset_result.stderr)
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"⚠️ Ошибка при сбросе: {str(e)}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) != 3:
|
||||
print("Использование: python stlink_flash.py <bootloader.hex> <application.hex>")
|
||||
print("Пример: python stlink_flash.py bootloader.hex firmware.hex")
|
||||
sys.exit(1)
|
||||
|
||||
bootloader_path = sys.argv[1]
|
||||
app_path = sys.argv[2]
|
||||
|
||||
# Прошиваем сначала бутлоадер
|
||||
if not flash_hex_with_stlink(bootloader_path, "Bootloader"):
|
||||
print("\n💥 Ошибка прошивки бутлоадера!")
|
||||
sys.exit(1)
|
||||
|
||||
# Сбрасываем устройство после прошивки бутлоадера
|
||||
reset_device()
|
||||
time.sleep(1) # Короткая пауза
|
||||
|
||||
# Прошиваем основное приложение
|
||||
if not flash_hex_with_stlink(app_path, "Application"):
|
||||
print("\n💥 Ошибка прошивки основного приложения!")
|
||||
sys.exit(1)
|
||||
|
||||
# Финальный сброс устройства
|
||||
reset_device()
|
||||
|
||||
print("\n🎉 Все компоненты успешно прошиты!")
|
||||
sys.exit(0)
|
||||
Loading…
Add table
Add a link
Reference in a new issue