Assalamu'alaikum warohmatullohi wabarakatuh,
Pada tutorial sebelumnya, Komunikasi CubeSat Space Protocol (libcsp) lewat CAN Bus FDCAN1 pada STM32H743 dengan FreeRTOS, kita membuat dua board STM32H743 berperan sebagai OBC (On-Board Computer) dan EPS (Electrical Power System) yang saling bertukar paket CSP lewat CAN bus. Masalahnya, kalau komunikasi gagal, kita hanya bisa menebak-nebak dari log UART masing-masing board.
Kali ini kita membuat "telinga" ketiga di bus: Korlan USB2CAN dari 8devices yang dicolokkan ke laptop, ditambah program Python dengan GUI untuk sniffing (menyadap) bus CAN. Program ini tidak hanya menampilkan frame CAN mentah, tetapi juga menyusun ulang frame menjadi paket CSP, menampilkan data housekeeping EPS, dan bisa mengirim perintah (ping, payload ON/OFF) langsung dari laptop.
Apa itu sniffing CAN bus?
Sniffing berarti ikut mendengarkan semua lalu lintas di bus tanpa mengganggu node lain. Di CAN, semua node berbagi dua kabel (CANH dan CANL), jadi adapter yang tersambung ke bus otomatis "mendengar" setiap frame. Dengan sniffer kita bisa:
- memastikan sebuah node benar-benar mengirim (misalnya OBC sudah mengirim ping atau belum),
- melihat isi data sebenarnya di kabel, bukan hanya yang dilaporkan firmware,
- mendeteksi masalah fisik: terminator, kabel tertukar, bitrate salah, dan
- menguji node satu per satu dengan mengirim perintah dari laptop.
Perangkat yang dibutuhkan
| Perangkat | Keterangan |
|---|---|
| Korlan USB2CAN (8devices) | adapter USB ke CAN, konektor DB9 |
| 2 board STM32H743 (DevEBox) + modul MCP2551 | OBC dan EPS dari tutorial sebelumnya |
| 2 resistor 120 Ω | terminator di kedua ujung bus |
| Konektor DB9 female / adaptor terminal sekrup | untuk menyambung kabel ke Korlan |
| Laptop Windows + Python 3 (64-bit) | python-can, pywin32, Tkinter |
Korlan USB2CAN dan pinout konektor DB9: pin 7 = CAN_H, pin 2 = CAN_L, pin 3 = GND.
Langkah 1 – Instal driver dan library
Semua file resmi ada di halaman produk 8devices Korlan USB2CAN. Gulir ke bawah tabel Specifications sampai bagian Resources (file disimpan di Google Drive). Unduh tiga file berikut:
- Windows Driver (WinUSB) → jalankan
usb2can_winusb.msi(butuh izin Administrator). Setelah selesai, Korlan muncul sebagai USB2CAN di Device Manager tanpa tanda segitiga kuning. - USB2CAN CANAL DLL v2.0.0 → di dalam zip ada
x64/Release/usb2can.dlldanx86/Release/usb2can.dll. - Windows python examples → contoh resmi 8devices (berguna untuk perbandingan).
Salin usb2can.dll versi x64 ke folder program kita. Versi DLL harus sama dengan Python: Python 64-bit memakai DLL x64. Lalu pasang library Python:
pip install python-can pywin32pywin32 dipakai python-can untuk mendeteksi serial number Korlan secara otomatis lewat WMI.
Langkah 2 – Menyambung Korlan ke bus CAN
Korlan dipasang paralel (sebagai cabang) di tengah bus, seperti node ketiga. Bus tetap memakai dua resistor 120 Ω di kedua ujung (di modul OBC dan modul EPS). Korlan sendiri tidak perlu resistor.
Wiring: CANH–CANH, CANL–CANL, GND bersama, terminator 120 Ω di kedua ujung, Korlan menyadap di tengah.
| DB9 Korlan | Bus CAN |
|---|---|
| pin 7 | CANH |
| pin 2 | CANL |
| pin 3 | GND board |
Cek terminator dengan multimeter saat semua board mati: hambatan CANH–CANL seharusnya sekitar 60 Ω (dua resistor 120 Ω paralel).
Langkah 3 – Tes koneksi dengan Python
Sebelum membuat GUI, pastikan Python bisa membuka Korlan. Skrip pendek ini mencari adapter lalu menampilkan frame selama 5 detik:
import can
# 1) cari Korlan yang terpasang
print(can.detect_available_configs("usb2can"))
# contoh hasil: [{'interface': 'usb2can', 'channel': '04EF18C4'}]
# 2) buka bus 1 Mbit/s dan tampilkan frame selama 5 detik
bus = can.Bus(interface="usb2can", channel="04EF18C4",
bitrate=1000000, dll="./usb2can.dll")
import time
t0 = time.time()
while time.time() - t0 < 5:
msg = bus.recv(timeout=0.5)
if msg:
print(msg)
bus.shutdown()Serial number (contoh 04EF18C4) adalah 8 karakter yang juga tertera di stiker Korlan. Kecepatan bitrate harus sama dengan firmware, di proyek ini 1 Mbit/s.
Langkah 4 – Membaca ID CAN sebagai header CFP
libcsp mengirim paket CSP lewat CAN memakai CFP (CAN Fragmentation Protocol). Semua informasi alamat ada di ID extended 29-bit:
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
Alur data dari Korlan sampai GUI.
Kunci agar GUI tidak macet adalah memisahkan pembacaan bus dari tampilan. Thread penerima membaca bus.recv() terus-menerus dan memasukkan hasilnya ke antrean:
def _rx_loop(self):
while not self._stop.is_set():
msg = self.bus.recv(timeout=0.1)
if msg is None:
continue
if msg.is_error_frame:
self.stats["errors"] += 1
self.inbox.put(("error", msg))
continue
if not msg.is_extended_id:
self.inbox.put(("std", msg))
continue
self.stats["frames"] += 1
pkt, err = self.rx.push(msg.arbitration_id, msg.data)
self.inbox.put(("frame", msg, err))
if pkt:
pkt["t"] = time.time()
self.stats["packets"] += 1
self.inbox.put(("packet", pkt))GUI (Tkinter) mengambil isi antrean setiap 40 ms dengan after(), paling banyak 500 item sekali ambil, sehingga jendela tetap responsif meskipun bus sangat ramai:
def _poll(self):
if self.node:
n = 0
while n < 500:
try:
item = self.node.inbox.get_nowait()
except queue.Empty:
break
n += 1
self._handle_item(item)
self._check_timeouts()
self._update_stats()
self.after(40, self._poll)Koneksi ke Korlan cukup satu baris can.Bus(interface="usb2can", ...). Fungsi berikut otomatis memakai usb2can.dll yang ada di folder program:
def open_bus(args):
if args.interface == "virtual":
return can.Bus(interface="virtual", channel=args.channel or "csp")
kw = {"interface": args.interface, "bitrate": args.bitrate}
if args.channel:
kw["channel"] = args.channel
if args.interface == "usb2can":
here = os.path.dirname(os.path.abspath(__file__))
dll = os.path.join(here, "usb2can.dll")
if os.path.exists(dll):
kw["dll"] = dll
return can.Bus(**kw)Serial number Korlan dicari otomatis saat GUI dibuka (tombol Cari untuk mencari ulang):
def _scan_serial(self, quiet=False):
"""Cari Korlan USB2CAN yang terpasang dan isi kolom Serial."""
try:
cfgs = can.detect_available_configs("usb2can")
except Exception as e: # noqa: BLE001
cfgs = []
if not quiet:
self._log(f"Gagal mencari adapter: {e}")
serials = [str(c.get("channel")) for c in cfgs if c.get("channel")]
self.cb_ch.configure(values=serials)
if serials:
if self.var_ch.get().strip() not in serials:
self.var_ch.set(serials[0])
self._log("Korlan USB2CAN terdeteksi: S/N " + ", ".join(serials))
elif not quiet:
self._log("Korlan USB2CAN tidak terdeteksi (cek kabel USB / driver 8devices / pywin32).")
return serialsLangkah 7 – Menjalankan GUI
Simpan csp_can_tool.py, csp_can_gui.py dan usb2can.dll dalam satu folder, lalu jalankan:
python csp_can_gui.pyPilih Adapter: usb2can untuk Korlan sungguhan, atau simulasi untuk mencoba tampilan tanpa hardware (program membuat OBC dan EPS tiruan yang perilakunya sama dengan firmware). Klik Hubungkan.
Tab Paket CSP saat terhubung ke Korlan asli (S/N 04EF18C4): biru = OBC, merah = EPS. Panel kiri menampilkan housekeeping EPS (suhu chip 32,8 °C, VDDA 3338 mV, Payload ON, uptime) dan hasil Ping EPS 1,2 ms.
Yang terlihat di tab Paket CSP:
- OBC:30 → EPS:1 PING lalu EPS:1 → OBC:30 re:PING: OBC mem-ping EPS dan EPS membalas.
- OBC → EPS:12 EPS_HK dan balasan 20 byte berisi uptime, suhu chip, VDDA dan status payload, yang langsung diterjemahkan.
- OBC → EPS:11 TLM: paket telemetri 200 byte (26 frame CAN) dengan checksum.
- Baris dari/ke PC: perintah yang kita kirim dari tombol Ping, Payload ON/OFF, dan Kirim pesan.
Tab Frame CAN saat terhubung ke Korlan asli (S/N 04EF18C4): setiap frame dengan ID yang sudah diuraikan (Src, Dst, BEGIN/MORE, Remain, CFP id). Terlihat paket telemetri 200 byte dari OBC (1) ke EPS (2) dengan CFP id 952: BEGIN remain=25 lalu MORE 24…0, serta frame balasan EPS (src 2) ke OBC (dst 1) dan ke PC (dst 10).
Hasil uji nyata: OBC dan EPS terbaca bersih
Setelah dua resistor 120 Ω terpasang di ujung bus (modul OBC dan modul EPS), semua stuff error hilang. Korlan langsung menerima frame valid dan GUI menampilkan seluruh lalu lintas CubeSat:
Hasil nyata dengan Korlan asli: kedua node aktif, Kondisi bus OK, error frame 0. Tombol Payload ON dari laptop dijawab EPS "status=OK rail=ON" dalam 10 ms, panel housekeeping menampilkan Payload ON, suhu chip 33,2 °C dan VDDA 3336 mV.
Yang terlihat pada pengujian dengan hardware asli (Korlan S/N 04EF18C4, 1 Mbit/s):
- Kondisi bus OK, error frame 0 dari 1.133 frame CAN (±48 frame/detik), status node OBC dan EPS sama-sama aktif.
- OBC → EPS PING dan EPS → OBC re:PING: OBC mem-ping EPS setiap 2 detik dan EPS membalas.
- OBC → EPS:10 ECHO berisi "Halo EPS, pesan ke-127", dibalas EPS "ACK dari node 2 (EPS)".
- OBC → EPS:12 EPS_HK dan balasan 20 byte berisi uptime, status payload, serta jumlah frame CAN yang diterima/dikirim EPS.
- OBC → EPS:11 TLM: telemetri 200 byte (26 frame) dengan checksum, dan EPS_CMD setiap 10 detik untuk membalik status payload.
Perintah langsung dari laptop lewat Korlan juga berhasil:
| Tombol di GUI | Hasil |
|---|---|
| Ping EPS | 2,0 ms |
| Ping OBC | OBC membalas (OBC:1 → PC re:PING) |
| Payload ON | status OK dalam 9,7 ms, LED PA1 di board EPS menyala |
| Kirim pesan ke EPS | balasan "ACK dari node 2 (EPS)" dalam 6,8 ms |
Update – suhu dan VDDA EPS sudah terbaca. Pada pengujian pertama nilai suhu chip dan VDDA EPS masih 0. Penyebabnya ada di firmware EPS: kalibrasi linearitas ADC3 STM32H743 butuh waktu lebih lama dari batas tunggu 10 ms di kode awal, sehingga inisialisasi ADC dianggap gagal. Setelah batas tunggu kalibrasi dinaikkan menjadi 200 ms, EPS mengirim nilai sensor yang benar:
Setelah perbaikan firmware: panel EPS housekeeping menampilkan suhu chip 33,1 °C dan VDDA 3337 mV.
- Suhu chip STM32H743 (EPS): ±32,7–33,2 °C, dari sensor suhu internal ADC3 channel 18 dengan kalibrasi pabrik TS_CAL1/TS_CAL2.
- VDDA: ±3337 mV, dihitung dari VREFINT (ADC3 channel 19) dan VREFINT_CAL.
- Ping EPS dari laptop: 2,3 ms; error frame tetap 0 dan Kondisi bus OK.
Log serial EPS. Dari sisi EPS, log UART (USART1, PA9/PA10, 115200 baud) dibaca lewat Mini Debugger di COM6. Log ini memastikan apa yang terlihat di Korlan memang diterima dan diproses oleh EPS:
Log serial EPS (COM6) di console STM32CubeIDE: permintaan housekeeping dari OBC dan laptop, pesan dan telemetri dari OBC, perintah payload, serta status CAN TEC=0 REC=0.
- [EPS-HK] permintaan housekeeping dari node 10 (laptop lewat Korlan) dan node 1 (OBC), masing-masing tiap 2 detik, lengkap dengan suhu chip ±31 °C dan VDDA ±3338 mV.
- [SERVER] pesan "Halo EPS, pesan ke-…" di port 10 dan paket telemetri 200 byte di port 11 beserta checksum.
- [EPS-CMD] perintah payload dari OBC: ON (LED PA1 menyala) lalu OFF (LED PA1 mati) setiap 10 detik.
- [STAT] kesehatan CAN di EPS:
TEC=0 REC=0 OK, 0 kiriman gagal, buffer CSP bebas 19–20 dari 20.
Troubleshooting: membaca status error dari Korlan
Kalau sinyal di bus rusak, Korlan tidak menerima frame data, melainkan pesan status/error. GUI meringkasnya per detik di tab Log dan menampilkannya di baris Kondisi bus. Berikut contoh nyata saat bus OBC–EPS saya belum dipasangi terminator:
Koneksi nyata ke Korlan (S/N 04EF18C4): Frame CAN 0, ribuan error frame, Kondisi bus "ERROR: stuff error".
Arti kode status yang dikirim adapter (byte data pertama):
# Kode status/error yang dikirim adapter 8devices USB2CAN (byte data pertama)
USB2CAN_STATUS = {
0x00: "OK", 0x01: "overrun", 0x02: "bus warning", 0x03: "error passive",
0x04: "BUS-OFF", 0x20: "stuff error", 0x21: "form error", 0x23: "ACK error",
0x24: "bit0 error", 0x25: "bit1 error", 0x27: "CRC error", -1: "error",
}| Gejala | Kemungkinan penyebab | Solusi |
|---|---|---|
| Frame CAN = 0, tidak ada error | Korlan belum tersambung ke bus / node tidak mengirim | cek kabel DB9 (pin 7, 2, 3) dan pastikan board menyala |
| Banyak stuff error, bus warning, error passive | tanpa terminator 120 Ω, CANH/CANL tertukar, atau bitrate berbeda | pasang 2×120 Ω (ukur ±60 Ω), cek CANH–CANH, samakan bitrate |
| Hanya ACK error | tidak ada node lain yang membalas | nyalakan node lain; Korlan sendiri juga memberi ACK |
| 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 |
| "DLL failed to load" | usb2can.dll tidak ada / beda arsitektur | salin DLL x64 ke folder program, Python 64-bit |
Catatan: Korlan ikut memberi ACK di bus. Jadi kalau hanya satu board yang terpasang bersama Korlan, frame board itu tetap terbaca dengan benar. Ini cara cepat menguji OBC dan EPS satu per satu.
Kode lengkap
Dua file berikut disimpan dalam satu folder bersama usb2can.dll. Klik judul untuk membuka kodenya.
csp_can_tool.py – library CFP/CSP + versi command line (436 baris)
#!/usr/bin/env python3
"""
csp_can_tool.py - membaca dan menguji bus CAN CubeSat (libcsp 1.6 / CFP)
memakai Korlan USB2CAN (8devices) dan python-can.
Bisa dipakai untuk:
monitor : menampilkan semua frame CAN + paket CSP yang sudah disusun ulang
ping : ping node CSP (OBC = 1, EPS = 2)
hk : minta data housekeeping EPS (port 12)
payload : perintah ON/OFF rail payload EPS (port 13)
echo : kirim pesan teks ke port 10 dan tampilkan balasannya
Contoh (Windows):
python csp_can_tool.py list
python csp_can_tool.py monitor
python csp_can_tool.py ping 2
python csp_can_tool.py hk
python csp_can_tool.py payload on
python csp_can_tool.py echo 1 "Halo OBC"
Kebutuhan:
pip install python-can pywin32
usb2can.dll (64-bit) dari driver 8devices diletakkan di folder yang sama
dengan script ini, atau di folder yang ada di PATH.
"""
import argparse
import os
import queue
import struct
import sys
import threading
import time
import can
# ---------------------------------------------------------------------------
# Konstanta sistem (harus sama dengan firmware: cubesat_proto.h)
# ---------------------------------------------------------------------------
ADDR_OBC = 1
ADDR_EPS = 2
ADDR_PC = 10 # alamat CSP untuk PC/Korlan (bebas, asal unik 0..30)
PORT_PING = 1
PORT_ECHO = 10
PORT_TLM = 11
PORT_EPS_HK = 12
PORT_EPS_CMD = 13
NODE_NAMES = {ADDR_OBC: "OBC", ADDR_EPS: "EPS", ADDR_PC: "PC", 31: "BCAST"}
PORT_NAMES = {0: "CMP", 1: "PING", 2: "PS", 3: "MEMFREE", 4: "REBOOT",
5: "BUF_FREE", 6: "UPTIME", PORT_ECHO: "ECHO",
PORT_TLM: "TLM", PORT_EPS_HK: "EPS_HK", PORT_EPS_CMD: "EPS_CMD"}
# eps_hk_t (20 byte, little endian, packed)
EPS_HK_FMT = "<IhHBBHII"
EPS_HK_SIZE = struct.calcsize(EPS_HK_FMT)
# ---------------------------------------------------------------------------
# CFP: CAN Fragmentation Protocol (ID extended 29 bit)
# [SRC 5][DST 5][TYPE 1][REMAIN 8][ID 10]
# ---------------------------------------------------------------------------
CFP_BEGIN, CFP_MORE = 0, 1
CFP_OVERHEAD = 6 # 4 byte header CSP + 2 byte panjang
def cfp_fields(can_id):
return {
"src": (can_id >> 24) & 0x1F,
"dst": (can_id >> 19) & 0x1F,
"type": (can_id >> 18) & 0x01,
"remain": (can_id >> 10) & 0xFF,
"id": can_id & 0x3FF,
}
def cfp_make_id(src, dst, typ, remain, ident):
return ((src & 0x1F) << 24) | ((dst & 0x1F) << 19) | ((typ & 1) << 18) \
| ((remain & 0xFF) << 10) | (ident & 0x3FF)
# ---------------------------------------------------------------------------
# Header CSP 1.6 (32 bit, dikirim big endian)
# [PRIO 2][SRC 5][DST 5][DPORT 6][SPORT 6][FLAGS 8]
# ---------------------------------------------------------------------------
def csp_header_unpack(raw4):
h = struct.unpack(">I", raw4)[0]
return {
"prio": (h >> 30) & 0x3,
"src": (h >> 25) & 0x1F,
"dst": (h >> 20) & 0x1F,
"dport": (h >> 14) & 0x3F,
"sport": (h >> 8) & 0x3F,
"flags": h & 0xFF,
}
def csp_header_pack(prio, src, dst, dport, sport, flags=0):
h = ((prio & 3) << 30) | ((src & 0x1F) << 25) | ((dst & 0x1F) << 20) \
| ((dport & 0x3F) << 14) | ((sport & 0x3F) << 8) | (flags & 0xFF)
return struct.pack(">I", h)
def cfp_fragment(src, dst, payload, header4, ident):
"""Pecah satu paket CSP menjadi daftar (can_id, data) sesuai csp_can_tx()."""
length = len(payload)
frames = []
first = header4 + struct.pack(">H", length) + payload[:2]
remain = (length + CFP_OVERHEAD - 1) // 8
frames.append((cfp_make_id(src, dst, CFP_BEGIN, remain, ident), first))
sent = min(length, 2)
while sent < length:
chunk = payload[sent:sent + 8]
remain = (length - sent - len(chunk) + 7) // 8
frames.append((cfp_make_id(src, dst, CFP_MORE, remain, ident), chunk))
sent += len(chunk)
return frames
class CfpReassembler:
"""Menyusun ulang frame CFP menjadi paket CSP (meniru csp_can_rx())."""
def __init__(self, timeout=1.0):
self.buffers = {}
self.timeout = timeout
def push(self, can_id, data):
f = cfp_fields(can_id)
key = (f["src"], f["dst"], f["id"])
now = time.time()
for k in [k for k, b in self.buffers.items() if now - b["t"] > self.timeout]:
del self.buffers[k]
if f["type"] == CFP_BEGIN:
if len(data) < CFP_OVERHEAD:
return None, "BEGIN terlalu pendek"
hdr = csp_header_unpack(bytes(data[:4]))
length = struct.unpack(">H", bytes(data[4:6]))[0]
buf = {"hdr": hdr, "len": length, "data": bytearray(data[6:]),
"remain": f["remain"], "t": now}
self.buffers[key] = buf
else:
buf = self.buffers.get(key)
if buf is None:
return None, "frame MORE tanpa BEGIN"
if f["remain"] != buf["remain"] - 1:
del self.buffers[key]
return None, "frame hilang (remain tidak urut)"
buf["remain"] = f["remain"]
buf["data"] += bytes(data)
if len(buf["data"]) >= buf["len"]:
del self.buffers[key]
return {"hdr": buf["hdr"], "data": bytes(buf["data"][:buf["len"]])}, None
return None, None
# ---------------------------------------------------------------------------
# Node CSP sederhana di PC
# ---------------------------------------------------------------------------
class CspNode:
def __init__(self, bus, addr=ADDR_PC, verbose=False):
self.bus = bus
self.addr = addr
self.verbose = verbose
self.rx = CfpReassembler()
self.ident = 0
self.sport = 20
self.inbox = queue.Queue()
self.stats = {"frames": 0, "packets": 0, "errors": 0}
self._stop = threading.Event()
self._thread = threading.Thread(target=self._rx_loop, daemon=True)
self._thread.start()
def close(self):
self._stop.set()
self._thread.join(timeout=1)
def _rx_loop(self):
while not self._stop.is_set():
msg = self.bus.recv(timeout=0.1)
if msg is None:
continue
if msg.is_error_frame:
self.stats["errors"] += 1
self.inbox.put(("error", msg))
continue
if not msg.is_extended_id:
self.inbox.put(("std", msg))
continue
self.stats["frames"] += 1
pkt, err = self.rx.push(msg.arbitration_id, msg.data)
self.inbox.put(("frame", msg, err))
if pkt:
pkt["t"] = time.time()
self.stats["packets"] += 1
self.inbox.put(("packet", pkt))
def alloc_sport(self):
self.sport = 17 + (self.sport - 16) % 40 # port sementara 17..56
return self.sport
def send(self, dst, dport, payload, sport=None, prio=2):
if sport is None:
sport = self.alloc_sport()
hdr = csp_header_pack(prio, self.addr, dst, dport, sport)
self.ident = (self.ident + 1) & 0x3FF
for can_id, data in cfp_fragment(self.addr, dst, payload, hdr, self.ident):
self.bus.send(can.Message(arbitration_id=can_id, data=data,
is_extended_id=True), timeout=0.2)
return sport
def wait_reply(self, src, sport, timeout):
"""Tunggu paket dari node 'src' yang ditujukan ke port 'sport' kita."""
end = time.time() + timeout
while time.time() < end:
try:
item = self.inbox.get(timeout=max(0.0, end - time.time()))
except queue.Empty:
break
if item[0] == "packet":
h = item[1]["hdr"]
if h["src"] == src and h["dst"] == self.addr and h["dport"] == sport:
return item[1]
return None
def transaction(self, dst, dport, payload, timeout=1.0):
t0 = time.time()
sport = self.send(dst, dport, payload)
reply = self.wait_reply(dst, sport, timeout)
return reply, (time.time() - t0) * 1000.0
# ---------------------------------------------------------------------------
# Tampilan
# ---------------------------------------------------------------------------
def node_name(a):
return f"{NODE_NAMES.get(a, '?')}({a})"
def describe_packet(pkt):
h, d = pkt["hdr"], pkt["data"]
port = PORT_NAMES.get(h["dport"]) or (PORT_NAMES.get(h["sport"], str(h["dport"])) + " balasan")
line = (f"CSP {node_name(h['src'])}:{h['sport']} -> {node_name(h['dst'])}:{h['dport']} "
f"[{port}] prio={h['prio']} len={len(d)}")
extra = ""
if h["dport"] == PORT_PING or h["sport"] == PORT_PING:
extra = f"ping {len(d)} byte"
elif h["dport"] == PORT_ECHO or h["sport"] == PORT_ECHO:
extra = f'teks="{d.decode(errors="replace")}"'
elif h["dport"] == PORT_TLM:
extra = f"checksum={sum(d)}"
elif h["src"] == ADDR_EPS and h["sport"] == PORT_EPS_HK and len(d) == EPS_HK_SIZE:
extra = format_hk(d)
elif h["dport"] == PORT_EPS_CMD and len(d) == 2:
extra = f"perintah rail={d[0]} -> {'ON' if d[1] else 'OFF'}"
elif h["sport"] == PORT_EPS_CMD and len(d) == 2:
extra = f"status={'OK' if d[0] == 0 else 'DITOLAK'} rail={'ON' if d[1] else 'OFF'}"
elif d:
extra = "data=" + d[:16].hex(" ") + (" ..." if len(d) > 16 else "")
return line + (" " + extra if extra else "")
def format_hk(d):
up, t10, vdda, rail, _, cnt, rxf, txf = struct.unpack(EPS_HK_FMT, d)
return (f"uptime={up}s suhu={t10 / 10:.1f}C VDDA={vdda}mV payload={'ON' if rail else 'OFF'} "
f"HK#{cnt} CAN rx={rxf} tx={txf}")
# ---------------------------------------------------------------------------
# Koneksi ke Korlan USB2CAN
# ---------------------------------------------------------------------------
def open_bus(args):
if args.interface == "virtual":
return can.Bus(interface="virtual", channel=args.channel or "csp")
kw = {"interface": args.interface, "bitrate": args.bitrate}
if args.channel:
kw["channel"] = args.channel
if args.interface == "usb2can":
here = os.path.dirname(os.path.abspath(__file__))
dll = os.path.join(here, "usb2can.dll")
if os.path.exists(dll):
kw["dll"] = dll
return can.Bus(**kw)
# ---------------------------------------------------------------------------
# Perintah
# ---------------------------------------------------------------------------
def cmd_list(args):
cfgs = can.detect_available_configs(args.interface)
if not cfgs:
print("Tidak ada adapter terdeteksi. Cek driver 8devices, usb2can.dll dan pywin32.")
for c in cfgs:
print(c)
def cmd_monitor(args):
bus = open_bus(args)
node = CspNode(bus)
print(f"Monitor CAN {args.bitrate} bit/s. Tekan Ctrl+C untuk berhenti.\n")
t_last = time.time()
try:
while True:
try:
item = node.inbox.get(timeout=1.0)
except queue.Empty:
item = None
ts = time.strftime("%H:%M:%S")
if item:
kind = item[0]
if kind == "frame" and args.frames:
msg, err = item[1], item[2]
f = cfp_fields(msg.arbitration_id)
print(f"{ts} frame 0x{msg.arbitration_id:08X} "
f"src={f['src']} dst={f['dst']} {'BEGIN' if f['type'] == 0 else 'MORE '} "
f"remain={f['remain']:3d} id={f['id']:4d} | {msg.data.hex(' ')}"
+ (f" !! {err}" if err else ""))
elif kind == "packet":
print(f"{ts} {describe_packet(item[1])}")
elif kind == "error":
print(f"{ts} ERROR FRAME {item[1]}")
elif kind == "std":
m = item[1]
print(f"{ts} frame standar 0x{m.arbitration_id:03X} {m.data.hex(' ')} (bukan CFP)")
if time.time() - t_last >= 10:
t_last = time.time()
s = node.stats
print(f"--- statistik: frame={s['frames']} paket CSP={s['packets']} error={s['errors']}")
except KeyboardInterrupt:
pass
finally:
node.close()
bus.shutdown()
def cmd_ping(args):
bus = open_bus(args)
node = CspNode(bus)
try:
for i in range(args.count):
payload = bytes(range(args.size))
reply, ms = node.transaction(args.addr, PORT_PING, payload, args.timeout)
if reply and reply["data"] == payload:
print(f"ping {node_name(args.addr)} #{i}: {ms:.1f} ms ({args.size} byte)")
elif reply:
print(f"ping {node_name(args.addr)} #{i}: balasan berbeda ({len(reply['data'])} byte)")
else:
print(f"ping {node_name(args.addr)} #{i}: TIMEOUT")
time.sleep(0.5)
finally:
node.close()
bus.shutdown()
def cmd_hk(args):
bus = open_bus(args)
node = CspNode(bus)
try:
reply, ms = node.transaction(ADDR_EPS, PORT_EPS_HK, b"\x00", args.timeout)
if reply and len(reply["data"]) == EPS_HK_SIZE:
print(f"EPS housekeeping ({ms:.1f} ms): {format_hk(reply['data'])}")
else:
print("Housekeeping EPS: TIMEOUT / format salah")
finally:
node.close()
bus.shutdown()
def cmd_payload(args):
bus = open_bus(args)
node = CspNode(bus)
try:
on = 1 if args.state == "on" else 0
reply, ms = node.transaction(ADDR_EPS, PORT_EPS_CMD, bytes([0, on]), args.timeout)
if reply and len(reply["data"]) == 2:
st, rail = reply["data"]
print(f"Perintah payload {args.state.upper()}: {'OK' if st == 0 else 'DITOLAK'}, "
f"rail sekarang {'ON' if rail else 'OFF'} ({ms:.1f} ms)")
else:
print("Perintah payload: TIMEOUT")
finally:
node.close()
bus.shutdown()
def cmd_echo(args):
bus = open_bus(args)
node = CspNode(bus)
try:
reply, ms = node.transaction(args.addr, PORT_ECHO, args.text.encode(), args.timeout)
if reply:
print(f'Balasan {node_name(args.addr)} ({ms:.1f} ms): "{reply["data"].decode(errors="replace")}"')
else:
print("Echo: TIMEOUT")
finally:
node.close()
bus.shutdown()
def main():
p = argparse.ArgumentParser(description="Alat uji CSP lewat CAN (Korlan USB2CAN)")
p.add_argument("--interface", default="usb2can", help="usb2can (default), virtual, dll")
p.add_argument("--channel", default=None, help="serial number Korlan, mis. ED000200 (opsional)")
p.add_argument("--bitrate", type=int, default=1000000, help="default 1000000 (1 Mbit/s)")
p.add_argument("--timeout", type=float, default=1.0)
sub = p.add_subparsers(dest="cmd", required=True)
sub.add_parser("list", help="cari adapter yang terpasang").set_defaults(fn=cmd_list)
m = sub.add_parser("monitor", help="tampilkan lalu lintas bus")
m.add_argument("--frames", action="store_true", help="tampilkan juga setiap frame CAN")
m.set_defaults(fn=cmd_monitor)
pg = sub.add_parser("ping", help="ping node CSP")
pg.add_argument("addr", type=int, help="1 = OBC, 2 = EPS")
pg.add_argument("--count", type=int, default=4)
pg.add_argument("--size", type=int, default=16)
pg.set_defaults(fn=cmd_ping)
sub.add_parser("hk", help="minta housekeeping EPS").set_defaults(fn=cmd_hk)
pl = sub.add_parser("payload", help="ON/OFF rail payload EPS")
pl.add_argument("state", choices=["on", "off"])
pl.set_defaults(fn=cmd_payload)
e = sub.add_parser("echo", help="kirim teks ke port 10")
e.add_argument("addr", type=int)
e.add_argument("text")
e.set_defaults(fn=cmd_echo)
args = p.parse_args()
args.fn(args)
if __name__ == "__main__":
main()
csp_can_gui.py – GUI Tkinter (622 baris)
#!/usr/bin/env python3
"""
csp_can_gui.py - GUI pemantau bus CAN CubeSat (libcsp 1.6 / CFP)
untuk Korlan USB2CAN (8devices) + python-can.
Jalankan: python csp_can_gui.py
Butuh : pip install python-can pywin32 (+ usb2can.dll di folder ini)
Fitur:
- Hubungkan ke Korlan USB2CAN (atau mode Simulasi tanpa hardware)
- Tabel paket CSP yang sudah disusun ulang (warna per node)
- Tabel frame CAN mentah (ID CFP diuraikan)
- Panel EPS: suhu, VDDA, status payload, uptime (auto-refresh)
- Tombol Ping OBC / Ping EPS, Minta HK, Payload ON/OFF, kirim pesan
- Statistik bus + simpan log ke CSV
"""
import csv
import queue
import struct
import threading
import time
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
import can
import csp_can_tool as T
APP_TITLE = "CubeSat CAN Monitor - Korlan USB2CAN"
MAX_ROWS = 2000
# Kode status/error yang dikirim adapter 8devices USB2CAN (byte data pertama)
USB2CAN_STATUS = {
0x00: "OK", 0x01: "overrun", 0x02: "bus warning", 0x03: "error passive",
0x04: "BUS-OFF", 0x20: "stuff error", 0x21: "form error", 0x23: "ACK error",
0x24: "bit0 error", 0x25: "bit1 error", 0x27: "CRC error", -1: "error",
}
NODE_COLORS = {T.ADDR_OBC: "#1f5fd6", T.ADDR_EPS: "#c0392b", T.ADDR_PC: "#1a8a3a"}
# ---------------------------------------------------------------------------
# Simulator OBC + EPS (mode tanpa hardware, memakai bus 'virtual' python-can)
# ---------------------------------------------------------------------------
class Simulator:
def __init__(self, channel):
self.bus = can.Bus(interface="virtual", channel=channel)
self.rx = T.CfpReassembler()
self.ident = 0
self.rail = 0
self.hk_count = 0
self.t0 = time.time()
self.stop = threading.Event()
threading.Thread(target=self._rx, daemon=True).start()
threading.Thread(target=self._obc, daemon=True).start()
def close(self):
self.stop.set()
time.sleep(0.15)
self.bus.shutdown()
def _send(self, src, dst, dport, sport, data):
self.ident = (self.ident + 1) & 0x3FF
hdr = T.csp_header_pack(2, src, dst, dport, sport)
for cid, d in T.cfp_fragment(src, dst, data, hdr, self.ident):
self.bus.send(can.Message(arbitration_id=cid, data=d, is_extended_id=True))
time.sleep(0.0002)
def _hk(self):
temp = int(380 + 15 * ((time.time() * 0.2) % 1))
return struct.pack(T.EPS_HK_FMT, int(time.time() - self.t0), temp, 3290,
self.rail, 0, self.hk_count, 0, 0)
def _reply(self, node, pkt):
h, d = pkt["hdr"], pkt["data"]
if h["dport"] == T.PORT_PING:
out = d
elif h["dport"] == T.PORT_ECHO:
out = f"ACK dari node {node} ({T.NODE_NAMES[node]})".encode()
elif h["dport"] == T.PORT_TLM:
return
elif node == T.ADDR_EPS and h["dport"] == T.PORT_EPS_HK:
self.hk_count += 1
out = self._hk()
elif node == T.ADDR_EPS and h["dport"] == T.PORT_EPS_CMD and len(d) == 2:
self.rail = 1 if d[1] else 0
out = bytes([0, self.rail])
else:
return
self._send(node, h["src"], h["sport"], h["dport"], out)
def _rx(self):
while not self.stop.is_set():
m = self.bus.recv(0.05)
if m is None or not m.is_extended_id:
continue
pkt, _ = self.rx.push(m.arbitration_id, m.data)
if pkt and pkt["hdr"]["dst"] in (T.ADDR_OBC, T.ADDR_EPS):
self._reply(pkt["hdr"]["dst"], pkt)
def _obc(self):
"""OBC tiruan: tiap 2 detik ping, HK dan paket 200 byte ke EPS."""
n = 0
while not self.stop.wait(2.0):
self._send(T.ADDR_OBC, T.ADDR_EPS, T.PORT_PING, 30, bytes(range(32)))
self._send(T.ADDR_EPS, T.ADDR_OBC, 30, T.PORT_PING, bytes(range(32)))
self._send(T.ADDR_OBC, T.ADDR_EPS, T.PORT_EPS_HK, 31, b"\x00")
self.hk_count += 1
self._send(T.ADDR_EPS, T.ADDR_OBC, 31, T.PORT_EPS_HK, self._hk())
self._send(T.ADDR_OBC, T.ADDR_EPS, T.PORT_TLM, 32, bytes((i + n) & 0xFF for i in range(200)))
n += 1
# ---------------------------------------------------------------------------
# GUI
# ---------------------------------------------------------------------------
class App(tk.Tk):
def __init__(self):
super().__init__()
self.title(APP_TITLE)
self.geometry("1280x800")
self.minsize(1000, 640)
self.bus = None
self.node = None
self.sim = None
self.pending = {} # sport -> (kind, info, t0, deadline)
self.log_rows = [] # untuk simpan CSV
self.node_seen = {T.ADDR_OBC: 0.0, T.ADDR_EPS: 0.0}
self.rate_count = 0
self.rate_t = time.time()
self.err_counts = {}
self.err_total = 0
self.last_err = None
self.err_log_t = 0.0
self.send_errors = []
self._style()
self._build()
self.after(40, self._poll)
self.after(2000, self._auto_hk)
self.after(300, lambda: self._scan_serial(quiet=True))
self.protocol("WM_DELETE_WINDOW", self._on_close)
# ---------------------------------------------------------------- style
def _style(self):
import tkinter.font as tkfont
fams = set(tkfont.families(self))
ui = next((f for f in ("Segoe UI", "Helvetica", "DejaVu Sans", "Arial") if f in fams), "TkDefaultFont")
mono = next((f for f in ("Consolas", "DejaVu Sans Mono", "Courier New") if f in fams), "TkFixedFont")
self.ui_font, self.mono_font = ui, mono
s = ttk.Style(self)
try:
s.theme_use("clam")
except tk.TclError:
pass
self.configure(bg="#eef1f5")
s.configure(".", font=(ui, 10), background="#eef1f5")
s.configure("Card.TFrame", background="white", relief="flat")
s.configure("Card.TLabel", background="white")
s.configure("Title.TLabel", font=(ui, 11, "bold"), background="white")
s.configure("Big.TLabel", font=(ui, 22, "bold"), background="white")
s.configure("Unit.TLabel", font=(ui, 9), foreground="#666", background="white")
s.configure("Treeview", rowheight=22, font=(mono, 10))
s.configure("Treeview.Heading", font=(ui, 10, "bold"))
s.configure("Accent.TButton", font=(ui, 10, "bold"))
def _card(self, parent, title):
f = ttk.Frame(parent, style="Card.TFrame", padding=10)
ttk.Label(f, text=title, style="Title.TLabel").pack(anchor="w", pady=(0, 6))
return f
# ---------------------------------------------------------------- layout
def _build(self):
# ---- baris koneksi
top = ttk.Frame(self, padding=(10, 8))
top.pack(fill="x")
ttk.Label(top, text="Adapter:").pack(side="left")
self.var_if = tk.StringVar(value="usb2can")
ttk.Combobox(top, textvariable=self.var_if, width=10, state="readonly",
values=["usb2can", "simulasi"]).pack(side="left", padx=(4, 12))
ttk.Label(top, text="Serial:").pack(side="left")
self.var_ch = tk.StringVar()
self.cb_ch = ttk.Combobox(top, textvariable=self.var_ch, width=11)
self.cb_ch.pack(side="left", padx=(4, 2))
ttk.Button(top, text="Cari", width=5, command=self._scan_serial).pack(side="left", padx=(0, 12))
ttk.Label(top, text="Bitrate:").pack(side="left")
self.var_br = tk.StringVar(value="1000000")
ttk.Combobox(top, textvariable=self.var_br, width=9,
values=["1000000", "500000", "250000", "125000"]).pack(side="left", padx=(4, 12))
self.btn_conn = ttk.Button(top, text="Hubungkan", style="Accent.TButton", command=self._toggle)
self.btn_conn.pack(side="left")
self.lbl_status = tk.Label(top, text=" TIDAK TERHUBUNG ", bg="#999", fg="white",
font=(self.ui_font, 10, "bold"))
self.lbl_status.pack(side="left", padx=12)
ttk.Button(top, text="Simpan log CSV", command=self._save_csv).pack(side="right")
ttk.Button(top, text="Bersihkan", command=self._clear).pack(side="right", padx=6)
body = ttk.Frame(self, padding=(10, 0, 10, 10))
body.pack(fill="both", expand=True)
# ---- panel kiri
left = ttk.Frame(body, width=330)
left.pack(side="left", fill="y", padx=(0, 10))
left.pack_propagate(False)
# node status
c = self._card(left, "Status node")
c.pack(fill="x", pady=(0, 8))
self.node_lbl = {}
for addr in (T.ADDR_OBC, T.ADDR_EPS):
row = ttk.Frame(c, style="Card.TFrame")
row.pack(fill="x", pady=2)
dot = tk.Label(row, text="●", fg="#bbb", bg="white", font=(self.ui_font, 14))
dot.pack(side="left")
ttk.Label(row, text=f"{T.NODE_NAMES[addr]} (alamat {addr})", style="Card.TLabel",
width=16).pack(side="left")
info = ttk.Label(row, text="belum terlihat", style="Card.TLabel", foreground="#777")
info.pack(side="left")
self.node_lbl[addr] = (dot, info)
# EPS housekeeping
c = self._card(left, "EPS housekeeping")
c.pack(fill="x", pady=(0, 8))
grid = ttk.Frame(c, style="Card.TFrame")
grid.pack(fill="x")
self.hk = {}
for i, (key, label, unit) in enumerate([("temp", "Suhu chip", "°C"), ("vdda", "VDDA", "mV"),
("rail", "Payload", ""), ("uptime", "Uptime", "detik")]):
f = ttk.Frame(grid, style="Card.TFrame", padding=(0, 2))
f.grid(row=i // 2, column=i % 2, sticky="w", padx=(0, 18), pady=2)
ttk.Label(f, text=label, style="Unit.TLabel").pack(anchor="w")
v = ttk.Label(f, text="–", style="Big.TLabel")
v.pack(anchor="w")
ttk.Label(f, text=unit, style="Unit.TLabel").pack(anchor="w")
self.hk[key] = v
self.lbl_hk_time = ttk.Label(c, text="belum ada data", style="Unit.TLabel")
self.lbl_hk_time.pack(anchor="w", pady=(4, 4))
r = ttk.Frame(c, style="Card.TFrame")
r.pack(fill="x")
ttk.Button(r, text="Minta HK", command=self._req_hk).pack(side="left")
self.var_auto = tk.BooleanVar(value=True)
ttk.Checkbutton(r, text="otomatis tiap 2 s", variable=self.var_auto).pack(side="left", padx=8)
# perintah
c = self._card(left, "Perintah")
c.pack(fill="x", pady=(0, 8))
r = ttk.Frame(c, style="Card.TFrame")
r.pack(fill="x", pady=2)
ttk.Button(r, text="Ping OBC", command=lambda: self._ping(T.ADDR_OBC)).pack(side="left", expand=True, fill="x")
ttk.Button(r, text="Ping EPS", command=lambda: self._ping(T.ADDR_EPS)).pack(side="left", expand=True, fill="x", padx=(6, 0))
r = ttk.Frame(c, style="Card.TFrame")
r.pack(fill="x", pady=2)
ttk.Button(r, text="Payload ON", command=lambda: self._payload(1)).pack(side="left", expand=True, fill="x")
ttk.Button(r, text="Payload OFF", command=lambda: self._payload(0)).pack(side="left", expand=True, fill="x", padx=(6, 0))
r = ttk.Frame(c, style="Card.TFrame")
r.pack(fill="x", pady=(6, 2))
self.var_dst = tk.StringVar(value="EPS (2)")
ttk.Combobox(r, textvariable=self.var_dst, width=8, state="readonly",
values=["OBC (1)", "EPS (2)"]).pack(side="left")
self.var_msg = tk.StringVar(value="Halo dari PC")
e = ttk.Entry(r, textvariable=self.var_msg)
e.pack(side="left", fill="x", expand=True, padx=6)
e.bind("<Return>", lambda _: self._echo())
ttk.Button(r, text="Kirim", command=self._echo).pack(side="left")
self.lbl_result = tk.Label(c, text="", bg="white", anchor="w", justify="left",
font=(self.ui_font, 10), wraplength=300)
self.lbl_result.pack(fill="x", pady=(6, 0))
# statistik
c = self._card(left, "Statistik bus")
c.pack(fill="both", expand=True)
self.stat_lbl = {}
for key, label in [("frames", "Frame CAN"), ("packets", "Paket CSP"), ("rate", "Frame/detik"),
("errors", "Error frame"), ("cfp_err", "Error susun CFP")]:
row = ttk.Frame(c, style="Card.TFrame")
row.pack(fill="x")
ttk.Label(row, text=label, style="Card.TLabel").pack(side="left")
v = ttk.Label(row, text="0", style="Card.TLabel", font=(self.ui_font, 10, "bold"))
v.pack(side="right")
self.stat_lbl[key] = v
self.cfp_err = 0
row = ttk.Frame(c, style="Card.TFrame")
row.pack(fill="x", pady=(6, 0))
ttk.Label(row, text="Kondisi bus", style="Card.TLabel").pack(side="left")
self.lbl_bus = tk.Label(row, text="–", bg="white", font=(self.ui_font, 10, "bold"))
self.lbl_bus.pack(side="right")
# ---- panel kanan: tab
nb = ttk.Notebook(body)
nb.pack(side="left", fill="both", expand=True)
t1 = ttk.Frame(nb)
nb.add(t1, text=" Paket CSP ")
cols = ("time", "src", "dst", "port", "len", "info")
self.tv_pkt = self._tree(t1, cols, {"time": ("Waktu", 95), "src": ("Dari", 90), "dst": ("Ke", 90),
"port": ("Port", 110), "len": ("Len", 50), "info": ("Isi", 520)})
for addr, col in NODE_COLORS.items():
self.tv_pkt.tag_configure(f"n{addr}", foreground=col)
t2 = ttk.Frame(nb)
nb.add(t2, text=" Frame CAN ")
self.var_frames = tk.BooleanVar(value=True)
ttk.Checkbutton(t2, text="Tampilkan frame (matikan jika bus sangat ramai)",
variable=self.var_frames).pack(anchor="w", padx=4, pady=2)
cols = ("time", "id", "src", "dst", "type", "remain", "cfp", "dlc", "data")
self.tv_frm = self._tree(t2, cols, {"time": ("Waktu", 95), "id": ("CAN ID", 95), "src": ("Src", 45),
"dst": ("Dst", 45), "type": ("Tipe", 60), "remain": ("Remain", 60),
"cfp": ("CFP id", 60), "dlc": ("DLC", 40), "data": ("Data", 300)})
self.tv_frm.tag_configure("err", foreground="#c0392b")
t3 = ttk.Frame(nb)
nb.add(t3, text=" Log ")
self.txt_log = tk.Text(t3, font=(self.mono_font, 10), wrap="word", bg="#fbfbfb")
self.txt_log.pack(fill="both", expand=True)
self._log("Siap. Pilih adapter lalu klik Hubungkan. Pilih 'simulasi' untuk mencoba tanpa hardware.")
def _tree(self, parent, cols, spec):
frame = ttk.Frame(parent)
frame.pack(fill="both", expand=True)
tv = ttk.Treeview(frame, columns=cols, show="headings")
for c in cols:
title, w = spec[c]
tv.heading(c, text=title)
tv.column(c, width=w, anchor="w", stretch=(c in ("info", "data")))
sb = ttk.Scrollbar(frame, orient="vertical", command=tv.yview)
tv.configure(yscrollcommand=sb.set)
tv.pack(side="left", fill="both", expand=True)
sb.pack(side="right", fill="y")
return tv
# ---------------------------------------------------------------- koneksi
def _toggle(self):
if self.bus:
self._disconnect()
else:
self._connect()
def _connect(self):
try:
br = int(self.var_br.get())
if self.var_if.get() == "simulasi":
self.sim = Simulator("csp_sim")
self.bus = can.Bus(interface="virtual", channel="csp_sim")
else:
if not self.var_ch.get().strip():
self._scan_serial(quiet=True)
class A: # argumen untuk open_bus()
pass
a = A()
a.interface, a.channel, a.bitrate = "usb2can", (self.var_ch.get().strip() or None), br
self.bus = T.open_bus(a)
self.node = T.CspNode(self.bus)
except Exception as e: # noqa: BLE001
self.bus = None
if self.sim:
self.sim.close()
self.sim = None
messagebox.showerror("Gagal terhubung",
f"{e}\n\nCek: driver 8devices terpasang, usb2can.dll (64-bit) ada di "
"folder script, pywin32 terpasang, dan serial number benar.")
return
self.btn_conn.configure(text="Putuskan")
sn = "" if self.var_if.get() == "simulasi" else (self.var_ch.get().strip() or "?")
self.lbl_status.configure(text=f" TERHUBUNG {int(self.var_br.get()) // 1000} kbit/s"
+ (f" · S/N {sn}" if sn else "") + " ", bg="#1a8a3a")
self.title(f"{APP_TITLE}" + (f" [S/N {sn}]" if sn else ""))
self._log(f"Terhubung ({self.var_if.get()}{', S/N ' + sn if sn else ''}), alamat CSP PC = {T.ADDR_PC}")
def _scan_serial(self, quiet=False):
"""Cari Korlan USB2CAN yang terpasang dan isi kolom Serial."""
try:
cfgs = can.detect_available_configs("usb2can")
except Exception as e: # noqa: BLE001
cfgs = []
if not quiet:
self._log(f"Gagal mencari adapter: {e}")
serials = [str(c.get("channel")) for c in cfgs if c.get("channel")]
self.cb_ch.configure(values=serials)
if serials:
if self.var_ch.get().strip() not in serials:
self.var_ch.set(serials[0])
self._log("Korlan USB2CAN terdeteksi: S/N " + ", ".join(serials))
elif not quiet:
self._log("Korlan USB2CAN tidak terdeteksi (cek kabel USB / driver 8devices / pywin32).")
return serials
def _disconnect(self):
try:
if self.node:
self.node.close()
if self.bus:
self.bus.shutdown()
if self.sim:
self.sim.close()
finally:
self.node = self.bus = self.sim = None
self.pending.clear()
self.btn_conn.configure(text="Hubungkan")
self.lbl_status.configure(text=" TIDAK TERHUBUNG ", bg="#999")
self._log("Koneksi diputus")
def _on_close(self):
if self.bus:
self._disconnect()
self.destroy()
# ---------------------------------------------------------------- perintah
def _request(self, kind, dst, dport, payload, info=None, timeout=1.0):
if not self.node:
self._result("Belum terhubung ke adapter.", "#c0392b")
return
node = self.node
sport = node.alloc_sport()
self.pending[sport] = (kind, dst, info, time.time(), time.time() + timeout)
def worker():
# dikirim di thread terpisah agar GUI tidak macet bila adapter lambat
try:
node.send(dst, dport, payload, sport=sport)
except Exception as e: # noqa: BLE001
self.send_errors.append(str(e))
threading.Thread(target=worker, daemon=True).start()
def _ping(self, addr):
self._request("ping", addr, T.PORT_PING, bytes(range(16)), bytes(range(16)))
def _req_hk(self):
self._request("hk", T.ADDR_EPS, T.PORT_EPS_HK, b"\x00")
def _payload(self, on):
self._request("cmd", T.ADDR_EPS, T.PORT_EPS_CMD, bytes([0, on]), on)
def _echo(self):
dst = T.ADDR_OBC if self.var_dst.get().startswith("OBC") else T.ADDR_EPS
self._request("echo", dst, T.PORT_ECHO, self.var_msg.get().encode())
def _auto_hk(self):
if self.node and self.var_auto.get():
if not any(p[0] == "hk" for p in self.pending.values()):
self._req_hk()
self.after(2000, self._auto_hk)
def _handle_reply(self, pkt):
h = pkt["hdr"]
p = self.pending.get(h["dport"])
if not p or h["src"] != p[1] or h["dst"] != T.ADDR_PC:
return
del self.pending[h["dport"]]
kind, dst, info, t0, _ = p
ms = (pkt.get("t", time.time()) - t0) * 1000
name = f"{T.NODE_NAMES[dst]} ({dst})"
d = pkt["data"]
if kind == "ping":
ok = d == info
self._result(f"Ping {name}: {ms:.1f} ms" + ("" if ok else " (data berbeda!)"),
"#1a8a3a" if ok else "#c0392b")
elif kind == "hk" and len(d) == T.EPS_HK_SIZE:
self._show_hk(d, ms)
elif kind == "cmd" and len(d) == 2:
ok = d[0] == 0
self._result(f"Payload {'ON' if d[1] else 'OFF'}: {'OK' if ok else 'DITOLAK'} ({ms:.1f} ms)",
"#1a8a3a" if ok else "#c0392b")
elif kind == "echo":
self._result(f'Balasan {name} ({ms:.1f} ms):\n"{d.decode(errors="replace")}"', "#1f5fd6")
def _show_hk(self, d, ms=None):
up, t10, vdda, rail, _, cnt, rxf, txf = struct.unpack(T.EPS_HK_FMT, d)
self.hk["temp"].configure(text=f"{t10 / 10:.1f}")
self.hk["vdda"].configure(text=f"{vdda}")
self.hk["rail"].configure(text="ON" if rail else "OFF", foreground="#1a8a3a" if rail else "#777")
self.hk["uptime"].configure(text=f"{up}")
extra = f", {ms:.0f} ms" if ms is not None else ""
self.lbl_hk_time.configure(text=f"diperbarui {time.strftime('%H:%M:%S')}{extra} · HK #{cnt}")
def _check_timeouts(self):
now = time.time()
for sport, p in list(self.pending.items()):
if now > p[4]:
del self.pending[sport]
kind, dst = p[0], p[1]
label = {"ping": "Ping", "hk": "Housekeeping", "cmd": "Perintah payload",
"echo": "Pesan"}[kind]
self._result(f"{label} ke {T.NODE_NAMES[dst]} ({dst}): TIMEOUT", "#c0392b")
self._log(f"{label} ke node {dst}: timeout")
# ---------------------------------------------------------------- loop GUI
def _poll(self):
if self.node:
n = 0
while n < 500:
try:
item = self.node.inbox.get_nowait()
except queue.Empty:
break
n += 1
self._handle_item(item)
self._check_timeouts()
self._update_stats()
self.after(40, self._poll)
def _handle_item(self, item):
ts = time.strftime("%H:%M:%S") + f".{int(time.time() * 1000) % 1000:03d}"
kind = item[0]
if kind == "frame":
msg, err = item[1], item[2]
self.rate_count += 1
f = T.cfp_fields(msg.arbitration_id)
if f["src"] in self.node_seen:
self.node_seen[f["src"]] = time.time()
if err:
self.cfp_err += 1
self._log(f"CFP: {err} (ID 0x{msg.arbitration_id:08X})")
if self.var_frames.get():
self._insert(self.tv_frm, (ts, f"0x{msg.arbitration_id:08X}", f["src"], f["dst"],
"BEGIN" if f["type"] == 0 else "MORE", f["remain"], f["id"],
len(msg.data), msg.data.hex(" ")), ("err",) if err else ())
elif kind == "packet":
pkt = item[1]
h = pkt["hdr"]
desc = T.describe_packet(pkt)
info = desc.split(" ", 1)[1] if " " in desc else ""
if h["dport"] in T.PORT_NAMES:
port = f"{h['dport']} {T.PORT_NAMES[h['dport']]}"
elif h["sport"] in T.PORT_NAMES:
port = f"{h['dport']} re:{T.PORT_NAMES[h['sport']]}"
else:
port = str(h["dport"])
self._insert(self.tv_pkt, (ts, f"{T.NODE_NAMES.get(h['src'], '?')}:{h['sport']}",
f"{T.NODE_NAMES.get(h['dst'], '?')}:{h['dport']}", port,
len(pkt["data"]), info), (f"n{h['src']}",))
self.log_rows.append([ts, h["src"], h["sport"], h["dst"], h["dport"], len(pkt["data"]),
pkt["data"].hex(), info])
# HK EPS (dari siapa pun yang meminta, termasuk OBC) -> perbarui panel
if h["src"] == T.ADDR_EPS and h["sport"] == T.PORT_EPS_HK and len(pkt["data"]) == T.EPS_HK_SIZE:
self._show_hk(pkt["data"])
self._handle_reply(pkt)
elif kind == "error":
data = bytes(item[1].data)
code = data[0] if data else -1
self.err_counts[code] = self.err_counts.get(code, 0) + 1
self.err_total += 1
self.last_err = code
elif kind == "std":
m = item[1]
self._log(f"Frame standar 0x{m.arbitration_id:03X} {m.data.hex(' ')} (bukan CFP)")
def _insert(self, tv, values, tags=()):
tv.insert("", "end", values=values, tags=tags)
kids = tv.get_children()
if len(kids) > MAX_ROWS:
tv.delete(*kids[:len(kids) - MAX_ROWS])
tv.yview_moveto(1.0)
def _update_stats(self):
now_t = time.time()
while self.send_errors:
self._result(f"Gagal kirim: {self.send_errors.pop()}", "#c0392b")
if self.err_counts and now_t - self.err_log_t >= 1.0:
self.err_log_t = now_t
txt = ", ".join(f"{USB2CAN_STATUS.get(c, hex(c))} x{n}" for c, n in sorted(self.err_counts.items()))
self._log(f"Status/error dari adapter (1 detik terakhir): {txt}")
self.lbl_bus.configure(text="ERROR: " + USB2CAN_STATUS.get(self.last_err, hex(self.last_err)),
fg="#c0392b")
self.err_counts.clear()
self.bus_err_t = now_t
elif now_t - getattr(self, "bus_err_t", 0) > 3 and self.node.stats["frames"]:
self.lbl_bus.configure(text="OK", fg="#1a8a3a")
s = self.node.stats
self.stat_lbl["frames"].configure(text=str(s["frames"]))
self.stat_lbl["packets"].configure(text=str(s["packets"]))
self.stat_lbl["errors"].configure(text=str(s["errors"]))
self.stat_lbl["cfp_err"].configure(text=str(self.cfp_err))
now = time.time()
if now - self.rate_t >= 1.0:
self.stat_lbl["rate"].configure(text=f"{self.rate_count / (now - self.rate_t):.0f}")
self.rate_count, self.rate_t = 0, now
for addr, (dot, info) in self.node_lbl.items():
seen = self.node_seen[addr]
if seen == 0:
continue
age = now - seen
if age < 5:
dot.configure(fg="#1a8a3a")
info.configure(text="aktif", foreground="#1a8a3a")
else:
dot.configure(fg="#e67e22")
info.configure(text=f"diam {age:.0f} s", foreground="#e67e22")
# ---------------------------------------------------------------- util
def _result(self, text, color="#333"):
self.lbl_result.configure(text=text, fg=color)
def _log(self, text):
self.txt_log.insert("end", f"[{time.strftime('%H:%M:%S')}] {text}\n")
self.txt_log.see("end")
def _clear(self):
for tv in (self.tv_pkt, self.tv_frm):
tv.delete(*tv.get_children())
self.log_rows.clear()
self.txt_log.delete("1.0", "end")
def _save_csv(self):
if not self.log_rows:
messagebox.showinfo("Simpan log", "Belum ada paket CSP untuk disimpan.")
return
path = filedialog.asksaveasfilename(defaultextension=".csv",
initialfile=time.strftime("csp_log_%Y%m%d_%H%M%S.csv"),
filetypes=[("CSV", "*.csv")])
if not path:
return
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["waktu", "src", "sport", "dst", "dport", "len", "data_hex", "keterangan"])
w.writerows(self.log_rows)
self._log(f"Log disimpan: {path}")
if __name__ == "__main__":
App().mainloop()
csp_can_tool.py juga bisa dipakai tanpa GUI, misalnya python csp_can_tool.py monitor, python csp_can_tool.py ping 2 atau python csp_can_tool.py hk.
Penutup
Dengan Korlan USB2CAN dan sedikit Python, kita punya alat diagnosis bus CAN yang memahami protokol CubeSat Space Protocol: frame mentah, paket CSP, data housekeeping EPS, sampai status error fisik bus. Hasil uji dengan dua board sungguhan menunjukkan OBC dan EPS berkomunikasi tanpa error setelah terminator 120 Ω terpasang. Semoga bermanfaat.
Wassalamu'alaikum warohmatullohi wabarakatuh.
