LIN Bus
otomotiv alt ağı.

def calc_pid(frame_id: int) -> int:
    """6-bit frame ID'den PID (Protected ID) hesapla."""
    id_bits = frame_id & 0x3F
    p0 = ((id_bits >> 0) ^ (id_bits >> 1) ^ (id_bits >> 2) ^ (id_bits >> 4)) & 1
    p1 = ~((id_bits >> 1) ^ (id_bits >> 3) ^ (id_bits >> 4) ^ (id_bits >> 5)) & 1
    return id_bits | (p0 << 6) | (p1 << 7)

print(hex(calc_pid(0x01)))   # → 0x41
print(hex(calc_pid(0x10)))   # → 0x50

def lin_checksum(data: bytes, pid: int = None) -> int:
    """Enhanced checksum: PID varsa dahil et, yoksa classic."""
    total = sum(data)
    if pid is not None:
        total += pid          # enhanced: PID de dahil
    # Carry propagation
    while total > 0xFF:
        total = (total & 0xFF) + (total >> 8)
    return (~total) & 0xFF   # 1'in tümleyeni

LIN_description_file;
LIN_protocol_version = "2.2";
LIN_language_version = "2.2";
LIN_speed = 9600 bps;

Nodes {
    Master: BCM, 5 ms, 0.1 ms;    /* BCM = Body Control Module */
    Slaves: SeatModule, MirrorModule, WindowModule;
}

Signals {
    SeatPos:       8, 0,   BCM,          SeatModule;
    MirrorAngleH:  8, 0,   BCM,          MirrorModule;
    WindowPos:     8, 255, WindowModule, BCM;
    WindowSwitch:  2, 0,   BCM,          WindowModule;
}

Frames {
    SeatControl:   0x01, BCM,          2 {SeatPos, 0;}
    MirrorControl: 0x02, BCM,          1 {MirrorAngleH, 0;}
    WindowStatus:  0x03, WindowModule, 2 {WindowPos, 0; WindowSwitch, 8;}
}

Schedule_tables {
    Normal_Schedule {
        SeatControl   delay 10 ms;
        MirrorControl delay 10 ms;
        WindowStatus  delay 10 ms;
    }
}

# PEAK Linux driver indir ve derle:
wget https://www.peak-system.com/fileadmin/media/linux/files/peak-linux-driver.tar.gz
tar xzf peak-linux-driver.tar.gz
cd peak-linux-driver-*
make
sudo make install

# PLIN-USB tak ve kontrol et:
dmesg | grep plin
# [12.345] plin: PLIN-USB attached to /dev/plin0

ls /dev/plin*
# /dev/plin0

#include <plin_api.h>
#include <stdio.h>

int main(void)
{
    TPLinHandle hClient = 0;
    TPLinHandle hHardware = 0;
    TPLinHandle hNet = 0;

    /* İstemci oluştur */
    LINRESULT res = LIN_RegisterClient("MyApp", 0, &hClient);
    if (res != TLIN_ERROR_OK) { printf("Client hatası\n"); return 1; }

    /* Donanımı bağla */
    res = LIN_ConnectClient(hClient, hHardware);

    /* Network oluştur: LIN 2.2, 9600 bps, master */
    res = LIN_CreateMasterNet(hClient, hHardware, 9600, TLIN_FRAMEVERSION_22, &hNet);

    /* Frame yapılandır: PID 0x01, 2 byte data */
    TLINFrameEntry frame = {0};
    frame.FrameId      = 0x01;
    frame.Direction    = TLIN_DIRECTION_PUBLISHER;
    frame.ChecksumType = TLIN_CHECKSUMTYPE_ENHANCED;
    frame.Length       = 2;
    LIN_SetFrameEntry(hNet, &frame);

    /* Veri yaz */
    BYTE data[2] = {0x50, 0x00};   /* koltuk pozisyon = 80, seçenek = 0 */
    LIN_UpdateByteArray(hNet, 0x01, 0, 2, data);

    printf("LIN frame gönderildi (PID 0x01)\n");

    LIN_RemoveNet(hNet);
    LIN_UnregisterClient(hClient);
    return 0;
}

pip install pyserial

import serial, time, struct

class LINMaster:
    def __init__(self, port: str, baudrate: int = 9600):
        self.ser = serial.Serial(port, baudrate, timeout=0.05)
        self.baudrate = baudrate

    def _calc_pid(self, frame_id: int) -> int:
        """6-bit ID → Protected Identifier."""
        i = frame_id & 0x3F
        p0 = ((i >> 0) ^ (i >> 1) ^ (i >> 2) ^ (i >> 4)) & 1
        p1 = ~((i >> 1) ^ (i >> 3) ^ (i >> 4) ^ (i >> 5)) & 1
        return i | (p0 << 6) | (p1 << 7)

    def _checksum(self, data: bytes, pid: int) -> int:
        """Enhanced checksum hesabı."""
        total = sum(data) + pid
        while total > 0xFF:
            total = (total & 0xFF) + (total >> 8)
        return (~total) & 0xFF

    def send_break(self):
        """LIN break field: geçici olarak daha düşük baud'da 0x00 gönder."""
        self.ser.baudrate = self.baudrate // 2
        self.ser.write(bytes([0x00]))   # 8 dominant + stop → 9 bit dominant
        self.ser.flush()
        self.ser.baudrate = self.baudrate

    def send_header(self, frame_id: int):
        """Break + Sync(0x55) + PID gönder."""
        self.send_break()
        pid = self._calc_pid(frame_id)
        self.ser.write(bytes([0x55, pid]))

    def send_frame(self, frame_id: int, data: bytes):
        """Tam LIN frame (header + data + checksum) gönder."""
        pid = self._calc_pid(frame_id)
        chk = self._checksum(data, pid)
        self.send_break()
        self.ser.write(bytes([0x55, pid]) + data + bytes([chk]))

    def request_response(self, frame_id: int, length: int) -> bytes:
        """Header gönder, slave response'u oku."""
        self.send_header(frame_id)
        resp = self.ser.read(length + 1)  # data + checksum
        if len(resp) == length + 1:
            return resp[:length]
        return b''

    def close(self):
        self.ser.close()

# Kullanım örneği:
master = LINMaster('/dev/ttyAMA0', baudrate=9600)

try:
    while True:
        # Slot 1: koltuk konum komutunu gönder (publisher)
        master.send_frame(0x01, bytes([0x50, 0x00]))
        time.sleep(0.01)

        # Slot 2: pencere durumunu oku (subscriber)
        data = master.request_response(0x03, 2)
        if data:
            window_pos = data[0]
            print(f'Pencere: {window_pos}% açık')
        time.sleep(0.01)

        # Slot 3: ayna açısı ayarla
        master.send_frame(0x02, bytes([0x7F]))   # merkez konum
        time.sleep(0.03)

except KeyboardInterrupt:
    master.close()

/* MRF (Master Request Frame): */
/* [NAD][PCI][SID][D1][D2][D3][D4][D5] */
/*  ↑    ↑    ↑              
/*  │    │    └─ Service ID (UDS: 0x22 = ReadDataByIdentifier)
/*  │    └────── Protocol Control Information
/*  └─────────── Node Address (1-127, 0x7F = broadcast)

def lin_read_data_by_id(master, nad: int, data_id: int) -> bytes:
    """
    UDS ReadDataByIdentifier (0x22) over LIN-TP.
    nad: node address (1-127)
    data_id: 2-byte identifier (örn. 0xF190 = VIN)
    """
    # MRF: NAD | 0x06 | 0x22 | ID_high | ID_low | 0xFF | 0xFF | 0xFF
    req = bytes([
        nad,          # NAD
        0x06,         # PCI: SF (Single Frame), uzunluk 6
        0x22,         # SID: ReadDataByIdentifier
        (data_id >> 8) & 0xFF,
        data_id & 0xFF,
        0xFF, 0xFF, 0xFF   # padding
    ])
    master.send_frame(0x3C, req)   # Master Request
    time.sleep(0.005)
    resp = master.request_response(0x3D, 8)   # Slave Response
    if resp and resp[2] == 0x62:   # pozitif yanıt
        return resp[5:]            # veri
    return b''

# Slave node'dan parça numarasını oku (DID 0xF187):
part_no = lin_read_data_by_id(master, nad=2, data_id=0xF187)
print('Parça No:', part_no.hex())

Raspberry Pi 4              TJA1021             LIN Bus
─────────────               ──────────          ────────
GPIO14 (TXD) ──────────────► TXD (pin 1)
GPIO15 (RXD) ◄────────────── RXD (pin 4)
3.3V ─────────────────────── INH (pin 3)  ← enable
GND  ─────────────────────── GND (pin 2)
                              VBAT ─────────── 12V
                              LIN  ─────────── LIN bus (single wire)
                              GND  ─────────── chassis GND

import time
from lin_master import LINMaster   # önceki bölümdeki sınıf

master = LINMaster('/dev/ttyAMA0', baudrate=9600)

# Frame ID atamaları:
#   0x01 → Koltuk kontrolü (master → slave1)
#   0x02 → Ayna kontrolü  (master → slave2)
#   0x03 → Pencere durum  (slave3 → master)
#   0x04 → Klima durum    (slave4 → master)

seat_position = 0x50    # %80 ileri
mirror_h      = 0x7F    # merkez yatay

try:
    while True:
        # Slot 1: koltuk pozisyon komutu (10ms)
        master.send_frame(0x01, bytes([seat_position, 0x00]))
        time.sleep(0.010)

        # Slot 2: ayna açısı (10ms)
        master.send_frame(0x02, bytes([mirror_h]))
        time.sleep(0.010)

        # Slot 3: pencere durumu oku
        win_data = master.request_response(0x03, 2)
        if win_data:
            print(f'Pencere: sol={win_data[0]}% sağ={win_data[1]}%')
        time.sleep(0.010)

        # Slot 4: klima durumu oku
        clim_data = master.request_response(0x04, 3)
        if clim_data:
            temp = clim_data[0] - 40   # -40°C offset
            fan  = clim_data[1]        # fan hızı %
            print(f'Klima: {temp}°C  Fan: {fan}%')
        time.sleep(0.010)

except KeyboardInterrupt:
    master.close()

import ldfparser

ldf = ldfparser.parse_ldf('door_module.ldf')

print('LIN hızı:', ldf.get_baudrate())
print('Slave'ler:', [s.name for s in ldf.get_slaves()])

for frame in ldf.get_frames():
    print(f'Frame {frame.frame_id:#04x}: {frame.name} ({frame.length}B)')

# Encoding ile sinyal decode:
seat_frame = ldf.get_frame('SeatControl')
raw_data = bytes([0x50, 0x00])
signals = seat_frame.parse_raw(raw_data)
print('Sinyal değerleri:', signals)