udsoncan

import udsoncan from udsoncan.client import Client from udsoncan.server import Server from udsoncan.connections import LoopbackConnection from udsoncan.services import ReadDataByIdentifier, DiagnosticSessionControl from udsoncan.exceptions import NegativeResponseException # Define a simple UDS server application logic class MyServerApplication: def __init__(self): self.data_store = { 0xF190: b'PythonUDS', # Example VIN 0xF180: b'v1.0' # Example Software version } self.current_session = 1 # Default session def read_data_by_identifier(self, did, access_level=None): if did in self.data_store: return self.data_store[did] raise NegativeResponseException(0x10) # SubFunction Not Supported if DID is unknown def diagnostic_session_control(self, session_id, access_level=None): if session_id in [1, 2, 3]: # Default, Programming, Extended self.current_session = session_id return b'' raise NegativeResponseException(0x10) # SubFunction Not Supported if session is unknown def get_did_config(self): # Define how DIDs are encoded/decoded for the server return { 0xF190: {'data_size': 9, 'codec': udsoncan.AsciiCodec}, 0xF180: {'data_size': 4, 'codec': udsoncan.AsciiCodec}, } # Configure client and server to use an in-memory loopback connection conn = LoopbackConnection(name='test_bus') # Server setup server_app = MyServerApplication() server_config = udsoncan.ServerConfiguration() server_config.default_response_pending_timeout = 200 # ms server_config.set_did_config(server_app.get_did_config()) # Link server services to the application methods server_config.request_handler = { ReadDataByIdentifier: server_app.read_data_by_identifier, DiagnosticSessionControl: server_app.diagnostic_session_control } server = Server(conn, server_config) server.start() # Client setup client = Client(conn, request_timeout=2) # 2 seconds timeout try: client.open() # Example 1: Read VIN (DID F190) response = client.read_data_by_identifier([0xF190]) vin = response.values[0xF190] print(f"VIN: {vin.decode('ascii')}") # Example 2: Change diagnostic session to Extended Diagnostic Session (ID 0x03) response = client.diagnostic_session_control(3) print(f"Session changed to: {response.service_data.session_id} (Success)") # Example 3: Try reading an unsupported DID (will raise NegativeResponseException) try: client.read_data_by_identifier([0x1234]) except NegativeResponseException as e: print(f"Tried reading unsupported DID 0x1234: Received NRC {hex(e.response.code)}") except Exception as e: print(f"An unexpected error occurred: {e}") finally: client.close() server.stop() print("Client and server stopped.")