Sunday, October 18, 2026

Implementasi Software Defined Radio xSDR pada Raspberry Pi Compute Module 5 melalui Antarmuka PCIe M.2 dengan Uji Coba Penerima Radio FM

Tutorial ini membawa Anda dari Raspberry Pi CM5 yang masih bersih sampai bisa mengakses SDR xSDR lewat slot M.2 (PCIe) dan mendengarkan radio FM dengan Python. Semua langkah sudah saya uji di CM5 dengan carrier board piBrick PocketCM5.

Pengenalan xSDR

xSDR adalah modul Software Defined Radio (SDR) buatan Wavelet Lab yang dibuat dalam format kartu M.2 2230, ukurannya hanya 30 × 22 mm, sama dengan kartu WiFi laptop. Walaupun kecil, xSDR adalah SDR full duplex dengan dua kanal penerima (RX) dan dua kanal pemancar (TX), sehingga mendukung konfigurasi MIMO 2×2.

Modul xSDR M.2 2230
Modul xSDR. Chip besar di atas adalah transceiver RF LMS7002M, di bawahnya FPGA Xilinx Artix-7. Sumber gambar: Wavelet Lab / Crowd Supply.

Pada SDR, sebagian besar fungsi radio seperti filter, demodulasi dan modulasi dikerjakan oleh perangkat lunak. Perangkat keras hanya bertugas mengubah sinyal RF menjadi sampel digital I/Q (dan sebaliknya untuk pemancar). Karena itu satu modul xSDR bisa dipakai sebagai penerima FM, penganalisis spektrum, pemancar uji, sampai base station seluler, cukup dengan mengganti programnya.

Arsitektur

xSDR dibangun dari dua komponen utama:

  • Lime Microsystems LMS7002M: transceiver RF yang berisi LNA, mixer, filter, penguat (TIA dan PGA), serta ADC/DAC 12-bit untuk dua kanal RX dan dua kanal TX.

  • FPGA AMD (Xilinx) Artix-7: mengatur LMS7002M, menyusun sampel I/Q menjadi paket data, dan mengirimkannya ke komputer lewat PCIe atau USB.

Diagram blok xSDR
Diagram blok xSDR. Sumber gambar: Wavelet Lab / Crowd Supply.

Spesifikasi utama

Transceiver RFLime Microsystems LMS7002M
FPGAAMD (Xilinx) Artix-7
Rentang frekuensi30 MHz sampai 3,8 GHz
Lebar kanal0,5 MHz sampai 90 MHz
Laju sampelHingga 122,88 MS/s (SISO), sekitar 80 MS/s (MIMO)
Kanal2 RX + 2 TX (MIMO 2×2)
Antarmuka hostPCIe 2.0 x2 dan USB 2.0, lewat konektor M.2 A+E key
Konektor RF4 × MHF4
Referensi clockStabilitas 0,5 ppm, bisa disinkronkan ke clock eksternal
GPIO12 + 2 pin, level 1,8 V
Ukuran30 × 22 mm (M.2 2230)

Dukungan perangkat lunak

Seluruh perangkat lunak xSDR bersifat open source. Library host usdr-lib berlisensi MIT, sedangkan gateware FPGA berlisensi CERN-OHL-P-2.0. xSDR bisa dipakai dengan SoapySDR (driver=usdr), GNU Radio dan srsRAN. Selain itu Wavelet Lab menyediakan platform web wsdr.io yang berjalan langsung di browser lewat WebUSB, berisi aplikasi siap pakai seperti penerima AM/FM, penganalisis spektrum dan waterfall.

Aplikasi web wsdr.io
Daftar aplikasi di platform web wsdr.io. Sumber gambar: Wavelet Lab / Crowd Supply.

Mengapa lewat PCIe?

USB 2.0 hanya sanggup mengalirkan data sekitar 40 MB/s, sedangkan PCIe 2.0 jauh lebih cepat dengan latensi lebih rendah. Itulah alasan pada tutorial ini xSDR dipasang di slot M.2 Raspberry Pi CM5 sebagai perangkat PCIe, bukan lewat USB.

Hasil akhir tutorial ini: driver kernel termuat otomatis saat boot, /dev/usdr0 muncul, dan xSDR bisa diakses dari Python lewat PCIe Gen2 (5 GT/s). Sebagai uji coba awal, kita membuat program penerima radio FM sederhana.

Pengenalan Carrier Board piBrick PocketCM5

Raspberry Pi Compute Module 5 (CM5) adalah versi modul dari Raspberry Pi 5. CM5 tidak punya port sendiri, semua antarmukanya (USB, HDMI, PCIe, dan lain-lain) dikeluarkan lewat dua konektor board-to-board di bagian bawah modul. Karena itu CM5 selalu membutuhkan carrier board.

Pada tutorial ini saya memakai piBrick PocketCM5, carrier board open source karya Ahmad Amarullah (amarullz), maker dari Indonesia. piBrick mengubah CM5 menjadi komputer genggam seukuran ponsel lengkap dengan layar AMOLED dan keyboard QWERTY.

piBrick PocketCM5
piBrick PocketCM5 dalam casing. Sumber gambar: piBrick oleh Ahmad Amarullah (amarullz).

Fitur utama piBrick PocketCM5:

  • Layar AMOLED 3,92 inci 1080 × 1240, multitouch, lewat MIPI DSI

  • Keyboard QWERTY BBQ20 dengan trackpad, dikendalikan RP2040 sebagai perangkat USB HID

  • Slot M.2 M-key untuk SSD NVMe ukuran 2230 atau 2242

  • USB 3.0 Type-A, USB 3.0 Type-C, USB 2.0 Type-A, USB 2.0 Type-C, dan header USB 2.0 internal

  • HDMI ukuran penuh dan micro HDMI

  • Kamera MIPI CSI, kartu suara internal dengan speaker stereo, jack 3,5 mm, akselerometer, RTC

  • Konektor I2C dan GPIO untuk ekspansi

  • Baterai LiPo 5000 mAh, ukuran 145 × 80 × 19,6 mm

  • Desain hardware, casing, dan firmware open source (GPL-3.0)

Peta konektor piBrick PocketCM5
Peta konektor piBrick PocketCM5. Slot M.2 bertanda "PCIe NVMe 2230/2242" ada di sisi kiri board. Sumber gambar: piBrick oleh Ahmad Amarullah (amarullz).

Antarmuka M.2 dan PCIe pada CM5

M.2 adalah standar konektor untuk kartu ekspansi kecil. Satu konektor M.2 bisa membawa beberapa jenis sinyal, tergantung key atau posisi takiknya:

KeySinyal utamaPenggunaan umum
A+EPCIe x2, USB 2.0Kartu WiFi/Bluetooth, xSDR
BPCIe x2, USB, SATAModem seluler, SSD SATA
MPCIe x4, SATASSD NVMe, slot M.2 di piBrick

Angka ukuran seperti 2230 atau 2242 berarti lebar 22 mm dan panjang 30 mm atau 42 mm. xSDR berukuran 2230 dengan key A+E, sedangkan slot di piBrick adalah M-key. Takiknya tidak cocok, jadi dibutuhkan adapter M.2 B+M-Key dari Wavelet Lab. Adapter ini hanya meneruskan jalur PCIe tanpa komponen aktif. Sesuaikan panjang adapter dengan lubang sekrup di board; piBrick menyediakan posisi 2230 dan 2242.

PCB piBrick PocketCM5 sisi atas dan bawah
PCB piBrick PocketCM5 sisi atas dan bawah. Slot M.2 dengan tanda posisi sekrup 2230 dan 2242 terlihat di kiri bawah sisi atas. Sumber gambar: CNX Software / amarullz.

Prosesor BCM2712 di CM5 menyediakan satu jalur (x1) PCIe 2.0 berkecepatan 5 GT/s yang dikeluarkan ke slot M.2 piBrick. xSDR sebenarnya mendukung PCIe x2, tetapi di CM5 ia otomatis berjalan di x1. Bandwidth satu jalur PCIe 2.0 sekitar 500 MB/s, masih jauh di atas USB 2.0, sehingga cukup untuk laju sampel tinggi.

Raspberry PiCM5 (BCM2712) Slot M.2 M-keypiBrick (2230/2242) AdapterB+M ke A+E key xSDRM.2 2230 Antena PCIe 2.0 x15 GT/sMHF4 RX
Jalur koneksi dari CM5 ke xSDR melalui slot M.2 piBrick.

Beberapa hal yang perlu diperhatikan:

  • Karena slot M.2 dipakai xSDR, sistem operasi harus boot dari eMMC CM5 atau kartu microSD, bukan dari SSD NVMe.

  • PCIe pada slot M.2 tidak aktif secara default. Kita mengaktifkannya lewat config.txt di Langkah 1.

  • Driver xSDR membutuhkan DMA 32-bit dan kernel dengan page 4 KB, yang juga diatur di Langkah 1.

Dokumentasi pemasangan perangkat

Sebelum masuk ke konfigurasi perangkat lunak, berikut dokumentasi perangkat keras yang saya pakai. Pengujian dilakukan dalam dua tahap: xSDR dicoba dulu lewat adapter M.2 ke USB untuk memastikan modulnya berfungsi, lalu dipindahkan ke slot M.2 pada carrier board piBrick PocketCM5 agar terhubung langsung ke jalur PCIe CM5.

Modul xSDR yang digunakan

Foto di bawah adalah modul xSDR yang dipakai dalam tutorial ini. PCB biru adalah xSDR berukuran 2230, dengan dua chip besar: transceiver RF LMS7002M dan FPGA Artix-7. Di bagian atas, xSDR sudah terpasang pada adapter M.2 kecil (PCB hijau dengan soket abu-abu) yang dipakai untuk menancapkannya ke slot M.2 piBrick. Di bagian bawah terlihat konektor antena MHF4 yang sudah terhubung ke kabel pigtail.

Modul xSDR terpasang pada adapter M.2
Modul xSDR yang dipakai, terpasang pada adapter M.2 kecil, dengan kabel pigtail MHF4 di bagian bawah. Foto: dokumentasi penulis.

Uji awal lewat adapter M.2 ke USB

Untuk uji pertama, xSDR dipasang pada adapter USB3_to_M2_E (PCB biru di bagian atas foto berikut). Adapter ini punya soket M.2 E-key, colokan USB, dan konektor SMA berlabel RX dan TX untuk antena. Dengan adapter ini xSDR cukup dicolokkan ke port USB CM5 dan dikenali oleh driver usdr lewat USB, tanpa perlu mengubah konfigurasi PCIe. Cara ini berguna untuk memastikan modul dan driver sudah benar sebelum memakai slot M.2. Pada tahap ini siaran radio FM sudah berhasil didengar.

Adapter USB3_to_M2_E dan piBrick PocketCM5 dengan xSDR
Atas: adapter USB3_to_M2_E untuk uji awal lewat USB. Bawah: carrier board piBrick PocketCM5 dengan xSDR di slot M.2 (kanan bawah). Foto: dokumentasi penulis.

Pemasangan pada slot M.2 carrier board piBrick

Setelah uji USB berhasil, xSDR dipindahkan ke slot M.2 di piBrick PocketCM5. Pada bagian bawah foto di atas terlihat xSDR sudah terpasang di sisi kanan bawah board, di sebelah konektor baterai, sementara CM5 tertutup heatsink dan kipas. Urutan pemasangannya:

  1. Matikan CM5 dan lepas catu daya.
  2. Masukkan xSDR (bersama adapter M.2-nya) ke slot M.2 dengan sudut kira-kira 30 derajat sampai kontak emasnya masuk penuh.
  3. Tekan modul rata ke board lalu kencangkan dengan sekrup pada lubang setengah lingkaran di ujung modul.
  4. Hubungkan kabel pigtail MHF4 ke konektor antena RX pada xSDR (lihat label di board), lalu pasang antena pada konektor SMA di ujung kabel.
  5. Nyalakan kembali CM5, lalu lanjutkan ke Langkah 1 untuk mengaktifkan PCIe.

Foto berikut memperlihatkan board piBrick dari sisi CM5 dengan xSDR di slot M.2 (kanan atas), kabel pigtail, dan antena all band yang dipakai untuk menerima siaran FM.

piBrick PocketCM5 dengan CM5, xSDR dan antena
piBrick PocketCM5 dengan Raspberry Pi CM5, xSDR di slot M.2 (kanan atas), kabel pigtail dan antena all band. Foto: dokumentasi penulis.

Konektor MHF4 sangat kecil dan mudah rusak. Pasang dan lepas dengan menekan lurus dari atas, jangan ditarik dari kabelnya.

Yang dibutuhkan

KomponenKeterangan
Raspberry Pi CM5Raspberry Pi OS 64-bit (Bookworm atau lebih baru)
Carrier board dengan slot M.2 M-keySaya memakai piBrick PocketCM5; slot NVMe biasa juga bisa
xSDR (Wavelet Lab)Modul M.2 2230 A+E key
M.2 B+M-Key AdapterMengubah A+E key xSDR ke M-key; tidak ada komponen aktif, boleh dipotong ke panjang 2242 atau 2260
Antena FMAntena teleskop atau kabel ±75 cm, ke konektor MHF4 RX xSDR (pigtail MHF4 ke SMA)
SpeakerJack, HDMI, atau speaker USB
InternetUntuk mengunduh paket dan kode sumber

Slot M.2 di carrier CM5 umumnya M-key untuk SSD NVMe, sedangkan xSDR memakai A+E key. Karena itu adapter B+M wajib dipakai. Pastikan adapter masuk penuh dan tersekrup rata sejajar board.

Adapter M.2 B+M-Key untuk xSDR
M.2 B+M-Key Adapter, pengubah konektor A+E key xSDR agar bisa dipasang di slot M-key. Sumber gambar: Wavelet Lab / Crowd Supply.

Langkah 1: Aktifkan PCIe di config.txt

Tiga baris di config.txt membuat CM5 bisa melihat xSDR. Tanpa baris ketiga, slot M.2 di CM5 saya selalu melaporkan link down.

Perbarui sistem dulu:

sudo apt update && sudo apt full-upgrade -y
sudo reboot

Cek nama overlay 32-bit DMA yang tersedia:

ls /boot/firmware/overlays | grep -i 32bit

Lalu tambahkan di bagian [all] paling bawah /boot/firmware/config.txt (sudo nano /boot/firmware/config.txt):

# Aktifkan PCIe ke slot M.2
dtparam=pciex1
# Kernel dengan page 4 KB (CM5 default memakai 16 KB)
kernel=kernel8.img
# DMA 32-bit untuk PCIe, sesuai panduan resmi Wavelet untuk Pi 5
dtoverlay=pcie-32bit-dma-pi5

Kalau yang muncul di perintah ls tadi hanya pcie-32bit-dma.dtbo, tulis dtoverlay=pcie-32bit-dma. Reboot, lalu cek:

getconf PAGESIZE                      # harus 4096
dmesg | grep -i "brcm-pcie.*link"     # harus: link up, 5.0 GT/s PCIe x1
lspci -nn | grep 10ee                 # xSDR muncul sebagai Xilinx [10ee:7049]

Contoh hasil yang benar:

brcm-pcie 1000110000.pcie: link up, 5.0 GT/s PCIe x1
0001:01:00.0 Memory controller: Xilinx Corporation Device [10ee:7049]

Kalau lspci belum menampilkan 10ee, jangan lanjut dulu. Lihat bagian Pemecahan masalah di bawah.

Langkah 2: Pasang library, modul SoapySDR, dan driver kernel

Kita membangun usdr-lib dari kode sumber. Cara ini jalan di Raspberry Pi OS Bookworm maupun Trixie. Waktu build di CM5 sekitar 5 sampai 15 menit.

2a. Paket yang dibutuhkan

sudo apt install -y git build-essential cmake pkg-config python3 python3-yaml \
  dwarves libusb-1.0-0-dev check dkms \
  libsoapysdr-dev soapysdr-tools python3-soapysdr python3-numpy python3-scipy \
  alsa-utils pciutils
sudo apt install -y linux-headers-$(uname -r) || sudo apt install -y linux-headers-rpi-v8

2b. Build dan install libusdr, alat CLI, dan modul SoapySDR

git clone https://github.com/wavelet-lab/usdr-lib.git ~/usdr-lib
cd ~/usdr-lib
cmake -S src -B build -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=/usr \
      -DENABLE_GUI=OFF -DENABLE_DMONITOR=OFF
cmake --build build -j$(nproc)
sudo cmake --install build
sudo ldconfig

-DCMAKE_INSTALL_PREFIX=/usr penting supaya modul SoapySDR terpasang di folder yang dicari SoapySDR bawaan Debian.

2c. Driver kernel PCIe lewat DKMS

Dengan DKMS, driver usdr_pcie_uram otomatis dibangun ulang setiap kali kernel diperbarui.

VER=$(git describe --tags --abbrev=0 | sed 's/^v//')
sudo mkdir -p /usr/src/usdr-dkms-$VER
sudo cp src/lib/lowlevel/pcie_uram/driver/usdr_pcie_uram.c \
        src/lib/lowlevel/pcie_uram/driver/Makefile \
        src/lib/lowlevel/pcie_uram/pcie_uram_driver_if.h \
        src/lib/device/device_cores.h \
        src/lib/ipblks/si2c.c src/lib/ipblks/si2c.h \
        src/lib/ipblks/spiext.c src/lib/ipblks/spiext.h \
        /usr/src/usdr-dkms-$VER/
sed "s/#MODULE_VERSION#/$VER/g" packaging/debian-bookworm/usdr-dkms.dkms.template \
  | sudo tee /usr/src/usdr-dkms-$VER/dkms.conf
sudo dkms add -m usdr-dkms -v $VER
sudo dkms build -m usdr-dkms -v $VER
sudo dkms install -m usdr-dkms -v $VER

2d. Izin akses dan muat otomatis saat boot

sudo cp src/lib/lowlevel/pcie_uram/driver/helpers/50-usdr-pcie-driver.rules /etc/udev/rules.d/
echo usdr_pcie_uram | sudo tee /etc/modules-load.d/usdr.conf
sudo udevadm control --reload-rules
sudo modprobe usdr_pcie_uram
sudo udevadm trigger

Langkah 3: Pastikan xSDR terdeteksi

Empat perintah ini harus berhasil semua sebelum lanjut ke Python:

lsmod | grep usdr                   # usdr_pcie_uram termuat
ls -l /dev/usdr0                    # node perangkat ada, izin crw-rw-rw-
dmesg | grep usdr                   # usdr 0001:01:00.0: is linked to /dev/usdr0
SoapySDRUtil --find="driver=usdr"   # ditemukan dengan bus=pci, device=usdr0

Untuk tes rekam IQ mentah tanpa Python (sekitar 1 detik di 100 MHz):

usdr_dm_create -r 2e6 -e 100e6 -w 1.5e6 -c 500 -S 4096 -f /tmp/rx.data
ls -l /tmp/rx.data

Kalau ukuran file lebih dari 0, streaming dari xSDR lewat PCIe sudah jalan.

Cek xSDR otomatis setiap masuk desktop

Kartu M.2 kadang tidak terdeteksi setelah CM5 dinyalakan, misalnya karena adapter sedikit bergeser. Supaya tidak perlu mengetik ulang, empat perintah pengecekan ini saya jalankan otomatis setiap kali masuk desktop Raspbian:

lspci | grep 10ee        # untuk M.2, harus muncul 10ee:7049
ls /dev/usdr0            # device M.2 harus ada
lsusb | grep 3727        # kalau pakai adapter USB
SoapySDRUtil --find="driver=usdr"

Simpan skrip berikut sebagai ~/xsdr/cek_xsdr.sh. Skrip ini menjalankan keempat perintah di atas, menuliskan kesimpulan (siap lewat M.2, siap lewat USB, atau tidak terdeteksi), dan menyimpan hasilnya ke ~/xsdr/cek_xsdr.log:

#!/bin/bash
# Cek apakah xSDR terdeteksi (M.2 / USB / SoapySDR). Hasil juga disimpan ke ~/xsdr/cek_xsdr.log
LOG=~/xsdr/cek_xsdr.log
mkdir -p ~/xsdr
{
echo "=== Cek xSDR  $(date '+%Y-%m-%d %H:%M:%S') ==="
echo "[M.2] lspci | grep 10ee"
lspci | grep 10ee && M2=1 || echo "  -> tidak ada 10ee:7049 (slot M.2 link down / kartu belum terpasang rata)"
echo "[M.2] ls /dev/usdr0"
ls /dev/usdr0 2>/dev/null || echo "  -> /dev/usdr0 tidak ada (coba: sudo modprobe usdr_pcie_uram)"
echo "[USB] lsusb | grep 3727"
lsusb | grep 3727 && USB=1 || echo "  -> xSDR USB tidak terdeteksi"
echo "[SoapySDR] SoapySDRUtil --find=\"driver=usdr\""
SoapySDRUtil --find="driver=usdr" 2>&1 | grep -v "^$"
echo
if [ -n "$M2" ] && [ -e /dev/usdr0 ]; then echo "HASIL: xSDR siap lewat M.2  ->  python3 ~/xsdr/fm_gui.py --device m2"
elif [ -n "$USB" ]; then echo "HASIL: xSDR siap lewat USB  ->  python3 ~/xsdr/fm_gui.py --device usb"
else echo "HASIL: xSDR TIDAK terdeteksi. Matikan, cabut daya, pasang ulang xSDR/adapter, lalu nyalakan lagi."
fi
} 2>&1 | tee "$LOG"

Lalu buat file autostart ~/xsdr/cek-xsdr.desktop. File ini membuka jendela terminal "Cek xSDR" saat login, dan jendela itu menunggu sampai Enter ditekan:

[Desktop Entry]
Type=Application
Name=Cek xSDR
Comment=Cek deteksi xSDR setiap login
Exec=lxterminal --title="Cek xSDR" -e bash -c 'sleep 3; ~/xsdr/cek_xsdr.sh; echo; read -p "Tekan Enter untuk menutup..."'
Terminal=false
X-GNOME-Autostart-enabled=true

Pasang keduanya (tidak perlu sudo):

chmod +x ~/xsdr/cek_xsdr.sh
mkdir -p ~/.config/autostart
cp ~/xsdr/cek-xsdr.desktop ~/.config/autostart/

Kalau hasilnya "xSDR TIDAK terdeteksi", matikan CM5 dengan sudo poweroff, cabut daya, pasang ulang xSDR dan adapternya sampai rata, lalu nyalakan lagi. Untuk mematikan pengecekan otomatis: rm ~/.config/autostart/cek-xsdr.desktop.

Langkah 4: Akses xSDR dari Python dan penerima radio FM

Setelah driver terpasang, xSDR bisa diakses dari Python lewat SoapySDR dengan driver=usdr. Pola dasarnya seperti ini:

import numpy as np
import SoapySDR
from SoapySDR import SOAPY_SDR_RX, SOAPY_SDR_CF32

dev = SoapySDR.Device("driver=usdr")
dev.setSampleRate(SOAPY_SDR_RX, 0, 1.92e6)
dev.setFrequency(SOAPY_SDR_RX, 0, 100e6)
dev.setAntenna(SOAPY_SDR_RX, 0, "LNAW")
for name, val in (("LNA", 30), ("TIA", 12), ("PGA", 0)):
    dev.setGain(SOAPY_SDR_RX, 0, name, val)

rx = dev.setupStream(SOAPY_SDR_RX, SOAPY_SDR_CF32, [0])
dev.activateStream(rx)
buf = np.zeros(65536, np.complex64)
sr = dev.readStream(rx, [buf], len(buf), timeoutUs=1_000_000)
print("sampel diterima:", sr.ret)
dev.deactivateStream(rx)
dev.closeStream(rx)

Catatan penting: pada driver usdr, setGain() dengan satu angka total hanya mengatur LNA. Supaya penguatan penuh, atur tiap tahap secara terpisah: LNA (0 sampai 30 dB), TIA (0 sampai 12 dB) dan PGA (−12 sampai 19 dB). Untuk pita FM gunakan antena LNAW.

Program penerima radio FM

Program di bawah menerima siaran FM mono dan memutarnya lewat aplay. Alurnya:

  1. Tuning 250 kHz di bawah frekuensi stasiun agar spike DC tidak mengganggu, lalu sinyal digeser kembali ke 0 Hz secara digital.

  2. Sample rate 1,92 MS/s, filter low-pass 100 kHz untuk memilih satu stasiun, lalu decimate ×8 menjadi 240 kS/s.

  3. Demodulasi FM dengan quadrature discriminator (beda fase antar sampel).

  4. De-emphasis 50 µs (standar Indonesia dan Eropa).

  5. Filter audio 15 kHz, decimate ×5 menjadi 48 kHz.

  6. Kirim audio 16-bit ke aplay.

Simpan sebagai fm_mini.py:

#!/usr/bin/env python3
# Penerima FM mono sederhana untuk xSDR (SoapySDR, driver=usdr)
# Pakai: python3 fm_mini.py 98.7   (Ctrl+C untuk berhenti)
import subprocess, sys
import numpy as np
from scipy import signal
import SoapySDR
from SoapySDR import SOAPY_SDR_RX, SOAPY_SDR_CF32

FREQ = float(sys.argv[1]) * 1e6 if len(sys.argv) > 1 else 98.7e6
FS, OFFSET = 1.92e6, 250e3          # tune 250 kHz di bawah stasiun, hindari spike DC
D1, D2 = 8, 5                       # 1.92 MS/s -> 240 kS/s -> 48 kHz
BLOCK = 96000                       # 50 ms

dev = SoapySDR.Device("driver=usdr")
dev.setSampleRate(SOAPY_SDR_RX, 0, FS)
dev.setFrequency(SOAPY_SDR_RX, 0, FREQ - OFFSET)
dev.setAntenna(SOAPY_SDR_RX, 0, "LNAW")
# setGain() total hanya mengatur LNA, jadi atur tiap tahap
for name, val in (("LNA", 30), ("TIA", 12), ("PGA", 19)):
    dev.setGain(SOAPY_SDR_RX, 0, name, val)

rx = dev.setupStream(SOAPY_SDR_RX, SOAPY_SDR_CF32, [0])
dev.activateStream(rx)
aplay = subprocess.Popen(["aplay", "-q", "-t", "raw", "-f", "S16_LE", "-c", "1", "-r", "48000"],
                         stdin=subprocess.PIPE)

lp1 = signal.firwin(65, 100e3, fs=FS)              # pilih satu stasiun (+-100 kHz)
lp2 = signal.firwin(65, 15e3, fs=FS / D1)          # audio 15 kHz
z1 = np.zeros(len(lp1) - 1, complex)
z2 = np.zeros(len(lp2) - 1)
alpha = 1 - np.exp(-1 / (FS / D1 * 50e-6))         # de-emphasis 50 us
zd = np.zeros(1)
mix = np.exp(-2j * np.pi * OFFSET / FS * np.arange(BLOCK))
phase, last = 1 + 0j, 1 + 0j
buf = np.zeros(BLOCK, np.complex64)

try:
    while True:
        got = 0
        while got < BLOCK:
            sr = dev.readStream(rx, [buf[got:]], BLOCK - got, timeoutUs=1_000_000)
            if sr.ret > 0:
                got += sr.ret
        x = buf * mix * phase                          # geser stasiun ke 0 Hz
        phase *= np.exp(-2j * np.pi * OFFSET / FS * BLOCK)
        x, z1 = signal.lfilter(lp1, 1, x, zi=z1)
        x = x[::D1]                                    # 240 kS/s
        d = np.angle(x * np.conj(np.r_[last, x[:-1]])) # demodulasi FM
        last = x[-1]
        d, zd = signal.lfilter([alpha], [1, alpha - 1], d, zi=zd)
        a, z2 = signal.lfilter(lp2, 1, d, zi=z2)
        a = a[::D2] / (2 * np.pi * 75e3 / (FS / D1))   # 48 kHz, skala ke +-1
        aplay.stdin.write((np.clip(a * 0.8, -1, 1) * 32767).astype("<i2").tobytes())
except KeyboardInterrupt:
    pass
finally:
    dev.deactivateStream(rx)
    dev.closeStream(rx)
    aplay.stdin.close()

Jalankan dengan frekuensi stasiun dalam MHz:

python3 fm_mini.py 98.7

Jika audio keluar ke perangkat yang salah (misalnya Anda memakai speaker USB), lihat daftar kartu suara dengan aplay -l, lalu tambahkan "-D", "plughw:2,0" (sesuaikan nomornya) ke daftar argumen aplay di skrip. Jangan lupa pasang antena di konektor MHF4 RX xSDR; tanpa antena yang terdengar hanya desis.

Versi GUI: penerima FM bergaya wsdr.io

Program di atas sengaja dibuat sederhana dan berjalan di terminal. Untuk pemakaian sehari-hari saya juga membuat versi GUI, fm_gui.py, yang tampilannya meniru FM receiver di wsdr.io.

Fungsi setiap bagiannya saya jelaskan di subbagian berikutnya, lengkap dengan tangkapan layar langsung dari CM5.

Tampilan fm_gui.py bergaya wsdr.io
Tampilan fm_gui.py dalam mode simulasi di 101,1 MHz: waterfall, spektrum RF, dan spektrum MPX dengan pilot 19 kHz.

Program ini membutuhkan Tkinter. Pasang lalu jalankan:

sudo apt install -y python3-tk
python3 fm_gui.py --audio plughw:2,0

Kode lengkap fm_gui.py:

#!/usr/bin/env python3
"""
Penerima radio FM dengan GUI untuk xSDR (dan SDR lain yang didukung SoapySDR).

Fitur:
  - pilihan input SDR (hasil SoapySDR enumerate) + mode simulasi tanpa hardware
  - pilihan output audio (aplay)
  - slider frekuensi, gain, volume
  - tampilan frekuensi gaya wsdr.io: waterfall +-500 kHz di sekitar stasiun,
    spektrum RF (PSD) dan spektrum MPX hasil demodulasi (pilot 19 kHz, stereo, RDS)
  - klik waterfall untuk pindah ke stasiun di posisi itu

Pakai:  python3 fm_gui.py            (butuh: sudo apt install python3-tk)
        python3 fm_gui.py --freq 98.7
"""

import argparse
import queue
import re
import subprocess
import threading
import time
import tkinter as tk
from tkinter import ttk

import numpy as np
from scipy import signal

try:
    import SoapySDR
    from SoapySDR import SOAPY_SDR_RX, SOAPY_SDR_CF32
except ImportError:  # GUI tetap bisa jalan dalam mode simulasi
    SoapySDR = None

FS = 1.92e6            # sample rate SDR
OFFSET = 250e3         # tune 250 kHz di bawah stasiun, hindari spike DC
D1, D2 = 8, 5          # 1.92 MS/s -> 240 kS/s -> 48 kHz
AUDIO_RATE = 48000
BLOCK = 96000          # 50 ms
FMIN, FMAX = 87.5, 108.0
SIM_LABEL = "Simulasi (tanpa SDR)"


# ---------------------------------------------------------------- sumber sinyal
class SoapySource:
    """SDR asli lewat SoapySDR. Untuk xSDR gain dibagi manual ke LNA/TIA/PGA."""

    def __init__(self, kwargs, ch=0):
        # beberapa binding SoapySDR gagal/crash bila diberi dict, jadi pakai string "k=v,..."
        args = {k: str(kwargs[k]) for k in ("driver", "bus", "device", "serial") if k in kwargs}
        args = {k: v for k, v in args.items() if "," not in v and "=" not in v}
        self.dev = SoapySDR.Device(",".join(f"{k}={v}" for k, v in args.items()))
        self.driver = self.dev.getDriverKey()
        # xSDR punya dua kanal penerima: kanal 0 = port RX A, kanal 1 = port RX B
        n = self.dev.getNumChannels(SOAPY_SDR_RX)
        if ch >= n:
            raise RuntimeError(f"{self.driver} hanya punya {n} kanal RX, RX B tidak tersedia")
        self.ch = ch
        self.dev.setSampleRate(SOAPY_SDR_RX, ch, FS)
        ants = list(self.dev.listAntennas(SOAPY_SDR_RX, ch))
        if "LNAW" in ants:
            self.dev.setAntenna(SOAPY_SDR_RX, ch, "LNAW")
        self.gains = list(self.dev.listGains(SOAPY_SDR_RX, ch))
        r = self.dev.getGainRange(SOAPY_SDR_RX, ch)
        self.gain_range = (r.minimum(), r.maximum())
        self.stream = self.dev.setupStream(SOAPY_SDR_RX, SOAPY_SDR_CF32, [ch])
        self.dev.activateStream(self.stream)

    def set_freq(self, hz):
        self.dev.setFrequency(SOAPY_SDR_RX, self.ch, hz)

    def set_gain(self, db):
        # setGain() total di driver usdr hanya mengatur LNA, jadi dibagi per tahap
        if {"LNA", "TIA", "PGA"} <= set(self.gains):
            lna = min(max(db, 0.0), 30.0)
            tia = min(max(db - lna, 0.0), 12.0)
            pga = min(max(db - lna - tia, -12.0), 19.0)
            for name, val in (("LNA", lna), ("TIA", tia), ("PGA", pga)):
                self.dev.setGain(SOAPY_SDR_RX, self.ch, name, val)
        else:
            self.dev.setGain(SOAPY_SDR_RX, self.ch, db)

    def read(self, buf):
        got = 0
        while got < len(buf):
            sr = self.dev.readStream(self.stream, [buf[got:]], len(buf) - got, timeoutUs=1_000_000)
            if sr.ret > 0:
                got += sr.ret
            elif sr.ret not in (-1, -4):  # -1 timeout, -4 overflow: cukup lanjut
                raise RuntimeError(f"readStream error {sr.ret}")
        return buf

    def close(self):
        self.dev.deactivateStream(self.stream)
        self.dev.closeStream(self.stream)


class SimSource:
    """Beberapa stasiun FM tiruan (nada audio berbeda) + noise, untuk mencoba GUI tanpa SDR."""

    STATIONS = {95.0e6: 440.0, 98.7e6: 1000.0, 99.0e6: 550.0, 101.1e6: 660.0, 101.4e6: 770.0,
                105.3e6: 880.0}
    gain_range = (0.0, 61.0)
    driver = "sim"

    def __init__(self):
        self.lo = 98.7e6 - OFFSET
        self.gain = 40.0
        self.n = 0
        self.t_next = time.monotonic()
        self.rng = np.random.default_rng()

    def set_freq(self, hz):
        self.lo = hz

    def set_gain(self, db):
        self.gain = db

    def read(self, buf):
        n = len(buf)
        t = (self.n + np.arange(n)) / FS
        x = (self.rng.standard_normal(n) + 1j * self.rng.standard_normal(n)) * 0.003
        for f_st, tone in self.STATIONS.items():
            off = f_st - self.lo
            if abs(off) < FS / 2 - 100e3:
                # FM: deviasi 75 kHz, nada audio
                ph = (2 * np.pi * off * t + (67.5e3 / tone) * np.sin(2 * np.pi * tone * t)
                      + (7.5e3 / 19e3) * np.sin(2 * np.pi * 19e3 * t))   # + pilot 19 kHz
                x += 0.05 * np.exp(1j * ph)
        x *= 10 ** ((self.gain - 40.0) / 20)
        buf[:] = np.clip(x.real, -1, 1) + 1j * np.clip(x.imag, -1, 1)
        self.n += n
        # jaga laju data seperti hardware sungguhan
        self.t_next += n / FS
        delay = self.t_next - time.monotonic()
        if delay > 0:
            time.sleep(delay)
        else:
            self.t_next = time.monotonic()
        return buf

    def close(self):
        pass


# ---------------------------------------------------------------- DSP & audio
NFFT_RF, NFFT_MPX = 2048, 1024
SPAN = 500e3           # lebar tampilan waterfall: +-500 kHz di sekitar frekuensi tengah
BW = 200e3             # lebar kanal awal (filter 100 kHz kiri-kanan)
BW_MIN, BW_MAX = 30e3, 220e3   # batas lebar kanal (dibatasi laju 240 kS/s setelah decimate)


def psd_db(x, nfft, nseg, scale=1.0):
    """PSD rata-rata beberapa segmen, dalam dBFS (nada skala penuh = 0 dB)."""
    w = np.hanning(nfft)
    seg = x[:nfft * nseg].reshape(nseg, nfft) * w
    p = np.mean(np.abs(np.fft.fft(seg, axis=1)) ** 2, axis=0) / (w.sum() * scale) ** 2
    return 10 * np.log10(p + 1e-15)


class FMDemod:
    """Geser frekuensi tengah ke 0 Hz (untuk spektrum), lalu geser lagi sejauh offset kanal df,
    filter selebar kanal, demodulasi FM, de-emphasis 50 us, decimate ke 48 kHz.
    Sekalian menghitung spektrum RF (sebelum filter) dan spektrum MPX (sebelum de-emphasis)."""

    def __init__(self):
        self.set_channel(0.0, BW)
        self.lp2 = signal.firwin(65, 15e3, fs=FS / D1)
        self.z1 = np.zeros(len(self.lp1) - 1, complex)
        self.z2 = np.zeros(len(self.lp2) - 1)
        self.alpha = 1 - np.exp(-1 / (FS / D1 * 50e-6))
        self.zd = np.zeros(1)
        self.mix = np.exp(-2j * np.pi * OFFSET / FS * np.arange(BLOCK))
        self.step = np.exp(-2j * np.pi * OFFSET / FS * BLOCK)
        self.phase = 1 + 0j
        self.last = 1 + 0j

    def set_channel(self, df, bw):
        """Pindah kanal ke df Hz dari frekuensi tengah, dengan lebar bw Hz."""
        self.df, self.bw = df, bw
        self.lp1 = signal.firwin(65, bw / 2, fs=FS)
        self.mix2 = np.exp(-2j * np.pi * df / FS * np.arange(BLOCK))
        self.step2 = np.exp(-2j * np.pi * df / FS * BLOCK)
        self.phase2 = 1 + 0j

    def process(self, iq):
        x = iq * self.mix * self.phase
        self.phase *= self.step
        self.phase /= abs(self.phase)
        rf = np.fft.fftshift(psd_db(x, NFFT_RF, 8))
        if self.df:
            x = x * self.mix2 * self.phase2
            self.phase2 *= self.step2
            self.phase2 /= abs(self.phase2)
        x, self.z1 = signal.lfilter(self.lp1, 1, x, zi=self.z1)
        x = x[::D1]
        rssi = 10 * np.log10(np.mean(np.abs(x) ** 2) + 1e-12)   # dBFS di kanal
        d = np.angle(x * np.conj(np.r_[self.last, x[:-1]]))
        self.last = x[-1]
        mpx = psd_db(d, NFFT_MPX, 8, scale=2 * np.pi * 75e3 / (FS / D1))[:NFFT_MPX // 2]
        d, self.zd = signal.lfilter([self.alpha], [1, self.alpha - 1], d, zi=self.zd)
        a, self.z2 = signal.lfilter(self.lp2, 1, d, zi=self.z2)
        a = a[::D2] / (2 * np.pi * 75e3 / (FS / D1))
        return a, rssi, rf, mpx


class AudioSink:
    def __init__(self, device):
        cmd = ["aplay", "-q", "-t", "raw", "-f", "S16_LE", "-c", "1", "-r", str(AUDIO_RATE)]
        if device and device != "default":
            cmd += ["-D", device]
        try:
            self.p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.DEVNULL)
        except FileNotFoundError:
            self.p = None  # tanpa aplay: audio dibuang, GUI tetap jalan

    def write(self, a):
        if self.p and self.p.poll() is None:
            self.p.stdin.write((np.clip(a, -1, 1) * 32767).astype("<i2").tobytes())

    def close(self):
        if self.p:
            try:
                self.p.stdin.close()
            except OSError:
                pass
            self.p.wait(timeout=2)


def list_audio_devices():
    devs = ["default"]
    try:
        out = subprocess.run(["aplay", "-l"], capture_output=True, text=True, timeout=3).stdout
        for card, name, dev in re.findall(r"card (\d+): \S+ \[(.+?)\], device (\d+):", out):
            devs.append(f"plughw:{card},{dev}  ({name})")
    except (FileNotFoundError, subprocess.TimeoutExpired):
        pass
    return devs


XSDR_M2, XSDR_USB = "xSDR M.2 (PCIe)", "xSDR USB"
RX_PORTS = ["RX A", "RX B"]   # port antena xSDR = kanal RX 0 / 1


def xsdr_kind(d):
    """'usb' atau 'pci' untuk perangkat usdr hasil enumerate."""
    info = " ".join(str(v) for v in d.values()).lower()
    return "usb" if "usb" in info else "pci"


def list_sdr_devices():
    """Kembalikan [(label, kwargs)]: xSDR M.2, xSDR USB, SDR lain, lalu mode simulasi.
    xSDR yang tidak terdeteksi tetap dicantumkan supaya pilihannya selalu ada."""
    found = []
    if SoapySDR is not None:
        try:
            found = [{key: k[key] for key in k.keys()} for k in SoapySDR.Device.enumerate()]
        except Exception:
            found = []
    items, others = [], []
    for kind, name in (("pci", XSDR_M2), ("usb", XSDR_USB)):
        hits = [d for d in found if d.get("driver") == "usdr" and xsdr_kind(d) == kind]
        for d in hits:
            tag = d.get("device") or d.get("serial") or ""
            items.append((f"{name}  {tag}".rstrip(), d))
        if not hits:
            items.append((f"{name}  (tidak terdeteksi)", {"driver": "usdr", "bus": kind}))
    for d in found:
        if d.get("driver") != "usdr":
            label = d.get("label") or d.get("driver", "?")
            extra = ", ".join(f"{x}={d[x]}" for x in ("bus", "device", "serial") if x in d)
            others.append((f"{label} [{d.get('driver', '?')}{', ' + extra if extra else ''}]", d))
    return items + others + [(SIM_LABEL, None)]


class Receiver(threading.Thread):
    """Thread pembaca SDR + DSP. Perubahan frekuensi/gain diterapkan di antara blok."""

    def __init__(self, source, audio_dev, freq_hz, gain_db, volume):
        super().__init__(daemon=True)
        self.src = source
        self.audio = AudioSink(audio_dev)
        self.demod = FMDemod()
        self.cmds = queue.Queue()
        self.volume = volume
        self.rssi = -120.0
        self.rf = self.mpx = None     # spektrum terbaru, diambil GUI
        self.error = None
        self.running = True
        self.src.set_freq(freq_hz - OFFSET)
        self.src.set_gain(gain_db)

    def tune(self, freq_hz):
        self.cmds.put(("freq", freq_hz))

    def gain(self, db):
        self.cmds.put(("gain", db))

    def channel(self, df, bw):
        self.cmds.put(("chan", (df, bw)))

    def run(self):
        buf = np.zeros(BLOCK, np.complex64)
        try:
            while self.running:
                while not self.cmds.empty():
                    what, val = self.cmds.get_nowait()
                    if what == "freq":
                        self.src.set_freq(val - OFFSET)
                    elif what == "chan":
                        self.demod.set_channel(*val)
                    else:
                        self.src.set_gain(val)
                a, self.rssi, self.rf, self.mpx = self.demod.process(self.src.read(buf))
                self.audio.write(a * self.volume)
        except Exception as e:  # tampilkan di status bar
            self.error = str(e)
        finally:
            self.src.close()
            self.audio.close()

    def stop(self):
        self.running = False


# ---------------------------------------------------------------- GUI
BLUE, BG, PANEL = "#3d5fa6", "#eef1f8", "#ffffff"
FONT, FONT_B, FONT_S = ("DejaVu Sans", 11), ("DejaVu Sans", 11, "bold"), ("DejaVu Sans", 8)


def make_cmap():
    """Peta warna biru tua -> cyan -> hijau -> kuning -> merah, seperti colorbar wsdr."""
    anchors = [(0.0, (0, 0, 150)), (0.25, (0, 120, 220)), (0.45, (60, 190, 160)),
               (0.6, (90, 180, 40)), (0.78, (220, 200, 40)), (1.0, (170, 20, 10))]
    t = np.linspace(0, 1, 256)
    xs = [a[0] for a in anchors]
    return np.stack([np.interp(t, xs, [a[1][i] for a in anchors]) for i in range(3)], 1).astype(np.uint8)


CMAP = make_cmap()


class Waterfall(tk.Canvas):
    """Waterfall +-SPAN di sekitar frekuensi tengah, colorbar dB di kiri, sumbu kHz di bawah.
    Kanal yang didemodulasi disorot; geser dengan mouse, tarik tepinya untuk melebarkan/menyempitkan,
    klik di luar kanal untuk memindahkannya ke sana, scroll untuk geser 5 kHz,
    klik ganda untuk menjadikan kanal sebagai frekuensi tengah."""

    EDGE = 6                 # jarak (piksel) untuk menangkap tepi kanal

    W, H = 640, 245          # ukuran gambar waterfall (muat di layar 1280x600)
    L, B = 70, 50            # margin kiri (colorbar + label) dan bawah (sumbu)

    def __init__(self, master, on_channel, on_recenter):
        super().__init__(master, width=self.L + self.W + 15, height=self.H + self.B + 15,
                         bg=PANEL, highlightthickness=0)
        self.lo, self.hi = -110.0, -10.0
        self.rows = np.zeros((self.H, self.W, 3), np.uint8)
        self.rows[:] = CMAP[0]
        self.img = tk.PhotoImage(width=self.W, height=self.H)
        self.create_image(self.L, 10, image=self.img, anchor="nw")
        self.fx = np.linspace(-SPAN, SPAN, self.W)
        self.df, self.bw = 0.0, BW
        self.band = np.abs(self.fx) <= BW / 2
        self.on_channel, self.on_recenter = on_channel, on_recenter
        self.drag = None
        self.cbar = tk.PhotoImage(width=12, height=self.H)
        self.create_image(self.L - 22, 10, image=self.cbar, anchor="nw")
        self._axes()
        self.edges = [self.create_line(0, 10, 0, self.H + 10, fill="#ffffff", width=1, dash=(3, 3))
                      for _ in range(2)]
        self.cline = self.create_line(0, 10, 0, self.H + 10, fill="#ff3030", width=2)
        self.bind("<Motion>", self._hover)
        self.bind("<Button-1>", self._press)
        self.bind("<B1-Motion>", self._move)
        self.bind("<ButtonRelease-1>", lambda e: setattr(self, "drag", None))
        self.bind("<Double-Button-1>", lambda e: self.on_recenter())
        self.bind("<MouseWheel>", lambda e: self._wheel(1 if e.delta > 0 else -1))
        self.bind("<Button-4>", lambda e: self._wheel(1))
        self.bind("<Button-5>", lambda e: self._wheel(-1))
        self.set_channel(0.0, BW)
        self.render()

    def _axes(self):
        self.delete("axis")
        col = CMAP[np.linspace(255, 0, self.H).astype(int)]
        self.cbar.put(" ".join("{" + " ".join(["#%02x%02x%02x" % tuple(c)] * 12) + "}" for c in col))
        y0 = 10
        for db in range(int(np.ceil(self.lo / 10) * 10), int(self.hi) + 1, 10):
            y = y0 + (self.hi - db) / (self.hi - self.lo) * self.H
            self.create_text(self.L - 28, y, text=str(db), anchor="e", font=FONT_S, tags="axis")
        for k in range(-400, 401, 200):
            x = self.L + (k * 1e3 + SPAN) / (2 * SPAN) * self.W
            self.create_line(x, self.H + 10, x, self.H + 15, tags="axis")
            self.create_text(x, self.H + 25, text=str(k), font=FONT, tags="axis")
        self.create_text(self.L + self.W / 2, self.H + 45, text="Frekuensi, kHz", font=FONT, tags="axis")

    def set_range(self, lo, hi):
        self.lo, self.hi = float(lo), float(hi)
        self._axes()

    def push(self, f_hz, p_db):
        """Tambah satu baris spektrum (f relatif stasiun, dB) di atas waterfall."""
        p = np.interp(self.fx, f_hz, p_db)
        idx = np.clip((p - self.lo) / (self.hi - self.lo) * 255, 0, 255).astype(int)
        self.rows = np.roll(self.rows, 1, axis=0)
        self.rows[0] = CMAP[idx]
        self.render()

    def render(self):
        rgb = self.rows.copy()
        rgb[:, self.band] = rgb[:, self.band] // 2 + 60      # sorot lebar kanal yang didemodulasi
        ppm = b"P6 %d %d 255\n" % (self.W, self.H) + rgb.tobytes()
        self.img.configure(data=ppm, format="PPM")

    # --- konversi piksel <-> frekuensi relatif tengah
    def _f(self, x):
        return (min(max(x, self.L), self.L + self.W) - self.L) / self.W * 2 * SPAN - SPAN

    def _x(self, f):
        return self.L + (f + SPAN) / (2 * SPAN) * self.W

    def set_channel(self, df, bw):
        """Gambar ulang sorotan kanal (dipanggil App setiap kanal berubah)."""
        self.df, self.bw = df, bw
        self.band = np.abs(self.fx - df) <= bw / 2
        for item, f in zip(self.edges, (df - bw / 2, df + bw / 2)):
            self.coords(item, self._x(f), 10, self._x(f), self.H + 10)
        self.coords(self.cline, self._x(df), 10, self._x(df), self.H + 10)
        self.render()

    def _where(self, x):
        xl, xr = self._x(self.df - self.bw / 2), self._x(self.df + self.bw / 2)
        if abs(x - xl) <= self.EDGE or abs(x - xr) <= self.EDGE:
            return "edge"
        return "inside" if xl < x < xr else "outside"

    def _hover(self, e):
        if self.drag is None:
            w = self._where(e.x)
            self.configure(cursor={"edge": "sb_h_double_arrow", "inside": "fleur"}.get(w, "crosshair"))

    def _press(self, e):
        if not self.L <= e.x < self.L + self.W or e.y > self.H + 10:
            return
        w = self._where(e.x)
        f = self._f(e.x)
        if w == "outside":                       # lompat: tengah kanal ke titik klik
            self._emit(f, self.bw)
            w = "inside"
        self.drag = ("edge", 0.0) if w == "edge" else ("move", f - self.df)

    def _move(self, e):
        if self.drag is None:
            return
        mode, grab = self.drag
        f = self._f(e.x)
        if mode == "edge":
            self._emit(self.df, 2 * abs(f - self.df))
        else:
            self._emit(f - grab, self.bw)

    def _wheel(self, sign):
        self._emit(self.df + sign * 5e3, self.bw)

    def _emit(self, df, bw):
        bw = min(max(bw, BW_MIN), BW_MAX)
        lim = SPAN - bw / 2
        df = round(min(max(df, -lim), lim) / 100) * 100      # resolusi 0,1 kHz
        self.on_channel(df, bw)


class Plot(tk.Canvas):
    """Grafik garis sederhana (PSD) dengan area terisi, sumbu dan grid."""

    def __init__(self, master, xlim, ylim, xticks, yticks, xlabel, ylabel, w=420, h=168):
        super().__init__(master, width=w, height=h, bg=PANEL, highlightthickness=0)
        self.xlim, self.ylim = xlim, ylim
        self.x0, self.y0, self.x1, self.y1 = 62, 10, w - 12, h - 40
        self.create_rectangle(self.x0, self.y0, self.x1, self.y1, fill="#e3edf9", outline="#999")
        for v in xticks:
            x = self._x(v)
            self.create_line(x, self.y0, x, self.y1, fill="#cfd8e6")
            self.create_text(x, self.y1 + 10, text=f"{v:g}", font=FONT_S)
        for v in yticks:
            y = self._y(v)
            self.create_line(self.x0, y, self.x1, y, fill="#cfd8e6")
            self.create_text(self.x0 - 5, y, text=f"{v:g}", anchor="e", font=FONT_S)
        self.create_text((self.x0 + self.x1) / 2, h - 12, text=xlabel, font=FONT_S)
        self.create_text(14, (self.y0 + self.y1) / 2, text=ylabel, angle=90, font=FONT_S)
        self.bandrect = self.create_rectangle(0, 0, 0, 0, fill="#8fb4e6", stipple="gray25", outline="#2f6fd0")
        self.area = self.create_polygon(0, 0, 0, 0, fill="#b9d3f2", outline="")
        self.line = self.create_line(0, 0, 0, 0, fill="#2f6fd0", width=1)

    def _x(self, v):
        return self.x0 + (v - self.xlim[0]) / (self.xlim[1] - self.xlim[0]) * (self.x1 - self.x0)

    def _y(self, v):
        v = min(max(v, self.ylim[0]), self.ylim[1])
        return self.y1 - (v - self.ylim[0]) / (self.ylim[1] - self.ylim[0]) * (self.y1 - self.y0)

    def set_band(self, lo, hi):
        self.coords(self.bandrect, self._x(lo), self.y0, self._x(hi), self.y1)

    def set_data(self, xs, ys):
        m = (xs >= self.xlim[0]) & (xs <= self.xlim[1])
        pts = [c for x, y in zip(xs[m], ys[m]) for c in (self._x(x), self._y(y))]
        if len(pts) >= 4:
            self.coords(self.line, *pts)
            self.coords(self.area, self._x(xs[m][0]), self.y1, *pts, self._x(xs[m][-1]), self.y1)


class App:
    def __init__(self, root, args):
        self.root = root
        self.rx = None
        self.avg_rf = self.avg_mpx = None
        root.title("Penerima FM xSDR")
        root.configure(bg=BG)
        style = ttk.Style()
        style.configure("TLabel", background=BG, font=FONT)
        style.configure("H.TLabel", background=BG, font=FONT_B)

        # --- header biru dengan tombol play dan judul
        head = tk.Frame(root, bg=BLUE)
        head.grid(row=0, column=0, columnspan=2, sticky="ew")
        self.start_btn = tk.Button(head, text="▶", font=("DejaVu Sans", 16), width=3, bg="#5a9e9a",
                                   fg="white", activebackground="#4c8c88", relief="flat", command=self.toggle)
        self.start_btn.pack(side="left", padx=14, pady=8)
        tk.Label(head, text="FM receiver", bg=BLUE, fg="white",
                 font=("DejaVu Sans", 18, "bold")).pack(side="right", padx=20)
        tk.Label(head, text="xSDR  ·  FM broadcast 87,5 - 108 MHz", bg=BLUE, fg="white",
                 font=FONT).pack(side="right")

        # --- baris kolom isian seperti wsdr
        bar = tk.Frame(root, bg=BG)
        bar.grid(row=1, column=0, columnspan=2, sticky="ew", padx=10, pady=(8, 4))
        self.freq = tk.DoubleVar(value=args.freq)
        self.gain = tk.DoubleVar(value=args.gain)
        self.vol = tk.DoubleVar(value=60)
        self.sdr_var, self.audio_var = tk.StringVar(), tk.StringVar()
        self.rx_var = tk.StringVar(value=RX_PORTS[0])
        self.freq_txt, self.tune_txt, self.bw_txt = tk.StringVar(), tk.StringVar(), tk.StringVar()
        self.df, self.bw = 0.0, BW          # posisi dan lebar kanal relatif frekuensi tengah

        def field(col, title, widget, **kw):
            lbl = ttk.Label(bar, text=title, style="H.TLabel")
            lbl.grid(row=0, column=col, sticky="w", padx=6)
            widget.grid(row=1, column=col, sticky="w", padx=6, **kw)
            return lbl

        f0 = tk.Frame(bar, bg=BG)
        self.sdr_box = ttk.Combobox(f0, textvariable=self.sdr_var, state="readonly", width=21)
        self.sdr_box.pack(side="left")
        ttk.Button(f0, text="⟳", width=2, command=self.refresh_devices).pack(side="left", padx=2)
        field(0, "Sumber (input SDR)", f0)
        self.rx_box = ttk.Combobox(bar, textvariable=self.rx_var, values=RX_PORTS, state="readonly", width=5)
        self.rx_box.bind("<<ComboboxSelected>>", self.on_rx_port)
        field(1, "Antena", self.rx_box)
        ent = ttk.Entry(bar, textvariable=self.freq_txt, width=10, font=FONT)
        ent.bind("<Return>", self.on_entry)
        field(2, "Frek. tengah", ent)
        ent_t = ttk.Entry(bar, textvariable=self.tune_txt, width=11, font=FONT)
        ent_t.bind("<Return>", self.on_tune_entry)
        field(3, "Tune Frequency", ent_t)
        ent_b = ttk.Entry(bar, textvariable=self.bw_txt, width=8, font=FONT)
        ent_b.bind("<Return>", self.on_bw_entry)
        field(4, "Bandwidth", ent_b)
        self.gain_scale = ttk.Scale(bar, variable=self.gain, from_=0, to=61, length=110,
                                    command=self.on_gain)
        self.gain_lbl = field(5, "Gain", self.gain_scale)
        field(6, "Samplerate", ttk.Label(bar, text=f"{FS / 1e6:.3f} MHz", width=10, relief="groove"))
        self.audio_box = ttk.Combobox(bar, textvariable=self.audio_var, width=15)
        field(7, "Output audio", self.audio_box)
        self.vol_scale = ttk.Scale(bar, variable=self.vol, from_=0, to=100, length=90, command=self.on_vol)
        self.vol_lbl = field(8, "Volume", self.vol_scale)

        # --- kiri: waterfall + slider frekuensi
        left = tk.Frame(root, bg=BG)
        left.grid(row=2, column=0, sticky="n", padx=(10, 4))
        self.wf = Waterfall(left, self.on_channel, self.recenter)
        self.wf.pack()
        tuner = tk.Frame(left, bg=BG)
        tuner.pack(fill="x", pady=6)
        ttk.Button(tuner, text="◀", width=3, command=lambda: self.step(-0.1)).pack(side="left")
        tk.Scale(tuner, variable=self.freq, from_=FMIN, to=FMAX, resolution=0.1, orient="horizontal",
                 length=560, bg=BG, highlightthickness=0, font=FONT_S,
                 showvalue=True, command=self.on_freq).pack(side="left", padx=4)
        ttk.Button(tuner, text="▶", width=3, command=lambda: self.step(0.1)).pack(side="left")

        # --- kanan: spektrum RF, pengaturan, spektrum MPX
        right = tk.Frame(root, bg=BG)
        right.grid(row=2, column=1, sticky="n", padx=(4, 10))
        self.p_rf = Plot(right, (-SPAN / 1e3, SPAN / 1e3), (-110, -10), range(-400, 401, 200),
                         range(-100, -9, 10), "Frekuensi, kHz", "PSD, dBFS")
        self.p_rf.pack()
        opt = tk.Frame(right, bg=BG)
        opt.pack(fill="x", pady=6)
        self.fps, self.alpha = tk.StringVar(value="10"), tk.StringVar(value="0.3")
        self.db_lo, self.db_hi = tk.StringVar(value="-110"), tk.StringVar(value="-10")
        for i, (t, v, vals) in enumerate((("FPS", self.fps, ("5", "10", "20")),
                                          ("Alpha", self.alpha, ("0.1", "0.3", "0.5", "1")),
                                          ("dB min", self.db_lo, ("-130", "-120", "-110", "-100", "-90")),
                                          ("dB maks", self.db_hi, ("-40", "-30", "-20", "-10", "0")))):
            ttk.Label(opt, text=t, style="H.TLabel").grid(row=0, column=i, sticky="w", padx=6)
            cb = ttk.Combobox(opt, textvariable=v, values=vals, width=6, state="readonly")
            cb.grid(row=1, column=i, padx=6, sticky="w")
            cb.bind("<<ComboboxSelected>>", self.on_range)
        self.p_mpx = Plot(right, (0, 60), (-80, 0), range(0, 61, 10), range(-80, 1, 10),
                          "Frekuensi MPX, kHz (pilot 19, stereo 38, RDS 57)", "PSD, dBFS")
        self.p_mpx.pack()

        self.status = ttk.Label(root, text="Pilih sumber SDR lalu tekan ▶.", anchor="w")
        self.status.grid(row=3, column=0, columnspan=2, sticky="ew", padx=12, pady=(0, 8))

        root.bind("<Left>", lambda e: self.step(-0.1))
        root.bind("<Right>", lambda e: self.step(0.1))
        root.protocol("WM_DELETE_WINDOW", self.quit)

        self.refresh_devices(prefer=args.device)
        self.audio_box["values"] = list_audio_devices()
        self.audio_var.set(args.audio or self.audio_box["values"][0])
        self.rx_var.set(f"RX {getattr(args, 'rx', 'A').upper()}")
        self.on_freq(); self.on_gain(); self.on_vol()
        self.f_rf = (np.arange(NFFT_RF) - NFFT_RF / 2) * FS / NFFT_RF
        self.f_mpx = np.arange(NFFT_MPX // 2) * (FS / D1) / NFFT_MPX
        self.update_view()

    # --- perangkat
    def refresh_devices(self, prefer=None):
        self.devices = list_sdr_devices()
        labels = [l for l, _ in self.devices]
        self.sdr_box["values"] = labels
        alias = {"m2": XSDR_M2, "m.2": XSDR_M2, "pci": XSDR_M2, "pcie": XSDR_M2, "usb": XSDR_USB}
        want = alias.get((prefer or "").lower(), prefer)
        found = [l for l in labels if "tidak terdeteksi" not in l and l != SIM_LABEL]
        pick = next((l for l in labels if want and want.lower() in l.lower()
                     and "tidak terdeteksi" not in l), None)
        pick = pick or (found[0] if found else labels[0])
        self.sdr_var.set(pick)
        self.status["text"] = (f"Terdeteksi: {', '.join(found)}" if found
                               else "Tidak ada SDR terdeteksi. Pilih mode simulasi untuk mencoba GUI.")

    def toggle(self):
        if self.rx:
            self.stop()
        else:
            self.start()

    def start(self):
        kwargs = dict(self.devices)[self.sdr_var.get()]
        try:
            ch = RX_PORTS.index(self.rx_var.get())
            src = SimSource() if kwargs is None else SoapySource(kwargs, ch)
        except Exception as e:
            hint = (" Cek kartu di slot M.2 (lspci) atau pilih xSDR USB." if "M.2" in self.sdr_var.get()
                    else " Cek colokan adapter USB (lsusb)." if "USB" in self.sdr_var.get() else "")
            self.status["text"] = f"Gagal membuka {self.sdr_var.get()}: {e}.{hint}"
            return
        lo, hi = src.gain_range
        self.gain_scale.configure(from_=lo, to=hi)
        self.gain.set(min(max(self.gain.get(), lo), hi))
        audio = self.audio_var.get().split()[0]
        self.rx = Receiver(src, audio, self.freq.get() * 1e6, self.gain.get(), self.vol.get() / 100)
        self.rx.channel(self.df, self.bw)
        self.rx.start()
        self.start_btn["text"] = "■"
        self.sdr_box.configure(state="disabled")
        self.status["text"] = f"Menerima lewat {self.sdr_var.get()}, antena {self.rx_var.get()}"

    def stop(self):
        if self.rx:
            self.rx.stop()
            self.rx.join(timeout=3)
            self.rx = None
        self.start_btn["text"] = "▶"
        self.sdr_box.configure(state="readonly")
        self.avg_rf = self.avg_mpx = None

    # --- kontrol
    def step(self, d):
        self.freq.set(round(min(max(self.freq.get() + d, FMIN), FMAX), 1))
        self.on_freq()

    def on_channel(self, df, bw):
        """Kanal digeser/dilebarkan (dari mouse atau kolom isian)."""
        self.df, self.bw = df, bw
        self.wf.set_channel(df, bw)
        self.p_rf.set_band((df - bw / 2) / 1e3, (df + bw / 2) / 1e3)
        self.tune_txt.set(f"{(self.freq.get() * 1e6 + df) / 1e3:.1f}kHz")
        self.bw_txt.set(f"{bw / 1e3:.1f}kHz")
        if self.rx:
            self.rx.channel(df, bw)

    def on_rx_port(self, *_):
        """Ganti port antena RX A/RX B; kalau sedang menerima, stream dibuka ulang."""
        if self.rx:
            self.stop()
            self.start()

    def recenter(self):
        """Jadikan kanal yang dipilih sebagai frekuensi tengah."""
        f = min(max(self.freq.get() + self.df / 1e6, FMIN), FMAX)
        self.freq.set(round(f, 4))
        self.df = 0.0
        self.on_freq()

    @staticmethod
    def _num(txt, unit):
        return float(txt.replace(",", ".").lower().replace(unit, "").strip())

    def on_tune_entry(self, *_):
        try:
            f = self._num(self.tune_txt.get(), "khz") * 1e3
        except ValueError:
            return
        df = f - self.freq.get() * 1e6
        if abs(df) > SPAN - self.bw / 2:       # di luar tampilan: pindahkan frekuensi tengah
            self.freq.set(round(min(max(f / 1e6, FMIN), FMAX), 4))
            self.df = 0.0
            self.on_freq()
        else:
            self.on_channel(round(df / 100) * 100, self.bw)

    def on_bw_entry(self, *_):
        try:
            bw = self._num(self.bw_txt.get(), "khz") * 1e3
        except ValueError:
            return
        bw = min(max(bw, BW_MIN), BW_MAX)
        self.on_channel(min(max(self.df, -(SPAN - bw / 2)), SPAN - bw / 2), bw)

    def on_entry(self, *_):
        try:
            f = float(self.freq_txt.get().replace(",", ".").lower().replace("mhz", ""))
        except ValueError:
            return
        self.freq.set(round(min(max(f, FMIN), FMAX), 4))
        self.on_freq()

    def on_freq(self, *_):
        f = round(self.freq.get(), 4)
        self.freq_txt.set(f"{f:.3f}MHz")
        if self.rx:
            self.rx.tune(f * 1e6)
        self.on_channel(self.df, self.bw)

    def on_gain(self, *_):
        self.gain_lbl["text"] = f"Gain {self.gain.get():.0f} dB"
        if self.rx:
            self.rx.gain(round(self.gain.get()))

    def on_vol(self, *_):
        self.vol_lbl["text"] = f"Volume {self.vol.get():.0f} %"
        if self.rx:
            self.rx.volume = self.vol.get() / 100

    def on_range(self, *_):
        self.wf.set_range(self.db_lo.get(), self.db_hi.get())

    def update_view(self):
        if self.rx:
            if self.rx.error:
                self.status["text"] = f"Error: {self.rx.error}"
                self.stop()
            elif self.rx.rf is not None:
                rf, mpx, a = self.rx.rf, self.rx.mpx, float(self.alpha.get())
                self.avg_rf = rf if self.avg_rf is None else (1 - a) * self.avg_rf + a * rf
                self.avg_mpx = mpx if self.avg_mpx is None else (1 - a) * self.avg_mpx + a * mpx
                self.wf.push(self.f_rf, rf)
                self.p_rf.set_data(self.f_rf / 1e3, self.avg_rf)
                self.p_mpx.set_data(self.f_mpx / 1e3, self.avg_mpx)
                self.status["text"] = (f"Menerima {(self.freq.get() * 1e6 + self.df) / 1e6:.4f} MHz"
                                       f" (lebar {self.bw / 1e3:.0f} kHz) lewat {self.sdr_var.get()}"
                                       f" {self.rx_var.get()}"
                                       f"   |   sinyal kanal {self.rx.rssi:.0f} dBFS")
        self.root.after(int(1000 / int(self.fps.get())), self.update_view)

    def quit(self):
        self.stop()
        self.root.destroy()


def main():
    p = argparse.ArgumentParser(description="Penerima FM dengan GUI untuk xSDR")
    p.add_argument("--freq", type=float, default=98.7, help="frekuensi awal (MHz)")
    p.add_argument("--gain", type=float, default=40, help="gain awal (dB)")
    p.add_argument("--device", help="sumber SDR: m2, usb, atau teks yang ada di labelnya")
    p.add_argument("--rx", choices=["A", "B", "a", "b"], default="A", help="port antena penerima: A atau B")
    p.add_argument("--audio", help="perangkat aplay, mis. plughw:2,0")
    p.add_argument("--autostart", action="store_true", help="langsung mulai menerima")
    args = p.parse_args()
    root = tk.Tk()
    app = App(root, args)
    if args.autostart:
        root.after(300, app.start)
    root.mainloop()


if __name__ == "__main__":
    main()

Tampilan GUI di CM5: sumber input dan fungsi tiap bagian

Penerima FM ini ditulis dengan Python (tkinter, NumPy, SciPy) dan membaca xSDR lewat SoapySDR. Tangkapan layar di bawah diambil langsung di Raspberry Pi CM5 (piBrick PocketCM5) saat menerima 98,7 MHz dengan xSDR terpasang di slot M.2.

Sumber input yang bisa dipilih

Dropdown Sumber (input SDR) berisi xSDR M.2, xSDR USB dan Simulasi

Dropdown Sumber (input SDR) selalu menampilkan dua jalur xSDR, walaupun salah satunya sedang tidak terpasang:

  • xSDR M.2 (PCIe) – xSDR di slot M.2 lewat adapter M.2 B+M-key. Butuh driver kernel usdr_pcie_uram dan muncul sebagai /dev/usdr0.
  • xSDR USB – xSDR lewat konverter M.2→USB (terdeteksi sebagai USB 3727:1011). Tidak butuh driver kernel.
  • SDR lain yang ditemukan SoapySDR (kalau ada) ikut dicantumkan.
  • Simulasi (tanpa SDR) – beberapa stasiun FM tiruan untuk mencoba GUI tanpa hardware.

Jalur yang tidak terdeteksi diberi keterangan "(tidak terdeteksi)". Tombol ⟳ di sebelahnya memindai ulang perangkat. Dropdown terkunci selama menerima, jadi tekan ■ dulu untuk berganti sumber. Dari terminal, sumber bisa dipilih dengan --device m2 atau --device usb.

Revisi: pilihan antena RX A / RX B

xSDR punya dua port penerima, RX A dan RX B. Di versi terbaru fm_gui.py ada kolom Antena di sebelah Sumber untuk memilih port tempat antena FM dipasang. RX A memakai kanal penerima 0 dan RX B kanal 1. Port bisa diganti saat sedang menerima, karena stream otomatis dibuka ulang di port yang baru, dan port yang dipakai ikut tampil di status bar. Kolom ini belum tampak di tangkapan layar di bawah, yang diambil sebelum revisi. Dari terminal, port dipilih dengan --rx A atau --rx B:

python3 fm_gui.py --device m2 --rx B --audio plughw:2,0 --freq 98.7 --gain 40 --autostart

Fungsi setiap bagian GUI

GUI penerima FM xSDR dengan nomor penanda tiap bagian
  1. ▶ / ■ – mulai atau berhenti menerima.
  2. Sumber (input SDR) – memilih xSDR M.2, xSDR USB, SDR lain, atau simulasi.
  3. Frek. tengah – frekuensi pusat tampilan (MHz). Ketik lalu tekan Enter.
  4. Tune Frequency – frekuensi kanal yang sedang didemodulasi (kHz). Ikut berubah saat kanal digeser di waterfall, dan bisa juga diketik.
  5. Bandwidth – lebar kanal/filter, 30–220 kHz. Ubah dengan menarik tepi kanal atau mengetik nilainya.
  6. Gain – penguatan penerima 0–61 dB, dibagi otomatis ke tahap LNA, TIA dan PGA xSDR.
  7. Samplerate – laju sampel SDR, 1,920 MHz.
  8. Output audio – kartu suara tujuan (misalnya speaker USB plughw:2,0).
  9. Volume – volume audio 0–100 %.
  10. Waterfall – spektrum ±500 kHz di sekitar frekuensi tengah terhadap waktu (baris terbaru di atas). Kanal yang didemodulasi disorot: garis merah menandai Tune Frequency, garis putus-putus menandai tepi kanal.
  11. Skala warna dB – arti warna waterfall, dari biru (lemah) sampai merah (kuat).
  12. Slider frekuensi dan tombol ◀ ▶ – memindah frekuensi tengah di pita FM 87,5–108 MHz, dengan langkah 0,1 MHz.
  13. Spektrum RF – daya sinyal (dBFS) di seluruh lebar tampilan, dengan kotak biru untuk kanal yang dipilih.
  14. FPS, Alpha, dB min, dB maks – laju pembaruan tampilan, perataan grafik, dan rentang warna waterfall.
  15. Spektrum MPX – isi sinyal hasil demodulasi: pilot stereo 19 kHz, sinyal stereo 38 kHz dan RDS 57 kHz. Puncak di 19 kHz menandakan siaran stereo tertangkap.
  16. Status bar – frekuensi kanal, lebar kanal, sumber yang dipakai, dan kuat sinyal kanal dalam dBFS.

Mengatur kanal dengan mouse

Kanal digeser ke kiri, Tune Frequency 98506,2 kHz

Menggeser kanal. Tarik area kanal yang disorot (kursor berubah menjadi panah empat arah). Pada gambar ini kanal digeser sekitar 194 kHz ke kiri, sehingga Tune Frequency menjadi 98506,2 kHz sementara frekuensi tengah tetap 98,700 MHz. Klik di luar kanal untuk melompat ke titik itu, scroll mouse untuk menggeser 5 kHz, dan klik ganda untuk menjadikan kanal sebagai frekuensi tengah.

Lebar kanal dipersempit menjadi 150 kHz

Melebarkan atau menyempitkan kanal. Arahkan mouse ke garis putus-putus di tepi kanal (kursor berubah menjadi panah kiri-kanan), lalu tarik. Pada gambar ini lebar kanal dipersempit dari 200 kHz menjadi 150 kHz. Kolom Bandwidth dan kotak biru di spektrum RF ikut berubah.

Pemecahan masalah

GejalaPenyebabSolusi
dmesg menunjukkan "link down", lspci tidak menampilkan 10ee:7049Overlay DMA 32-bit belum aktif, atau adapter tidak terpasang rapatTambahkan dtoverlay=pcie-32bit-dma-pi5 di config.txt, pastikan adapter masuk penuh dan sekrup terpasang, lalu reboot
Driver gagal dimuat atau error memori saat streamingKernel memakai page 16K (kernel_2712.img)Tambahkan kernel=kernel8.img, reboot, lalu cek getconf PAGESIZE harus 4096
modprobe: FATAL: Module usdr_pcie_uram not foundHeader kernel tidak cocok atau DKMS belum membangun modulPasang linux-headers-$(uname -r), cek dkms status, lalu ulangi dkms build dan dkms install
/dev/usdr0 hanya bisa dibuka dengan sudoAturan udev belum terpasangSalin 50-usdr-pcie-driver.rules ke /etc/udev/rules.d/, lalu sudo udevadm control --reload-rules && sudo udevadm trigger
Gain seolah tidak berubah, sinyal sangat lemahsetGain() total hanya mengatur LNAAtur LNA, TIA dan PGA satu per satu seperti di skrip
Hanya terdengar desisAntena belum terpasang atau port antena salahPasang antena di konektor MHF4 RX dan gunakan antena LNAW untuk FM
Suara putus-putusCPU tidak sempat memprosesTutup program lain, jalankan tanpa desktop, atau turunkan sample rate

Untuk diagnosa cepat, jalankan perintah berikut lalu periksa hasilnya dari atas ke bawah:

getconf PAGESIZE
sudo dmesg | grep -iE 'pcie|usdr' | tail -20
lspci -nn | grep 10ee
lsmod | grep usdr
ls -l /dev/usdr*
SoapySDRUtil --find="driver=usdr"

Penutup dan referensi

Dengan tiga baris di config.txt, driver usdr_pcie_uram dan modul SoapySDR, xSDR sudah bisa dipakai sebagai SDR PCIe di Raspberry Pi CM5. Dari sini Anda bisa melanjutkan ke penerima stereo, RDS, pemindai spektrum, atau memakai GNU Radio dengan blok Soapy.

Referensi:

Friday, October 16, 2026

Sniffing CAN Bus CubeSat (libcsp) dengan Korlan USB2CAN dan GUI Python

Assalamu'alaikum warohmatullohi wabarakatuh,

Pada tutorial sebelumnya, Komunikasi CubeSat Space Protocol (libcsp) lewat CAN Bus FDCAN1 pada STM32H743 dengan FreeRTOS, kita membuat dua board STM32H743 berperan sebagai OBC (On-Board Computer) dan EPS (Electrical Power System) yang saling bertukar paket CSP lewat CAN bus. Masalahnya, kalau komunikasi gagal, kita hanya bisa menebak-nebak dari log UART masing-masing board.

Kali ini kita membuat "telinga" ketiga di bus: Korlan USB2CAN dari 8devices yang dicolokkan ke laptop, ditambah program Python dengan GUI untuk sniffing (menyadap) bus CAN. Program ini tidak hanya menampilkan frame CAN mentah, tetapi juga menyusun ulang frame menjadi paket CSP, menampilkan data housekeeping EPS, dan bisa mengirim perintah (ping, payload ON/OFF) langsung dari laptop.

Apa itu sniffing CAN bus?

Sniffing berarti ikut mendengarkan semua lalu lintas di bus tanpa mengganggu node lain. Di CAN, semua node berbagi dua kabel (CANH dan CANL), jadi adapter yang tersambung ke bus otomatis "mendengar" setiap frame. Dengan sniffer kita bisa:

  • memastikan sebuah node benar-benar mengirim (misalnya OBC sudah mengirim ping atau belum),
  • melihat isi data sebenarnya di kabel, bukan hanya yang dilaporkan firmware,
  • mendeteksi masalah fisik: terminator, kabel tertukar, bitrate salah, dan
  • menguji node satu per satu dengan mengirim perintah dari laptop.

Perangkat yang dibutuhkan

PerangkatKeterangan
Korlan USB2CAN (8devices)adapter USB ke CAN, konektor DB9
2 board STM32H743 (DevEBox) + modul MCP2551OBC dan EPS dari tutorial sebelumnya
2 resistor 120 Ωterminator di kedua ujung bus
Konektor DB9 female / adaptor terminal sekrupuntuk menyambung kabel ke Korlan
Laptop Windows + Python 3 (64-bit)python-can, pywin32, Tkinter
Korlan USB2CAN dan pinout DB9

Korlan USB2CAN dan pinout konektor DB9: pin 7 = CAN_H, pin 2 = CAN_L, pin 3 = GND.

Langkah 1 – Instal driver dan library

Semua file resmi ada di halaman produk 8devices Korlan USB2CAN. Gulir ke bawah tabel Specifications sampai bagian Resources (file disimpan di Google Drive). Unduh tiga file berikut:

  1. Windows Driver (WinUSB) → jalankan usb2can_winusb.msi (butuh izin Administrator). Setelah selesai, Korlan muncul sebagai USB2CAN di Device Manager tanpa tanda segitiga kuning.
  2. USB2CAN CANAL DLL v2.0.0 → di dalam zip ada x64/Release/usb2can.dll dan x86/Release/usb2can.dll.
  3. Windows python examples → contoh resmi 8devices (berguna untuk perbandingan).

Salin usb2can.dll versi x64 ke folder program kita. Versi DLL harus sama dengan Python: Python 64-bit memakai DLL x64. Lalu pasang library Python:

pip install python-can pywin32

pywin32 dipakai python-can untuk mendeteksi serial number Korlan secara otomatis lewat WMI.

Langkah 2 – Menyambung Korlan ke bus CAN

Korlan dipasang paralel (sebagai cabang) di tengah bus, seperti node ketiga. Bus tetap memakai dua resistor 120 Ω di kedua ujung (di modul OBC dan modul EPS). Korlan sendiri tidak perlu resistor.

Wiring Korlan USB2CAN ke bus CAN OBC dan EPS

Wiring: CANH–CANH, CANL–CANL, GND bersama, terminator 120 Ω di kedua ujung, Korlan menyadap di tengah.

DB9 KorlanBus CAN
pin 7CANH
pin 2CANL
pin 3GND board

Cek terminator dengan multimeter saat semua board mati: hambatan CANH–CANL seharusnya sekitar 60 Ω (dua resistor 120 Ω paralel).

Langkah 3 – Tes koneksi dengan Python

Sebelum membuat GUI, pastikan Python bisa membuka Korlan. Skrip pendek ini mencari adapter lalu menampilkan frame selama 5 detik:

import can

# 1) cari Korlan yang terpasang
print(can.detect_available_configs("usb2can"))
# contoh hasil: [{'interface': 'usb2can', 'channel': '04EF18C4'}]

# 2) buka bus 1 Mbit/s dan tampilkan frame selama 5 detik
bus = can.Bus(interface="usb2can", channel="04EF18C4",
              bitrate=1000000, dll="./usb2can.dll")
import time
t0 = time.time()
while time.time() - t0 < 5:
    msg = bus.recv(timeout=0.5)
    if msg:
        print(msg)
bus.shutdown()

Serial number (contoh 04EF18C4) adalah 8 karakter yang juga tertera di stiker Korlan. Kecepatan bitrate harus sama dengan firmware, di proyek ini 1 Mbit/s.

Langkah 4 – Membaca ID CAN sebagai header CFP

libcsp mengirim paket CSP lewat CAN memakai CFP (CAN Fragmentation Protocol). Semua informasi alamat ada di ID extended 29-bit:

Membaca ID CAN sebagai header CFP

Contoh ID 0x0110645D: dari OBC (1) ke EPS (2), frame BEGIN, masih 25 frame lagi, paket nomor 93.

Fungsi Python untuk memecah ID dan header CSP 32-bit (dikirim big-endian di frame BEGIN):

def cfp_fields(can_id):
    return {
        "src": (can_id >> 24) & 0x1F,
        "dst": (can_id >> 19) & 0x1F,
        "type": (can_id >> 18) & 0x01,
        "remain": (can_id >> 10) & 0xFF,
        "id": can_id & 0x3FF,
    }


def csp_header_unpack(raw4):
    h = struct.unpack(">I", raw4)[0]
    return {
        "prio": (h >> 30) & 0x3,
        "src": (h >> 25) & 0x1F,
        "dst": (h >> 20) & 0x1F,
        "dport": (h >> 14) & 0x3F,
        "sport": (h >> 8) & 0x3F,
        "flags": h & 0xFF,
    }

Langkah 5 – Menyusun ulang frame menjadi paket CSP

Satu paket CSP bisa terdiri dari banyak frame. Kelas CfpReassembler meniru fungsi csp_can_rx() di libcsp: frame dikelompokkan berdasarkan (SRC, DST, ID), frame BEGIN membuka buffer baru, frame MORE ditambahkan selama nilai REMAIN turun berurutan, dan paket dianggap lengkap ketika jumlah byte sudah sama dengan panjang di header.

class CfpReassembler:
    """Menyusun ulang frame CFP menjadi paket CSP (meniru csp_can_rx())."""

    def __init__(self, timeout=1.0):
        self.buffers = {}
        self.timeout = timeout

    def push(self, can_id, data):
        f = cfp_fields(can_id)
        key = (f["src"], f["dst"], f["id"])
        now = time.time()
        for k in [k for k, b in self.buffers.items() if now - b["t"] > self.timeout]:
            del self.buffers[k]

        if f["type"] == CFP_BEGIN:
            if len(data) < CFP_OVERHEAD:
                return None, "BEGIN terlalu pendek"
            hdr = csp_header_unpack(bytes(data[:4]))
            length = struct.unpack(">H", bytes(data[4:6]))[0]
            buf = {"hdr": hdr, "len": length, "data": bytearray(data[6:]),
                   "remain": f["remain"], "t": now}
            self.buffers[key] = buf
        else:
            buf = self.buffers.get(key)
            if buf is None:
                return None, "frame MORE tanpa BEGIN"
            if f["remain"] != buf["remain"] - 1:
                del self.buffers[key]
                return None, "frame hilang (remain tidak urut)"
            buf["remain"] = f["remain"]
            buf["data"] += bytes(data)

        if len(buf["data"]) >= buf["len"]:
            del self.buffers[key]
            return {"hdr": buf["hdr"], "data": bytes(buf["data"][:buf["len"]])}, None
        return None, None

Langkah 6 – Arsitektur program

Arsitektur software sniffer

Alur data dari Korlan sampai GUI.

Kunci agar GUI tidak macet adalah memisahkan pembacaan bus dari tampilan. Thread penerima membaca bus.recv() terus-menerus dan memasukkan hasilnya ke antrean:

    def _rx_loop(self):
        while not self._stop.is_set():
            msg = self.bus.recv(timeout=0.1)
            if msg is None:
                continue
            if msg.is_error_frame:
                self.stats["errors"] += 1
                self.inbox.put(("error", msg))
                continue
            if not msg.is_extended_id:
                self.inbox.put(("std", msg))
                continue
            self.stats["frames"] += 1
            pkt, err = self.rx.push(msg.arbitration_id, msg.data)
            self.inbox.put(("frame", msg, err))
            if pkt:
                pkt["t"] = time.time()
                self.stats["packets"] += 1
                self.inbox.put(("packet", pkt))

GUI (Tkinter) mengambil isi antrean setiap 40 ms dengan after(), paling banyak 500 item sekali ambil, sehingga jendela tetap responsif meskipun bus sangat ramai:

    def _poll(self):
        if self.node:
            n = 0
            while n < 500:
                try:
                    item = self.node.inbox.get_nowait()
                except queue.Empty:
                    break
                n += 1
                self._handle_item(item)
            self._check_timeouts()
            self._update_stats()
        self.after(40, self._poll)

Koneksi ke Korlan cukup satu baris can.Bus(interface="usb2can", ...). Fungsi berikut otomatis memakai usb2can.dll yang ada di folder program:

def open_bus(args):
    if args.interface == "virtual":
        return can.Bus(interface="virtual", channel=args.channel or "csp")
    kw = {"interface": args.interface, "bitrate": args.bitrate}
    if args.channel:
        kw["channel"] = args.channel
    if args.interface == "usb2can":
        here = os.path.dirname(os.path.abspath(__file__))
        dll = os.path.join(here, "usb2can.dll")
        if os.path.exists(dll):
            kw["dll"] = dll
    return can.Bus(**kw)

Serial number Korlan dicari otomatis saat GUI dibuka (tombol Cari untuk mencari ulang):

    def _scan_serial(self, quiet=False):
        """Cari Korlan USB2CAN yang terpasang dan isi kolom Serial."""
        try:
            cfgs = can.detect_available_configs("usb2can")
        except Exception as e:  # noqa: BLE001
            cfgs = []
            if not quiet:
                self._log(f"Gagal mencari adapter: {e}")
        serials = [str(c.get("channel")) for c in cfgs if c.get("channel")]
        self.cb_ch.configure(values=serials)
        if serials:
            if self.var_ch.get().strip() not in serials:
                self.var_ch.set(serials[0])
            self._log("Korlan USB2CAN terdeteksi: S/N " + ", ".join(serials))
        elif not quiet:
            self._log("Korlan USB2CAN tidak terdeteksi (cek kabel USB / driver 8devices / pywin32).")
        return serials

Langkah 7 – Menjalankan GUI

Simpan csp_can_tool.py, csp_can_gui.py dan usb2can.dll dalam satu folder, lalu jalankan:

python csp_can_gui.py

Pilih Adapter: usb2can untuk Korlan sungguhan, atau simulasi untuk mencoba tampilan tanpa hardware (program membuat OBC dan EPS tiruan yang perilakunya sama dengan firmware). Klik Hubungkan.

GUI CubeSat CAN Monitor tab Paket CSP dengan Korlan asli

Tab Paket CSP saat terhubung ke Korlan asli (S/N 04EF18C4): biru = OBC, merah = EPS. Panel kiri menampilkan housekeeping EPS (suhu chip 32,8 °C, VDDA 3338 mV, Payload ON, uptime) dan hasil Ping EPS 1,2 ms.

Yang terlihat di tab Paket CSP:

  • OBC:30 → EPS:1 PING lalu EPS:1 → OBC:30 re:PING: OBC mem-ping EPS dan EPS membalas.
  • OBC → EPS:12 EPS_HK dan balasan 20 byte berisi uptime, suhu chip, VDDA dan status payload, yang langsung diterjemahkan.
  • OBC → EPS:11 TLM: paket telemetri 200 byte (26 frame CAN) dengan checksum.
  • Baris dari/ke PC: perintah yang kita kirim dari tombol Ping, Payload ON/OFF, dan Kirim pesan.
GUI CubeSat CAN Monitor tab Frame CAN dengan Korlan asli

Tab Frame CAN saat terhubung ke Korlan asli (S/N 04EF18C4): setiap frame dengan ID yang sudah diuraikan (Src, Dst, BEGIN/MORE, Remain, CFP id). Terlihat paket telemetri 200 byte dari OBC (1) ke EPS (2) dengan CFP id 952: BEGIN remain=25 lalu MORE 24…0, serta frame balasan EPS (src 2) ke OBC (dst 1) dan ke PC (dst 10).

Hasil uji nyata: OBC dan EPS terbaca bersih

Setelah dua resistor 120 Ω terpasang di ujung bus (modul OBC dan modul EPS), semua stuff error hilang. Korlan langsung menerima frame valid dan GUI menampilkan seluruh lalu lintas CubeSat:

GUI CubeSat CAN Monitor membaca OBC dan EPS lewat Korlan, payload ON

Hasil nyata dengan Korlan asli: kedua node aktif, Kondisi bus OK, error frame 0. Tombol Payload ON dari laptop dijawab EPS "status=OK rail=ON" dalam 10 ms, panel housekeeping menampilkan Payload ON, suhu chip 33,2 °C dan VDDA 3336 mV.

Yang terlihat pada pengujian dengan hardware asli (Korlan S/N 04EF18C4, 1 Mbit/s):

  • Kondisi bus OK, error frame 0 dari 1.133 frame CAN (±48 frame/detik), status node OBC dan EPS sama-sama aktif.
  • OBC → EPS PING dan EPS → OBC re:PING: OBC mem-ping EPS setiap 2 detik dan EPS membalas.
  • OBC → EPS:10 ECHO berisi "Halo EPS, pesan ke-127", dibalas EPS "ACK dari node 2 (EPS)".
  • OBC → EPS:12 EPS_HK dan balasan 20 byte berisi uptime, status payload, serta jumlah frame CAN yang diterima/dikirim EPS.
  • OBC → EPS:11 TLM: telemetri 200 byte (26 frame) dengan checksum, dan EPS_CMD setiap 10 detik untuk membalik status payload.

Perintah langsung dari laptop lewat Korlan juga berhasil:

Tombol di GUIHasil
Ping EPS2,0 ms
Ping OBCOBC membalas (OBC:1 → PC re:PING)
Payload ONstatus OK dalam 9,7 ms, LED PA1 di board EPS menyala
Kirim pesan ke EPSbalasan "ACK dari node 2 (EPS)" dalam 6,8 ms

Update – suhu dan VDDA EPS sudah terbaca. Pada pengujian pertama nilai suhu chip dan VDDA EPS masih 0. Penyebabnya ada di firmware EPS: kalibrasi linearitas ADC3 STM32H743 butuh waktu lebih lama dari batas tunggu 10 ms di kode awal, sehingga inisialisasi ADC dianggap gagal. Setelah batas tunggu kalibrasi dinaikkan menjadi 200 ms, EPS mengirim nilai sensor yang benar:

Housekeeping EPS suhu dan VDDA terbaca lewat Korlan

Setelah perbaikan firmware: panel EPS housekeeping menampilkan suhu chip 33,1 °C dan VDDA 3337 mV.

  • Suhu chip STM32H743 (EPS): ±32,7–33,2 °C, dari sensor suhu internal ADC3 channel 18 dengan kalibrasi pabrik TS_CAL1/TS_CAL2.
  • VDDA: ±3337 mV, dihitung dari VREFINT (ADC3 channel 19) dan VREFINT_CAL.
  • Ping EPS dari laptop: 2,3 ms; error frame tetap 0 dan Kondisi bus OK.

Log serial EPS. Dari sisi EPS, log UART (USART1, PA9/PA10, 115200 baud) dibaca lewat Mini Debugger di COM6. Log ini memastikan apa yang terlihat di Korlan memang diterima dan diproses oleh EPS:

Log serial EPS di console STM32CubeIDE COM6

Log serial EPS (COM6) di console STM32CubeIDE: permintaan housekeeping dari OBC dan laptop, pesan dan telemetri dari OBC, perintah payload, serta status CAN TEC=0 REC=0.

  • [EPS-HK] permintaan housekeeping dari node 10 (laptop lewat Korlan) dan node 1 (OBC), masing-masing tiap 2 detik, lengkap dengan suhu chip ±31 °C dan VDDA ±3338 mV.
  • [SERVER] pesan "Halo EPS, pesan ke-…" di port 10 dan paket telemetri 200 byte di port 11 beserta checksum.
  • [EPS-CMD] perintah payload dari OBC: ON (LED PA1 menyala) lalu OFF (LED PA1 mati) setiap 10 detik.
  • [STAT] kesehatan CAN di EPS: TEC=0 REC=0 OK, 0 kiriman gagal, buffer CSP bebas 19–20 dari 20.

Troubleshooting: membaca status error dari Korlan

Kalau sinyal di bus rusak, Korlan tidak menerima frame data, melainkan pesan status/error. GUI meringkasnya per detik di tab Log dan menampilkannya di baris Kondisi bus. Berikut contoh nyata saat bus OBC–EPS saya belum dipasangi terminator:

GUI menampilkan stuff error dari Korlan

Koneksi nyata ke Korlan (S/N 04EF18C4): Frame CAN 0, ribuan error frame, Kondisi bus "ERROR: stuff error".

Arti kode status yang dikirim adapter (byte data pertama):

# Kode status/error yang dikirim adapter 8devices USB2CAN (byte data pertama)
USB2CAN_STATUS = {
    0x00: "OK", 0x01: "overrun", 0x02: "bus warning", 0x03: "error passive",
    0x04: "BUS-OFF", 0x20: "stuff error", 0x21: "form error", 0x23: "ACK error",
    0x24: "bit0 error", 0x25: "bit1 error", 0x27: "CRC error", -1: "error",
}
GejalaKemungkinan penyebabSolusi
Frame CAN = 0, tidak ada errorKorlan belum tersambung ke bus / node tidak mengirimcek kabel DB9 (pin 7, 2, 3) dan pastikan board menyala
Banyak stuff error, bus warning, error passivetanpa terminator 120 Ω, CANH/CANL tertukar, atau bitrate berbedapasang 2×120 Ω (ukur ±60 Ω), cek CANH–CANH, samakan bitrate
Hanya ACK errortidak ada node lain yang membalasnyalakan node lain; Korlan sendiri juga memberi ACK
Error terus walau kabel benarMCP2551 dicatu 5 V butuh sinyal TXD ≥ 3,5 V, STM32 hanya 3,3 Vpakai transceiver 3,3 V (SN65HVD230) atau pull-up 2,2 kΩ CTX ke 5 V + PA12 open-drain
"DLL failed to load"usb2can.dll tidak ada / beda arsitektursalin DLL x64 ke folder program, Python 64-bit

Catatan: Korlan ikut memberi ACK di bus. Jadi kalau hanya satu board yang terpasang bersama Korlan, frame board itu tetap terbaca dengan benar. Ini cara cepat menguji OBC dan EPS satu per satu.

Kode lengkap

Dua file berikut disimpan dalam satu folder bersama usb2can.dll. Klik judul untuk membuka kodenya.

csp_can_tool.py – library CFP/CSP + versi command line (436 baris)
#!/usr/bin/env python3
"""
csp_can_tool.py - membaca dan menguji bus CAN CubeSat (libcsp 1.6 / CFP)
memakai Korlan USB2CAN (8devices) dan python-can.

Bisa dipakai untuk:
  monitor  : menampilkan semua frame CAN + paket CSP yang sudah disusun ulang
  ping     : ping node CSP (OBC = 1, EPS = 2)
  hk       : minta data housekeeping EPS (port 12)
  payload  : perintah ON/OFF rail payload EPS (port 13)
  echo     : kirim pesan teks ke port 10 dan tampilkan balasannya

Contoh (Windows):
  python csp_can_tool.py list
  python csp_can_tool.py monitor
  python csp_can_tool.py ping 2
  python csp_can_tool.py hk
  python csp_can_tool.py payload on
  python csp_can_tool.py echo 1 "Halo OBC"

Kebutuhan:
  pip install python-can pywin32
  usb2can.dll (64-bit) dari driver 8devices diletakkan di folder yang sama
  dengan script ini, atau di folder yang ada di PATH.
"""
import argparse
import os
import queue
import struct
import sys
import threading
import time

import can

# ---------------------------------------------------------------------------
# Konstanta sistem (harus sama dengan firmware: cubesat_proto.h)
# ---------------------------------------------------------------------------
ADDR_OBC = 1
ADDR_EPS = 2
ADDR_PC = 10            # alamat CSP untuk PC/Korlan (bebas, asal unik 0..30)

PORT_PING = 1
PORT_ECHO = 10
PORT_TLM = 11
PORT_EPS_HK = 12
PORT_EPS_CMD = 13

NODE_NAMES = {ADDR_OBC: "OBC", ADDR_EPS: "EPS", ADDR_PC: "PC", 31: "BCAST"}
PORT_NAMES = {0: "CMP", 1: "PING", 2: "PS", 3: "MEMFREE", 4: "REBOOT",
              5: "BUF_FREE", 6: "UPTIME", PORT_ECHO: "ECHO",
              PORT_TLM: "TLM", PORT_EPS_HK: "EPS_HK", PORT_EPS_CMD: "EPS_CMD"}

# eps_hk_t (20 byte, little endian, packed)
EPS_HK_FMT = "<IhHBBHII"
EPS_HK_SIZE = struct.calcsize(EPS_HK_FMT)

# ---------------------------------------------------------------------------
# CFP: CAN Fragmentation Protocol (ID extended 29 bit)
#   [SRC 5][DST 5][TYPE 1][REMAIN 8][ID 10]
# ---------------------------------------------------------------------------
CFP_BEGIN, CFP_MORE = 0, 1
CFP_OVERHEAD = 6                        # 4 byte header CSP + 2 byte panjang


def cfp_fields(can_id):
    return {
        "src": (can_id >> 24) & 0x1F,
        "dst": (can_id >> 19) & 0x1F,
        "type": (can_id >> 18) & 0x01,
        "remain": (can_id >> 10) & 0xFF,
        "id": can_id & 0x3FF,
    }


def cfp_make_id(src, dst, typ, remain, ident):
    return ((src & 0x1F) << 24) | ((dst & 0x1F) << 19) | ((typ & 1) << 18) \
        | ((remain & 0xFF) << 10) | (ident & 0x3FF)


# ---------------------------------------------------------------------------
# Header CSP 1.6 (32 bit, dikirim big endian)
#   [PRIO 2][SRC 5][DST 5][DPORT 6][SPORT 6][FLAGS 8]
# ---------------------------------------------------------------------------
def csp_header_unpack(raw4):
    h = struct.unpack(">I", raw4)[0]
    return {
        "prio": (h >> 30) & 0x3,
        "src": (h >> 25) & 0x1F,
        "dst": (h >> 20) & 0x1F,
        "dport": (h >> 14) & 0x3F,
        "sport": (h >> 8) & 0x3F,
        "flags": h & 0xFF,
    }


def csp_header_pack(prio, src, dst, dport, sport, flags=0):
    h = ((prio & 3) << 30) | ((src & 0x1F) << 25) | ((dst & 0x1F) << 20) \
        | ((dport & 0x3F) << 14) | ((sport & 0x3F) << 8) | (flags & 0xFF)
    return struct.pack(">I", h)


def cfp_fragment(src, dst, payload, header4, ident):
    """Pecah satu paket CSP menjadi daftar (can_id, data) sesuai csp_can_tx()."""
    length = len(payload)
    frames = []
    first = header4 + struct.pack(">H", length) + payload[:2]
    remain = (length + CFP_OVERHEAD - 1) // 8
    frames.append((cfp_make_id(src, dst, CFP_BEGIN, remain, ident), first))
    sent = min(length, 2)
    while sent < length:
        chunk = payload[sent:sent + 8]
        remain = (length - sent - len(chunk) + 7) // 8
        frames.append((cfp_make_id(src, dst, CFP_MORE, remain, ident), chunk))
        sent += len(chunk)
    return frames


class CfpReassembler:
    """Menyusun ulang frame CFP menjadi paket CSP (meniru csp_can_rx())."""

    def __init__(self, timeout=1.0):
        self.buffers = {}
        self.timeout = timeout

    def push(self, can_id, data):
        f = cfp_fields(can_id)
        key = (f["src"], f["dst"], f["id"])
        now = time.time()
        for k in [k for k, b in self.buffers.items() if now - b["t"] > self.timeout]:
            del self.buffers[k]

        if f["type"] == CFP_BEGIN:
            if len(data) < CFP_OVERHEAD:
                return None, "BEGIN terlalu pendek"
            hdr = csp_header_unpack(bytes(data[:4]))
            length = struct.unpack(">H", bytes(data[4:6]))[0]
            buf = {"hdr": hdr, "len": length, "data": bytearray(data[6:]),
                   "remain": f["remain"], "t": now}
            self.buffers[key] = buf
        else:
            buf = self.buffers.get(key)
            if buf is None:
                return None, "frame MORE tanpa BEGIN"
            if f["remain"] != buf["remain"] - 1:
                del self.buffers[key]
                return None, "frame hilang (remain tidak urut)"
            buf["remain"] = f["remain"]
            buf["data"] += bytes(data)

        if len(buf["data"]) >= buf["len"]:
            del self.buffers[key]
            return {"hdr": buf["hdr"], "data": bytes(buf["data"][:buf["len"]])}, None
        return None, None


# ---------------------------------------------------------------------------
# Node CSP sederhana di PC
# ---------------------------------------------------------------------------
class CspNode:
    def __init__(self, bus, addr=ADDR_PC, verbose=False):
        self.bus = bus
        self.addr = addr
        self.verbose = verbose
        self.rx = CfpReassembler()
        self.ident = 0
        self.sport = 20
        self.inbox = queue.Queue()
        self.stats = {"frames": 0, "packets": 0, "errors": 0}
        self._stop = threading.Event()
        self._thread = threading.Thread(target=self._rx_loop, daemon=True)
        self._thread.start()

    def close(self):
        self._stop.set()
        self._thread.join(timeout=1)

    def _rx_loop(self):
        while not self._stop.is_set():
            msg = self.bus.recv(timeout=0.1)
            if msg is None:
                continue
            if msg.is_error_frame:
                self.stats["errors"] += 1
                self.inbox.put(("error", msg))
                continue
            if not msg.is_extended_id:
                self.inbox.put(("std", msg))
                continue
            self.stats["frames"] += 1
            pkt, err = self.rx.push(msg.arbitration_id, msg.data)
            self.inbox.put(("frame", msg, err))
            if pkt:
                pkt["t"] = time.time()
                self.stats["packets"] += 1
                self.inbox.put(("packet", pkt))

    def alloc_sport(self):
        self.sport = 17 + (self.sport - 16) % 40   # port sementara 17..56
        return self.sport

    def send(self, dst, dport, payload, sport=None, prio=2):
        if sport is None:
            sport = self.alloc_sport()
        hdr = csp_header_pack(prio, self.addr, dst, dport, sport)
        self.ident = (self.ident + 1) & 0x3FF
        for can_id, data in cfp_fragment(self.addr, dst, payload, hdr, self.ident):
            self.bus.send(can.Message(arbitration_id=can_id, data=data,
                                      is_extended_id=True), timeout=0.2)
        return sport

    def wait_reply(self, src, sport, timeout):
        """Tunggu paket dari node 'src' yang ditujukan ke port 'sport' kita."""
        end = time.time() + timeout
        while time.time() < end:
            try:
                item = self.inbox.get(timeout=max(0.0, end - time.time()))
            except queue.Empty:
                break
            if item[0] == "packet":
                h = item[1]["hdr"]
                if h["src"] == src and h["dst"] == self.addr and h["dport"] == sport:
                    return item[1]
        return None

    def transaction(self, dst, dport, payload, timeout=1.0):
        t0 = time.time()
        sport = self.send(dst, dport, payload)
        reply = self.wait_reply(dst, sport, timeout)
        return reply, (time.time() - t0) * 1000.0


# ---------------------------------------------------------------------------
# Tampilan
# ---------------------------------------------------------------------------
def node_name(a):
    return f"{NODE_NAMES.get(a, '?')}({a})"


def describe_packet(pkt):
    h, d = pkt["hdr"], pkt["data"]
    port = PORT_NAMES.get(h["dport"]) or (PORT_NAMES.get(h["sport"], str(h["dport"])) + " balasan")
    line = (f"CSP {node_name(h['src'])}:{h['sport']} -> {node_name(h['dst'])}:{h['dport']} "
            f"[{port}] prio={h['prio']} len={len(d)}")
    extra = ""
    if h["dport"] == PORT_PING or h["sport"] == PORT_PING:
        extra = f"ping {len(d)} byte"
    elif h["dport"] == PORT_ECHO or h["sport"] == PORT_ECHO:
        extra = f'teks="{d.decode(errors="replace")}"'
    elif h["dport"] == PORT_TLM:
        extra = f"checksum={sum(d)}"
    elif h["src"] == ADDR_EPS and h["sport"] == PORT_EPS_HK and len(d) == EPS_HK_SIZE:
        extra = format_hk(d)
    elif h["dport"] == PORT_EPS_CMD and len(d) == 2:
        extra = f"perintah rail={d[0]} -> {'ON' if d[1] else 'OFF'}"
    elif h["sport"] == PORT_EPS_CMD and len(d) == 2:
        extra = f"status={'OK' if d[0] == 0 else 'DITOLAK'} rail={'ON' if d[1] else 'OFF'}"
    elif d:
        extra = "data=" + d[:16].hex(" ") + (" ..." if len(d) > 16 else "")
    return line + ("  " + extra if extra else "")


def format_hk(d):
    up, t10, vdda, rail, _, cnt, rxf, txf = struct.unpack(EPS_HK_FMT, d)
    return (f"uptime={up}s suhu={t10 / 10:.1f}C VDDA={vdda}mV payload={'ON' if rail else 'OFF'} "
            f"HK#{cnt} CAN rx={rxf} tx={txf}")


# ---------------------------------------------------------------------------
# Koneksi ke Korlan USB2CAN
# ---------------------------------------------------------------------------
def open_bus(args):
    if args.interface == "virtual":
        return can.Bus(interface="virtual", channel=args.channel or "csp")
    kw = {"interface": args.interface, "bitrate": args.bitrate}
    if args.channel:
        kw["channel"] = args.channel
    if args.interface == "usb2can":
        here = os.path.dirname(os.path.abspath(__file__))
        dll = os.path.join(here, "usb2can.dll")
        if os.path.exists(dll):
            kw["dll"] = dll
    return can.Bus(**kw)


# ---------------------------------------------------------------------------
# Perintah
# ---------------------------------------------------------------------------
def cmd_list(args):
    cfgs = can.detect_available_configs(args.interface)
    if not cfgs:
        print("Tidak ada adapter terdeteksi. Cek driver 8devices, usb2can.dll dan pywin32.")
    for c in cfgs:
        print(c)


def cmd_monitor(args):
    bus = open_bus(args)
    node = CspNode(bus)
    print(f"Monitor CAN {args.bitrate} bit/s. Tekan Ctrl+C untuk berhenti.\n")
    t_last = time.time()
    try:
        while True:
            try:
                item = node.inbox.get(timeout=1.0)
            except queue.Empty:
                item = None
            ts = time.strftime("%H:%M:%S")
            if item:
                kind = item[0]
                if kind == "frame" and args.frames:
                    msg, err = item[1], item[2]
                    f = cfp_fields(msg.arbitration_id)
                    print(f"{ts}  frame 0x{msg.arbitration_id:08X} "
                          f"src={f['src']} dst={f['dst']} {'BEGIN' if f['type'] == 0 else 'MORE '} "
                          f"remain={f['remain']:3d} id={f['id']:4d} | {msg.data.hex(' ')}"
                          + (f"  !! {err}" if err else ""))
                elif kind == "packet":
                    print(f"{ts}  {describe_packet(item[1])}")
                elif kind == "error":
                    print(f"{ts}  ERROR FRAME {item[1]}")
                elif kind == "std":
                    m = item[1]
                    print(f"{ts}  frame standar 0x{m.arbitration_id:03X} {m.data.hex(' ')} (bukan CFP)")
            if time.time() - t_last >= 10:
                t_last = time.time()
                s = node.stats
                print(f"--- statistik: frame={s['frames']} paket CSP={s['packets']} error={s['errors']}")
    except KeyboardInterrupt:
        pass
    finally:
        node.close()
        bus.shutdown()


def cmd_ping(args):
    bus = open_bus(args)
    node = CspNode(bus)
    try:
        for i in range(args.count):
            payload = bytes(range(args.size))
            reply, ms = node.transaction(args.addr, PORT_PING, payload, args.timeout)
            if reply and reply["data"] == payload:
                print(f"ping {node_name(args.addr)} #{i}: {ms:.1f} ms ({args.size} byte)")
            elif reply:
                print(f"ping {node_name(args.addr)} #{i}: balasan berbeda ({len(reply['data'])} byte)")
            else:
                print(f"ping {node_name(args.addr)} #{i}: TIMEOUT")
            time.sleep(0.5)
    finally:
        node.close()
        bus.shutdown()


def cmd_hk(args):
    bus = open_bus(args)
    node = CspNode(bus)
    try:
        reply, ms = node.transaction(ADDR_EPS, PORT_EPS_HK, b"\x00", args.timeout)
        if reply and len(reply["data"]) == EPS_HK_SIZE:
            print(f"EPS housekeeping ({ms:.1f} ms): {format_hk(reply['data'])}")
        else:
            print("Housekeeping EPS: TIMEOUT / format salah")
    finally:
        node.close()
        bus.shutdown()


def cmd_payload(args):
    bus = open_bus(args)
    node = CspNode(bus)
    try:
        on = 1 if args.state == "on" else 0
        reply, ms = node.transaction(ADDR_EPS, PORT_EPS_CMD, bytes([0, on]), args.timeout)
        if reply and len(reply["data"]) == 2:
            st, rail = reply["data"]
            print(f"Perintah payload {args.state.upper()}: {'OK' if st == 0 else 'DITOLAK'}, "
                  f"rail sekarang {'ON' if rail else 'OFF'} ({ms:.1f} ms)")
        else:
            print("Perintah payload: TIMEOUT")
    finally:
        node.close()
        bus.shutdown()


def cmd_echo(args):
    bus = open_bus(args)
    node = CspNode(bus)
    try:
        reply, ms = node.transaction(args.addr, PORT_ECHO, args.text.encode(), args.timeout)
        if reply:
            print(f'Balasan {node_name(args.addr)} ({ms:.1f} ms): "{reply["data"].decode(errors="replace")}"')
        else:
            print("Echo: TIMEOUT")
    finally:
        node.close()
        bus.shutdown()


def main():
    p = argparse.ArgumentParser(description="Alat uji CSP lewat CAN (Korlan USB2CAN)")
    p.add_argument("--interface", default="usb2can", help="usb2can (default), virtual, dll")
    p.add_argument("--channel", default=None, help="serial number Korlan, mis. ED000200 (opsional)")
    p.add_argument("--bitrate", type=int, default=1000000, help="default 1000000 (1 Mbit/s)")
    p.add_argument("--timeout", type=float, default=1.0)
    sub = p.add_subparsers(dest="cmd", required=True)

    sub.add_parser("list", help="cari adapter yang terpasang").set_defaults(fn=cmd_list)

    m = sub.add_parser("monitor", help="tampilkan lalu lintas bus")
    m.add_argument("--frames", action="store_true", help="tampilkan juga setiap frame CAN")
    m.set_defaults(fn=cmd_monitor)

    pg = sub.add_parser("ping", help="ping node CSP")
    pg.add_argument("addr", type=int, help="1 = OBC, 2 = EPS")
    pg.add_argument("--count", type=int, default=4)
    pg.add_argument("--size", type=int, default=16)
    pg.set_defaults(fn=cmd_ping)

    sub.add_parser("hk", help="minta housekeeping EPS").set_defaults(fn=cmd_hk)

    pl = sub.add_parser("payload", help="ON/OFF rail payload EPS")
    pl.add_argument("state", choices=["on", "off"])
    pl.set_defaults(fn=cmd_payload)

    e = sub.add_parser("echo", help="kirim teks ke port 10")
    e.add_argument("addr", type=int)
    e.add_argument("text")
    e.set_defaults(fn=cmd_echo)

    args = p.parse_args()
    args.fn(args)


if __name__ == "__main__":
    main()
csp_can_gui.py – GUI Tkinter (622 baris)
#!/usr/bin/env python3
"""
csp_can_gui.py - GUI pemantau bus CAN CubeSat (libcsp 1.6 / CFP)
untuk Korlan USB2CAN (8devices) + python-can.

Jalankan:   python csp_can_gui.py
Butuh   :   pip install python-can pywin32   (+ usb2can.dll di folder ini)

Fitur:
  - Hubungkan ke Korlan USB2CAN (atau mode Simulasi tanpa hardware)
  - Tabel paket CSP yang sudah disusun ulang (warna per node)
  - Tabel frame CAN mentah (ID CFP diuraikan)
  - Panel EPS: suhu, VDDA, status payload, uptime (auto-refresh)
  - Tombol Ping OBC / Ping EPS, Minta HK, Payload ON/OFF, kirim pesan
  - Statistik bus + simpan log ke CSV
"""
import csv
import queue
import struct
import threading
import time
import tkinter as tk
from tkinter import ttk, filedialog, messagebox

import can

import csp_can_tool as T

APP_TITLE = "CubeSat CAN Monitor - Korlan USB2CAN"
MAX_ROWS = 2000

# Kode status/error yang dikirim adapter 8devices USB2CAN (byte data pertama)
USB2CAN_STATUS = {
    0x00: "OK", 0x01: "overrun", 0x02: "bus warning", 0x03: "error passive",
    0x04: "BUS-OFF", 0x20: "stuff error", 0x21: "form error", 0x23: "ACK error",
    0x24: "bit0 error", 0x25: "bit1 error", 0x27: "CRC error", -1: "error",
}

NODE_COLORS = {T.ADDR_OBC: "#1f5fd6", T.ADDR_EPS: "#c0392b", T.ADDR_PC: "#1a8a3a"}


# ---------------------------------------------------------------------------
# Simulator OBC + EPS (mode tanpa hardware, memakai bus 'virtual' python-can)
# ---------------------------------------------------------------------------
class Simulator:
    def __init__(self, channel):
        self.bus = can.Bus(interface="virtual", channel=channel)
        self.rx = T.CfpReassembler()
        self.ident = 0
        self.rail = 0
        self.hk_count = 0
        self.t0 = time.time()
        self.stop = threading.Event()
        threading.Thread(target=self._rx, daemon=True).start()
        threading.Thread(target=self._obc, daemon=True).start()

    def close(self):
        self.stop.set()
        time.sleep(0.15)
        self.bus.shutdown()

    def _send(self, src, dst, dport, sport, data):
        self.ident = (self.ident + 1) & 0x3FF
        hdr = T.csp_header_pack(2, src, dst, dport, sport)
        for cid, d in T.cfp_fragment(src, dst, data, hdr, self.ident):
            self.bus.send(can.Message(arbitration_id=cid, data=d, is_extended_id=True))
            time.sleep(0.0002)

    def _hk(self):
        temp = int(380 + 15 * ((time.time() * 0.2) % 1))
        return struct.pack(T.EPS_HK_FMT, int(time.time() - self.t0), temp, 3290,
                           self.rail, 0, self.hk_count, 0, 0)

    def _reply(self, node, pkt):
        h, d = pkt["hdr"], pkt["data"]
        if h["dport"] == T.PORT_PING:
            out = d
        elif h["dport"] == T.PORT_ECHO:
            out = f"ACK dari node {node} ({T.NODE_NAMES[node]})".encode()
        elif h["dport"] == T.PORT_TLM:
            return
        elif node == T.ADDR_EPS and h["dport"] == T.PORT_EPS_HK:
            self.hk_count += 1
            out = self._hk()
        elif node == T.ADDR_EPS and h["dport"] == T.PORT_EPS_CMD and len(d) == 2:
            self.rail = 1 if d[1] else 0
            out = bytes([0, self.rail])
        else:
            return
        self._send(node, h["src"], h["sport"], h["dport"], out)

    def _rx(self):
        while not self.stop.is_set():
            m = self.bus.recv(0.05)
            if m is None or not m.is_extended_id:
                continue
            pkt, _ = self.rx.push(m.arbitration_id, m.data)
            if pkt and pkt["hdr"]["dst"] in (T.ADDR_OBC, T.ADDR_EPS):
                self._reply(pkt["hdr"]["dst"], pkt)

    def _obc(self):
        """OBC tiruan: tiap 2 detik ping, HK dan paket 200 byte ke EPS."""
        n = 0
        while not self.stop.wait(2.0):
            self._send(T.ADDR_OBC, T.ADDR_EPS, T.PORT_PING, 30, bytes(range(32)))
            self._send(T.ADDR_EPS, T.ADDR_OBC, 30, T.PORT_PING, bytes(range(32)))
            self._send(T.ADDR_OBC, T.ADDR_EPS, T.PORT_EPS_HK, 31, b"\x00")
            self.hk_count += 1
            self._send(T.ADDR_EPS, T.ADDR_OBC, 31, T.PORT_EPS_HK, self._hk())
            self._send(T.ADDR_OBC, T.ADDR_EPS, T.PORT_TLM, 32, bytes((i + n) & 0xFF for i in range(200)))
            n += 1


# ---------------------------------------------------------------------------
# GUI
# ---------------------------------------------------------------------------
class App(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title(APP_TITLE)
        self.geometry("1280x800")
        self.minsize(1000, 640)

        self.bus = None
        self.node = None
        self.sim = None
        self.pending = {}            # sport -> (kind, info, t0, deadline)
        self.log_rows = []           # untuk simpan CSV
        self.node_seen = {T.ADDR_OBC: 0.0, T.ADDR_EPS: 0.0}
        self.rate_count = 0
        self.rate_t = time.time()
        self.err_counts = {}
        self.err_total = 0
        self.last_err = None
        self.err_log_t = 0.0
        self.send_errors = []

        self._style()
        self._build()
        self.after(40, self._poll)
        self.after(2000, self._auto_hk)
        self.after(300, lambda: self._scan_serial(quiet=True))
        self.protocol("WM_DELETE_WINDOW", self._on_close)

    # ---------------------------------------------------------------- style
    def _style(self):
        import tkinter.font as tkfont
        fams = set(tkfont.families(self))
        ui = next((f for f in ("Segoe UI", "Helvetica", "DejaVu Sans", "Arial") if f in fams), "TkDefaultFont")
        mono = next((f for f in ("Consolas", "DejaVu Sans Mono", "Courier New") if f in fams), "TkFixedFont")
        self.ui_font, self.mono_font = ui, mono
        s = ttk.Style(self)
        try:
            s.theme_use("clam")
        except tk.TclError:
            pass
        self.configure(bg="#eef1f5")
        s.configure(".", font=(ui, 10), background="#eef1f5")
        s.configure("Card.TFrame", background="white", relief="flat")
        s.configure("Card.TLabel", background="white")
        s.configure("Title.TLabel", font=(ui, 11, "bold"), background="white")
        s.configure("Big.TLabel", font=(ui, 22, "bold"), background="white")
        s.configure("Unit.TLabel", font=(ui, 9), foreground="#666", background="white")
        s.configure("Treeview", rowheight=22, font=(mono, 10))
        s.configure("Treeview.Heading", font=(ui, 10, "bold"))
        s.configure("Accent.TButton", font=(ui, 10, "bold"))

    def _card(self, parent, title):
        f = ttk.Frame(parent, style="Card.TFrame", padding=10)
        ttk.Label(f, text=title, style="Title.TLabel").pack(anchor="w", pady=(0, 6))
        return f

    # ---------------------------------------------------------------- layout
    def _build(self):
        # ---- baris koneksi
        top = ttk.Frame(self, padding=(10, 8))
        top.pack(fill="x")
        ttk.Label(top, text="Adapter:").pack(side="left")
        self.var_if = tk.StringVar(value="usb2can")
        ttk.Combobox(top, textvariable=self.var_if, width=10, state="readonly",
                     values=["usb2can", "simulasi"]).pack(side="left", padx=(4, 12))
        ttk.Label(top, text="Serial:").pack(side="left")
        self.var_ch = tk.StringVar()
        self.cb_ch = ttk.Combobox(top, textvariable=self.var_ch, width=11)
        self.cb_ch.pack(side="left", padx=(4, 2))
        ttk.Button(top, text="Cari", width=5, command=self._scan_serial).pack(side="left", padx=(0, 12))
        ttk.Label(top, text="Bitrate:").pack(side="left")
        self.var_br = tk.StringVar(value="1000000")
        ttk.Combobox(top, textvariable=self.var_br, width=9,
                     values=["1000000", "500000", "250000", "125000"]).pack(side="left", padx=(4, 12))
        self.btn_conn = ttk.Button(top, text="Hubungkan", style="Accent.TButton", command=self._toggle)
        self.btn_conn.pack(side="left")
        self.lbl_status = tk.Label(top, text="  TIDAK TERHUBUNG  ", bg="#999", fg="white",
                                   font=(self.ui_font, 10, "bold"))
        self.lbl_status.pack(side="left", padx=12)
        ttk.Button(top, text="Simpan log CSV", command=self._save_csv).pack(side="right")
        ttk.Button(top, text="Bersihkan", command=self._clear).pack(side="right", padx=6)

        body = ttk.Frame(self, padding=(10, 0, 10, 10))
        body.pack(fill="both", expand=True)

        # ---- panel kiri
        left = ttk.Frame(body, width=330)
        left.pack(side="left", fill="y", padx=(0, 10))
        left.pack_propagate(False)

        # node status
        c = self._card(left, "Status node")
        c.pack(fill="x", pady=(0, 8))
        self.node_lbl = {}
        for addr in (T.ADDR_OBC, T.ADDR_EPS):
            row = ttk.Frame(c, style="Card.TFrame")
            row.pack(fill="x", pady=2)
            dot = tk.Label(row, text="●", fg="#bbb", bg="white", font=(self.ui_font, 14))
            dot.pack(side="left")
            ttk.Label(row, text=f"{T.NODE_NAMES[addr]} (alamat {addr})", style="Card.TLabel",
                      width=16).pack(side="left")
            info = ttk.Label(row, text="belum terlihat", style="Card.TLabel", foreground="#777")
            info.pack(side="left")
            self.node_lbl[addr] = (dot, info)

        # EPS housekeeping
        c = self._card(left, "EPS housekeeping")
        c.pack(fill="x", pady=(0, 8))
        grid = ttk.Frame(c, style="Card.TFrame")
        grid.pack(fill="x")
        self.hk = {}
        for i, (key, label, unit) in enumerate([("temp", "Suhu chip", "°C"), ("vdda", "VDDA", "mV"),
                                                ("rail", "Payload", ""), ("uptime", "Uptime", "detik")]):
            f = ttk.Frame(grid, style="Card.TFrame", padding=(0, 2))
            f.grid(row=i // 2, column=i % 2, sticky="w", padx=(0, 18), pady=2)
            ttk.Label(f, text=label, style="Unit.TLabel").pack(anchor="w")
            v = ttk.Label(f, text="–", style="Big.TLabel")
            v.pack(anchor="w")
            ttk.Label(f, text=unit, style="Unit.TLabel").pack(anchor="w")
            self.hk[key] = v
        self.lbl_hk_time = ttk.Label(c, text="belum ada data", style="Unit.TLabel")
        self.lbl_hk_time.pack(anchor="w", pady=(4, 4))
        r = ttk.Frame(c, style="Card.TFrame")
        r.pack(fill="x")
        ttk.Button(r, text="Minta HK", command=self._req_hk).pack(side="left")
        self.var_auto = tk.BooleanVar(value=True)
        ttk.Checkbutton(r, text="otomatis tiap 2 s", variable=self.var_auto).pack(side="left", padx=8)

        # perintah
        c = self._card(left, "Perintah")
        c.pack(fill="x", pady=(0, 8))
        r = ttk.Frame(c, style="Card.TFrame")
        r.pack(fill="x", pady=2)
        ttk.Button(r, text="Ping OBC", command=lambda: self._ping(T.ADDR_OBC)).pack(side="left", expand=True, fill="x")
        ttk.Button(r, text="Ping EPS", command=lambda: self._ping(T.ADDR_EPS)).pack(side="left", expand=True, fill="x", padx=(6, 0))
        r = ttk.Frame(c, style="Card.TFrame")
        r.pack(fill="x", pady=2)
        ttk.Button(r, text="Payload ON", command=lambda: self._payload(1)).pack(side="left", expand=True, fill="x")
        ttk.Button(r, text="Payload OFF", command=lambda: self._payload(0)).pack(side="left", expand=True, fill="x", padx=(6, 0))
        r = ttk.Frame(c, style="Card.TFrame")
        r.pack(fill="x", pady=(6, 2))
        self.var_dst = tk.StringVar(value="EPS (2)")
        ttk.Combobox(r, textvariable=self.var_dst, width=8, state="readonly",
                     values=["OBC (1)", "EPS (2)"]).pack(side="left")
        self.var_msg = tk.StringVar(value="Halo dari PC")
        e = ttk.Entry(r, textvariable=self.var_msg)
        e.pack(side="left", fill="x", expand=True, padx=6)
        e.bind("<Return>", lambda _: self._echo())
        ttk.Button(r, text="Kirim", command=self._echo).pack(side="left")
        self.lbl_result = tk.Label(c, text="", bg="white", anchor="w", justify="left",
                                   font=(self.ui_font, 10), wraplength=300)
        self.lbl_result.pack(fill="x", pady=(6, 0))

        # statistik
        c = self._card(left, "Statistik bus")
        c.pack(fill="both", expand=True)
        self.stat_lbl = {}
        for key, label in [("frames", "Frame CAN"), ("packets", "Paket CSP"), ("rate", "Frame/detik"),
                           ("errors", "Error frame"), ("cfp_err", "Error susun CFP")]:
            row = ttk.Frame(c, style="Card.TFrame")
            row.pack(fill="x")
            ttk.Label(row, text=label, style="Card.TLabel").pack(side="left")
            v = ttk.Label(row, text="0", style="Card.TLabel", font=(self.ui_font, 10, "bold"))
            v.pack(side="right")
            self.stat_lbl[key] = v
        self.cfp_err = 0
        row = ttk.Frame(c, style="Card.TFrame")
        row.pack(fill="x", pady=(6, 0))
        ttk.Label(row, text="Kondisi bus", style="Card.TLabel").pack(side="left")
        self.lbl_bus = tk.Label(row, text="–", bg="white", font=(self.ui_font, 10, "bold"))
        self.lbl_bus.pack(side="right")

        # ---- panel kanan: tab
        nb = ttk.Notebook(body)
        nb.pack(side="left", fill="both", expand=True)

        t1 = ttk.Frame(nb)
        nb.add(t1, text="  Paket CSP  ")
        cols = ("time", "src", "dst", "port", "len", "info")
        self.tv_pkt = self._tree(t1, cols, {"time": ("Waktu", 95), "src": ("Dari", 90), "dst": ("Ke", 90),
                                           "port": ("Port", 110), "len": ("Len", 50), "info": ("Isi", 520)})
        for addr, col in NODE_COLORS.items():
            self.tv_pkt.tag_configure(f"n{addr}", foreground=col)

        t2 = ttk.Frame(nb)
        nb.add(t2, text="  Frame CAN  ")
        self.var_frames = tk.BooleanVar(value=True)
        ttk.Checkbutton(t2, text="Tampilkan frame (matikan jika bus sangat ramai)",
                        variable=self.var_frames).pack(anchor="w", padx=4, pady=2)
        cols = ("time", "id", "src", "dst", "type", "remain", "cfp", "dlc", "data")
        self.tv_frm = self._tree(t2, cols, {"time": ("Waktu", 95), "id": ("CAN ID", 95), "src": ("Src", 45),
                                           "dst": ("Dst", 45), "type": ("Tipe", 60), "remain": ("Remain", 60),
                                           "cfp": ("CFP id", 60), "dlc": ("DLC", 40), "data": ("Data", 300)})
        self.tv_frm.tag_configure("err", foreground="#c0392b")

        t3 = ttk.Frame(nb)
        nb.add(t3, text="  Log  ")
        self.txt_log = tk.Text(t3, font=(self.mono_font, 10), wrap="word", bg="#fbfbfb")
        self.txt_log.pack(fill="both", expand=True)

        self._log("Siap. Pilih adapter lalu klik Hubungkan. Pilih 'simulasi' untuk mencoba tanpa hardware.")

    def _tree(self, parent, cols, spec):
        frame = ttk.Frame(parent)
        frame.pack(fill="both", expand=True)
        tv = ttk.Treeview(frame, columns=cols, show="headings")
        for c in cols:
            title, w = spec[c]
            tv.heading(c, text=title)
            tv.column(c, width=w, anchor="w", stretch=(c in ("info", "data")))
        sb = ttk.Scrollbar(frame, orient="vertical", command=tv.yview)
        tv.configure(yscrollcommand=sb.set)
        tv.pack(side="left", fill="both", expand=True)
        sb.pack(side="right", fill="y")
        return tv

    # ---------------------------------------------------------------- koneksi
    def _toggle(self):
        if self.bus:
            self._disconnect()
        else:
            self._connect()

    def _connect(self):
        try:
            br = int(self.var_br.get())
            if self.var_if.get() == "simulasi":
                self.sim = Simulator("csp_sim")
                self.bus = can.Bus(interface="virtual", channel="csp_sim")
            else:
                if not self.var_ch.get().strip():
                    self._scan_serial(quiet=True)
                class A:  # argumen untuk open_bus()
                    pass
                a = A()
                a.interface, a.channel, a.bitrate = "usb2can", (self.var_ch.get().strip() or None), br
                self.bus = T.open_bus(a)
            self.node = T.CspNode(self.bus)
        except Exception as e:  # noqa: BLE001
            self.bus = None
            if self.sim:
                self.sim.close()
                self.sim = None
            messagebox.showerror("Gagal terhubung",
                                 f"{e}\n\nCek: driver 8devices terpasang, usb2can.dll (64-bit) ada di "
                                 "folder script, pywin32 terpasang, dan serial number benar.")
            return
        self.btn_conn.configure(text="Putuskan")
        sn = "" if self.var_if.get() == "simulasi" else (self.var_ch.get().strip() or "?")
        self.lbl_status.configure(text=f"  TERHUBUNG {int(self.var_br.get()) // 1000} kbit/s"
                                       + (f" · S/N {sn}" if sn else "") + "  ", bg="#1a8a3a")
        self.title(f"{APP_TITLE}" + (f"  [S/N {sn}]" if sn else ""))
        self._log(f"Terhubung ({self.var_if.get()}{', S/N ' + sn if sn else ''}), alamat CSP PC = {T.ADDR_PC}")

    def _scan_serial(self, quiet=False):
        """Cari Korlan USB2CAN yang terpasang dan isi kolom Serial."""
        try:
            cfgs = can.detect_available_configs("usb2can")
        except Exception as e:  # noqa: BLE001
            cfgs = []
            if not quiet:
                self._log(f"Gagal mencari adapter: {e}")
        serials = [str(c.get("channel")) for c in cfgs if c.get("channel")]
        self.cb_ch.configure(values=serials)
        if serials:
            if self.var_ch.get().strip() not in serials:
                self.var_ch.set(serials[0])
            self._log("Korlan USB2CAN terdeteksi: S/N " + ", ".join(serials))
        elif not quiet:
            self._log("Korlan USB2CAN tidak terdeteksi (cek kabel USB / driver 8devices / pywin32).")
        return serials

    def _disconnect(self):
        try:
            if self.node:
                self.node.close()
            if self.bus:
                self.bus.shutdown()
            if self.sim:
                self.sim.close()
        finally:
            self.node = self.bus = self.sim = None
            self.pending.clear()
        self.btn_conn.configure(text="Hubungkan")
        self.lbl_status.configure(text="  TIDAK TERHUBUNG  ", bg="#999")
        self._log("Koneksi diputus")

    def _on_close(self):
        if self.bus:
            self._disconnect()
        self.destroy()

    # ---------------------------------------------------------------- perintah
    def _request(self, kind, dst, dport, payload, info=None, timeout=1.0):
        if not self.node:
            self._result("Belum terhubung ke adapter.", "#c0392b")
            return
        node = self.node
        sport = node.alloc_sport()
        self.pending[sport] = (kind, dst, info, time.time(), time.time() + timeout)

        def worker():
            # dikirim di thread terpisah agar GUI tidak macet bila adapter lambat
            try:
                node.send(dst, dport, payload, sport=sport)
            except Exception as e:  # noqa: BLE001
                self.send_errors.append(str(e))
        threading.Thread(target=worker, daemon=True).start()

    def _ping(self, addr):
        self._request("ping", addr, T.PORT_PING, bytes(range(16)), bytes(range(16)))

    def _req_hk(self):
        self._request("hk", T.ADDR_EPS, T.PORT_EPS_HK, b"\x00")

    def _payload(self, on):
        self._request("cmd", T.ADDR_EPS, T.PORT_EPS_CMD, bytes([0, on]), on)

    def _echo(self):
        dst = T.ADDR_OBC if self.var_dst.get().startswith("OBC") else T.ADDR_EPS
        self._request("echo", dst, T.PORT_ECHO, self.var_msg.get().encode())

    def _auto_hk(self):
        if self.node and self.var_auto.get():
            if not any(p[0] == "hk" for p in self.pending.values()):
                self._req_hk()
        self.after(2000, self._auto_hk)

    def _handle_reply(self, pkt):
        h = pkt["hdr"]
        p = self.pending.get(h["dport"])
        if not p or h["src"] != p[1] or h["dst"] != T.ADDR_PC:
            return
        del self.pending[h["dport"]]
        kind, dst, info, t0, _ = p
        ms = (pkt.get("t", time.time()) - t0) * 1000
        name = f"{T.NODE_NAMES[dst]} ({dst})"
        d = pkt["data"]
        if kind == "ping":
            ok = d == info
            self._result(f"Ping {name}: {ms:.1f} ms" + ("" if ok else " (data berbeda!)"),
                         "#1a8a3a" if ok else "#c0392b")
        elif kind == "hk" and len(d) == T.EPS_HK_SIZE:
            self._show_hk(d, ms)
        elif kind == "cmd" and len(d) == 2:
            ok = d[0] == 0
            self._result(f"Payload {'ON' if d[1] else 'OFF'}: {'OK' if ok else 'DITOLAK'} ({ms:.1f} ms)",
                         "#1a8a3a" if ok else "#c0392b")
        elif kind == "echo":
            self._result(f'Balasan {name} ({ms:.1f} ms):\n"{d.decode(errors="replace")}"', "#1f5fd6")

    def _show_hk(self, d, ms=None):
        up, t10, vdda, rail, _, cnt, rxf, txf = struct.unpack(T.EPS_HK_FMT, d)
        self.hk["temp"].configure(text=f"{t10 / 10:.1f}")
        self.hk["vdda"].configure(text=f"{vdda}")
        self.hk["rail"].configure(text="ON" if rail else "OFF", foreground="#1a8a3a" if rail else "#777")
        self.hk["uptime"].configure(text=f"{up}")
        extra = f", {ms:.0f} ms" if ms is not None else ""
        self.lbl_hk_time.configure(text=f"diperbarui {time.strftime('%H:%M:%S')}{extra} · HK #{cnt}")

    def _check_timeouts(self):
        now = time.time()
        for sport, p in list(self.pending.items()):
            if now > p[4]:
                del self.pending[sport]
                kind, dst = p[0], p[1]
                label = {"ping": "Ping", "hk": "Housekeeping", "cmd": "Perintah payload",
                         "echo": "Pesan"}[kind]
                self._result(f"{label} ke {T.NODE_NAMES[dst]} ({dst}): TIMEOUT", "#c0392b")
                self._log(f"{label} ke node {dst}: timeout")

    # ---------------------------------------------------------------- loop GUI
    def _poll(self):
        if self.node:
            n = 0
            while n < 500:
                try:
                    item = self.node.inbox.get_nowait()
                except queue.Empty:
                    break
                n += 1
                self._handle_item(item)
            self._check_timeouts()
            self._update_stats()
        self.after(40, self._poll)

    def _handle_item(self, item):
        ts = time.strftime("%H:%M:%S") + f".{int(time.time() * 1000) % 1000:03d}"
        kind = item[0]
        if kind == "frame":
            msg, err = item[1], item[2]
            self.rate_count += 1
            f = T.cfp_fields(msg.arbitration_id)
            if f["src"] in self.node_seen:
                self.node_seen[f["src"]] = time.time()
            if err:
                self.cfp_err += 1
                self._log(f"CFP: {err} (ID 0x{msg.arbitration_id:08X})")
            if self.var_frames.get():
                self._insert(self.tv_frm, (ts, f"0x{msg.arbitration_id:08X}", f["src"], f["dst"],
                                           "BEGIN" if f["type"] == 0 else "MORE", f["remain"], f["id"],
                                           len(msg.data), msg.data.hex(" ")), ("err",) if err else ())
        elif kind == "packet":
            pkt = item[1]
            h = pkt["hdr"]
            desc = T.describe_packet(pkt)
            info = desc.split("  ", 1)[1] if "  " in desc else ""
            if h["dport"] in T.PORT_NAMES:
                port = f"{h['dport']} {T.PORT_NAMES[h['dport']]}"
            elif h["sport"] in T.PORT_NAMES:
                port = f"{h['dport']} re:{T.PORT_NAMES[h['sport']]}"
            else:
                port = str(h["dport"])
            self._insert(self.tv_pkt, (ts, f"{T.NODE_NAMES.get(h['src'], '?')}:{h['sport']}",
                                       f"{T.NODE_NAMES.get(h['dst'], '?')}:{h['dport']}", port,
                                       len(pkt["data"]), info), (f"n{h['src']}",))
            self.log_rows.append([ts, h["src"], h["sport"], h["dst"], h["dport"], len(pkt["data"]),
                                  pkt["data"].hex(), info])
            # HK EPS (dari siapa pun yang meminta, termasuk OBC) -> perbarui panel
            if h["src"] == T.ADDR_EPS and h["sport"] == T.PORT_EPS_HK and len(pkt["data"]) == T.EPS_HK_SIZE:
                self._show_hk(pkt["data"])
            self._handle_reply(pkt)
        elif kind == "error":
            data = bytes(item[1].data)
            code = data[0] if data else -1
            self.err_counts[code] = self.err_counts.get(code, 0) + 1
            self.err_total += 1
            self.last_err = code
        elif kind == "std":
            m = item[1]
            self._log(f"Frame standar 0x{m.arbitration_id:03X} {m.data.hex(' ')} (bukan CFP)")

    def _insert(self, tv, values, tags=()):
        tv.insert("", "end", values=values, tags=tags)
        kids = tv.get_children()
        if len(kids) > MAX_ROWS:
            tv.delete(*kids[:len(kids) - MAX_ROWS])
        tv.yview_moveto(1.0)

    def _update_stats(self):
        now_t = time.time()
        while self.send_errors:
            self._result(f"Gagal kirim: {self.send_errors.pop()}", "#c0392b")
        if self.err_counts and now_t - self.err_log_t >= 1.0:
            self.err_log_t = now_t
            txt = ", ".join(f"{USB2CAN_STATUS.get(c, hex(c))} x{n}" for c, n in sorted(self.err_counts.items()))
            self._log(f"Status/error dari adapter (1 detik terakhir): {txt}")
            self.lbl_bus.configure(text="ERROR: " + USB2CAN_STATUS.get(self.last_err, hex(self.last_err)),
                                   fg="#c0392b")
            self.err_counts.clear()
            self.bus_err_t = now_t
        elif now_t - getattr(self, "bus_err_t", 0) > 3 and self.node.stats["frames"]:
            self.lbl_bus.configure(text="OK", fg="#1a8a3a")
        s = self.node.stats
        self.stat_lbl["frames"].configure(text=str(s["frames"]))
        self.stat_lbl["packets"].configure(text=str(s["packets"]))
        self.stat_lbl["errors"].configure(text=str(s["errors"]))
        self.stat_lbl["cfp_err"].configure(text=str(self.cfp_err))
        now = time.time()
        if now - self.rate_t >= 1.0:
            self.stat_lbl["rate"].configure(text=f"{self.rate_count / (now - self.rate_t):.0f}")
            self.rate_count, self.rate_t = 0, now
        for addr, (dot, info) in self.node_lbl.items():
            seen = self.node_seen[addr]
            if seen == 0:
                continue
            age = now - seen
            if age < 5:
                dot.configure(fg="#1a8a3a")
                info.configure(text="aktif", foreground="#1a8a3a")
            else:
                dot.configure(fg="#e67e22")
                info.configure(text=f"diam {age:.0f} s", foreground="#e67e22")

    # ---------------------------------------------------------------- util
    def _result(self, text, color="#333"):
        self.lbl_result.configure(text=text, fg=color)

    def _log(self, text):
        self.txt_log.insert("end", f"[{time.strftime('%H:%M:%S')}] {text}\n")
        self.txt_log.see("end")

    def _clear(self):
        for tv in (self.tv_pkt, self.tv_frm):
            tv.delete(*tv.get_children())
        self.log_rows.clear()
        self.txt_log.delete("1.0", "end")

    def _save_csv(self):
        if not self.log_rows:
            messagebox.showinfo("Simpan log", "Belum ada paket CSP untuk disimpan.")
            return
        path = filedialog.asksaveasfilename(defaultextension=".csv",
                                            initialfile=time.strftime("csp_log_%Y%m%d_%H%M%S.csv"),
                                            filetypes=[("CSV", "*.csv")])
        if not path:
            return
        with open(path, "w", newline="", encoding="utf-8") as f:
            w = csv.writer(f)
            w.writerow(["waktu", "src", "sport", "dst", "dport", "len", "data_hex", "keterangan"])
            w.writerows(self.log_rows)
        self._log(f"Log disimpan: {path}")


if __name__ == "__main__":
    App().mainloop()

csp_can_tool.py juga bisa dipakai tanpa GUI, misalnya python csp_can_tool.py monitor, python csp_can_tool.py ping 2 atau python csp_can_tool.py hk.

Penutup

Dengan Korlan USB2CAN dan sedikit Python, kita punya alat diagnosis bus CAN yang memahami protokol CubeSat Space Protocol: frame mentah, paket CSP, data housekeeping EPS, sampai status error fisik bus. Hasil uji dengan dua board sungguhan menunjukkan OBC dan EPS berkomunikasi tanpa error setelah terminator 120 Ω terpasang. Semoga bermanfaat.

Wassalamu'alaikum warohmatullohi wabarakatuh.