Skip to content

Instantly share code, notes, and snippets.

@milisarge

milisarge/config.ini

Last active July 7, 2026 15:25
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save milisarge/91e2e26daf5b9dc84f006ced4fa583ef to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save milisarge/91e2e26daf5b9dc84f006ced4fa583ef to your computer and use it in GitHub Desktop.
Download ZIP
lora chat sx1262 waveshare usb to lora
Raw
config.ini
[serial]
port = /dev/ttyACM0
baudrate = 115200
[user]
name = Cihaz_0
[lora]
; --- Kanal / izolasyon parametreleri ---
; Bu değerler haberleşecek TÜM cihazlarda birebir aynı olmalı.
frequency = 868000000
spreading_factor = 7
; Bandwidth (kHz): 7.8, 10.4, 15.6, 20.8, 31.25, 41.7, 62.5, 125, 250, 500
bandwidth = 125
; Coding rate: 4/5, 4/6, 4/7, 4/8
coding_rate = 4/5
low_data_rate = 0
preamble_length = 16
; Sync word: aynı bantta çalışan diğer LoRa cihazlarından izole olmak için
; standart dışı bir değer seçin (0x12 genel varsayılandır, Meshtastic 0x2B kullanır)
sync_word = 0x12
crc_on = 1
invert_iq = 0
Raw
lora_chat.py
import argparse
import configparser
import os
import gi
import serial
import serial.tools.list_ports
import threading
import struct
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, GLib
# --- LORA PROTOKOL YARDIMCILARI ---
def crc16(data: bytes) -> int:
crc = 0
for b in data:
a = (crc >> 8) ^ b
crc = (a << 2) ^ (a << 1) ^ a ^ (crc << 8)
crc = crc & 0xFFFF
return crc
def escape_bytes(data: bytes) -> bytes:
out = bytearray()
for b in data:
if b == 0xAA:
out.extend([0x7D, 0x8A])
elif b == 0x7D:
out.extend([0x7D, 0x5D])
else:
out.append(b)
return bytes(out)
def unescape_bytes(data: bytes) -> bytes:
out = bytearray()
i = 0
while i < len(data):
if data[i] == 0x7D and i + 1 < len(data):
if data[i+1] == 0x8A:
out.append(0xAA)
elif data[i+1] == 0x5D:
out.append(0x7D)
i += 2
else:
out.append(data[i])
i += 1
return bytes(out)
def pack_message(cmd: int, payload: bytes = b'') -> bytes:
size = len(payload)
header_and_payload = struct.pack('<BH', cmd, size) + payload
crc = crc16(header_and_payload)
full_body = header_and_payload + struct.pack('<H', crc)
escaped_body = escape_bytes(full_body)
return b'\xAA' + escaped_body
# --- UYGULAMA SEVİYESİ MESAJ ZARFI ---
# Firmware ham LoRa verisi taşıyor, adresleme (toplu / özel) yok.
# Uygulama katmanında basit bir zarf kullanıyoruz: HEDEF \x1f GONDEREN \x1f MESAJ
ENVELOPE_SEP = '\x1f'
BROADCAST_TARGET = '*'
BROADCAST_LABEL = "Herkese Gönder"
# STDBY_XOSC_RX: TX bitince cihazı otomatik sürekli RX moduna geri döndürür.
FALLBACK_MODE_STDBY_XOSC_RX = 0x31
# --- KANAL / İZOLASYON PARAMETRE HARİTALARI (SX1262) ---
SF_OPTIONS = [5, 6, 7, 8, 9, 10, 11, 12] # SX126X_LORA_SF5..SF12, değer == SF numarası
BW_OPTIONS = [
("7.8 kHz", 0x00),
("10.4 kHz", 0x08),
("15.6 kHz", 0x01),
("20.8 kHz", 0x09),
("31.25 kHz", 0x02),
("41.7 kHz", 0x0A),
("62.5 kHz", 0x03),
("125 kHz", 0x04),
("250 kHz", 0x05),
("500 kHz", 0x06),
]
BW_LABEL_TO_CODE = {label: code for label, code in BW_OPTIONS}
CR_OPTIONS = [
("4/5", 0x01),
("4/6", 0x02),
("4/7", 0x03),
("4/8", 0x04),
]
CR_LABEL_TO_CODE = {label: code for label, code in CR_OPTIONS}
DEFAULT_CONFIG = {
'serial': {'port': '', 'baudrate': '115200'},
'user': {'name': 'Cihaz_1'},
'lora': {
'frequency': '868000000',
'spreading_factor': '7',
'bandwidth': '125', # kHz
'coding_rate': '4/5',
'low_data_rate': '0',
'preamble_length': '16',
'sync_word': '0x12',
'crc_on': '1',
'invert_iq': '0',
},
}
def load_config(path):
"""config.ini dosyasını okur, eksik alanlar için varsayılanları kullanır."""
cfg = configparser.ConfigParser()
for section, values in DEFAULT_CONFIG.items():
cfg[section] = dict(values)
if path and os.path.isfile(path):
cfg.read(path)
return cfg
def bandwidth_khz_to_code(khz_text):
"""'125' veya '125 kHz' gibi bir metni SX1262 bandwidth koduna çevirir."""
khz_text = khz_text.strip().replace('kHz', '').replace('KHz', '').strip()
for label, code in BW_OPTIONS:
label_khz = label.replace(' kHz', '').strip()
if label_khz == khz_text:
return code
try:
val = float(khz_text)
for label, code in BW_OPTIONS:
if abs(float(label.replace(' kHz', '')) - val) < 0.01:
return code
except ValueError:
pass
return 0x04 # 125 kHz varsayılan
def find_serial_ports():
"""/dev/ttyACM*, /dev/ttyUSB* (Linux) ve COM* (Windows) portlarını listeler."""
ports = []
for p in serial.tools.list_ports.comports():
dev = p.device
# Windows uyumluluğu için kontrolü genişletiyoruz
if 'ttyACM' in dev or 'ttyUSB' in dev or 'COM' in dev.upper():
ports.append(dev)
if not ports:
ports = [p.device for p in serial.tools.list_ports.comports()]
return ports
# --- ANA UYGULAMA PENCERESİ ---
class LoRaChatApp(Gtk.Window):
def __init__(self, config):
super().__init__(title="LoRa PyGTK3 Chat")
self.set_default_size(680, 560)
self.ser = None
self.running = False
self.known_users = set()
self.config = config
vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
vbox.set_margin_start(10)
vbox.set_margin_end(10)
vbox.set_margin_top(10)
vbox.set_margin_bottom(10)
self.add(vbox)
# --- AYARLAR BÖLÜMÜ (Grid) ---
grid = Gtk.Grid(column_spacing=10, row_spacing=6)
vbox.pack_start(grid, False, False, 0)
grid.attach(Gtk.Label(label="Seri Port:", xalign=0), 0, 0, 1, 1)
port_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=4)
self.port_combo = Gtk.ComboBoxText()
port_box.pack_start(self.port_combo, True, True, 0)
refresh_btn = Gtk.Button(label="⟳")
refresh_btn.set_tooltip_text("Portları yenile")
refresh_btn.connect("clicked", self.on_refresh_ports_clicked)
port_box.pack_start(refresh_btn, False, False, 0)
grid.attach(port_box, 1, 0, 1, 1)
grid.attach(Gtk.Label(label="Kullanıcı Adı:", xalign=0), 0, 1, 1, 1)
self.name_entry = Gtk.Entry(text=config['user'].get('name', 'Cihaz_1'))
grid.attach(self.name_entry, 1, 1, 1, 1)
grid.attach(Gtk.Label(label="Frekans (Hz):", xalign=0), 2, 0, 1, 1)
self.freq_entry = Gtk.Entry(text=config['lora'].get('frequency', '868000000'))
grid.attach(self.freq_entry, 3, 0, 1, 1)
self.conn_btn = Gtk.Button(label="Bağlan ve Ayarla")
self.conn_btn.connect("clicked", self.on_connect_clicked)
grid.attach(self.conn_btn, 3, 1, 1, 1)
# --- KANAL / İZOLASYON AYARLARI (Expander) ---
expander = Gtk.Expander(label="Kanal / İzolasyon Ayarları (SF, BW, CR, Sync Word...)")
vbox.pack_start(expander, False, False, 0)
ch_grid = Gtk.Grid(column_spacing=10, row_spacing=6)
ch_grid.set_margin_top(6)
expander.add(ch_grid)
ch_grid.attach(Gtk.Label(label="Spreading Factor:", xalign=0), 0, 0, 1, 1)
self.sf_combo = Gtk.ComboBoxText()
for sf in SF_OPTIONS:
self.sf_combo.append_text(f"SF{sf}")
cfg_sf = config['lora'].getint('spreading_factor', fallback=7)
try:
self.sf_combo.set_active(SF_OPTIONS.index(cfg_sf))
except ValueError:
self.sf_combo.set_active(SF_OPTIONS.index(7))
ch_grid.attach(self.sf_combo, 1, 0, 1, 1)
ch_grid.attach(Gtk.Label(label="Bandwidth:", xalign=0), 2, 0, 1, 1)
self.bw_combo = Gtk.ComboBoxText()
for label, _ in BW_OPTIONS:
self.bw_combo.append_text(label)
bw_code = bandwidth_khz_to_code(config['lora'].get('bandwidth', '125'))
bw_labels = [l for l, _ in BW_OPTIONS]
bw_codes = [c for _, c in BW_OPTIONS]
self.bw_combo.set_active(bw_codes.index(bw_code) if bw_code in bw_codes else bw_labels.index("125 kHz"))
ch_grid.attach(self.bw_combo, 3, 0, 1, 1)
ch_grid.attach(Gtk.Label(label="Coding Rate:", xalign=0), 0, 1, 1, 1)
self.cr_combo = Gtk.ComboBoxText()
for label, _ in CR_OPTIONS:
self.cr_combo.append_text(label)
cfg_cr_label = config['lora'].get('coding_rate', '4/5')
cr_labels = [l for l, _ in CR_OPTIONS]
self.cr_combo.set_active(cr_labels.index(cfg_cr_label) if cfg_cr_label in cr_labels else 0)
ch_grid.attach(self.cr_combo, 1, 1, 1, 1)
self.ldr_check = Gtk.CheckButton(label="Low Data Rate Optimization")
self.ldr_check.set_active(config['lora'].getboolean('low_data_rate', fallback=False))
ch_grid.attach(self.ldr_check, 2, 1, 2, 1)
ch_grid.attach(Gtk.Label(label="Sync Word (hex):", xalign=0), 0, 2, 1, 1)
self.sync_entry = Gtk.Entry(text=config['lora'].get('sync_word', '0x12'))
self.sync_entry.set_tooltip_text(
"Aynı Sync Word'ü kullanan cihazlar birbirini görür. "
"Farklı ağlardan izole olmak için standart dışı bir değer seçin."
)
ch_grid.attach(self.sync_entry, 1, 2, 1, 1)
ch_grid.attach(Gtk.Label(label="Preamble Length:", xalign=0), 2, 2, 1, 1)
self.preamble_spin = Gtk.SpinButton()
self.preamble_spin.set_adjustment(Gtk.Adjustment(value=config['lora'].getint('preamble_length', fallback=16),
lower=6, upper=65535, step_increment=1))
self.preamble_spin.set_numeric(True)
ch_grid.attach(self.preamble_spin, 3, 2, 1, 1)
self.crc_check = Gtk.CheckButton(label="Havada CRC (radyo katmanı)")
self.crc_check.set_active(config['lora'].getboolean('crc_on', fallback=True))
ch_grid.attach(self.crc_check, 0, 3, 2, 1)
self.iq_check = Gtk.CheckButton(label="Invert IQ")
self.iq_check.set_active(config['lora'].getboolean('invert_iq', fallback=False))
ch_grid.attach(self.iq_check, 2, 3, 2, 1)
vbox.pack_start(Gtk.Separator(orientation=Gtk.Orientation.HORIZONTAL), False, False, 6)
# --- CHAT PENCERESİ ---
scrolled_window = Gtk.ScrolledWindow()
scrolled_window.set_hexpand(True)
scrolled_window.set_vexpand(True)
vbox.pack_start(scrolled_window, True, True, 0)
self.chat_buffer = Gtk.TextBuffer()
self.chat_view = Gtk.TextView(buffer=self.chat_buffer, editable=False, wrap_mode=Gtk.WrapMode.WORD)
scrolled_window.add(self.chat_view)
# --- HEDEF SEÇİMİ (Toplu / Özel) ---
target_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
vbox.pack_start(target_box, False, False, 0)
target_box.pack_start(Gtk.Label(label="Hedef:"), False, False, 0)
self.target_combo = Gtk.ComboBoxText.new_with_entry()
self.target_combo.append_text(BROADCAST_LABEL)
self.target_combo.set_active(0)
target_box.pack_start(self.target_combo, True, True, 0)
# --- MESAJ GÖNDERME BÖLÜMÜ ---
hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
vbox.pack_start(hbox, False, False, 0)
self.msg_entry = Gtk.Entry(placeholder_text="Mesajınızı yazın...")
self.msg_entry.connect("activate", self.on_send_clicked)
hbox.pack_start(self.msg_entry, True, True, 0)
send_btn = Gtk.Button(label="Gönder")
send_btn.connect("clicked", self.on_send_clicked)
hbox.pack_start(send_btn, False, False, 0)
self.on_refresh_ports_clicked(None)
cfg_port = config['serial'].get('port', '').strip()
if cfg_port:
model = self.port_combo.get_model()
for i, row in enumerate(model):
if row[0] == cfg_port:
self.port_combo.set_active(i)
break
self.append_chat("Sistem: Uygulama başlatıldı. Portu seçip 'Bağlan' butonuna basın.")
# --- PORT LİSTELEME ---
def on_refresh_ports_clicked(self, widget):
current = self.port_combo.get_active_text()
self.port_combo.remove_all()
ports = find_serial_ports()
for p in ports:
self.port_combo.append_text(p)
if ports:
if current in ports:
self.port_combo.set_active(ports.index(current))
else:
self.port_combo.set_active(0)
else:
self.append_chat("Sistem: Hiç seri port bulunamadı.")
# --- BUTON VE TETİKLEYİCİ FONKSİYONLARI ---
def append_chat(self, text):
GLib.idle_add(self._unsafe_append_chat, text)
def _unsafe_append_chat(self, text):
end_iter = self.chat_buffer.get_end_iter()
self.chat_buffer.insert(end_iter, text + "\n")
adj = self.chat_view.get_vadjustment()
adj.set_value(adj.get_upper() - adj.get_page_size())
def remember_user(self, name):
if name and name not in self.known_users and name != self.name_entry.get_text().strip():
self.known_users.add(name)
GLib.idle_add(self._add_target_option, name)
def _add_target_option(self, name):
model = self.target_combo.get_model()
for row in model:
if row[0] == name:
return
self.target_combo.append_text(name)
def on_connect_clicked(self, widget):
if self.ser and self.ser.is_open:
self.running = False
self.ser.close()
self.conn_btn.set_label("Bağlan ve Ayarla")
self.append_chat("Sistem: Bağlantı kesildi.")
return
port = self.port_combo.get_active_text()
if not port:
self.append_chat("Hata: Seri port seçilmedi!")
return
try:
baudrate = self.config['serial'].getint('baudrate', fallback=115200)
# timeout=0.5 -> okuma bloklanır, busy-loop olmaz, CPU boşta beklerken yormaz
self.ser = serial.Serial(port, baudrate, timeout=0.5)
self.append_chat(f"Sistem: {port} portuna bağlandı.")
# 1. LoRa modülasyon parametreleri (CMD 0x02): SF, BW, CR, LDR
sf = SF_OPTIONS[self.sf_combo.get_active()]
bw_label = self.bw_combo.get_active_text()
bw_code = BW_LABEL_TO_CODE[bw_label]
cr_label = self.cr_combo.get_active_text()
cr_code = CR_LABEL_TO_CODE[cr_label]
ldr = 0x01 if self.ldr_check.get_active() else 0x00
mod_payload = struct.pack('<BBBB', sf, bw_code, cr_code, ldr)
self.ser.write(pack_message(0x02, mod_payload))
self.append_chat(f"Sistem: Modülasyon SF{sf}/{bw_label}/{cr_label} ayarlandı.")
# 2. LoRa paket parametreleri (CMD 0x03): preamble, header, sync word, crc, invert iq
preamble = int(self.preamble_spin.get_value())
header_type = 0x00 # explicit - değişken boyutlu mesajlarımız için gerekli
sync_word = int(self.sync_entry.get_text().strip(), 0) # "0x12" veya "18" kabul eder
crc_on = 0x01 if self.crc_check.get_active() else 0x00
invert_iq = 0x01 if self.iq_check.get_active() else 0x00
pkt_payload = struct.pack('<HBBBB', preamble, header_type, sync_word, crc_on, invert_iq)
self.ser.write(pack_message(0x03, pkt_payload))
self.append_chat(f"Sistem: Sync word 0x{sync_word:02X} ile kanal izolasyonu ayarlandı.")
# 3. Frekans Ayarla (CMD 0x06)
freq = int(self.freq_entry.get_text())
freq_payload = struct.pack('<I', freq)
self.ser.write(pack_message(0x06, freq_payload))
self.append_chat(f"Sistem: Frekans {freq} Hz olarak istendi.")
# 4. Fallback modunu STDBY_XOSC_RX yap (CMD 0x07)
fallback_payload = struct.pack('<B', FALLBACK_MODE_STDBY_XOSC_RX)
self.ser.write(pack_message(0x07, fallback_payload))
self.append_chat("Sistem: TX sonrası otomatik RX moduna dönüş (STDBY_XOSC_RX) ayarlandı.")
# 5. Sürekli Dinleme Moduna Al (CMD 0x09)
rx_payload = struct.pack('<IB', 0x00FFFFFF, 0x00)
self.ser.write(pack_message(0x09, rx_payload))
self.append_chat("Sistem: Cihaz dinleme (RX) moduna alındı.")
self.running = True
self.read_thread = threading.Thread(target=self.serial_listen_worker, daemon=True)
self.read_thread.start()
self.conn_btn.set_label("Bağlantıyı Kes")
except ValueError as e:
self.append_chat(f"Hata: Geçersiz parametre değeri! ({str(e)})")
except Exception as e:
self.append_chat(f"Hata: Bağlantı kurulamadı! ({str(e)})")
def on_send_clicked(self, widget):
if not self.ser or not self.ser.is_open:
self.append_chat("Sistem: Önce cihaza bağlanmalısınız!")
return
msg_text = self.msg_entry.get_text().strip()
if not msg_text:
return
name = self.name_entry.get_text().strip()
target_text = self.target_combo.get_child().get_text().strip()
if not target_text or target_text == BROADCAST_LABEL:
target = BROADCAST_TARGET
else:
target = target_text
self.remember_user(target)
envelope = f"{target}{ENVELOPE_SEP}{name}{ENVELOPE_SEP}{msg_text}"
payload_bytes = envelope.encode('utf-8')
try:
tx_payload = struct.pack('<I', 0) + payload_bytes
self.ser.write(pack_message(0x0A, tx_payload))
if target == BROADCAST_TARGET:
self.append_chat(f"Siz (Herkese): {msg_text}")
else:
self.append_chat(f"Siz -> {target}: {msg_text}")
self.msg_entry.set_text("")
except Exception as e:
self.append_chat(f"Hata: Mesaj gönderilemedi! ({str(e)})")
# --- ARKA PLAN DİNLEME THREAD'İ ---
def serial_listen_worker(self):
buffer = bytearray()
while self.running:
try:
# ser.read(1) port timeout'u kadar BLOKLAR -> busy-wait / gereksiz CPU yok.
# Önceki sürümdeki "in_waiting > 0" kontrolü veri yokken sürekli
# dönerek CPU'yu boşuna meşgul ediyordu.
b = self.ser.read(1)
if not b:
continue # timeout, veri yok - tekrar bloklayarak bekle
if b == b'\xAA':
buffer = bytearray()
continue
buffer.extend(b)
try:
unescaped = unescape_bytes(buffer)
if len(unescaped) >= 3:
cmd, size = struct.unpack('<BH', unescaped[0:3])
expected_len = 1 + 2 + size + 2
if len(unescaped) == expected_len:
calc_crc = crc16(unescaped[:-2])
packet_crc, = struct.unpack('<H', unescaped[-2:])
if calc_crc == packet_crc:
self.process_incoming_packet(cmd, unescaped[3:3+size])
else:
self.append_chat("Sistem: CRC Hatası, paket bozuk.")
buffer = bytearray()
except Exception:
pass
except Exception as e:
self.append_chat(f"Sistem: Okuma hatası ({str(e)})")
break
def process_incoming_packet(self, cmd, payload):
if cmd == 0x91:
if len(payload) >= 3:
rssi, snr, sig_rssi = struct.unpack('<bbb', payload[0:3])
data_bytes = payload[3:]
try:
text = data_bytes.decode('utf-8')
except UnicodeDecodeError:
self.append_chat(f"Bilinmeyen Veri (Hex): {data_bytes.hex()}")
return
parts = text.split(ENVELOPE_SEP, 2)
my_name = self.name_entry.get_text().strip()
if len(parts) == 3:
target, sender, message = parts
self.remember_user(sender)
if target == BROADCAST_TARGET:
self.append_chat(f"{sender} (Herkese): {message}")
elif target == my_name:
self.append_chat(f"{sender} -> Size (özel): {message}")
else:
pass # başka bir cihaza yönelik özel mesaj, gösterme
else:
self.append_chat(text)
elif cmd == 0x92:
pass # STDBY_XOSC_RX sayesinde cihaz otomatik RX'e döner
# --- UYGULAMAYI ÇALIŞTIR ---
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="LoRa PyGTK3 Chat")
parser.add_argument('-c', '--config', default=None,
help="config.ini dosya yolu (ör. -c config.ini)")
args = parser.parse_args()
config = load_config(args.config)
win = LoRaChatApp(config)
win.connect("destroy", Gtk.main_quit)
win.show_all()
Gtk.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment