I'll get those to you tomorrow. Unfortunately, the Jeep dealer held my 4xe since Thursday to repair a bad A/C Compressor (Module). They called yesterday to confirm a Monday pickup. Then I can pull those off of my current Pi.
Thanks! And when you've got your vehicle back, if you could, please let me know in raspi-config if (under advanced settings) you have predictable network names enabled or not. I think that they should be enabled so that can0 and can1 don't get swapped?
I know you know already, but please be very, very, careful with generative AI, especially for something that has write access to your vehicle's CAN bus.
I meant to respond to your post and @jmccorm above.
#!/usr/bin/python3
import udsoncan
with udsoncan.Connection('can0', txid=0x7BF, rxid=0x53F) as conn:
request = udsoncan.Request(service=udsoncan.services.ECUReset)
request.data = b'\x01'
response = conn.send(request)#!/usr/bin/python3
import can
import time
# define the bus
bus = can.interface.Bus(channel='can0', bustype='socketcan_native')
# sync and sleep
bus.send(can.Message(arbitration_id=0x314, data=[0x5A, 0x03, 0x04]))
time.sleep(0.1)
bus.send(can.Message(arbitration_id=0x314, data=[0x5A, 0x03, 0x02]))
time.sleep(0.1)
# sync and sleep
bus.send(can.Message(arbitration_id=0x314, data=[0x59, 0x03, 0x04]))
time.sleep(0.1)
bus.send(can.Message(arbitration_id=0x314, data=[0x59, 0x03, 0x02]))#!/usr/bin/python3
import can
import isotp
import udsoncan
import time
# Set up the CAN buses
can0 = can.interface.Bus(channel='can0', bustype='socketcan_native')
can1 = can.interface.Bus(channel='can1', bustype='socketcan_native')
# Send the CAN message to wake up the bus
msg = can.Message(arbitration_id=0x2D3, data=[0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00], is_extended_id=False)
can0.send(msg)
time.sleep(0.2)
# Set up the ISO-TP socket to communicate over the second CAN bus
tp_sock = isotp.socket()
tp_sock.set_fc_opts(stmin=0x01, bs=0x10)
tp_sock.bind(interface=can1, rxid=0x504, txid=0x620)
# Set up the UDS client and enter into an extended diagnostic session
uds_client = udsoncan.UDSClient(tp_sock)
uds_client.change_session(udsoncan.Session.ExtendedDiagnosticSession)
# Honk the horn three times
for i in range(3):
# Turn the horn on
uds_client.io_control(0xd0ad, b"\x03\x01", timeout=0.1)
time.sleep(0.01)
# Turn the horn off
uds_client.io_control(0xd0ad, b"\x03\x00", timeout=0.1)
time.sleep(0.025)#!/usr/bin/python3
import can
import time
import isotp
import udsoncan
# Set up the CAN buses
#can0 = can.interface.Bus(channel='can0', bustype='socketcan_native')
#can1 = can.interface.Bus(channel='can1', bustype='socketcan_native')
can0 = can.interface.Bus(channel='can0', bustype='socketcan')
can1 = can.interface.Bus(channel='can1', bustype='socketcan')
# Send the CAN message to wake up the bus
msg = can.Message(arbitration_id=0x2D3, data=[0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00], is_extended_id=False)
can0.send(msg)
time.sleep(0.1)
# Set up the ISO-TP socket to communicate over the second CAN bus
tp_sock = isotp.socket()
tp_sock.set_fc_opts(stmin=0x01, bs=0x10)
tp_sock.bind(interface=can1, rxid=0x504, txid=0x620)
# Set up the UDS client and enter into an extended diagnostic session
uds_client = udsoncan.UDSClient(tp_sock)
uds_client.change_session(udsoncan.Session.ExtendedDiagnosticSession)
time.sleep(0.1)
# Honk the horn three times
for i in range(3):
uds_client.io_control(0xd0ad, b"\x03\x01", timeout=0.1)
time.sleep(0.015)
uds_client.io_control(0xd0ad, b"\x03\x00", timeout=0.1)
time.sleep(0.035)
# Send an extra OFF just in case the previous one gets lost
uds_client.io_control(0xd0ad, b"\x03\x00", timeout=0.5)
time.sleep(0.5)#!/usr/bin/env python3
import random
import can
import time
# create CAN bus object
can0_bus = can.interface.Bus(bustype='socketcan', channel='can0')
# generate a random number between 0 and 3 to simulate a button press
fake_button = random.randint(0, 3)
# create a message to wake up the bus
wake_message = can.Message(
arbitration_id=0x2D3,
data=[0x07, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 + fake_button],
extended_id=False
)
# send the wake-up message
can0_bus.send(wake_message)
# Sleep for 0.1 seconds
time.sleep(0.1)#!/usr/bin/python3
import isotp
import logging
import udsoncan
# Set up logging
logging.basicConfig(level=logging.DEBUG)
# Define constants
CAN_INTERFACE = 'can0'
CAN_TX_ADDRESS = 0x7BF
CAN_RX_ADDRESS = 0x53F
# Initialize ISO-TP stack
isotp_params = {
'stmin': 0, # minimum separation time between consecutive frames (ms)
'blocksize': 1, # number of consecutive frames per block
'wftmax': 0, # maximum number of wait frame transmissions
'll_data_length': 8 # length of each CAN message payload
}
tp = isotp.TransportLayer(CAN_INTERFACE, rxid=CAN_TX_ADDRESS, txid=CAN_RX_ADDRESS, params=isotp_params)
# Initialize UDS client
with udsoncan.UDSClient(tp) as client:
# Send ECU reset request with data of 01
response = client.ecu_reset(udsoncan.DidLengthFormat.LONG, b'\x01')
# Check response status
if response.positive:
print('ECU reset successful')
else:
print(f'ECU reset failed with negative response: {response.negative_response.service} - {response.negative_response.code}')I haven't enabled write access or the can C network, so I can't test your code, but if you post the error message, I might be able to help figure out the cause anyway.
#!/usr/bin/env python3
import can
# Set up the CAN bus
bus = can.interface.Bus(channel='can1', bustype='socketcan')
# Set the ID we want to filter for
CAN_ID = 0x023
# Infinite loop to listen for messages
while True:
message = bus.recv()
if message.arbitration_id == CAN_ID:
# Get the value of the first word
value = (message.data[0] << 8) + message.data[1]
print(f"Received message with ID 0x{CAN_ID:x}. First word value is 0x{value:04x}")#!/usr/bin/python3
import can
# Initialize the CAN bus
bus = can.interface.Bus(bustype='socketcan', channel='can1')
# Set the ID to filter for
CAN_ID = 0x127
# Define the message handler function
def handle_message(message):
if message.arbitration_id == CAN_ID:
data = message.data
print("Intake Air Temperature:", data[0])
print("Engine Coolant Temperature:", data[1])
print("Barometric Pressure:", data[2])
# Set up the message filter
can_filter = {"can_id": CAN_ID, "can_mask": 0x7FF}
# Add the filter to the bus
bus.set_filters([can_filter])
# Run the loop
while True:
message = bus.recv()
handle_message(message)#!/usr/bin/python3
import can
params = [
("Intake Air Temperature", "can1", 0x127, 0, 1, "decimal"),
("Engine Coolant Temperature", "can1", 0x127, 1, 1, "decimal"),
("Barometric Pressure", "can1", 0x127, 2, 1, "decimal")
]
bus = can.interface.Bus(bustype='socketcan', channel='can1')
can_ids = list(set([p[2] for p in params]))
filters = [can.Filter(can_id=id, can_mask=0x7FF) for id in can_ids]
msgs = [can.Message(arbitration_id=p[2], is_extended_id=False) for p in params]
bus.set_filters(filters)
while True:
recv_msg = bus.recv()
for i, msg in enumerate(msgs):
if recv_msg.arbitration_id == msg.arbitration_id:
data = recv_msg.data[p[3]:p[3]+p[4]]
if p[5] == "decimal":
value = int.from_bytes(data, byteorder='big')
print(params[i][0], ":", value)#!/usr/bin/env python3
import can
import time
params = [
("Steering Angle", "half degrees", "can1", 0x023, 0, 2, "decimal", "-4096"),
("Intake Air Temperature", "C", "can1", 0x127, 0, 1, "decimal", "*1"),
("Engine Coolant Temperature", "C", "can1", 0x127, 1, 1, "decimal", "*1"),
("Barometric Pressure", "kPa", "can1", 0x127, 2, 1, "decimal", "*1"),
("Engine RPM", "rpm", "can1", 0x0B5, 0, 2, "decimal", "*1"),
("Vehicle Speed", "MPH", "can1", 0x1F7, 6, 2, "decimal", "*1"),
("Ignition Status", "hex", "can0", 0x122, 0, 2, "hexadecimal", "*1"),
("Aux Battery", "volts", "can0", 0x2C2, 2, 1, "decimal", "/10")
]
can_bus1 = can.interface.Bus(bustype='socketcan', channel='can1')
can_bus0 = can.interface.Bus(bustype='socketcan', channel='can0')
can_ids1 = list(set([p[3] for p in params if p[2] == "can1"]))
can_ids0 = list(set([p[3] for p in params if p[2] == "can0"]))
filters1 = [{"can_id": id, "can_mask": 0x7FF} for id in can_ids1]
filters0 = [{"can_id": id, "can_mask": 0x7FF} for id in can_ids0]
can_bus1.set_filters(filters1)
can_bus0.set_filters(filters0)
while True:
data_dict = {}
start_time = time.time()
while time.time() - start_time < 1:
recv_msg = can_bus1.recv() or can_bus0.recv()
for param in params:
if recv_msg.arbitration_id == param[3] and recv_msg.channel == param[2]:
data = recv_msg.data[param[4]:param[4]+param[5]]
if param[6] == "decimal":
value = int.from_bytes(data, byteorder='big', signed=True)
elif param[6] == "hexadecimal":
value = "0x" + data.hex()
for op in param[7:]:
if isinstance(op, str) and op.startswith("*"):
value *= int(op[1:])
elif isinstance(op, str) and op.startswith("/"):
value /= float(op[1:])
data_dict[param[0]] = f"{value} {param[1]}"
for k, v in data_dict.items():
print(f"{k}: {v}")Found part of the bug (why it was only seeing one bus). It needs a timeout...
recv_msg = can_bus1.recv(timeout=0.0001) or can_bus0.recv(timeout=0.0001) if recv_msg is not None:
#!/usr/bin/python3
import can
import isotp
import udsoncan
import logging
from udsoncan.connections import PythonIsoTpConnection
from udsoncan.client import Client
# Written by Josh McCormick and Hallucinated by ChatGPT 4.0 on
# March 14th, 2023 for the Jeep Wrangler JL Community.
# Configure ISO-TP and UDS settings
isotp_params = {
'stmin': 32,
'blocksize': 8,
'wftmax': 0,
'll_data_length': 8,
'tx_padding': 0,
}
# AN OPTIONAL DEBUG CONFIGURATION. UNCOMMENT IF NEEDED.
# logging.basicConfig(level=logging.DEBUG)
# CONFIGURE UDS WITH DATA IDENTIFIER
config = {
'data_identifiers': {
0xF190: udsoncan.AsciiCodec(17)
}
}
# Initialize SocketCAN interface without specifying channel or bitrate
bus = can.interface.Bus(channel="can1", bustype='socketcan')
# Define the CAN IDs
tx_id = 0x620
rx_id = 0x504
# Create an ISO-TP address and connection, and UDS client
tp_addr = isotp.Address(isotp.AddressingMode.Normal_11bits, txid=tx_id, rxid=rx_id)
stack = isotp.CanStack(bus=bus, address=tp_addr, params=isotp_params)
conn = PythonIsoTpConnection(stack)
conn.open() # Open the connection explicitly
client = Client(conn, request_timeout=2, config=config)
# Perform Read Data by Identifier (UDS) with ID 0xf190
try:
identifier = 0xf190
response = client.read_data_by_identifier_first(identifier)
print(f"Req for Identifier 0x{identifier:x}: {response}")
except udsoncan.exceptions.NegativeResponseException as e:
print(f"Server responded with a negative response. {e}")
except udsoncan.exceptions.TimeoutException:
print("Server did not respond within the timeout.")
finally:
conn.close() # Close the connection explicitly#!/usr/bin/python3
import can
def rpm_warning(msg):
rpm = int.from_bytes(msg.data[0:2], byteorder='big')
if rpm > 500:
print(f"Warning: Engine RPM exceeds 500! Current RPM: {rpm}")
def main():
can0 = can.interface.Bus(bustype='socketcan', channel='can0')
can0.set_filters([{"can_id": 0x322, "can_mask": 0x7FF}])
while True:
msg = can0.recv(timeout=0.001)
if msg:
rpm_warning(msg)#!/usr/bin/python3
import can
def rpm_warning(msg):
rpm = int.from_bytes(msg.data[0:2], byteorder='big')
if rpm > 500:
print(f"Warning: Engine RPM exceeds 500! Current RPM: {rpm}")
def temp_warning(msg):
temp_f = msg.data[0]
if temp_f > 20:
print(f"Warning: Intake air temp exceeds 20°F! Current temperature: {temp_f}°F")
def main():
can0 = can.interface.Bus(bustype='socketcan', channel='can0')
can1 = can.interface.Bus(bustype='socketcan', channel='can1')
can0.set_filters([{"can_id": 0x322, "can_mask": 0x7FF}])
can1.set_filters([{"can_id": 0x127, "can_mask": 0x7FF}])
while True:
msg_can0 = can0.recv(timeout=0.001)
msg_can1 = can1.recv(timeout=0.001)
if msg_can0:
rpm_warning(msg_can0)
if msg_can1:
temp_warning(msg_can1)#!/usr/bin/python3
import can
def rpm_warning(msg):
rpm = int.from_bytes(msg.data[0:2], byteorder='big')
if rpm > 500:
print(f"Warning: Engine RPM exceeds 500! Current RPM: {rpm}")
def temp_warning(msg):
temp_f = msg.data[0]
if temp_f > 20:
print(f"Warning: Intake air temperature exceeds 20°F! Current temperature: {temp_f}°F")
def speed_warning(msg):
speed_mph = int.from_bytes(msg.data[6:8], byteorder='big')
if speed_mph > 90:
print(f"Warning: Vehicle speed exceeds 90 MPH! Current speed: {speed_mph} MPH")
def main():
can0 = can.interface.Bus(bustype='socketcan', channel='can0')
can1 = can.interface.Bus(bustype='socketcan', channel='can1')
can0.set_filters([{"can_id": 0x322, "can_mask": 0x7FF}])
can1.set_filters([{"can_id": 0x127, "can_mask": 0x7FF}, {"can_id": 0x1F7, "can_mask": 0x7FF}])
while True:
msg_can0 = can0.recv(timeout=0.001)
msg_can1 = can1.recv(timeout=0.001)
if msg_can0:
rpm_warning(msg_can0)
if msg_can1:
if msg_can1.arbitration_id == 0x127:
temp_warning(msg_can1)
elif msg_can1.arbitration_id == 0x1F7:
speed_warning(msg_can1)Yes. If you don't specify, it will just pick something. Sometimes APIs don't change enough to matter, but often at least one thing will break.
Is likely the bug you're looking for (and after I typed that, I realize you found it below). `recv` will block until it receives a message, so `can_bus0.recv()` will never be called unless `can_bus1.recv()` returns None, which can only happen if there's a timeout, which isn't specified.
with can.Bus(...) as can1, can.Bus(...) as bus2:
...That's never left me as an easy way to reason about one thread -- one worker -- handling many concurrent (but not parallel) tasks. Playing "chess" with threads would require one thread per chess table -- one Judith per table, and that would be inefficient (and infeasible) if there were many tables.RealPython said:
