CAN driver for Geschwister Schneider USB/CAN devices.
Support multichannel and can fd.
pip install python-can-candle
If your system does not have any usb backend installed, you can borrow a libusb1 backend from libusb.
pip install python-can-candle[libusb]
We also provide a handy GUI tool, which you can install with the following command.
pip install python-can-candle[viewer]
Once installed you can launch it via candle_viewer
or python -m candle.candle_viewer
and you will see the following window.
(unstable) Installing a backend written in C can improve performance.
pip install "candle_api @ git+https://github.com/BIRLab/candle_api.git@main"
This library implements the plugin interface in python-can, aiming to replace the gs_usb interface within it.
import can
from candle import CandleBus
bus: CandleBus # This line is added to provide type hints.
# Create a CandleBus instance in the python-can API.
with can.Bus(interface='candle', channel=0, fd=True, bitrate=1000000, data_bitrate=5000000) as bus:
# Note that bus is an instance of CandleBus.
assert isinstance(bus, CandleBus)
# Send normal can message without data.
bus.send(can.Message(arbitration_id=1, is_extended_id=False))
# Send normal can message with extended id
bus.send(can.Message(arbitration_id=2, is_extended_id=True))
# Send normal can message with data.
bus.send(can.Message(arbitration_id=3, is_extended_id=False, data=[i for i in range(8)]))
# Send can fd message.
if bus.is_fd_supported:
bus.send(can.Message(arbitration_id=4, is_extended_id=False, is_fd=True, bitrate_switch=True,
error_state_indicator=True, data=[i for i in range(64)]))
# Read messages from bus.
for message in bus:
print(message)
Using the API directly can be very cumbersome. However, we still provide a simple example for developers to refer to.
from candle.candle_api import CandleDevice, GSHostFrame, GSHostFrameHeader, GSCANFlag, GSCANIDFlag
# Scan available devices.
available_devices = list(CandleDevice.scan())
# Select a device.
for i, device in enumerate(available_devices):
print(f'{i}: {device}')
device = available_devices[int(input('Select a device by index: '))]
# Select a interface.
# Only single interface devices are supported currently.
interface = device[0]
# Select a channel.
for i in range(len(interface)):
print(f'{i}: channel {i}')
channel = interface[int(input(f'Select a channel by index: '))]
# Set bit timing.
# channel.set_bit_timing(...)
# channel.set_data_bit_timing(...)
# Open the channel.
channel.open(fd=channel.is_fd_supported)
# Send a normal frame.
channel.write(
GSHostFrame(
header=GSHostFrameHeader(
echo_id=0,
can_id=1,
can_dlc=8,
channel=channel.index,
flags=GSCANFlag(0)
),
data=bytes([i for i in range(8)])
)
)
# Send a can fd frame with extended id.
channel.write(
GSHostFrame(
header=GSHostFrameHeader(
echo_id=0,
can_id=1 | GSCANIDFlag.EFF,
can_dlc=15,
channel=channel.index,
flags=GSCANFlag.FD | GSCANFlag.BRS | GSCANFlag.ESI
),
data=bytes([i for i in range(64)])
)
)
try:
while True:
# Polling message.
device.polling()
# Receive a frame.
frame = channel.read()
if frame is not None:
print(frame.data)
except KeyboardInterrupt:
pass
# Close the channel.
channel.close()