Assalamu'alaikum warohmatullohi wabarakatuh,
Pada tutorial Sniffing CAN Bus CubeSat (libcsp) dengan Korlan USB2CAN dan GUI Python, kita memakai Korlan USB2CAN dari 8devices sebagai "telinga" ketiga di bus CAN antara board OBC (On-Board Computer) dan EPS (Electrical Power System) dari tutorial Komunikasi CubeSat Space Protocol (libcsp) lewat CAN Bus FDCAN1 pada STM32H743 dengan FreeRTOS.
Kali ini adapter-nya diganti dengan Makerbase CANable V2.0 Pro, adapter USB ke CAN yang murah, open-source, dan sudah terisolasi galvanik. Program Python dengan GUI-nya tetap sama: menampilkan frame CAN mentah, menyusun ulang frame menjadi paket CSP, menampilkan data housekeeping EPS, dan mengirim perintah (ping, payload ON/OFF) langsung dari laptop. Yang berubah hanya cara terhubung ke adapter.
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.
Korlan USB2CAN vs Makerbase CANable V2.0 Pro
| Korlan USB2CAN (8devices) | Makerbase CANable V2.0 Pro | |
|---|---|---|
| Mikrokontroler | STM32 | STM32G431 (mendukung CAN FD) |
| Konektor | DB9 | terminal sekrup CANH / CANL / GND |
| Isolasi | tidak | ya (galvanik), lebih aman untuk laptop |
| Terminator 120 Ω | tidak ada | ada di board, jumper 120R |
| Driver Windows | WinUSB + usb2can.dll | tidak perlu (firmware slcan = port COM) |
| python-can | interface="usb2can" | interface="slcan" |
| Laporan jenis error bus | ada (stuff, ACK, CRC, dll.) | tidak ada di firmware slcan |
Perangkat yang dibutuhkan
| Perangkat | Keterangan |
|---|---|
| Makerbase CANable V2.0 Pro | adapter USB ke CAN terisolasi, firmware slcan |
| 2 board STM32H743 (DevEBox) + modul MCP2551 | OBC dan EPS dari tutorial sebelumnya |
| 2 resistor 120 Ω | terminator di kedua ujung bus |
| Kabel jumper | CANH, CANL, GND ke terminal sekrup CANable |
| Laptop Windows + Python 3 | python-can, pyserial, Tkinter |
Makerbase CANable V2.0 Pro tampak atas: konektor USB-C, STM32G431, LED PWR / STATA / WORK, modul isolasi, terminal sekrup hijau dan jumper terminator 120 Ω di ujung kiri.
Tampak bawah: urutan terminal CANL, CANH, GND, pad jumper 120R (terminator), pad BOOT untuk update firmware, dan header SWD (SWC, SWD, GND, 3V3).
Langkah 1 – Cek firmware dan pasang library
CANable 2.0 bisa memakai dua jenis firmware. Colokkan CANable ke laptop lalu buka Device Manager:
- slcan → muncul di Ports (COM & LPT) sebagai port COM baru (misalnya COM7). Inilah yang dipakai di tutorial ini. Di Windows 10/11 tidak perlu driver tambahan dan tidak perlu DLL.
- candleLight (gs_usb) → muncul sebagai perangkat USB "candleLight", bukan port COM. Firmware bisa diganti ke slcan lewat canable.io/updater (buka di Chrome, pilih firmware CANable 2.0 slcan, board masuk mode boot/DFU dengan menghubungkan (short) dua pad BOOT di bagian bawah board saat USB dicolokkan, lalu klik Update).
Setelah port COM terlihat, pasang library Python:
pip install python-can pyserialpyserial dipakai python-can untuk berbicara dengan CANable lewat port COM, sekaligus untuk mencari port CANable secara otomatis. pywin32 dan usb2can.dll dari versi Korlan tidak dibutuhkan lagi.
Langkah 2 – Menyambung CANable ke bus CAN
CANable 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).
| Terminal CANable (urutan di board) | Bus CAN |
|---|---|
| CANL | CANL |
| CANH | CANH |
| GND | GND board |
Penting – matikan terminator 120 Ω di CANable. Berbeda dengan Korlan, CANable V2.0 Pro punya resistor 120 Ω di board yang diaktifkan dengan jumper berlabel 120R di samping terminal sekrup (lihat foto: jumper hitam di ujung kiri masih terpasang). Karena CANable menyadap di tengah bus, terminator ini harus OFF: cabut jumper 120R. Kalau ON, ada tiga resistor 120 Ω paralel sehingga hambatan bus turun menjadi sekitar 40 Ω.
Cek dengan multimeter saat semua board mati: hambatan CANH–CANL seharusnya sekitar 60 Ω (dua resistor 120 Ω paralel). Kalau terbaca sekitar 40 Ω, terminator di CANable masih ON (jumper 120R masih terpasang).
Karena CANable V2.0 Pro terisolasi, GND sisi CAN tidak terhubung ke GND laptop. Tetap sambungkan terminal GND CANable ke GND board supaya level sinyal punya acuan yang sama.
Langkah 3 – Tes koneksi dengan Python
Sebelum membuat GUI, pastikan Python bisa membuka CANable. Skrip pendek ini mencari port COM lalu menampilkan frame selama 5 detik:
import time
import can
from serial.tools import list_ports
# 1) tampilkan port COM; CANable (firmware slcan) punya VID 16D0
for p in list_ports.comports():
print(p.device, p.description, hex(p.vid or 0))
# contoh hasil: COM7 USB Serial Device (COM7) 0x16d0
# 2) buka bus 1 Mbit/s dan tampilkan frame selama 5 detik
bus = can.Bus(interface="slcan", channel="COM7", bitrate=1000000)
t0 = time.time()
while time.time() - t0 < 5:
msg = bus.recv(timeout=0.5)
if msg:
print(msg)
bus.shutdown()Bitrate harus sama dengan firmware, di proyek ini 1 Mbit/s. Pada firmware slcan, python-can mengirim bitrate ini ke CANable sebagai perintah S8. Nilai tty_baudrate tidak berpengaruh karena CANable memakai USB CDC (port COM virtual).
Langkah 4 – Membaca ID CAN sebagai header CFP
Bagian ini tidak berubah dari versi Korlan, karena tidak bergantung pada adapter. libcsp mengirim paket CSP lewat CAN memakai CFP (CAN Fragmentation Protocol). Semua informasi alamat ada di ID extended 29-bit:
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, NoneLangkah 6 – Arsitektur program dan perubahan untuk CANable
Alur datanya sama seperti versi Korlan: thread penerima membaca bus.recv() terus-menerus dan memasukkan hasilnya ke antrean, lalu GUI Tkinter mengambil isi antrean setiap 40 ms dengan after(), paling banyak 500 item sekali ambil. Ada tiga perubahan untuk CANable.
1. Membuka bus lewat slcan. Fungsi open_bus() sekarang memakai interface="slcan" dan port COM. Kalau port tidak diisi, port CANable dicari otomatis:
def open_bus(args):
if args.interface == "virtual":
return can.Bus(interface="virtual", channel=args.channel or "csp")
if args.interface == "slcan":
port = args.channel
if not port:
ports = find_canable_ports()
if not ports:
raise RuntimeError("Port COM CANable tidak ditemukan (cek kabel USB / pyserial).")
port = ports[0]
# CANable memakai USB CDC: tty_baudrate tidak berpengaruh,
# bitrate CAN dikirim sebagai perintah slcan (S8 = 1 Mbit/s).
return can.Bus(interface="slcan", channel=port, bitrate=args.bitrate)
kw = {"interface": args.interface, "bitrate": args.bitrate}
if args.channel:
kw["channel"] = args.channel
return can.Bus(**kw)2. Mencari port COM CANable. Firmware slcan CANable memakai USB VID:PID 16D0:117E. Port dengan VID ini diletakkan paling depan di daftar, jadi GUI langsung memilihnya:
def find_canable_ports():
"""Daftar port COM; port CANable (VID 16D0) diletakkan paling depan."""
if list_ports is None:
return []
canable, others = [], []
for p in list_ports.comports():
desc = (p.description or "") + " " + (p.product or "")
if (p.vid == CANABLE_SLCAN_VID and p.pid == CANABLE_SLCAN_PID) or "canable" in desc.lower():
canable.append(p.device)
else:
others.append(p.device)
return canable + others3. Satu pengirim pada satu waktu. slcan adalah protokol teks lewat port serial. Tombol di GUI mengirim dari thread terpisah, jadi pengiriman frame dilindungi threading.Lock() supaya perintah dua tombol yang ditekan bersamaan tidak bercampur di port serial:
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)
with self._tx_lock: # serial slcan: satu pengirim pada satu waktu
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 sportLangkah 7 – Menjalankan GUI
Simpan csp_can_tool.py dan csp_can_gui.py dalam satu folder, lalu jalankan:
python csp_can_gui.pyPilih Adapter: slcan untuk CANable sungguhan, pastikan kolom Port berisi port COM CANable (tombol Cari untuk mencari ulang), atau pilih simulasi untuk mencoba tampilan tanpa hardware (program membuat OBC dan EPS tiruan yang perilakunya sama dengan firmware). Klik Hubungkan.
Tab Paket CSP saat terhubung ke CANable di COM3 (1 Mbit/s): biru = OBC, merah = EPS. Kedua node aktif, panel kiri menampilkan housekeeping EPS (suhu chip 32,3 °C, VDDA 3338 mV) dan hasil Ping OBC 3,6 ms.
Yang terlihat di tab Paket CSP sama seperti versi Korlan:
- 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.
Tab Frame CAN: paket telemetri 200 byte dari OBC (1) ke EPS (2) dipecah menjadi frame BEGIN lalu MORE dengan remain turun 22…0, disusul balasan EPS (src 2) ke OBC (dst 1) dan ke PC (dst 10).
Hasil uji dengan CANable
Pengujian dengan Makerbase CANable V2.0 Pro di COM3, 1 Mbit/s, dengan board OBC dan EPS dari tutorial sebelumnya. Kedua node aktif dan seluruh lalu lintas CubeSat terbaca, sama seperti saat memakai Korlan:
- OBC → EPS PING dan EPS → OBC re:PING (32 byte) setiap 2 detik.
- OBC → EPS:10 ECHO berisi "Halo EPS, pesan ke-…", dibalas EPS "ACK dari node 2 (EPS)".
- OBC → EPS:12 EPS_HK dan balasan 20 byte: suhu chip ±30,5–32,5 °C, VDDA ±3337–3339 mV.
- OBC → EPS:11 TLM: telemetri 200 byte dengan checksum, dan EPS_CMD yang membalik status payload.
- Lalu lintas bus sekitar 47 frame/detik; dalam satu sesi tercatat lebih dari 6.000 frame CAN dan 1.000 paket CSP tersusun.
Perintah langsung dari laptop lewat CANable:
| Tombol di GUI | Hasil |
|---|---|
| Ping EPS | 3,5 ms |
| Ping OBC | 3,6 ms |
| Payload ON | status OK dalam 10,0 ms |
| Payload OFF | status OK dalam 45,4 ms |
| Kirim pesan ke EPS | balasan "ACK dari node 2 (EPS)" dalam 10,0 ms |
Catatan dari pengujian: di awal hanya EPS yang terbaca dan Ping OBC selalu TIMEOUT. Ternyata board OBC belum mendapat daya karena kabel USB-nya belum dicolokkan. Begitu OBC menyala, statusnya langsung "aktif". Perlu diingat juga, frame yang dikirim dari laptop (permintaan HK, ping) tidak muncul di tabel, karena firmware slcan tidak menampilkan ulang frame yang dikirimnya sendiri; yang terlihat adalah balasan dari node.
Troubleshooting
Firmware slcan tidak melaporkan jenis error bus seperti Korlan (stuff error, ACK error, CRC error, dan seterusnya). Karena itu baris Kondisi bus di GUI ditebak dari lalu lintas: OK kalau frame masuk dalam 3 detik terakhir, DIAM: tidak ada frame kalau tidak ada frame sama sekali. Kalau bus bermasalah, gejala utamanya adalah tidak ada frame yang masuk.
| Gejala | Kemungkinan penyebab | Solusi |
|---|---|---|
| Tidak ada port COM baru | firmware candleLight, kabel USB hanya untuk charging | ganti firmware ke slcan di canable.io/updater, pakai kabel USB data |
| Device Manager: "Unknown USB Device (Device Descriptor Request Failed)" | kabel USB-C hanya untuk charging atau kurang bagus | ganti kabel USB-C data; setelah diganti CANable langsung muncul sebagai USB Serial Device (COM3) |
| "could not open port" / "Access is denied" | port COM sedang dipakai program lain (serial monitor, program lain yang masih terbuka) | tutup program lain, cabut-colok CANable |
| Kondisi bus "DIAM", frame CAN = 0 | CANH/CANL tertukar atau belum tersambung, bitrate berbeda, GND tidak tersambung, board mati | cek CANH–CANH, CANL–CANL, GND bersama, samakan bitrate 1 Mbit/s |
| Frame hilang / error susun CFP, bus tidak stabil | terminator 120 Ω di CANable masih ON (bus ±40 Ω) atau tanpa terminator | matikan terminator CANable, pasang 2×120 Ω di ujung bus (ukur ±60 Ω) |
| Error terus walau kabel benar | MCP2551 dicatu 5 V butuh sinyal TXD ≥ 3,5 V, STM32 hanya 3,3 V | pakai transceiver 3,3 V (SN65HVD230) atau pull-up 2,2 kΩ CTX ke 5 V + PA12 open-drain |
| Hanya satu node "aktif", ping node lain TIMEOUT | board node itu belum mendapat daya / belum terhubung ke bus | pastikan semua board dicatu (USB/catu daya) dan CANH, CANL, GND tersambung |
Catatan: dalam mode normal, CANable juga ikut memberi ACK di bus seperti Korlan. Jadi kalau hanya satu board yang terpasang bersama CANable, frame board itu tetap terbaca. Ini cara cepat menguji OBC dan EPS satu per satu. Kalau butuh mencari error fisik secara detail (jenis error, TEC/REC), log UART di board STM32 tetap menjadi acuan, atau gunakan adapter yang melaporkan status error seperti Korlan.
Kode lengkap
Dua file berikut disimpan dalam satu folder. Klik judul untuk membuka kodenya.
csp_can_tool.py – library CFP/CSP + versi command line (476 baris)
#!/usr/bin/env python3
"""
csp_can_tool.py - membaca dan menguji bus CAN CubeSat (libcsp 1.6 / CFP)
memakai Makerbase CANable V2.0 Pro (firmware slcan) 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 --channel COM7 monitor
python csp_can_tool.py --channel COM7 ping 2
python csp_can_tool.py --channel COM7 hk
python csp_can_tool.py --channel COM7 payload on
python csp_can_tool.py --channel COM7 echo 1 "Halo OBC"
Tanpa --channel, port COM CANable dicari otomatis.
Kebutuhan:
pip install python-can pyserial
CANable dengan firmware slcan (muncul sebagai port COM, tanpa driver/DLL).
"""
import argparse
import queue
import struct
import threading
import time
import can
try:
from serial.tools import list_ports
except ImportError: # pyserial belum terpasang
list_ports = None
# ---------------------------------------------------------------------------
# Konstanta sistem (harus sama dengan firmware: cubesat_proto.h)
# ---------------------------------------------------------------------------
ADDR_OBC = 1
ADDR_EPS = 2
ADDR_PC = 10 # alamat CSP untuk PC/CANable (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)
# VID:PID CANable firmware slcan (USB CDC)
CANABLE_SLCAN_VID = 0x16D0
CANABLE_SLCAN_PID = 0x117E
# ---------------------------------------------------------------------------
# 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._tx_lock = threading.Lock()
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():
try:
msg = self.bus.recv(timeout=0.1)
except Exception as e: # noqa: BLE001 (mis. kabel USB dicabut)
self.inbox.put(("fatal", str(e)))
return
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)
with self._tx_lock: # serial slcan: satu pengirim pada satu waktu
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 CANable (slcan)
# ---------------------------------------------------------------------------
def find_canable_ports():
"""Daftar port COM; port CANable (VID 16D0) diletakkan paling depan."""
if list_ports is None:
return []
canable, others = [], []
for p in list_ports.comports():
desc = (p.description or "") + " " + (p.product or "")
if (p.vid == CANABLE_SLCAN_VID and p.pid == CANABLE_SLCAN_PID) or "canable" in desc.lower():
canable.append(p.device)
else:
others.append(p.device)
return canable + others
def open_bus(args):
if args.interface == "virtual":
return can.Bus(interface="virtual", channel=args.channel or "csp")
if args.interface == "slcan":
port = args.channel
if not port:
ports = find_canable_ports()
if not ports:
raise RuntimeError("Port COM CANable tidak ditemukan (cek kabel USB / pyserial).")
port = ports[0]
# CANable memakai USB CDC: tty_baudrate tidak berpengaruh,
# bitrate CAN dikirim sebagai perintah slcan (S8 = 1 Mbit/s).
return can.Bus(interface="slcan", channel=port, bitrate=args.bitrate)
kw = {"interface": args.interface, "bitrate": args.bitrate}
if args.channel:
kw["channel"] = args.channel
return can.Bus(**kw)
# ---------------------------------------------------------------------------
# Perintah
# ---------------------------------------------------------------------------
def cmd_list(args):
ports = find_canable_ports()
if list_ports is None:
print("pyserial belum terpasang: pip install pyserial")
return
if not ports:
print("Tidak ada port COM. Cek kabel USB dan firmware CANable (harus slcan).")
for p in list_ports.comports():
tag = " <- CANable" if p.vid == CANABLE_SLCAN_VID else ""
print(f"{p.device:8s} {p.description}{tag}")
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)")
elif kind == "fatal":
print(f"{ts} koneksi ke adapter terputus: {item[1]}")
break
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 (Makerbase CANable V2.0 Pro)")
p.add_argument("--interface", default="slcan", help="slcan (default), virtual, dll")
p.add_argument("--channel", default=None, help="port COM CANable, mis. COM7 (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="tampilkan port COM / CANable").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 (634 baris)
#!/usr/bin/env python3
"""
csp_can_gui.py - GUI pemantau bus CAN CubeSat (libcsp 1.6 / CFP)
untuk Makerbase CANable V2.0 Pro (firmware slcan) + python-can.
Jalankan: python csp_can_gui.py
Butuh : pip install python-can pyserial
Fitur:
- Hubungkan ke CANable lewat port COM (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 - Makerbase CANable V2.0 Pro"
MAX_ROWS = 2000
SILENT_WARN_S = 3.0 # bus dianggap diam bila tidak ada frame selama ini
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, dst, 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_new = 0
self.err_log_t = 0.0
self.bus_err_t = 0.0
self.last_frame_t = 0.0
self.connect_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_ports(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="slcan")
ttk.Combobox(top, textvariable=self.var_if, width=10, state="readonly",
values=["slcan", "simulasi"]).pack(side="left", padx=(4, 12))
ttk.Label(top, text="Port:").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_ports).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 port COM CANable 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_ports(quiet=True)
class A: # argumen untuk open_bus()
pass
a = A()
a.interface, a.channel, a.bitrate = "slcan", (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: CANable memakai firmware slcan (muncul sebagai port COM), "
"port COM benar dan tidak sedang dipakai program lain, "
"serta python-can dan pyserial sudah terpasang.")
return
self.connect_t = time.time()
self.last_frame_t = 0.0
self.btn_conn.configure(text="Putuskan")
port = "" if self.var_if.get() == "simulasi" else (self.var_ch.get().strip() or "auto")
self.lbl_status.configure(text=f" TERHUBUNG {int(self.var_br.get()) // 1000} kbit/s"
+ (f" · {port}" if port else "") + " ", bg="#1a8a3a")
self.title(f"{APP_TITLE}" + (f" [{port}]" if port else ""))
self._log(f"Terhubung ({self.var_if.get()}{', ' + port if port else ''}), alamat CSP PC = {T.ADDR_PC}")
def _scan_ports(self, quiet=False):
"""Cari port COM; port CANable diletakkan paling depan."""
if T.list_ports is None:
if not quiet:
self._log("pyserial belum terpasang: pip install pyserial")
return []
ports = T.find_canable_ports()
self.cb_ch.configure(values=ports)
canable = [p.device for p in T.list_ports.comports() if p.vid == T.CANABLE_SLCAN_VID]
if ports:
if self.var_ch.get().strip() not in ports:
self.var_ch.set(ports[0])
if canable:
self._log("CANable terdeteksi di " + ", ".join(canable))
elif not quiet:
self._log("Port COM: " + ", ".join(ports) + " (CANable tidak dikenali otomatis, pilih manual)")
elif not quiet:
self._log("Tidak ada port COM (cek kabel USB dan firmware CANable harus slcan).")
return ports
def _disconnect(self):
try:
if self.node:
self.node.close()
if self.bus:
self.bus.shutdown()
if self.sim:
self.sim.close()
except Exception: # noqa: BLE001
pass
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.lbl_bus.configure(text="–", fg="black")
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 and self.node:
try:
item = self.node.inbox.get_nowait()
except queue.Empty:
break
n += 1
self._handle_item(item)
if self.node:
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
self.last_frame_t = time.time()
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":
# firmware slcan umumnya tidak mengirim error frame; kalau ada, hitung saja
self.err_new += 1
elif kind == "std":
m = item[1]
self._log(f"Frame standar 0x{m.arbitration_id:03X} {m.data.hex(' ')} (bukan CFP)")
elif kind == "fatal":
self._log(f"Koneksi ke adapter terputus: {item[1]}")
messagebox.showerror("Adapter terputus", f"{item[1]}\n\nCek kabel USB CANable.")
self._disconnect()
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_bus_label(self, now):
"""slcan tidak melaporkan jenis error, jadi kondisi bus ditebak dari lalu lintas."""
if self.err_new and now - self.err_log_t >= 1.0:
self.err_log_t = now
self._log(f"Error frame dari adapter (1 detik terakhir): {self.err_new}")
self.err_new = 0
self.bus_err_t = now
self.lbl_bus.configure(text="ERROR frame", fg="#c0392b")
elif now - self.bus_err_t < 3:
return
elif self.last_frame_t and now - self.last_frame_t < SILENT_WARN_S:
self.lbl_bus.configure(text="OK", fg="#1a8a3a")
elif now - self.connect_t >= SILENT_WARN_S:
self.lbl_bus.configure(text="DIAM: tidak ada frame", fg="#e67e22")
def _update_stats(self):
now = time.time()
while self.send_errors:
self._result(f"Gagal kirim: {self.send_errors.pop()}", "#c0392b")
self._update_bus_label(now)
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))
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 list, python csp_can_tool.py --channel COM7 monitor, python csp_can_tool.py ping 2 atau python csp_can_tool.py hk.
Bonus – Satu GUI untuk Korlan USB2CAN dan CANable
Karena sekarang saya punya dua adapter, program di atas saya gabungkan dengan versi Korlan dari tutorial sebelumnya. Hasilnya satu GUI yang mendeteksi Korlan USB2CAN maupun CANable secara otomatis, dan adapter bisa diganti dari menu tanpa mengubah kode. Logika CFP/CSP, simulasi, dan tampilan tetap sama.
Demo GUI dual adapter
Demo nyata di laptop (±80 detik, dipercepat). Kotak oranye menandai bagian yang sedang didemokan, lingkaran merah adalah posisi kursor.
Urutan demo:
- Pemilihan adapter: mode otomatis, tombol Cari menampilkan Korlan USB2CAN dan CANable di daftar Perangkat.
- Pilih Korlan, klik Hubungkan: status bar hijau, OBC dan EPS menjadi aktif.
- Perintah dari laptop: Ping OBC, Ping EPS, Payload ON, Payload OFF, dan kirim pesan ke EPS.
- Tab Frame CAN: setiap frame dengan ID CFP yang sudah diuraikan.
- Tab Log: adapter yang terdeteksi, koneksi, dan hasil perintah.
- Ganti adapter: Putuskan, pilih CANable COM3, Hubungkan lagi.
- Terhubung lewat CANable, kedua node tetap aktif.
- Status node: kedua node aktif.
- USB board OBC dicabut: status OBC berubah oranye "diam … s" dan Ping OBC TIMEOUT.
- USB OBC dicolok lagi: OBC kembali aktif dan Ping OBC berhasil.
Persiapan untuk kedua adapter:
pip install python-can pyserial pywin32- Korlan: driver WinUSB 8devices +
usb2can.dll(x64) di folder program.pywin32dipakai untuk membaca serial number Korlan. - CANable: firmware slcan dan kabel USB-C data, tanpa driver atau DLL.
1. Mendeteksi adapter. Korlan dicari lewat can.detect_available_configs("usb2can"), CANable lewat VID USB 16D0. Port COM lain (misalnya STLink) tetap didaftar tapi ditandai bukan adapter yang dikenali, sehingga tidak dipilih otomatis:
def find_korlan():
"""Serial number Korlan USB2CAN yang terpasang (butuh pywin32 + driver 8devices)."""
try:
cfgs = can.detect_available_configs("usb2can")
except Exception: # noqa: BLE001 (pywin32 / driver belum terpasang)
return []
return [str(c["channel"]) for c in cfgs if c.get("channel")]
def find_canable_ports():
"""Daftar port COM; port CANable (VID 16D0) diletakkan paling depan."""
if list_ports is None:
return []
canable, others = [], []
for p in list_ports.comports():
desc = (p.description or "") + " " + (p.product or "")
if (p.vid == CANABLE_SLCAN_VID and p.pid == CANABLE_SLCAN_PID) or "canable" in desc.lower():
canable.append(p.device)
else:
others.append(p.device)
return canable + others
def find_adapters():
"""Semua adapter yang dikenali, urutan: Korlan, CANable, port COM lain.
Setiap item: {"interface", "channel", "label", "known"}.
"""
out = []
for sn in find_korlan():
out.append({"interface": "usb2can", "channel": sn, "label": f"Korlan USB2CAN S/N {sn}", "known": True})
if list_ports is not None:
for p in list_ports.comports():
desc = (p.description or "") + " " + (p.product or "")
is_canable = (p.vid == CANABLE_SLCAN_VID and p.pid == CANABLE_SLCAN_PID) or "canable" in desc.lower()
if is_canable:
out.append({"interface": "slcan", "channel": p.device, "label": f"CANable {p.device}", "known": True})
for p in list_ports.comports():
if not any(a["channel"] == p.device for a in out):
out.append({"interface": "slcan", "channel": p.device,
"label": f"{p.device} ({p.description})", "known": False})
return out2. Membuka bus. Fungsi open_bus() sekarang punya mode auto: adapter pertama yang dikenali langsung dipakai (Korlan diutamakan kalau keduanya ada). Untuk Korlan, usb2can.dll di folder program dipakai otomatis:
def open_bus(args):
if args.interface == "virtual":
return can.Bus(interface="virtual", channel=args.channel or "csp")
if args.interface == "auto":
known = [a for a in find_adapters() if a["known"]]
if args.channel:
known = [a for a in known if a["channel"] == args.channel] or known
if not known:
raise RuntimeError("Tidak ada Korlan USB2CAN atau CANable yang terdeteksi.")
args.interface, args.channel = known[0]["interface"], known[0]["channel"]
if args.interface == "usb2can":
kw = {"interface": "usb2can", "bitrate": args.bitrate}
if args.channel:
kw["channel"] = args.channel
if os.path.exists(USB2CAN_DLL):
kw["dll"] = USB2CAN_DLL
return can.Bus(**kw)3. Memilih adapter di GUI. Menu Adapter sekarang berisi otomatis, usb2can, slcan, dan simulasi. Kolom Perangkat diisi tombol Cari dengan semua adapter yang ditemukan. Nilainya juga boleh diketik: COM3 dianggap CANable, 04EF18C4 dianggap serial number Korlan.
def _selected_adapter(self):
"""(interface, channel) dari pilihan di GUI."""
mode = self.var_if.get()
text = self.var_ch.get().strip()
for a in self.adapters: # pilihan dari daftar hasil Cari
if text == a["label"]:
return (a["interface"] if mode == "otomatis" else mode), a["channel"]
if text: # diketik manual: COMx atau S/N Korlan
if mode in ("usb2can", "slcan"):
return mode, text
return ("slcan" if text.upper().startswith("COM") or text.startswith("/dev/") else "usb2can"), text
if mode in ("usb2can", "slcan"):
return mode, None
return "auto", None4. Kondisi bus. Saat memakai Korlan, kode status adapter kembali diterjemahkan (stuff error, ACK error, CRC error, BUS-OFF, dan seterusnya). Saat memakai CANable, kondisi tetap ditebak dari lalu lintas (OK atau DIAM), karena firmware slcan tidak melaporkan jenis error.
Hasil uji dua adapter
Korlan dan CANable dipasang bersamaan secara paralel di bus OBC–EPS yang sama. Hasil python csp_can_tool.py list:
usb2can 04EF18C4 Korlan USB2CAN S/N 04EF18C4
slcan COM3 CANable COM3
slcan COM6 COM6 (STMicroelectronics STLink Virtual COM Port (COM6)) (bukan adapter yang dikenali)
slcan COM7 COM7 (STMicroelectronics STLink Virtual COM Port (COM7)) (bukan adapter yang dikenali)- Dengan Adapter otomatis, daftar Perangkat berisi Korlan dan CANable sekaligus; Korlan terpilih lebih dulu.
- Korlan: status TERHUBUNG 1000 kbit/s · Korlan S/N 04EF18C4, OBC dan EPS aktif, semua paket CSP terbaca.
- CANable: setelah Putuskan lalu memilih CANable COM3, status TERHUBUNG · CANable COM3, OBC dan EPS aktif, Ping EPS 4,1 ms.
- Kedua adapter boleh tetap tersambung di bus bersamaan, karena keduanya hanya mendengar dan memberi ACK.
GUI dual adapter dengan Adapter "otomatis": daftar Perangkat berisi Korlan USB2CAN S/N 04EF18C4, CANable COM3, dan port STLink. Di sini GUI tersambung lewat Korlan, OBC dan EPS sama-sama aktif.
GUI yang sama setelah Putuskan lalu memilih CANable COM3: status berubah menjadi TERHUBUNG · CANable COM3, kedua node tetap aktif dan Ping EPS 4,1 ms.
Kode lengkap versi dua adapter (juga ada di GitHub, folder dual-adapter). Cadangan driver Korlan dan firmware CANable ada di folder drivers.
csp_can_tool.py – versi dua adapter (530 baris)
#!/usr/bin/env python3
"""
csp_can_tool.py - membaca dan menguji bus CAN CubeSat (libcsp 1.6 / CFP)
memakai Korlan USB2CAN (8devices) ATAU Makerbase CANable V2.0 / V2.0 Pro
(firmware slcan) dan python-can. Adapter dideteksi otomatis.
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 # tampilkan semua adapter
python csp_can_tool.py monitor # adapter dipilih otomatis
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"
python csp_can_tool.py --interface slcan --channel COM3 monitor
python csp_can_tool.py --interface usb2can --channel 04EF18C4 monitor
Kebutuhan:
pip install python-can pyserial pywin32
Korlan : driver 8devices + usb2can.dll (64-bit) di folder yang sama dengan script
CANable: firmware slcan (muncul sebagai port COM, tanpa driver/DLL)
"""
import argparse
import os
import queue
import struct
import threading
import time
import can
try:
from serial.tools import list_ports
except ImportError: # pyserial belum terpasang
list_ports = None
# ---------------------------------------------------------------------------
# Konstanta sistem (harus sama dengan firmware: cubesat_proto.h)
# ---------------------------------------------------------------------------
ADDR_OBC = 1
ADDR_EPS = 2
ADDR_PC = 10 # alamat CSP untuk PC/adapter (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)
# VID:PID CANable firmware slcan (USB CDC)
CANABLE_SLCAN_VID = 0x16D0
CANABLE_SLCAN_PID = 0x117E
# ---------------------------------------------------------------------------
# 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._tx_lock = threading.Lock()
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():
try:
msg = self.bus.recv(timeout=0.1)
except Exception as e: # noqa: BLE001 (mis. kabel USB dicabut)
self.inbox.put(("fatal", str(e)))
return
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)
with self._tx_lock: # serial slcan: satu pengirim pada satu waktu
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}")
# ---------------------------------------------------------------------------
# Deteksi adapter: Korlan USB2CAN (usb2can) dan CANable (slcan)
# ---------------------------------------------------------------------------
HERE = os.path.dirname(os.path.abspath(__file__))
USB2CAN_DLL = os.path.join(HERE, "usb2can.dll")
def find_korlan():
"""Serial number Korlan USB2CAN yang terpasang (butuh pywin32 + driver 8devices)."""
try:
cfgs = can.detect_available_configs("usb2can")
except Exception: # noqa: BLE001 (pywin32 / driver belum terpasang)
return []
return [str(c["channel"]) for c in cfgs if c.get("channel")]
def find_canable_ports():
"""Daftar port COM; port CANable (VID 16D0) diletakkan paling depan."""
if list_ports is None:
return []
canable, others = [], []
for p in list_ports.comports():
desc = (p.description or "") + " " + (p.product or "")
if (p.vid == CANABLE_SLCAN_VID and p.pid == CANABLE_SLCAN_PID) or "canable" in desc.lower():
canable.append(p.device)
else:
others.append(p.device)
return canable + others
def find_adapters():
"""Semua adapter yang dikenali, urutan: Korlan, CANable, port COM lain.
Setiap item: {"interface", "channel", "label", "known"}.
"""
out = []
for sn in find_korlan():
out.append({"interface": "usb2can", "channel": sn, "label": f"Korlan USB2CAN S/N {sn}", "known": True})
if list_ports is not None:
for p in list_ports.comports():
desc = (p.description or "") + " " + (p.product or "")
is_canable = (p.vid == CANABLE_SLCAN_VID and p.pid == CANABLE_SLCAN_PID) or "canable" in desc.lower()
if is_canable:
out.append({"interface": "slcan", "channel": p.device, "label": f"CANable {p.device}", "known": True})
for p in list_ports.comports():
if not any(a["channel"] == p.device for a in out):
out.append({"interface": "slcan", "channel": p.device,
"label": f"{p.device} ({p.description})", "known": False})
return out
def open_bus(args):
if args.interface == "virtual":
return can.Bus(interface="virtual", channel=args.channel or "csp")
if args.interface == "auto":
known = [a for a in find_adapters() if a["known"]]
if args.channel:
known = [a for a in known if a["channel"] == args.channel] or known
if not known:
raise RuntimeError("Tidak ada Korlan USB2CAN atau CANable yang terdeteksi.")
args.interface, args.channel = known[0]["interface"], known[0]["channel"]
if args.interface == "usb2can":
kw = {"interface": "usb2can", "bitrate": args.bitrate}
if args.channel:
kw["channel"] = args.channel
if os.path.exists(USB2CAN_DLL):
kw["dll"] = USB2CAN_DLL
return can.Bus(**kw)
if args.interface == "slcan":
port = args.channel
if not port:
ports = find_canable_ports()
if not ports:
raise RuntimeError("Port COM CANable tidak ditemukan (cek kabel USB / pyserial).")
port = ports[0]
# CANable memakai USB CDC: tty_baudrate tidak berpengaruh,
# bitrate CAN dikirim sebagai perintah slcan (S8 = 1 Mbit/s).
return can.Bus(interface="slcan", channel=port, bitrate=args.bitrate)
kw = {"interface": args.interface, "bitrate": args.bitrate}
if args.channel:
kw["channel"] = args.channel
return can.Bus(**kw)
# ---------------------------------------------------------------------------
# Perintah
# ---------------------------------------------------------------------------
def cmd_list(args):
if list_ports is None:
print("pyserial belum terpasang: pip install pyserial")
adapters = find_adapters()
if not any(a["known"] for a in adapters):
print("Tidak ada Korlan USB2CAN atau CANable yang terdeteksi.")
print(" Korlan : cek driver 8devices, usb2can.dll dan pywin32")
print(" CANable: cek kabel USB-C (harus kabel data) dan firmware slcan")
for a in adapters:
tag = "" if a["known"] else " (bukan adapter yang dikenali)"
print(f"{a['interface']:8s} {a['channel']:10s} {a['label']}{tag}")
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)")
elif kind == "fatal":
print(f"{ts} koneksi ke adapter terputus: {item[1]}")
break
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 / Makerbase CANable)")
p.add_argument("--interface", default="auto",
help="auto (default: Korlan atau CANable yang terdeteksi), usb2can, slcan, virtual")
p.add_argument("--channel", default=None,
help="serial number Korlan (mis. 04EF18C4) atau port COM CANable (mis. COM3); 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="tampilkan adapter Korlan / CANable").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 – versi dua adapter (682 baris)
#!/usr/bin/env python3
"""
csp_can_gui.py - GUI pemantau bus CAN CubeSat (libcsp 1.6 / CFP)
untuk Korlan USB2CAN (8devices) dan Makerbase CANable V2.0 / V2.0 Pro (slcan).
Jalankan: python csp_can_gui.py
Butuh : pip install python-can pyserial pywin32
(+ usb2can.dll 64-bit di folder ini bila memakai Korlan)
Fitur:
- Deteksi otomatis Korlan USB2CAN dan CANable (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 / Makerbase CANable"
MAX_ROWS = 2000
SILENT_WARN_S = 3.0 # bus dianggap diam bila tidak ada frame selama ini
ADAPTER_MODES = ["otomatis", "usb2can", "slcan", "simulasi"]
ADAPTER_NAMES = {"usb2can": "Korlan", "slcan": "CANable"}
# 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, dst, 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 = {} # kode error -> jumlah (1 detik terakhir)
self.last_err = None
self.cur_if = None # "usb2can" / "slcan" / "virtual" saat terhubung
self.adapters = [] # hasil T.find_adapters()
self.err_log_t = 0.0
self.bus_err_t = 0.0
self.last_frame_t = 0.0
self.connect_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_adapters(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="otomatis")
cb_if = ttk.Combobox(top, textvariable=self.var_if, width=9, state="readonly", values=ADAPTER_MODES)
cb_if.pack(side="left", padx=(4, 12))
cb_if.bind("<<ComboboxSelected>>", lambda _: self._scan_adapters(quiet=True))
ttk.Label(top, text="Perangkat:").pack(side="left")
self.var_ch = tk.StringVar()
self.cb_ch = ttk.Combobox(top, textvariable=self.var_ch, width=26)
self.cb_ch.pack(side="left", padx=(4, 2))
ttk.Button(top, text="Cari", width=5, command=self._scan_adapters).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. Adapter Korlan USB2CAN atau CANable dicari otomatis; 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 _selected_adapter(self):
"""(interface, channel) dari pilihan di GUI."""
mode = self.var_if.get()
text = self.var_ch.get().strip()
for a in self.adapters: # pilihan dari daftar hasil Cari
if text == a["label"]:
return (a["interface"] if mode == "otomatis" else mode), a["channel"]
if text: # diketik manual: COMx atau S/N Korlan
if mode in ("usb2can", "slcan"):
return mode, text
return ("slcan" if text.upper().startswith("COM") or text.startswith("/dev/") else "usb2can"), text
if mode in ("usb2can", "slcan"):
return mode, None
return "auto", None
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")
self.cur_if, desc = "virtual", "simulasi"
else:
if not self.var_ch.get().strip():
self._scan_adapters(quiet=True)
class A: # argumen untuk open_bus()
pass
a = A()
a.interface, a.channel = self._selected_adapter()
a.bitrate = br
self.bus = T.open_bus(a) # mode auto mengisi a.interface/a.channel
self.cur_if = a.interface
name = ADAPTER_NAMES.get(a.interface, a.interface)
ch = a.channel or "?"
desc = f"{name} S/N {ch}" if a.interface == "usb2can" else f"{name} {ch}"
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:\n"
"- Korlan: driver 8devices, usb2can.dll (64-bit) di folder script, pywin32, "
"serial number benar.\n"
"- CANable: kabel USB-C data, firmware slcan (muncul sebagai port COM), "
"port tidak dipakai program lain.")
return
self.connect_t = time.time()
self.last_frame_t = 0.0
self.err_counts.clear()
self.btn_conn.configure(text="Putuskan")
self.lbl_status.configure(text=f" TERHUBUNG {int(self.var_br.get()) // 1000} kbit/s · {desc} ",
bg="#1a8a3a")
self.title(f"{APP_TITLE} [{desc}]")
self._log(f"Terhubung ({desc}), alamat CSP PC = {T.ADDR_PC}")
def _scan_adapters(self, quiet=False):
"""Cari Korlan USB2CAN dan CANable, isi daftar Perangkat."""
mode = self.var_if.get()
if mode == "simulasi":
self.cb_ch.configure(values=[])
self.var_ch.set("")
return []
self.adapters = T.find_adapters()
shown = [a for a in self.adapters if mode == "otomatis" or a["interface"] == mode]
labels = [a["label"] for a in shown]
self.cb_ch.configure(values=labels)
known = [a for a in shown if a["known"]]
if self.var_ch.get().strip() not in labels:
self.var_ch.set(known[0]["label"] if known else (labels[0] if labels else ""))
if known:
self._log("Adapter terdeteksi: " + "; ".join(a["label"] for a in known))
elif not quiet:
self._log("Korlan USB2CAN / CANable tidak terdeteksi. Korlan: cek driver 8devices + pywin32. "
"CANable: cek kabel USB-C data dan firmware slcan.")
if T.list_ports is None and not quiet:
self._log("pyserial belum terpasang: pip install pyserial")
return known
def _disconnect(self):
try:
if self.node:
self.node.close()
if self.bus:
self.bus.shutdown()
if self.sim:
self.sim.close()
except Exception: # noqa: BLE001
pass
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.title(APP_TITLE)
self.cur_if = None
self.lbl_bus.configure(text="–", fg="black")
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 and self.node:
try:
item = self.node.inbox.get_nowait()
except queue.Empty:
break
n += 1
self._handle_item(item)
if self.node:
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
self.last_frame_t = time.time()
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":
# Korlan mengirim kode status di byte data pertama; slcan umumnya tidak mengirim error frame
data = bytes(item[1].data)
code = data[0] if (data and self.cur_if == "usb2can") else -1
self.err_counts[code] = self.err_counts.get(code, 0) + 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)")
elif kind == "fatal":
self._log(f"Koneksi ke adapter terputus: {item[1]}")
messagebox.showerror("Adapter terputus", f"{item[1]}\n\nCek kabel USB CANable.")
self._disconnect()
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_bus_label(self, now):
"""Korlan melaporkan jenis error; slcan tidak, jadi kondisinya ditebak dari lalu lintas."""
if self.err_counts and now - self.err_log_t >= 1.0:
self.err_log_t = now
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.err_counts.clear()
self.bus_err_t = now
self.lbl_bus.configure(text="ERROR: " + USB2CAN_STATUS.get(self.last_err, hex(self.last_err)),
fg="#c0392b")
elif now - self.bus_err_t < 3:
return
elif self.last_frame_t and now - self.last_frame_t < SILENT_WARN_S:
self.lbl_bus.configure(text="OK", fg="#1a8a3a")
elif now - self.connect_t >= SILENT_WARN_S:
self.lbl_bus.configure(text="DIAM: tidak ada frame", fg="#e67e22")
def _update_stats(self):
now = time.time()
while self.send_errors:
self._result(f"Gagal kirim: {self.send_errors.pop()}", "#c0392b")
self._update_bus_label(now)
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))
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()Penutup
Dengan Makerbase CANable V2.0 Pro, alat sniffing CAN bus CubeSat kita jadi lebih sederhana: tanpa driver, tanpa DLL, cukup port COM dan python-can. Bonusnya, adapter sudah terisolasi dan punya terminator bawaan (yang harus dimatikan saat menyadap di tengah bus). Kekurangannya, firmware slcan tidak melaporkan jenis error bus, jadi untuk diagnosis fisik yang detail Korlan masih lebih unggul. Kalau punya Korlan dan CANable sekaligus, versi dua adapter di bagian Bonus bisa memakai keduanya dari satu GUI. Semoga bermanfaat.
Wassalamu'alaikum warohmatullohi wabarakatuh.
No comments:
Post a Comment