#**********************************************************************************************************************************
# File : HamlibServer.py
# Project : Hamlib Rigtctld.exe GUI
# Description : Sets a rigctld.exe server, with radio preset option
# Date : 01-05-2025
# Authors : Bjorn Pasteuning - PD5DJ
# Website : https://wwww.pd5dj.nl
#
# Version history
# 01-05-2025 : 1.0.0 - Initial basics running
# 19-05-2025 : 1.0.1 - Preset options added
# 10-08-2025 : 1.0.2 - Changed filepath structure
# 28-09-2025 : 1.0.2 - Start and Stop button behaviour changed
# - Point folder to rigctld.exe added, default /hamlib
# 1.0.4 - default hamlib folder fix when path is not valid anymore.
# 1.0.5d - Changed monitor to use frequency poll and ignore RPRT responses. Prevents false disconnects on Icom
# 27-11-2025 : 1.0.5e - Added frame for Hamlib version and showing Server IP
# - FIX - Connection Error handling improved.
# 1.0.5f - ADD - Only one instance allowed to run now.
# 1.0.5g - CHANGE - rigs.ini file format changed, Rig list update added to file menu.
# 1.0.5j - FIX - Improved connection handler. specially for Bluetooth SPP devices
# 10-12-2025 1.0.5k - FIX - Autostart fixed
# 22-12-2025 1.0.5l - CHANGE Notebood tab added
# ADD UDP Frequency output broadcast (N1MM Style)
#**********************************************************************************************************************************
import tkinter as tk
from tkinter import ttk, messagebox, simpledialog
from pathlib import Path
import subprocess
import sys
import serial.tools.list_ports
import configparser
import ctypes
import socket
import logging
import msvcrt
import time
# --- Global debug toggle ---
DEBUG = True # Set to False to disable all console debug prints
def debug(msg):
"""Print debug info only when DEBUG=True."""
if DEBUG:
print(msg)
VERSION = "v1.0.5l"
# Determine application root based on this script's location
SCRIPT_DIR = Path(__file__).resolve().parent
SETTINGS_FOLDER = SCRIPT_DIR / "settings"
DATA_FOLDER = SCRIPT_DIR / "data"
LOG_FOLDER = SCRIPT_DIR / "logs"
SETTINGS_FOLDER.mkdir(parents=True, exist_ok=True)
DATA_FOLDER.mkdir(parents=True, exist_ok=True)
LOG_FOLDER.mkdir(parents=True, exist_ok=True)
SETTINGS_FILE = SETTINGS_FOLDER / "hamlibserver.ini"
RIGS_FILE = DATA_FOLDER / "rigs.ini"
LOG_FILE = LOG_FOLDER / "hamlibserver.log"
LOCKFILE = SCRIPT_DIR / SETTINGS_FOLDER / "_hamlibserver.lck"
logging.basicConfig(
filename=LOG_FILE,
level=logging.DEBUG,
format="%(asctime)s [%(levelname)s] %(message)s",
encoding="utf-8"
)
# --- Global Bluetooth/Serial timing configuration ---
BT_CONNECT_TIMEOUT = 0.7 # Fast timeout for TCP connect to rigctld (localhost)
BT_RECV_TIMEOUT = 3.0 # More relaxed timeout for first 'f' reply from the radio
BT_TEST_RETRIES = 10 # was 10
BT_TEST_DELAY = 0.7 # was 1.0
BT_INITIAL_DELAY = 1.0 # was 4.0
BT_MONITOR_TIMEOUT = 2.0 # was 3.0
BT_MONITOR_INTERVAL = 1000 # was 7000
BT_MONITOR_FAILURES = 3 # was 5
SERVER_STARTUP_GRACE = 8.0 # seconds: during this time, 10061 is treated as "not ready yet"
RADIO_INITIAL_GRACE = 15.0 # Time radio/BT may still be initializing before failure counts
class RigCtlGUI:
def __init__(self, root):
ttk.Style().theme_use("default")
self.root = root
self.root.title(f"Hamlib Server | {VERSION}")
self.root.resizable(False, False)
self.udp_sock = None
self.last_udp_band = None
self.last_udp_freq = None
self.udp_after_id = None
self.proc = None
self.server = None
self.udp_interval_var = tk.IntVar(value=5) # seconds, default
# Detect serial ports
self.serial_ports = [port.device for port in serial.tools.list_ports.comports()]
if not self.serial_ports:
self.serial_ports = ["No ports found"]
# Load rigs.ini FIRST (no GUI interaction here)
self.rigs = self.load_rigs()
# Presets (unchanged)
self.presets = {
"ICOM IC-7300": {
"serial_port": "",
"baudrate": "115200",
"server_port": "4532",
"rig_model": "ICOM IC-7300",
"data_bits": "8",
"stop_bits": "1",
"handshake": "None",
"rts": False,
"dtr": False
},
"ICOM IC-705": {
"serial_port": "",
"baudrate": "115200",
"server_port": "4532",
"rig_model": "ICOM IC-705",
"data_bits": "8",
"stop_bits": "1",
"handshake": "None",
"rts": False,
"dtr": False
},
"ICOM IC-9700": {
"serial_port": "",
"baudrate": "115200",
"server_port": "4532",
"rig_model": "ICOM IC-9700",
"data_bits": "8",
"stop_bits": "1",
"handshake": "None",
"rts": False,
"dtr": False
},
"ICOM IC-7610": {
"serial_port": "",
"baudrate": "115200",
"server_port": "4532",
"rig_model": "ICOM IC-7610",
"data_bits": "8",
"stop_bits": "1",
"handshake": "None",
"rts": False,
"dtr": False
},
"YAESU FTDX-10": {
"serial_port": "",
"baudrate": "38400",
"server_port": "4532",
"rig_model": "YAESU FTDX-10",
"data_bits": "8",
"stop_bits": "1",
"handshake": "None",
"rts": False,
"dtr": False
},
"Custom": {}
}
# Now build all GUI widgets (rig_combo is created here)
self.create_widgets()
# Now it is SAFE to load settings (rig_combo exists)
self.load_settings()
# Autostart server if enabled (delay until GUI + presets are fully ready)
if self.autostart_var.get():
self.root.after(200, self.autostart_safe)
# Display Hamlib version info
self.display_hamlib_version()
# Handle window close
self.root.protocol("WM_DELETE_WINDOW", self.exit_program)
# --- Runtime state flags ---
self.server_active = False
self.radio_ready = False
self.test_after_id = None
self.monitor_after_id = None
def send_udp_radio_status(self, freq_hz):
# Only send when enabled
if not self.udp_enabled_var.get():
debug("[UDP] Skipped send (UDP disabled)")
return
try:
ip = self.udp_ip_var.get().strip()
port = int(str(self.udp_port_var.get()).strip())
if not self.udp_sock:
self.udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# allow broadcast if you ever use 255.255.255.255
self.udp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
xml = (
"\n"
f" {int(freq_hz / 10)}\n"
"\n"
)
self.udp_sock.sendto(xml.encode("utf-8"), (ip, port))
debug(f"[UDP] Sent RadioInfo freq={int(freq_hz)} to {ip}:{port}")
except Exception as e:
debug(f"[UDP] Error sending RadioInfo: {e}")
def udp_periodic_update(self):
if not self.server_active or not self.udp_enabled_var.get():
return
if self.last_udp_freq:
self.send_udp_radio_status(self.last_udp_freq)
interval_ms = max(1, min(30, self.udp_interval_var.get())) * 1000
self.udp_after_id = self.root.after(interval_ms, self.udp_periodic_update)
def update_rigs_list(self):
"""
Regenerate rigs.ini using rigctl.exe --list.
GUI-safe: Only updates combobox if it exists.
"""
rigctl_path = SCRIPT_DIR / "hamlib" / "rigctl.exe"
if not rigctl_path.exists():
messagebox.showerror("Hamlib missing",
f"rigctl.exe not found:\n{rigctl_path}")
return
try:
result = subprocess.run(
[str(rigctl_path), "--list"],
capture_output=True,
text=True,
encoding="utf-8"
)
lines = result.stdout.splitlines()
if not lines:
messagebox.showerror("Hamlib error", "rigctl.exe returned no data.")
return
rigs_by_brand = {}
for line in lines:
line = line.strip()
if not line:
continue
parts = line.split()
if len(parts) < 3:
continue
rigid = parts[0].strip()
brand = parts[1].strip()
model = parts[2].strip()
base_name = f"{brand} {model}"
final_name = base_name
# Ensure name is unique inside brand
existing = rigs_by_brand.get(brand, [])
used_names = [n.lower() for n, _ in existing]
suffix = 2
while final_name.lower() in used_names:
final_name = f"{base_name}-{suffix}"
suffix += 1
rigs_by_brand.setdefault(brand, []).append((final_name, rigid))
# ---- WRITE rigs.ini ----
with open(RIGS_FILE, "w", encoding="utf-8") as f:
for brand in sorted(rigs_by_brand.keys()):
f.write(f"[{brand}]\n")
for name, rigid in sorted(rigs_by_brand[brand]):
f.write(f"{name} = {rigid}\n")
f.write("\n")
# ---- RELOAD rigs from file ----
self.rigs = self.load_rigs()
# ---- UPDATE GUI ONLY IF COMBOBOX EXISTS ----
if hasattr(self, "rig_combo"):
self.rig_combo["values"] = list(self.rigs.keys())
# Select the first rig by default
if self.rigs:
self.rig_var.set(list(self.rigs.keys())[0])
# Show success message (if GUI exists)
if hasattr(self, "root"):
messagebox.showinfo("Rigs updated",
f"Loaded {len(self.rigs)} rig models.")
except Exception as e:
messagebox.showerror("Error", str(e))
def load_rigs(self):
"""
Safely loads rigs.ini with full validation.
If rigs.ini is missing or invalid, automatically regenerate it
by calling update_rigs_list(), then reload.
"""
config = configparser.ConfigParser()
config.optionxform = str # preserve key casing
# If rigs.ini is missing → generate new list
if not RIGS_FILE.exists():
messagebox.showinfo(
"Missing rigs.ini",
"The rigs.ini file is missing.\nA new list will be generated."
)
self.update_rigs_list()
return self.load_rigs()
# First attempt to read
try:
config.read(RIGS_FILE, encoding="utf-8")
except Exception:
messagebox.showerror(
"Invalid rigs.ini",
"Failed to load rigs.ini.\nA new list will be generated."
)
self.update_rigs_list()
return self.load_rigs()
# VALIDATION 1: must contain at least one section
if not config.sections():
messagebox.showerror(
"Invalid rigs.ini",
"The rigs.ini file contains no sections.\nA new list will be generated."
)
self.update_rigs_list()
return self.load_rigs()
rigs = {}
# VALIDATION 2: validate every entry
for section in config.sections():
for key, value in config[section].items():
key_str = key.strip()
val_str = value.strip()
# Key must not be empty
if not key_str:
messagebox.showerror(
"Invalid rigs.ini",
f"Empty rig name found in section [{section}].\n"
"A new list will be generated."
)
self.update_rigs_list()
return self.load_rigs()
# Value must be numeric (Hamlib rig ID)
if not val_str.isdigit():
messagebox.showerror(
"Invalid rigs.ini",
f"Invalid entry in rigs.ini:\n[{section}]\n{key} = {value}\n\n"
"Rig ID must be numeric.\nA new list will be generated."
)
self.update_rigs_list()
return self.load_rigs()
# Check for duplicate rig names (case-insensitive)
if key_str.lower() in (k.lower() for k in rigs):
messagebox.showerror(
"Invalid rigs.ini",
f"Duplicate rig name detected:\n{key_str}\n\n"
"A new list will be generated."
)
self.update_rigs_list()
return self.load_rigs()
rigs[key_str] = val_str
return rigs
#------------------------------------------------------------------------------------------------------------------
# Test if rigctld is really communicating with the radio (with retry for delayed BT connection)
#------------------------------------------------------------------------------------------------------------------
def test_radio_connection(self, retries=BT_TEST_RETRIES, delay=BT_TEST_DELAY):
if not self.server_active:
debug("[test] Aborted: server inactive")
return
port = int(self.port_var.get())
self.server_tcp_ready = False
debug(f"[test] Start on port {port}")
def attempt(n):
if not self.server_active:
debug("[test] Aborted mid-attempt: server inactive")
return
debug(f"[test] Attempt {n}/{retries}")
try:
# --- Fast CONNECT timeout ---
s = socket.create_connection(
("127.0.0.1", port),
timeout=BT_CONNECT_TIMEOUT
)
self.server_tcp_ready = True
debug("[test] TCP connect OK")
# --- RECV phase ---
s.settimeout(BT_RECV_TIMEOUT)
try:
s.sendall(b"f\n")
data = s.recv(64).decode(errors="ignore").strip()
debug(f"[test] Raw reply: '{data}'")
# VALID FREQUENCY CHECK
if data.isdigit() and int(data) > 0:
self.radio_ready = True
debug("[test] RADIO READY (valid freq received)")
logging.info("Radio connected (valid frequency received during test)")
logging.info(f"Radio connected during test, freq={data}")
self.radio_indicator.config(bg="green")
self.status_var.set("Status: Radio connected")
self.status_label.config(fg="green")
self.last_udp_freq = int(data)
self.last_udp_band = None # force immediate send on first ready freq
self.udp_periodic_update()
self.send_udp_radio_status(self.last_udp_freq)
self.last_udp_band = self.determine_band(self.last_udp_freq)
# Start monitor
debug("[test] Starting monitor cycle")
self.monitor_after_id = self.root.after(
BT_MONITOR_INTERVAL,
self.monitor_radio_connection,
BT_MONITOR_INTERVAL,
BT_MONITOR_FAILURES
)
s.close()
return
else:
debug(f"[test] Invalid or error reply ignored: '{data}'")
except socket.timeout:
debug("[test] RECV timeout → radio not ready yet")
s.close()
except socket.timeout:
debug("[test] CONNECT timeout → rigctld not ready yet")
except Exception as e:
debug(f"[test] Exception: {e}")
# --- Retry logic ---
if n < retries and self.server_active:
self.radio_indicator.config(bg="orange")
self.test_after_id = self.root.after(
int(delay * 1000),
lambda: attempt(n + 1)
)
else:
if self.server_tcp_ready and self.server_active:
# Fallback to monitor
debug("[test] No valid freq after test window → fallback to monitor")
self.radio_indicator.config(bg="orange")
self.status_var.set("Status: Waiting for radio...")
self.monitor_after_id = self.root.after(
BT_MONITOR_INTERVAL,
self.monitor_radio_connection,
BT_MONITOR_INTERVAL,
BT_MONITOR_FAILURES
)
else:
debug("[test] FAILED: rigctld unreachable")
# Update UI error state
self.radio_indicator.config(bg="red")
self.status_var.set("Status: Radio test failed")
self.status_label.config(fg="red")
# Fully stop server to reset UI + internal state
debug("[test] Auto-stopping server because radio test failed")
self.stop_server()
attempt(1)
def determine_band(self, freq_hz):
"""
Determine amateur band from frequency in Hz.
Returns band string or None.
"""
bands = [
(1_800_000, 2_000_000, "160m"),
(3_500_000, 4_000_000, "80m"),
(5_200_000, 5_400_000, "60m"),
(7_000_000, 7_300_000, "40m"),
(10_100_000, 10_200_000, "30m"),
(14_000_000, 14_350_000, "20m"),
(18_068_000, 18_168_000, "17m"),
(21_000_000, 21_450_000, "15m"),
(24_890_000, 24_990_000, "12m"),
(28_000_000, 29_700_000, "10m"),
(50_000_000, 54_000_000, "6m"),
(144_000_000, 148_000_000, "2m"),
]
for low, high, band in bands:
if low <= freq_hz <= high:
return band
return None
#------------------------------------------------------------------------------------------------------------------
# Periodically check if the radio connection is still alive (real rig feedback)
#------------------------------------------------------------------------------------------------------------------
def monitor_radio_connection(self, interval=BT_MONITOR_INTERVAL, max_failures=BT_MONITOR_FAILURES):
if not self.server_active:
debug("[monitor] Aborted: server inactive")
return
if not self.proc:
debug("[monitor] No running process")
return
port = int(self.port_var.get())
elapsed = time.time() - self.server_start_time
debug(f"[monitor] Polling (elapsed={elapsed:.1f}s, failures={self.monitor_failures})")
# Allow radio to wake up (BT/COM warmup)
if not self.radio_ready and elapsed < RADIO_INITIAL_GRACE:
debug("[monitor] Radio not ready yet (within grace) → skip failure")
self.monitor_after_id = self.root.after(
interval,
self.monitor_radio_connection,
interval,
max_failures
)
return
got_valid_freq = False
try:
with socket.create_connection(("127.0.0.1", port), timeout=BT_MONITOR_TIMEOUT) as s:
s.sendall(b"f\n")
data = s.recv(128).decode(errors="ignore").strip()
debug(f"[monitor] Raw reply: '{data}'")
# VALID FREQUENCY ONLY
if data.isdigit() and int(data) > 0:
got_valid_freq = True
freq = int(data)
band = self.determine_band(freq)
# first time radio becomes ready
if not self.radio_ready:
self.radio_ready = True
debug("[monitor] RADIO READY (first valid frequency)")
logging.info("Radio connected (first valid frequency during monitor)")
logging.info(f"Radio connected during monitor, freq={data}")
# start periodic updates as soon as radio is ready
self.last_udp_band = None # force an immediate "band change" send below
self.udp_periodic_update()
# always update last freq
self.last_udp_freq = freq
# band change -> immediate send (also on first valid freq)
if band and band != self.last_udp_band:
debug(f"[UDP] Band change detected → immediate send ({self.last_udp_band} → {band})")
self.last_udp_band = band
self.send_udp_radio_status(freq)
except Exception as e:
debug(f"[monitor] Exception: {e}")
# During rigctld warmup → ignore temporary 10061
if "10061" in str(e) and elapsed < SERVER_STARTUP_GRACE:
debug("[monitor] 10061 within server grace → skip failure")
self.monitor_after_id = self.root.after(
interval,
self.monitor_radio_connection,
interval,
max_failures
)
return
# --- evaluate result ---
if got_valid_freq:
self.monitor_failures = 0
self.radio_indicator.config(bg="green")
self.status_var.set("Status: Radio connected")
self.status_label.config(fg="green")
self.monitor_after_id = self.root.after(
interval,
self.monitor_radio_connection,
interval,
max_failures
)
return
# No valid freq
self.monitor_failures += 1
debug(f"[monitor] Failure count = {self.monitor_failures}")
# Not yet max → retry
if self.monitor_failures < max_failures:
self.radio_indicator.config(bg="orange")
self.status_var.set("Status: Checking radio...")
self.monitor_after_id = self.root.after(
interval,
self.monitor_radio_connection,
interval,
max_failures
)
return
# ---- MAX FAILURES → RADIO DISCONNECTED ----
debug("[monitor] Max failures → RADIO DISCONNECTED")
self.radio_indicator.config(bg="red")
self.status_var.set("Status: Radio disconnected")
self.status_label.config(fg="red")
debug("[monitor] Stopping server")
self.stop_server()
#------------------------------------------------------------------------------------------------------------------
# Create all UI frames and widgets
#------------------------------------------------------------------------------------------------------------------
def create_widgets(self):
self.menubar = tk.Menu(self.root)
self.root.config(menu=self.menubar)
# ===== File menu =====
file_menu = tk.Menu(self.menubar, tearoff=0)
self.menubar.add_cascade(label="File", menu=file_menu)
file_menu.add_command(label="Exit", command=self.exit_program)
# ===== Tools menu =====
tools_menu = tk.Menu(self.menubar, tearoff=0)
self.menubar.add_cascade(label="Tools", menu=tools_menu)
tools_menu.add_command(label="Update Rigs List", command=self.update_rigs_list)
# ===== Help menu =====
help_menu = tk.Menu(self.menubar, tearoff=0)
self.menubar.add_cascade(label="Help", menu=help_menu)
help_menu.add_command(label="About", command=self.show_about)
# === Notebook ===
self.notebook = ttk.Notebook(self.root)
self.notebook.grid(row=0, column=0, padx=10, pady=5, sticky="ew")
# Tabs
tab_preset = ttk.Frame(self.notebook)
tab_udp = ttk.Frame(self.notebook)
tab_options = ttk.Frame(self.notebook)
self.notebook.add(tab_preset, text="Rig Preset Settings")
self.notebook.add(tab_udp, text="UDP Radio Status")
self.notebook.add(tab_options, text="Options")
# === Frame 1: Presets & Rig Settings ===
preset_frame = ttk.LabelFrame(tab_preset, text="Rig Preset Settings")
preset_frame.pack(fill="x", padx=10, pady=5)
# Buttons New / Save / Delete
button_frame = tk.Frame(preset_frame)
button_frame.pack(fill="x", pady=3)
self.new_preset_button = tk.Button(button_frame, text="New Preset", command=self.create_new_preset)
self.new_preset_button.pack(side="left", expand=True, fill="x", padx=2)
self.save_preset_button = tk.Button(button_frame, text="Save Preset", command=self.save_to_preset)
self.save_preset_button.pack(side="left", expand=True, fill="x", padx=2)
self.delete_preset_button = tk.Button(button_frame, text="Delete Preset", command=self.delete_preset)
self.delete_preset_button.pack(side="left", expand=True, fill="x", padx=2)
# Preset selection
self.preset_var = tk.StringVar()
preset_sel_frame = tk.Frame(preset_frame)
preset_sel_frame.pack(fill="x", pady=2)
tk.Label(preset_sel_frame, text="Preset:", width=12, anchor="w").pack(side="left")
self.preset_combo = ttk.Combobox(preset_sel_frame, textvariable=self.preset_var, values=list(self.presets.keys()), state="readonly")
self.preset_combo.pack(side="left", fill="x", expand=True)
self.preset_combo.bind("<>", lambda e: self.load_selected_preset())
self.rename_preset_button = tk.Button(preset_sel_frame, text="Rename", command=self.rename_preset)
self.rename_preset_button.pack(side="left", padx=3)
# Serial Port
self.serial_var = tk.StringVar()
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Serial Port:", width=12, anchor="w").pack(side="left")
self.serial_combo = ttk.Combobox(row, textvariable=self.serial_var, values=self.serial_ports)
self.serial_combo.pack(side="left", fill="x", expand=True)
# Baudrate
self.baud_var = tk.StringVar(value="115200")
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Baudrate:", width=12, anchor="w").pack(side="left")
self.baud_combo = ttk.Combobox(row, textvariable=self.baud_var, values=["4800", "9600", "14400", "19200", "38400", "57600", "115200"])
self.baud_combo.pack(side="left", fill="x", expand=True)
# Server Port
self.port_var = tk.StringVar(value="4532")
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Server Port:", width=12, anchor="w").pack(side="left")
self.port_combo = ttk.Combobox(row, textvariable=self.port_var, values=["4532", "4536", "4538", "4540"])
self.port_combo.pack(side="left", fill="x", expand=True)
# Rig Model
self.rig_var = tk.StringVar()
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Rig Model:", width=12, anchor="w").pack(side="left")
self.rig_combo = ttk.Combobox(row, textvariable=self.rig_var, values=list(self.rigs.keys()))
self.rig_combo.pack(side="left", fill="x", expand=True)
# Data Bits
self.databits_var = tk.StringVar(value="8")
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Data Bits:", width=12, anchor="w").pack(side="left")
tk.Radiobutton(row, text="7", variable=self.databits_var, value="7").pack(side="left")
tk.Radiobutton(row, text="8", variable=self.databits_var, value="8").pack(side="left")
# Stop Bits
self.stopbits_var = tk.StringVar(value="1")
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Stop Bits:", width=12, anchor="w").pack(side="left")
tk.Radiobutton(row, text="1", variable=self.stopbits_var, value="1").pack(side="left")
tk.Radiobutton(row, text="2", variable=self.stopbits_var, value="2").pack(side="left")
# Handshake
self.handshake_var = tk.StringVar(value="None")
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Handshake:", width=12, anchor="w").pack(side="left")
tk.Radiobutton(row, text="None", variable=self.handshake_var, value="None").pack(side="left")
tk.Radiobutton(row, text="RTS/CTS", variable=self.handshake_var, value="RTSCTS").pack(side="left")
tk.Radiobutton(row, text="XON/XOFF", variable=self.handshake_var, value="XONXOFF").pack(side="left")
# RTS/DTR
self.rts_var = tk.BooleanVar()
self.dtr_var = tk.BooleanVar()
row = tk.Frame(preset_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="RTS/DTR:", width=12, anchor="w").pack(side="left")
tk.Checkbutton(row, text="RTS ON", variable=self.rts_var).pack(side="left")
tk.Checkbutton(row, text="DTR ON", variable=self.dtr_var).pack(side="left")
# === Frame: UDP Radio Status ===
udp_frame = ttk.LabelFrame(tab_udp, text="UDP Radio Status")
udp_frame.pack(fill="x", padx=10, pady=5)
self.udp_enabled_var = tk.BooleanVar(value=False)
self.udp_ip_var = tk.StringVar(value="127.0.0.1")
self.udp_port_var = tk.StringVar(value="12060")
self.udp_interval_var = tk.IntVar(value=5)
# Enable UDP
row = tk.Frame(udp_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Enabled:", width=12, anchor="w").pack(side="left")
tk.Checkbutton(
row,
text="Enable UDP Radio Status output",
variable=self.udp_enabled_var
).pack(side="left")
# IP address
row = tk.Frame(udp_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="IP Address:", width=12, anchor="w").pack(side="left")
tk.Entry(row, textvariable=self.udp_ip_var).pack(side="left", fill="x", expand=True)
# Port
row = tk.Frame(udp_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Port:", width=12, anchor="w").pack(side="left")
tk.Entry(row, textvariable=self.udp_port_var, width=10).pack(side="left")
# Update interval
row = tk.Frame(udp_frame)
row.pack(fill="x", pady=1)
tk.Label(row, text="Update every:", width=12, anchor="w").pack(side="left")
self.udp_interval_entry = tk.Spinbox(
row,
from_=1,
to=30,
width=6,
textvariable=self.udp_interval_var
)
self.udp_interval_entry.pack(side="left")
tk.Label(row, text="seconds (1-30)").pack(side="left", padx=6)
# === Frame 2: Server Control ===
control_frame = ttk.LabelFrame(self.root, text="Server Control")
control_frame.grid(row=1, column=0, padx=10, pady=5, sticky="ew")
self.start_button = tk.Button(control_frame, text="Start Server", command=self.start_server)
self.start_button.pack(side="left", expand=True, fill="x", padx=5, pady=3)
self.stop_button = tk.Button(control_frame, text="Stop Server", command=self.stop_server, state="disabled")
self.stop_button.pack(side="left", expand=True, fill="x", padx=5, pady=3)
# === Frame 3: Autostart ===
auto_frame = ttk.LabelFrame(tab_options, text="Startup Options")
auto_frame.pack(fill="x", padx=10, pady=5)
self.autostart_var = tk.BooleanVar()
self.autostart_check = tk.Checkbutton(auto_frame, text="Autostart Server on startup", variable=self.autostart_var)
self.autostart_check.pack(anchor="w")
# === Frame 4: Status ===
status_frame = tk.Frame(self.root)
status_frame.grid(row=2, column=0, padx=10, pady=5, sticky="ew")
self.status_var = tk.StringVar(value="Status: Server stopped")
self.status_label = tk.Label(status_frame, textvariable=self.status_var, anchor="w", fg="red")
self.status_label.pack(side="left", expand=True, fill="x")
# Radio connection indicator between status and exit
self.radio_indicator = tk.Label(status_frame, width=2, height=1, bg="red", relief="groove")
self.radio_indicator.pack(side="left", padx=5)
self.exit_button = tk.Button(status_frame, text="Exit", command=self.exit_program, width=5)
self.exit_button.pack(side="left", padx=5)
# === Frame 5: Server Info ===
info_frame = ttk.LabelFrame(self.root, text="Hamlib Server Info")
info_frame.grid(row=3, column=0, padx=10, pady=5, sticky="ew")
self.version_label = tk.Label(info_frame, text="", anchor="w")
self.version_label.pack(fill="x", padx=5, pady=2)
self.ip_label = tk.Label(info_frame, text="Server IP:", anchor="w")
self.ip_label.pack(fill="x", padx=5, pady=2)
def show_about(self):
about_text = (
"Hamlib Server GUI\n\n"
f"Version: {VERSION}\n\n"
"Author: Bjorn Pasteuning (PD5DJ)\n"
"Website: https://www.pd5dj.nl\n\n"
"GUI for controlling rigctld.exe with\n"
"preset management and UDP radio status output.\n\n"
"© PD5DJ"
)
messagebox.showinfo(
"About Hamlib Server",
about_text
)
#------------------------------------------------------------------------------------------------------------------
# Show Hamlib version at bottom (rigctld -V)
#------------------------------------------------------------------------------------------------------------------
def display_hamlib_version(self):
rigctld_path = SCRIPT_DIR / "hamlib" / "rigctld.exe"
if not rigctld_path.exists():
self.version_label.config(text="Hamlib version: rigctld.exe not found")
return
try:
result = subprocess.run([str(rigctld_path), "-V"], capture_output=True, text=True, timeout=2)
version_line = result.stdout.strip().splitlines()[0] if result.stdout else "Unknown version"
self.version_label.config(text=version_line)
except Exception as e:
self.version_label.config(text=f"Hamlib version check failed: {e}")
def update_server_ip(self):
"""Detect local IP address and show it in the GUI."""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80)) # external pingless trick
local_ip = s.getsockname()[0]
s.close()
except Exception:
local_ip = "Unknown"
self.ip_label.config(text=f"Server IP: {local_ip}")
logging.info(f"Server IP detected: {local_ip}")
def disable_preset_controls(self):
self.new_preset_button.config(state="disabled")
self.save_preset_button.config(state="disabled")
self.delete_preset_button.config(state="disabled")
self.rename_preset_button.config(state="disabled")
self.preset_combo.config(state="disabled")
def enable_preset_controls(self):
self.new_preset_button.config(state="normal")
self.save_preset_button.config(state="normal")
self.delete_preset_button.config(state="normal")
self.rename_preset_button.config(state="normal")
self.preset_combo.config(state="readonly")
def disable_all_inputs(self):
for frame in self.root.winfo_children():
for widget in frame.winfo_children() if isinstance(frame, tk.Frame) else [frame]:
if isinstance(widget, (tk.Entry, ttk.Combobox, tk.Checkbutton, tk.Radiobutton)):
widget.configure(state="disabled")
def enable_all_inputs(self):
for frame in self.root.winfo_children():
for widget in frame.winfo_children() if isinstance(frame, tk.Frame) else [frame]:
if isinstance(widget, (tk.Entry, ttk.Combobox, tk.Checkbutton, tk.Radiobutton)):
widget.configure(state="normal")
def start_server(self):
debug("[start_server] Called")
# Save all settings when starting server (same logic as exit)
try:
self.save_settings()
except Exception as e:
logging.error(f"Failed to save settings on start: {e}")
if self.proc:
debug("[start_server] Process already running")
return
try:
rig_name = self.rig_var.get()
rig_id = self.rigs.get(rig_name, "UNKNOWN")
cmd = self.build_command()
debug(f"[start_server] Command: {cmd}")
CREATE_NO_WINDOW = 0x08000000 if sys.platform == "win32" else 0
self.proc = subprocess.Popen(cmd, creationflags=CREATE_NO_WINDOW)
debug(f"[start_server] rigctld launched for '{rig_name}' (ID {rig_id})")
# --- Reset runtime flags ---
self.server_active = True
self.radio_ready = False
self.monitor_failures = 0
self.test_after_id = None
self.monitor_after_id = None
# Timestamp for grace windows
self.server_start_time = time.time()
debug(f"[start_server] server_start_time={self.server_start_time:.3f}")
# UI updates
self.status_var.set("Status: Server started")
self.status_label.config(fg="green")
self.update_server_ip()
self.start_button.config(state="disabled")
self.stop_button.config(state="normal")
self.disable_all_inputs()
self.disable_preset_controls()
# Start test cycle
debug(f"[start_server] Scheduling first test in {BT_INITIAL_DELAY}s")
self.test_after_id = self.root.after(
int(BT_INITIAL_DELAY * 1000),
lambda: self.test_radio_connection()
)
except Exception as e:
debug(f"[start_server] ERROR: {e}")
self.server_active = False
messagebox.showerror("Error Starting Server", str(e))
def stop_server(self):
logging.info("Stopping rigctld server...")
debug("[stop_server] Called")
# --- Mark server as inactive and cancel pending after callbacks ---
self.server_active = False
debug("[stop_server] server_active set to False")
# Cancel pending test/monitor callbacks if any
if self.test_after_id is not None:
try:
self.root.after_cancel(self.test_after_id)
debug("[stop_server] Cancelled pending test_radio_connection callback")
except Exception as e:
debug(f"[stop_server] Error cancelling test_after_id: {e}")
self.test_after_id = None
if self.monitor_after_id is not None:
try:
self.root.after_cancel(self.monitor_after_id)
debug("[stop_server] Cancelled pending monitor_radio_connection callback")
except Exception as e:
debug(f"[stop_server] Error cancelling monitor_after_id: {e}")
self.monitor_after_id = None
# -------- rigctld process handling --------
if self.proc:
try:
if self.proc.poll() is None:
debug("[stop_server] Process is running, sending terminate()")
self.proc.terminate()
try:
self.proc.wait(timeout=1.0)
debug("[stop_server] Process terminated gracefully")
except subprocess.TimeoutExpired:
logging.warning("rigctld not exiting gracefully, forcing kill...")
debug("[stop_server] TimeoutExpired, forcing kill()")
self.proc.kill()
else:
debug("[stop_server] Process already not running (poll() != None)")
except Exception as e:
logging.exception("Error while stopping rigctld: %s", e)
debug(f"[stop_server] Exception while stopping process: {e}")
finally:
self.proc = None
debug("[stop_server] self.proc set to None")
# -------- kill TCP listener and client sockets --------
if getattr(self, "server", None):
debug("[stop_server] Closing TCP listener/server socket")
try:
self.server.shutdown()
self.server.server_close()
except Exception as e:
debug(f"[stop_server] Exception while closing server socket: {e}")
finally:
self.server = None
debug("[stop_server] self.server set to None")
# -------- UI updates --------
self.status_var.set("Status: Server stopped")
self.status_label.config(fg="red")
self.radio_indicator.config(bg="red")
self.ip_label.config(text="Server IP:")
debug("[stop_server] UI updated: Status=Server stopped, indicators red")
self.enable_all_inputs()
self.enable_preset_controls()
debug("[stop_server] Inputs and preset controls enabled")
# always enforce button states:
self.start_button.config(state="normal")
self.stop_button.config(state="disabled")
debug("[stop_server] Button state: start=normal, stop=disabled")
logging.info("Hamlib server fully stopped.")
debug("[stop_server] Hamlib server fully stopped")
if self.udp_after_id:
self.root.after_cancel(self.udp_after_id)
self.udp_after_id = None
if self.udp_sock:
self.udp_sock.close()
self.udp_sock = None
def exit_program(self):
if messagebox.askyesno("Exit", "Are you sure you want to exit?"):
# Try to save settings (including autostart and last_preset)
try:
self.save_settings()
except Exception as e:
logging.error(f"Failed to save settings on exit: {e}")
if self.proc:
self.stop_server()
self.root.destroy()
def build_command(self):
rig_number = self.rigs.get(self.rig_var.get(), "3078")
rts_state = "ON" if self.rts_var.get() else "OFF"
dtr_state = "ON" if self.dtr_var.get() else "OFF"
serial_parity = "None"
serial_handshake = self.handshake_var.get()
# Always use hamlib folder relative to the script location
rigctld_folder = SCRIPT_DIR / "hamlib"
executable = rigctld_folder / "rigctld.exe"
# Ensure hamlib folder exists (safety)
if not rigctld_folder.exists():
rigctld_folder.mkdir(parents=True, exist_ok=True)
# If the executable is missing, show a clear error popup with full path
if not executable.exists():
messagebox.showerror(
"Error Starting Server",
f"rigctld.exe not found in default folder:\n\n{executable}\n\n"
"Please place rigctld.exe in this folder and restart the app."
)
raise FileNotFoundError(f"rigctld.exe not found in default folder: {executable}")
return [
str(executable),
"-t", self.port_var.get(),
"-m", rig_number,
"-r", self.serial_var.get(),
"-s", self.baud_var.get(),
"--set-conf=data_bits={},stop_bits={},serial_parity={},serial_handshake={},dtr_state={},rts_state={}".format(
self.databits_var.get(), self.stopbits_var.get(), serial_parity, serial_handshake, dtr_state, rts_state
)
]
def delete_preset(self):
name = self.preset_var.get()
if name not in self.presets:
messagebox.showwarning("Delete Preset", "No valid preset selected.")
return
if messagebox.askyesno("Delete Preset", f"Are you sure you want to delete preset '{name}'?"):
del self.presets[name]
self.preset_combo['values'] = list(self.presets.keys())
if self.presets:
self.preset_var.set(list(self.presets.keys())[0])
else:
self.preset_var.set("")
self.save_settings()
messagebox.showinfo("Deleted", f"Preset '{name}' deleted.")
def create_new_preset(self):
new_name = simpledialog.askstring("New Preset", "Enter name for new preset:")
if not new_name or new_name in self.presets:
messagebox.showerror("Error", "Invalid or duplicate preset name.")
return
self.presets[new_name] = {
'serial_port': self.serial_var.get(),
'baudrate': self.baud_var.get(),
'server_port': self.port_var.get(),
'rig_model': self.rig_var.get(),
'data_bits': self.databits_var.get(),
'stop_bits': self.stopbits_var.get(),
'handshake': self.handshake_var.get(),
'rts': self.rts_var.get(),
'dtr': self.dtr_var.get()
}
self.preset_combo['values'] = list(self.presets.keys())
self.preset_var.set(new_name)
self.save_settings()
messagebox.showinfo("New Preset", f"Preset '{new_name}' created.")
def save_settings(self):
config = configparser.ConfigParser()
config['Settings'] = {
'autostart': str(self.autostart_var.get()),
'last_preset': self.preset_var.get(),
'udp_enabled': str(self.udp_enabled_var.get()),
'udp_ip': self.udp_ip_var.get(),
'udp_port': self.udp_port_var.get(),
'udp_interval': str(self.udp_interval_var.get())
}
if not config.has_section("Presets"):
config.add_section("Presets")
for name, p in self.presets.items():
config["Presets"][name] = '|'.join([
p.get("serial_port", ""), p.get("baudrate", ""), p.get("server_port", ""),
p.get("rig_model", ""), p.get("data_bits", ""), p.get("stop_bits", ""),
p.get("handshake", ""), str(p.get("rts", False)), str(p.get("dtr", False))
])
with open(SETTINGS_FILE, 'w') as f:
config.write(f)
def load_settings(self):
"""
Load settings from settings/hamlibserver.ini (next to the script).
Always ensure a hamlib folder exists under the script directory.
"""
# Default hamlib folder in the root of the app (script directory)
default_rigctld_folder = SCRIPT_DIR / "hamlib"
# Ensure hamlib folder exists so the UI/search path is consistent
if not default_rigctld_folder.exists():
default_rigctld_folder.mkdir(parents=True, exist_ok=True)
# If settings file doesn't exist, nothing to load (defaults remain)
if not SETTINGS_FILE.exists():
return
config = configparser.ConfigParser()
config.read(SETTINGS_FILE)
# --- Settings section ---
if config.has_section('Settings'):
self.autostart_var.set(config.getboolean('Settings', 'autostart', fallback=False))
last_preset = config.get('Settings', 'last_preset', fallback='')
self.udp_enabled_var.set(config.getboolean('Settings', 'udp_enabled', fallback=False))
self.udp_ip_var.set(config.get('Settings', 'udp_ip', fallback="127.0.0.1"))
self.udp_port_var.set(config.get('Settings', 'udp_port', fallback="12060"))
self.udp_interval_var.set(max(1, min(30, config.getint('Settings', 'udp_interval', fallback=5))))
else:
self.autostart_var.set(False)
last_preset = ''
# --- Presets section ---
if config.has_section('Presets'):
self.presets = {}
for name, val in config['Presets'].items():
parts = val.split('|')
if len(parts) == 9:
self.presets[name] = {
'serial_port': parts[0],
'baudrate': parts[1],
'server_port': parts[2],
'rig_model': parts[3],
'data_bits': parts[4],
'stop_bits': parts[5],
'handshake': parts[6],
'rts': parts[7] == 'True',
'dtr': parts[8] == 'True'
}
else:
print(f"⚠️ Preset '{name}' is malformed. Skipping.")
self.preset_combo['values'] = list(self.presets.keys())
# Restore and load the last used preset
if last_preset in self.presets:
self.preset_var.set(last_preset)
self.preset_combo.set(last_preset)
self.load_selected_preset()
elif self.presets:
fallback = list(self.presets.keys())[0]
self.preset_var.set(fallback)
self.preset_combo.set(fallback)
self.load_selected_preset()
def autostart_safe(self):
"""Start server only after GUI widgets and presets are fully initialized."""
try:
# Make sure the selected preset is really applied to all variables
self.load_selected_preset()
except Exception as e:
logging.error(f"Autostart: failed to load selected preset: {e}")
# Optional safety: do not autostart when required fields are empty
if not self.serial_var.get() or not self.rig_var.get():
self.status_var.set("Status: Autostart cancelled (incomplete preset)")
self.status_label.config(fg="red")
return
self.status_var.set("Status: Starting server...")
self.status_label.config(fg="green")
self.start_server()
def load_selected_preset(self):
"""Load the selected preset and auto-correct outdated rig names."""
name = self.preset_var.get()
if name not in self.presets:
messagebox.showwarning("Preset", "Preset not found.")
return
p = self.presets[name]
rig_name = p.get('rig_model', '')
# --- Auto-correct rig name if rigs.ini was updated ---
if rig_name not in self.rigs:
# Try to match by substring (case-insensitive)
corrected = None
for key in self.rigs.keys():
if rig_name.lower().replace("-", "").replace(" ", "") in key.lower().replace("-", "").replace(" ", ""):
corrected = key
break
if corrected:
rig_name = corrected
p['rig_model'] = corrected
# Optional: immediately resave corrected preset
self.save_settings()
# --- Apply preset values ---
self.serial_var.set(p.get('serial_port', ''))
self.baud_var.set(p.get('baudrate', ''))
self.port_var.set(p.get('server_port', ''))
self.rig_var.set(rig_name)
self.databits_var.set(p.get('data_bits', '8'))
self.stopbits_var.set(p.get('stop_bits', '1'))
self.handshake_var.set(p.get('handshake', 'None'))
self.rts_var.set(p.get('rts', False))
self.dtr_var.set(p.get('dtr', False))
def save_to_preset(self):
name = self.preset_var.get()
if not name:
messagebox.showwarning("Preset", "No preset selected")
return
self.presets[name] = {
'serial_port': self.serial_var.get(),
'baudrate': self.baud_var.get(),
'server_port': self.port_var.get(),
'rig_model': self.rig_var.get(),
'data_bits': self.databits_var.get(),
'stop_bits': self.stopbits_var.get(),
'handshake': self.handshake_var.get(),
'rts': self.rts_var.get(),
'dtr': self.dtr_var.get()
}
self.save_settings()
messagebox.showinfo("Preset Saved", f"Preset '{name}' saved.")
def rename_preset(self):
old = self.preset_var.get()
if old not in self.presets: return
new = simpledialog.askstring("Rename Preset", f"Enter new name for '{old}':")
if new and new.strip() and new not in self.presets:
self.presets[new] = self.presets.pop(old)
self.preset_combo['values'] = list(self.presets.keys())
self.preset_var.set(new)
self.save_settings()
def ensure_single_instance():
"""
Windows file lock based single instance.
Works with python.exe, pythonw.exe and as EXE.
"""
global _lock_handle
_lock_handle = open(LOCKFILE, "w")
try:
# Try to lock the file exclusively
msvcrt.locking(_lock_handle.fileno(), msvcrt.LK_NBLCK, 1)
except OSError:
# always show popup, even with pythonw
ctypes.windll.user32.MessageBoxW(
None,
"HamlibServer is already running.",
"Hamlib Server",
0x10
)
sys.exit(0)
if __name__ == "__main__":
ensure_single_instance()
root = tk.Tk()
app = RigCtlGUI(root)
root.mainloop()