UDS
ISO 14229 araç diagnostics protokolü.

# Tester → ECU (CAN ID: 0x7DF broadcast veya 0x7E0 + ECU offset)
# Single Frame: PCI=0x03 (3 byte), SID=0x22, DID=0xF190
CAN ID: 0x7DF
Data:   03 22 F1 90 00 00 00 00

# ECU → Tester (First Frame: toplam 20 byte VIN yanıtı)
CAN ID: 0x7E8
Data:   10 14 62 F1 90 31 47 31   # FF: len=0x14=20, SID+0x40=0x62

# Tester → ECU (Flow Control: devam et)
CAN ID: 0x7DF
Data:   30 00 00 00 00 00 00 00   # FC: BS=0 (sınırsız), STmin=0

# ECU → Tester (Consecutive Frame 1)
CAN ID: 0x7E8
Data:   21 46 4A 31 32 33 34 35   # CF sn=1

# ECU → Tester (Consecutive Frame 2)
CAN ID: 0x7E8
Data:   22 36 37 38 39 30 31 32   # CF sn=2, VIN tamamlandı

# Format: 7F [SID] [NRC]
# Örnek: 0x27 SecurityAccess'e hatalı key gönderildi
7F 27 35   # NRC 0x35 = invalidKey

# Yaygın NRC kodları
# 0x10 — generalReject
# 0x11 — serviceNotSupported
# 0x12 — subFunctionNotSupported
# 0x13 — incorrectMessageLengthOrInvalidFormat
# 0x22 — conditionsNotCorrect (wrong session)
# 0x24 — requestSequenceError (wrong order)
# 0x31 — requestOutOfRange (invalid DID)
# 0x33 — securityAccessDenied
# 0x35 — invalidKey
# 0x36 — exceededNumberOfAttempts
# 0x37 — requiredTimeDelayNotExpired
# 0x78 — requestCorrectlyReceivedResponsePending (ECU meşgul, bekle)

# 1. Default → Extended Session aç
# İstek:  10 03
# Yanıt:  50 03 00 32 01 F4  (P2=50ms, P2*=500ms)

# 2. Session canlı tut (her 4.5 saniyede bir)
# İstek:  3E 00  (TesterPresent, suppress response)
# Yanıt:  7E 00  (ya da suppress bit ile yanıt yok)

# 3. Default Session'a dön
# İstek:  10 01
# Yanıt:  50 01 ...

# 4. ECU Reset
# İstek:  11 01  (Hard Reset)
# Yanıt:  51 01  (sonra ECU yeniden başlar)

def calculate_key(seed: bytes, secret: int = 0xABCD1234) -> bytes:
    """
    Gerçek araçlarda bu algoritma OEM tarafından gizli tutulur.
    Tersine mühendislik veya NDA ile temin edilir.
    """
    seed_int = int.from_bytes(seed, 'big')
    key_int = seed_int ^ secret
    key_int = ((key_int << 3) | (key_int >> 29)) & 0xFFFFFFFF
    return key_int.to_bytes(4, 'big')

# Kullanım:
seed = bytes.fromhex('12 34 56 78'.replace(' ', ''))
key  = calculate_key(seed)
print(f"Seed: {seed.hex()}, Key: {key.hex()}")

# Tek DID: VIN oku
# İstek:  22 F1 90
# Yanıt:  62 F1 90 [17 byte VIN ASCII]

# İki DID aynı anda: VIN + SW Version
# İstek:  22 F1 90 F1 01
# Yanıt:  62 F1 90 [17 byte] F1 01 [N byte versiyon]

# Örnek yanıt (ASCII VIN: "1G1JC5444Y7123456")
62 F1 90 \
31 47 31 4A 43 35 34 34 \   # 1G1JC544
34 59 37 31 32 33 34 35 \   # 4Y712345
36                          # 6

# Önce Extended Session ve SecurityAccess gerekli
# İstek:  2E F1 90 [17 byte VIN]
# Yanıt:  6E F1 90  (başarılı)
# Hata:   7F 2E 22  (conditionsNotCorrect — wrong session)

# Yeni VIN: "WBAC5441X1LB12345"
2E F1 90 \
57 42 41 43 35 34 34 31 \   # WBAC5441
58 31 4C 42 31 32 33 34 \   # X1LB1234
35                          # 5

# Tüm aktif (testFailed) DTC'leri iste: statusMask = 0x01
# İstek:  19 02 01
# Yanıt:  59 02 FF [DTC1 3 byte] [status1] [DTC2 3 byte] [status2] ...

# Yanıt parse:
# 59 02 FF        — SID+0x40, sub-func, DTC format
# 00 12 34 09     — DTC: P0x1234, status: 0x09 (TF=1, confirmed=1)
# 00 45 67 0F     — DTC: P0x4567, status: 0x0F

# Tüm DTC'leri sil
# İstek:  14 FF FF FF
# Yanıt:  54

def decode_dtc(dtc_bytes: bytes) -> str:
    domain = {0: 'P', 1: 'C', 2: 'B', 3: 'U'}
    high = dtc_bytes[0]
    code = ((high & 0x3F) << 8) | dtc_bytes[1]
    d = domain[(high >> 6) & 0x03]
    return f"{d}{code:04X}"

# Örnek: 0x001234 → P1234
print(decode_dtc(bytes([0x00, 0x12, 0x34])))  # P1234
print(decode_dtc(bytes([0x40, 0x45, 0x67])))  # C4567

pip install udsoncan python-can
# DoIP transport için:
pip install doipclient

import udsoncan
from udsoncan.connections import PythonIsoTpConnection
from udsoncan.client import Client
import isotp
import can

# CAN bus ve ISO-TP bağlantısı kur (SocketCAN: vcan0)
bus = can.interface.Bus(channel='vcan0', bustype='socketcan')
tp_addr = isotp.Address(isotp.AddressingMode.Normal_11bits,
                        txid=0x7E0, rxid=0x7E8)
stack = isotp.CanStack(bus=bus, address=tp_addr)
conn  = PythonIsoTpConnection(stack)

config = udsoncan.configs.default_client_config.copy()
config['request_timeout'] = 5

with Client(conn, config=config) as client:

    # 1. Extended Session aç
    client.change_session(
        udsoncan.services.DiagnosticSessionControl.Session.extendedDiagnosticSession)
    print("Extended session açıldı")

    # 2. SecurityAccess — seed iste, key hesapla, gönder
    result = client.request_seed(0x01)
    seed = result.service_data.seed
    key  = calculate_key(seed)   # OEM'e özgü algoritma
    client.send_key(0x02, key)
    print("SecurityAccess başarılı")

    # 3. VIN oku
    result = client.read_data_by_identifier([0xF190])
    vin = result.service_data.values[0xF190].decode('ascii')
    print(f"VIN: {vin}")

    # 4. DTC listesini oku
    result = client.get_dtc_by_status_mask(0xFF)
    for dtc_entry in result.service_data.dtcs:
        print(f"DTC: {dtc_entry.id:06X}  Status: {dtc_entry.status.get_byte():02X}")

    # 5. DTC'leri temizle
    client.clear_dtc(0xFFFFFF)
    print("DTC'ler temizlendi")

    # 6. ECU soft reset
    client.ecu_reset(udsoncan.services.ECUReset.ResetType.softReset)

import threading, time

def keep_session_alive(client, interval=4.5):
    while not stop_event.is_set():
        try:
            client.tester_present(suppress_positive_response=True)
        except:
            pass
        time.sleep(interval)

stop_event = threading.Event()
tp_thread = threading.Thread(target=keep_session_alive,
                             args=(client,), daemon=True)
tp_thread.start()
# ... UDS işlemler ...
stop_event.set()

# /boot/config.txt eklemeleri
dtparam=spi=on
dtoverlay=mcp2515-can0,oscillator=8000000,interrupt=25

# CAN interface'i başlat
sudo ip link set can0 up type can bitrate 500000
ip -details link show can0

# Test: candump ile CAN trafiği izle
candump can0

import udsoncan
from udsoncan.client import Client

def flash_firmware(client, firmware_path: str, flash_addr: int):
    with open(firmware_path, 'rb') as f:
        firmware = f.read()

    # Programming session
    client.change_session(
        udsoncan.services.DiagnosticSessionControl.Session.programmingSession)

    # Security access
    seed = client.request_seed(0x11).service_data.seed
    client.send_key(0x12, calculate_key(seed))

    # RequestDownload: mem addr 4 byte, len 4 byte
    mem_addr = udsoncan.MemoryLocation(
        address=flash_addr,
        memorysize=len(firmware),
        address_format=32, memorysize_format=32)

    result = client.request_download(
        memory_location=mem_addr,
        dfi=udsoncan.DataFormatIdentifier(0x00))  # no compression

    max_block = result.service_data.max_length
    print(f"Max blok boyutu: {max_block} byte")

    # TransferData bloklar halinde
    block_seq = 1
    offset = 0
    while offset < len(firmware):
        chunk = firmware[offset:offset + max_block - 2]
        client.transfer_data(block_seq, chunk)
        offset += len(chunk)
        block_seq = (block_seq % 0xFF) + 1
        print(f"  {offset}/{len(firmware)} byte aktarıldı")

    # Transfer tamamla
    client.request_transfer_exit()
    print("Flash tamamlandı, ECU resetleniyor...")

    # ECU reset
    client.ecu_reset(udsoncan.services.ECUReset.ResetType.hardReset)

# Örnek kullanım:
# flash_firmware(client, 'firmware_v2.bin', 0x08000000)