Modül 2: Otomotiv Protokolleri ve İletişim Sistemleri
Konular
İlerleme0%
Hafta 5: CAN Bus Temelleri ve Frame Yapısı
Hafta 5: CAN Bus Temelleri ve Frame Yapısı
🎯 Öğrenme Hedefleri
Bu haftanın sonunda:
- CAN Bus'un tarihi ve çalışma prensiplerini anlayacaksınız
- CAN Frame yapısını detaylı öğreneceksiniz
- Classical CAN ve CAN FD farklarını kavrayacaksınız
- python-can kütüphanesi ile mesaj gönderip alabileceksiniz
1. CAN Bus Nedir?
Şekil: Modern bir araçtaki ECU'ların CAN bus üzerinden bağlantısını gösteren ağ topolojisi diyagramı.
1.1 Tarihçe ve Gelişim
📅 CAN Bus Tarihi:
├── 1983: Robert Bosch GmbH tarafından geliştirilmeye başlandı
├── 1986: Resmi olarak tanıtıldı (SAE Congress)
├── 1987: İlk CAN chip'leri (Intel 82526, Philips 82C200)
├── 1991: Mercedes-Benz W140 - ilk seri üretim CAN uygulaması
├── 1993: ISO 11898 standardı yayınlandı
├── 2012: CAN FD (Flexible Data-rate) tanıtıldı
├── 2015: ISO 11898-1:2015 (CAN FD dahil)
└── 2020+: CAN XL geliştirme (10+ Mbps)
1.2 CAN Bus Neden Kullanılır?
| Özellik | Açıklama | |---------|----------| | Daha Az Kablo | Tek bir veri yolu üzerinden tüm ECU'lar iletişim kurar | | Güvenilirlik | Hata tespiti ve düzeltme mekanizmaları | | Gerçek Zamanlı | Deterministik önceliklendirme (arbitration) | | Maliyet | Basit donanım gereksinimleri | | Esneklik | Yeni ECU ekleme kolay |
1.3 Fiziksel Katman
CAN Bus Fiziksel Bağlantı
ECU 1 ECU 2 ECU 3 ECU 4
│ │ │ │
╔══╩══╗ ╔══╩══╗ ╔══╩══╗ ╔══╩══╗
║ CAN ║ ║ CAN ║ ║ CAN ║ ║ CAN ║
║Trans║ ║Trans║ ║Trans║ ║Trans║
╚═┬═┬═╝ ╚═┬═┬═╝ ╚═┬═┬═╝ ╚═┬═┬═╝
│ │ │ │ │ │ │ │
═══┼─┼═══════┼─┼═══════┼─┼═══════┼─┼═══ CAN_H (Yüksek)
═══┼─┼═══════┼─┼═══════┼─┼═══════┼─┼═══ CAN_L (Düşük)
│ │ │ │
120Ω (iç) (iç) 120Ω
Terminator Terminator
Dominant (0): CAN_H = 3.5V, CAN_L = 1.5V (Fark = 2V)
Recessive (1): CAN_H = 2.5V, CAN_L = 2.5V (Fark = 0V)
2. CAN Frame Yapısı
2.1 Standart CAN Frame (11-bit ID)
┌─────┬────────────────┬─────┬─────┬───┬─────┬───────────┬───────┬─────┬─────┐
│ SOF │ Arbitration ID │ RTR │ IDE │ r0│ DLC │ DATA │ CRC │ ACK │ EOF │
│ 1 │ 11 bit │ 1 │ 1 │ 1 │ 4 │ 0-64 bit │ 15+1 │ 2 │ 7 │
└─────┴────────────────┴─────┴─────┴───┴─────┴───────────┴───────┴─────┴─────┘
• SOF (Start of Frame): Mesaj başlangıcı, dominant bit (0)
• Arbitration ID: Mesaj önceliği (düşük değer = yüksek öncelik)
• RTR (Remote Transmission Request): 0 = Data frame, 1 = Remote frame
• IDE (Identifier Extension): 0 = Standard (11-bit), 1 = Extended (29-bit)
• r0: Rezerve (0)
• DLC (Data Length Code): Veri uzunluğu (0-8 byte)
• DATA: Mesaj verisi (0-8 byte, her byte 8 bit)
• CRC: Cyclic Redundancy Check (15 bit + 1 bit delimiter)
• ACK: Acknowledgement (1 bit ACK + 1 bit delimiter)
• EOF (End of Frame): 7 recessive bit
2.2 Python ile CAN Frame Analizi
#!/usr/bin/env python3
"""
CAN Frame Yapısı Analizi
"""
from dataclasses import dataclass
from typing import List, Optional
from enum import IntEnum
class DLC(IntEnum):
"""Data Length Code - CAN FD için."""
DLC_0 = 0
DLC_1 = 1
DLC_2 = 2
DLC_3 = 3
DLC_4 = 4
DLC_5 = 5
DLC_6 = 6
DLC_7 = 7
DLC_8 = 8
DLC_12 = 9 # CAN FD
DLC_16 = 10 # CAN FD
DLC_20 = 11 # CAN FD
DLC_24 = 12 # CAN FD
DLC_32 = 13 # CAN FD
DLC_48 = 14 # CAN FD
DLC_64 = 15 # CAN FD
@property
def bytes(self) -> int:
"""DLC değerine karşılık gelen byte sayısı."""
mapping = {0:0, 1:1, 2:2, 3:3, 4:4, 5:5, 6:6, 7:7, 8:8,
9:12, 10:16, 11:20, 12:24, 13:32, 14:48, 15:64}
return mapping.get(self.value, 0)
@dataclass
class CANFrame:
"""CAN Frame veri yapısı."""
arbitration_id: int
data: bytes
is_extended_id: bool = False
is_remote_frame: bool = False
is_fd: bool = False
bitrate_switch: bool = False # CAN FD BRS
error_state_indicator: bool = False # CAN FD ESI
timestamp: float = 0.0
@property
def dlc(self) -> int:
"""Data Length Code."""
return len(self.data)
@property
def id_type(self) -> str:
"""ID tipi."""
return "Extended (29-bit)" if self.is_extended_id else "Standard (11-bit)"
@property
def frame_type(self) -> str:
"""Frame tipi."""
if self.is_fd:
return "CAN FD"
elif self.is_remote_frame:
return "Remote"
else:
return "Data"
def to_hex_string(self) -> str:
"""Frame'i hex string olarak formatla."""
id_format = f"0x{self.arbitration_id:08X}" if self.is_extended_id else f"0x{self.arbitration_id:03X}"
data_hex = ' '.join(f'{b:02X}' for b in self.data)
return f"{id_format} [{self.dlc}] {data_hex}"
def analyze(self) -> dict:
"""Frame'i analiz et."""
analysis = {
"id": hex(self.arbitration_id),
"id_type": self.id_type,
"frame_type": self.frame_type,
"dlc": self.dlc,
"data_hex": self.data.hex().upper(),
"data_ascii": ''.join(chr(b) if 32 <= b < 127 else '.' for b in self.data),
}
# UDS analizi
if len(self.data) >= 2:
first_byte = self.data[0]
second_byte = self.data[1]
# Single Frame (SF)
if (first_byte & 0xF0) == 0x00:
sf_length = first_byte & 0x0F
analysis["iso_tp"] = f"Single Frame, Length: {sf_length}"
if sf_length >= 1:
service_id = second_byte
analysis["uds_service"] = self._identify_uds_service(service_id)
# First Frame (FF)
elif (first_byte & 0xF0) == 0x10:
ff_length = ((first_byte & 0x0F) << 8) | second_byte
analysis["iso_tp"] = f"First Frame, Total Length: {ff_length}"
# Consecutive Frame (CF)
elif (first_byte & 0xF0) == 0x20:
seq_num = first_byte & 0x0F
analysis["iso_tp"] = f"Consecutive Frame, Sequence: {seq_num}"
# Flow Control (FC)
elif (first_byte & 0xF0) == 0x30:
flow_status = first_byte & 0x0F
status_map = {0: "CTS (Continue To Send)", 1: "Wait", 2: "Overflow"}
analysis["iso_tp"] = f"Flow Control, Status: {status_map.get(flow_status, 'Unknown')}"
return analysis
def _identify_uds_service(self, service_id: int) -> str:
"""UDS servisini tanımla."""
services = {
0x10: "DiagnosticSessionControl",
0x11: "ECUReset",
0x14: "ClearDiagnosticInformation",
0x19: "ReadDTCInformation",
0x22: "ReadDataByIdentifier",
0x23: "ReadMemoryByAddress",
0x24: "ReadScalingDataByIdentifier",
0x27: "SecurityAccess",
0x28: "CommunicationControl",
0x2A: "ReadDataByPeriodicIdentifier",
0x2C: "DynamicallyDefineDataIdentifier",
0x2E: "WriteDataByIdentifier",
0x2F: "InputOutputControlByIdentifier",
0x31: "RoutineControl",
0x34: "RequestDownload",
0x35: "RequestUpload",
0x36: "TransferData",
0x37: "RequestTransferExit",
0x38: "RequestFileTransfer",
0x3E: "TesterPresent",
0x85: "ControlDTCSetting",
# Pozitif yanıtlar
0x50: "DiagnosticSessionControl (+)",
0x51: "ECUReset (+)",
0x62: "ReadDataByIdentifier (+)",
0x67: "SecurityAccess (+)",
0x6E: "WriteDataByIdentifier (+)",
0x71: "RoutineControl (+)",
0x7E: "TesterPresent (+)",
0x7F: "Negative Response",
}
return services.get(service_id, f"Unknown (0x{service_id:02X})")
# Demo
if __name__ == "__main__":
print("📡 CAN FRAME ANALİZ DEMO")
print("=" * 60)
# Örnek frameler
frames = [
CANFrame(0x7E0, bytes([0x02, 0x10, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00])),
CANFrame(0x7E8, bytes([0x02, 0x50, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00])),
CANFrame(0x7E0, bytes([0x03, 0x22, 0xF1, 0x90, 0x00, 0x00, 0x00, 0x00])),
CANFrame(0x7E8, bytes([0x10, 0x14, 0x62, 0xF1, 0x90, 0x57, 0x42, 0x41])),
CANFrame(0x7E0, bytes([0x30, 0x00, 0x14, 0x00, 0x00, 0x00, 0x00, 0x00])),
CANFrame(0x7E8, bytes([0x21, 0x33, 0x41, 0x35, 0x47, 0x35, 0x39, 0x44])),
CANFrame(0x7E0, bytes([0x02, 0x27, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00])),
CANFrame(0x7E8, bytes([0x7F, 0x27, 0x35, 0x00, 0x00, 0x00, 0x00, 0x00])),
]
for i, frame in enumerate(frames, 1):
print(f"
📦 Frame {i}: {frame.to_hex_string()}")
analysis = frame.analyze()
for key, value in analysis.items():
print(f" {key}: {value}")
3. Classical CAN vs CAN FD
╔════════════════════════════════════════════════════════════╗
║ Classical CAN vs CAN FD ║
╠═════════════════════╬═══════════════════╬═══════════════════╣
║ Özellik ║ Classical CAN ║ CAN FD ║
╠═════════════════════╬═══════════════════╬═══════════════════╣
║ Maks. Veri ║ 8 byte ║ 64 byte ║
║ Maks. Hız ║ 1 Mbps ║ 8 Mbps (veri) ║
║ Arbitration Hızı ║ 1 Mbps ║ 1 Mbps (uyumluluk) ║
║ Güvenlik ║ CRC-15 ║ CRC-17/CRC-21 ║
║ DLC Değerleri ║ 0-8 ║ 0-8, 12, 16, 20, ║
║ ║ ║ 24, 32, 48, 64 ║
║ Uyumluluk ║ Tüm CAN ║ Geriye uyumlu ║
╚═════════════════════╩═══════════════════╩═══════════════════╝
4. python-can Kütüphanesi
#!/usr/bin/env python3
"""
python-can ile CAN Bus İletişimi
Kurulum: pip install python-can
"""
import can
from typing import Optional, List
import time
class CANInterface:
"""CAN Bus arayüz sınıfı."""
def __init__(self, channel: str = 'vcan0', bustype: str = 'socketcan'):
"""
CAN arayüzü başlat.
Args:
channel: CAN kanalı (vcan0, can0, PCAN_USBBUS1, vb.)
bustype: Arayüz tipi (socketcan, pcan, vector, vb.)
"""
self.channel = channel
self.bustype = bustype
self.bus: Optional[can.Bus] = None
self.filters: List[dict] = []
def connect(self) -> bool:
"""CAN bus'a bağlan."""
try:
self.bus = can.Bus(channel=self.channel, bustype=self.bustype)
print(f"✅ CAN Bus bağlandı: {self.channel} ({self.bustype})")
return True
except can.CanError as e:
print(f"❌ CAN Bus bağlantı hatası: {e}")
return False
def disconnect(self):
"""CAN bus bağlantısını kapat."""
if self.bus:
self.bus.shutdown()
print("CAN Bus bağlantısı kapatıldı")
def set_filters(self, can_ids: List[int], mask: int = 0x7FF):
"""
CAN ID filtreleri ayarla.
Args:
can_ids: Filtrelenecek CAN ID'leri
mask: Filtre maskesi
"""
self.filters = [{"can_id": cid, "can_mask": mask} for cid in can_ids]
if self.bus:
self.bus.set_filters(self.filters)
print(f"🔍 Filtreler ayarlandı: {[hex(f['can_id']) for f in self.filters]}")
def send(self, arbitration_id: int, data: bytes,
is_extended: bool = False) -> bool:
"""
CAN mesajı gönder.
Args:
arbitration_id: CAN ID
data: Mesaj verisi (max 8 byte)
is_extended: Extended ID (29-bit) kullan
Returns:
bool: Başarı durumu
"""
if not self.bus:
print("❌ CAN Bus bağlı değil!")
return False
msg = can.Message(
arbitration_id=arbitration_id,
data=data,
is_extended_id=is_extended
)
try:
self.bus.send(msg)
print(f"📤 Gönderildi: 0x{arbitration_id:03X} {data.hex().upper()}")
return True
except can.CanError as e:
print(f"❌ Gönderim hatası: {e}")
return False
def receive(self, timeout: float = 1.0) -> Optional[can.Message]:
"""
CAN mesajı al.
Args:
timeout: Bekleme süresi (saniye)
Returns:
can.Message veya None
"""
if not self.bus:
return None
try:
msg = self.bus.recv(timeout=timeout)
if msg:
print(f"📥 Alındı: 0x{msg.arbitration_id:03X} {bytes(msg.data).hex().upper()}")
return msg
except can.CanError as e:
print(f"❌ Alım hatası: {e}")
return None
def sniff(self, duration: float = 5.0, max_messages: int = 100) -> List[can.Message]:
"""
CAN trafiğini dinle.
Args:
duration: Dinleme süresi (saniye)
max_messages: Maksimum mesaj sayısı
Returns:
Yakalanan mesajlar
"""
messages = []
start_time = time.time()
print(f"
🔊 CAN Sniffing başladı ({duration}s, max {max_messages} mesaj)...")
print("-" * 60)
while (time.time() - start_time) < duration and len(messages) < max_messages:
msg = self.receive(timeout=0.1)
if msg:
messages.append(msg)
print("-" * 60)
print(f"✅ Toplam: {len(messages)} mesaj yakalandı")
return messages
# UDS Client simülasyonu
class UDSClientSimulator:
"""UDS Client simülasyonu (CAN bağlantısı olmadan)."""
def __init__(self, tx_id: int = 0x7E0, rx_id: int = 0x7E8):
self.tx_id = tx_id
self.rx_id = rx_id
self.session = 0x01 # Default session
def create_request(self, service: int, *data_bytes) -> bytes:
"""UDS request oluştur."""
# Single Frame format: [SF_DL, Service, Data...]
payload = bytes([service] + list(data_bytes))
sf_dl = len(payload)
frame_data = bytes([sf_dl]) + payload
# 8 byte'a doldur
padded = frame_data + bytes(8 - len(frame_data))
return padded[:8]
def diagnostic_session_control(self, session: int = 0x01) -> dict:
"""0x10 - DiagnosticSessionControl."""
request = self.create_request(0x10, session)
print(f"
🔧 DiagnosticSessionControl (Session: 0x{session:02X})")
print(f" Request: 0x{self.tx_id:03X} {request.hex().upper()}")
# Simüle edilmiş pozitif yanıt
response = bytes([0x02, 0x50, session, 0x00, 0x00, 0x00, 0x00, 0x00])
print(f" Response: 0x{self.rx_id:03X} {response.hex().upper()}")
self.session = session
return {"success": True, "session": session}
def read_data_by_identifier(self, did: int) -> dict:
"""0x22 - ReadDataByIdentifier."""
did_high = (did >> 8) & 0xFF
did_low = did & 0xFF
request = self.create_request(0x22, did_high, did_low)
print(f"
📋 ReadDataByIdentifier (DID: 0x{did:04X})")
print(f" Request: 0x{self.tx_id:03X} {request.hex().upper()}")
# Simüle edilmiş yanıtlar
did_responses = {
0xF190: b'WBA3A5G59DNP26082', # VIN
0xF191: b'ECU_HW_V1.0', # Hardware Version
0xF195: b'SW_V2.4.1', # Software Version
}
if did in did_responses:
data = did_responses[did]
# Multi-frame response simülasyonu
print(f" Response: {data.decode('utf-8', errors='replace')}")
return {"success": True, "did": hex(did), "data": data}
else:
# Negative Response
nrc = 0x31 # requestOutOfRange
print(f" Negative Response: NRC 0x{nrc:02X} (requestOutOfRange)")
return {"success": False, "nrc": nrc}
def security_access(self, level: int = 1) -> dict:
"""0x27 - SecurityAccess."""
print(f"
🔐 SecurityAccess (Level: {level})")
# Seed request
seed_request = self.create_request(0x27, (level * 2) - 1) # 0x01, 0x03, ...
print(f" Seed Request: 0x{self.tx_id:03X} {seed_request.hex().upper()}")
# Simüle edilmiş seed
seed = bytes([0xAB, 0xCD, 0x12, 0x34])
print(f" Seed: {seed.hex().upper()}")
# Key hesapla (basit XOR algoritması)
key = bytes([b ^ 0xFF for b in seed])
print(f" Key (XOR): {key.hex().upper()}")
# Key gönder
key_request = self.create_request(0x27, level * 2, *key) # 0x02, 0x04, ...
print(f" Key Request: 0x{self.tx_id:03X} {key_request.hex().upper()}")
return {"success": True, "level": level, "seed": seed, "key": key}
# Demo
if __name__ == "__main__":
print("🚗 CAN/UDS SİMÜLASYON DEMO")
print("=" * 60)
client = UDSClientSimulator()
# 1. Session aç
client.diagnostic_session_control(0x01) # Default
client.diagnostic_session_control(0x03) # Extended
# 2. Veri oku
client.read_data_by_identifier(0xF190) # VIN
client.read_data_by_identifier(0xF195) # Software Version
client.read_data_by_identifier(0x1234) # Bilinmeyen DID
# 3. Security Access
client.security_access(1)
5. Virtual CAN Kurulumu (Linux)
#!/bin/bash
# Virtual CAN kurulum scripti
# Modülleri yükle
sudo modprobe can
sudo modprobe can_raw
sudo modprobe vcan
# Virtual CAN arayüzü oluştur
sudo ip link add dev vcan0 type vcan
sudo ip link set up vcan0
# Durumu kontrol et
ip link show vcan0
echo "✅ vcan0 arayüzü hazır!"
echo "Örnek kullanım:"
echo " candump vcan0 # Mesajları dinle"
echo " cansend vcan0 7E0#021001 # Mesaj gönder"