zx/quisk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added UT3MK SDR radios

Browse Source
This commit is contained in:
zx
2026-10-18 22:02:37 +03:00
parent c97ffa155e
commit d39722b170
4 changed files with 541 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ut3mkpkg/quisk_hardware.py
+523
View File
@@ -0,0 +1,523 @@
# -*- coding: utf-8 -*-
from __future__ import print_function
import serial, time, logging
import re
import _quisk as QS
import quisk_conf_defaults as conf
import sys, wx, traceback
# ── ПОСЛЕДОВАТЕЛЬНЫЙ ПОРТ ДЛЯ CAT ───────────────────────────────────
sdr1000_serial_port = "/dev/ttyUSB0" # Windows: "COM3" / Linux: "/dev/ttyUSB0"
sdr1000_serial_rate = 57600 # PowerSDR CAT: до 115200
# ── ЛОГ ────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-5s %(message)s",
datefmt="%H:%M:%S",
)
log = logging.getLogger("SDR1000")
Mode2Index = {'CWL':0, 'CWU':1, 'LSB':2, 'USB':3, 'AM':4,
'FM':5, 'EXT':6, 'DGT-U':7, 'DGT-L':8, 'DGT-IQ':9,
'IMD':10, 'FDV-U':11, 'FDV-L':12, 'DGT-FM':13}
# ── КЛАСС УПРАВЛЕНИЯ ЖЕЛЕЗОМ ────────────────────────────────────────
from quisk_hardware_model import Hardware as BaseHardware
class Hardware(BaseHardware):
"""
Quisk Hardware-класс для FlexRadio SDR-1000.
Использует CAT-команды протокола PowerSDR (ZZ + Kenwood).
"""
# Буфер для приема данных с com-порта
serialbuf = '';
# Буфер для отправки команд на com-порт
cmdbuf = '';
# Буфер отправки IF
statusbuf ='';
statussending = False;
# Сопоставление размера шагов
STEP_MAP = {
'1': '0000', '10': '0001', '50': '1000', '100': '0010',
'250': '1001', '500': '1010', '1k': '0011', '5k': '1011',
'9k': '1100', '10k': '0100', '100k':'0101', '1M': '0110',
'10M': '0111',
}
STEP_VAL = {
'1': 1,
'10': 10,
'50': 50,
'100': 100,
'250': 250,
'500': 500,
'1k': 1000,
'5k': 5000,
'9k': 9000,
'10k': 10000,
'100k': 100000,
'1M': 1000000,
'10M': 10000000,
}
# STEP_ENA = ['1', '10', '50', '100', '250', '500', '1k', '5k', '10k', '100k' ]
STEP_ENA = [ '10', '50', '100', '250', '500', '1k', '5k', '10k', '100k' ]
BAND_ENA = ['80', '40', '30', '20', '17', '15', '12', '10']
watchdog = 0
watchdog_limit = 100
# Сопоставление режимов Quisk → код MD (PowerSDR/Kenwood)
# TODO
MODE_MAP = {
'LSB': '1', 'USB': '2', 'CWU': '3', 'FM': '4',
'AM': '5', 'DGT-L': '6', 'CWL': '7',
# Quisk-специфичные → стандартные
'DGT-U':'2', 'FMN': '4', 'AMN': '5',
'DGT-FM': '4', 'FDV-U':'2', 'FDV-L':'1',
'FSK' : '9', 'FSKR' : '2',
}
def __init__(self, app, conf):
BaseHardware.__init__(self, app, conf)
self.ser = None
self.current_freq = 0
self.current_vfo = 0
self.current_mode = 'USB'
self.current_band = '80'
self.current_step = '50'
self.rf_gain_labels = ('RF -20', 'RF -10', 'RF 0', 'RF +10' )
self.current_pre = 0
self.current_ritfreq = 0
self.rit_active = False
self.ptt_active = False
# ── Открытие/закрытие ────────────────────────────────────────────
def open(self):
"""Вызывается один раз при старте Quisk."""
try:
self.ser = serial.Serial(
sdr1000_serial_port,
sdr1000_serial_rate,
timeout=0, xonxoff=True,
)
log.debug("SDR-1000: COM-порт открыт %s @ %d",
sdr1000_serial_port, sdr1000_serial_rate)
except serial.SerialException as e:
log.error("SDR-1000: не удалось открыть порт: %s", e)
self.ser = None
return "SDR-1000: ошибка порта %s" % sdr1000_serial_port
msg = "SDR-1000 подключен (%s, %s)" % (
sdr1000_serial_port, "PowerSDR")
return msg
def close(self):
"""Вызывается один раз при завершении Quisk."""
if self.ptt_active:
self.ser.cmdbuf += 'RX;' # снять передачу
if self.ser and self.ser.is_open:
self.cmdbuf = ''
self.ser.flush
self.ser.close()
log.debug("SDR-1000: порт закрыт")
# ── Низкоуровневый CAT-обмен ────────────────────────────────────
def _cat(self):
"""
Вызываться будет периодически для чтения порта
"""
if not self.ser or not self.ser.is_open:
return ""
try:
#self.ser.write(frame)
if self.ser.in_waiting > 0:
sdata = self.ser.read(size=self.ser.in_waiting)
self.serialbuf += sdata.decode('ascii')
else:
self.watchdog += 1
if self.watchdog > self.watchdog_limit:
log.debug("Watchdog! Resetting arduino")
self.ser.dtr = 0
time.sleep(0.1)
self.ser.dtr = 1
self.ser.flush
self.ser.close()
time.sleep(0.1)
try:
self.ser = serial.Serial(
sdr1000_serial_port,
sdr1000_serial_rate,
timeout=0, xonxoff=True,
)
log.debug("SDR-1000: COM-порт открыт %s @ %d",
sdr1000_serial_port, sdr1000_serial_rate)
except serial.SerialException as e:
log.error("SDR-1000: не удалось открыть порт: %s", e)
self.ser = None
return "SDR-1000: ошибка порта %s" % sdr1000_serial_port
self.ser.flushInput()
self.ser.flushOutput()
self.watchdog = 0
ans = ''
while ';' in self.serialbuf:
cmds = self.serialbuf.partition(';')
cmd = cmds[0]
self.serialbuf = cmds[-1]
# log.info("Splitted to %s and %s", cmd, self.serialbuf)
if len(cmd) > 0:
answer = ''
cmddone = 0
match cmd:
case 'IF':
# Запрос полного статуса
# Возврат в statusbuf
self.statusbuf = self.ReturnInfo()
log.debug("Status refresh, buf: %s", self.serialbuf)
cmddone = 1
case 'ZZSM0':
# Запрос показаний S-метра
ans = self.ReturnSmeterA()
log.debug("S-meter refresh, buf: %s", ans)
cmddone = 1
case 'PS1':
# Включение
ans = 'PS1;'
log.debug("PowerON, buf: %s", ans)
cmddone = 1
case 'ZZPA':
# Запрос показаний PRE ! TODO смена режима
ans = self.ReturnPre()
log.debug("PRE answer, buf: %s", ans)
cmddone = 1
case 'UP':
# Запрос на увеличение частоты
self.application.ChangeHwFrequency(9000,
self.current_vfo+self.STEP_VAL[self.current_step])
cmddone = 1
case 'DN':
# Запрос на уменьшение частоты
self.application.ChangeHwFrequency(9000,
self.current_vfo-self.STEP_VAL[self.current_step])
cmddone = 1
case 'ZZSU':
# Запрос на увеличение шага
self.StepInc()
cmddone = 1
case 'ZZSD':
# Запрос на уменьшение шага
self.StepDec()
cmddone = 1
case 'BU':
# Запрос смены диапазона
self.BandUp()
cmddone = 1
case 'BD':
# Запрос смены диапазона
self.BandDn()
cmddone = 1
case 'TX':
# Transmit
QS.set_PTT(1)
self.ptt_active = True
log.info("PTT ON")
cmddone = 1
case 'RX':
# Recieve
QS.set_PTT(0)
self.ptt_active = False
log.info("PTT OFF")
cmddone = 1
case 'ZZQR':
# Запрос восстановление QuickSave
cmddone = 1
if len(cmd) > 4 and not cmddone:
log.debug("Try to match cmd: %s, buf: %s", cmd[0:4], self.serialbuf)
match cmd[0:4]:
case 'ZZPA':
# Запрос изменения PRE
self.SetPre(cmd)
ans = self.ReturnPre()
log.debug("PRE set, buf: %s", ans)
cmddone = 1
case 'ZZMD':
# Запрос изменения режима
self.SetMode(cmd)
log.debug("Mode set, buf: %s", cmd)
cmddone = 1
case 'FA00':
# Запрос изменения частоты
self.SetFreq(cmd)
log.debug("Freq set, buf: %s", cmd)
cmddone = 1
if not cmddone:
log.info("unknown cmd: %s, buf: %s", cmd, self.serialbuf)
ans = '?;'
if len(ans) > 0:
log.debug("Sending answer, buf: %s", ans)
self.ser.write(ans.encode('ascii'))
if len(self.serialbuf) == 0:
if len(self.cmdbuf) > 0:
cmds = self.cmdbuf.partition(';')
cmd = cmds[0]
self.cmdbuf = cmds[-1]
log.debug("Sending cmd %s, buf: %s", cmd, self.cmdbuf)
self.ser.write(cmd.encode('ascii'))
if (len(self.statusbuf) > 0) and not self.statussending:
self.statussending = True
log.debug("Sending status, buf: %s", self.statusbuf)
self.ser.write(self.statusbuf.encode('ascii'))
self.statusbuf = ''
self.statussending = False
# End of CAT
except Exception as e:
log.error("SDR-1000 CAT error: %s", e)
return ""
@staticmethod
def _fmt_freq(hz):
"""Частота → 11 цифр: 14200000 → '00014200000'."""
return "%011d" % int(hz)
# ── Quisk API: частота ──────────────────────────────────────────
def ChangeFrequency(self, tune, vfo, source='', band='', event=None):
"""Quisk вызывает при смене частоты. Возвращает (tune, vfo)."""
if tune and tune != self.current_freq:
self.current_freq = tune
self.current_vfo = tune-9000
if re.search('FA\\d{11};', self.cmdbuf,0):
log.debug("--> RE MATCH")
re.sub('FA\\d{11};', "FA" + self._fmt_freq(tune)+";", self.cmdbuf, 1)
# self.cmdbuf += ("FA" + self._fmt_freq(tune)+";")
else:
self.cmdbuf += ("FA" + self._fmt_freq(tune)+";")
log.debug("VFO A → %d Hz (%.6f MHz)", self.current_vfo, self.current_freq / 1e6)
return self.current_freq, self.current_vfo
def ReturnInfo(self):
ritstr = self.application.ritFreq
if ritstr >= 0:
ritsign = "+"
else:
ritsign = ""
answer = "IF%011d%4s%s%05d%01d0000%01d%1s0000000;" % (
self.current_freq, self.STEP_MAP[self.current_step],
ritsign,ritstr,self.application.ritButton.GetValue(),
self.ptt_active,self.MODE_MAP[self.current_mode])
# log.info(answer)
return answer
def ReturnSmeterA(self):
i = round((self.application.smeter_db + 67) * 2)
if i < 0:
i = 0
elif i > 260:
i = 260
answer = "ZZSM0%03d;" % ( i )
# log.info(answer)
return answer
def ReturnPre(self):
answer = "ZZPA%01d;" % ( self.current_pre )
return answer
def SetPre(self, precmd):
if len(precmd) == 5:
match precmd:
case 'ZZPA0': i = 0
case 'ZZPA1': i = 1
case 'ZZPA2': i = 2
case 'ZZPA3': i = 3
case _: i = 0
self.current_pre = i
def SetFreq(self, freqcmd):
if len(freqcmd) == 13:
newfreq = int(freqcmd[2:])
log.info("Request to change freq: %d", newfreq)
self.current_vfo=newfreq-9000
self.application.ChangeHwFrequency(9000, self.current_vfo)
def OnButtonRfGain(self, event):
btn = event.GetEventObject()
n = btn.index
self.current_pre = n
self.cmdbuf += ("ZZPA%1d;" % self.current_pre)
def SetMode(self, modecmd):
if len(modecmd) == 6:
log.info("Request to change mode: %s", modecmd[4:])
match modecmd[4:]:
# 00 = LSB => LSB
# 01 = USB => USB
# 02 = DSB => USB
# 03 = CWL => CW
# 04 = CWU => CW
# 05 = FMN => FMN
# 06 = AM => AM
# 07 = DIGU => FSKR
# 08 = SPEC => AMN
# 09 = DIGL => FSK
# 10 = SAM => USB
# 11 = DRM => USB
case '00': newmode = 'LSB'
case '01': newmode = 'USB'
case '02': newmode = 'USB'
case '03': newmode = 'CWL'
case '04': newmode = 'CWU'
case '05': newmode = 'FM'
case '06': newmode = 'AM'
case '07': newmode = 'DGT-U'
case '08': newmode = 'AM'
case '09': newmode = 'DGT-L'
case '10': newmode = 'USB'
case '11': newmode = 'USB'
case _: newmode = 'USB'
self.current_mode = newmode
self.application.mode = self.current_mode
self.application.MakeFilterButtons(self.application.Mode2Filters(self.current_mode))
QS.set_rx_mode(Mode2Index.get(self.current_mode, 3))
if self.current_mode == 'CWL':
self.application.SetRit(conf.cwTone)
elif self.current_mode == 'CWU':
self.application.SetRit(-conf.cwTone)
else:
self.application.SetRit(0)
if self.current_mode in ('CWL', 'CWU'):
self.application.SetFilterByMode('CW')
elif self.current_mode in ('LSB', 'USB'):
self.application.SetFilterByMode('SSB')
elif self.current_mode == 'AM':
self.application.SetFilterByMode('AM')
elif self.current_mode[0:2] == 'FM':
self.application.SetFilterByMode('FM')
elif self.current_mode[0:4] == 'DGT-':
self.application.SetFilterByMode('DGT')
self.application.sliderSquelch.DeleteSliderWindow()
if self.current_mode == 'FM':
self.application.sliderSquelch.SetSlider(self.application.levelSquelch)
else:
self.application.sliderSquelch.SetSlider(self.application.levelSquelchSSB)
self.application.OnBtnSquelch()
if self.current_mode not in ('FDV-L', 'FDV-U'):
self.application.graph.SetDisplayMsg()
self.application.waterfall.SetDisplayMsg()
self.application.SetTxAudio()
def StepInc(self):
i = self.STEP_ENA.index(self.current_step)
log.debug("Found index of %s, idx: %d", self.current_step, i)
if i < (len(self.STEP_ENA)-1):
self.current_step = self.STEP_ENA[i + 1]
log.debug("Step changed to %s, idx: %d", self.current_step, i+1)
else:
log.debug("Step not changed, remain %s", self.current_step)
def StepDec(self):
i = self.STEP_ENA.index(self.current_step)
log.info("Found index of %s, idx: %d", self.current_step, i)
if i > 0:
self.current_step = self.STEP_ENA[i - 1]
log.debug("Step changed to %s, idx: %d", self.current_step, i-1)
else:
log.debug("Step not changed, remain %s", self.current_step)
def BandUp(self):
i = self.BAND_ENA.index(self.current_band)
log.debug("Found index of %s, idx: %d", self.current_band, i)
if i < (len(self.BAND_ENA)-1):
self.current_band = self.BAND_ENA[i + 1]
else:
self.current_band = self.BAND_ENA[0]
log.info("Band changed to %s, idx: %d", self.current_band, i)
vfo, tune, mode = self.application.bandState[self.current_band]
self.application.ChangeHwFrequency(9000, vfo+tune)
def BandDn(self):
i = self.BAND_ENA.index(self.current_band)
log.debug("Found index of %s, idx: %d", self.current_band, i)
if i > 0:
self.current_band = self.BAND_ENA[i - 1]
else:
self.current_band = self.BAND_ENA[len(self.BAND_ENA)-1]
log.info("Band changed to %s, idx: %d", self.current_band, i)
vfo, tune, mode = self.application.bandState[self.current_band]
self.application.ChangeHwFrequency(9000, vfo+tune)
def ReturnFrequency(self):
"""Опрос ~10 Гц. Возвращает (tune, vfo) или (None, None)."""
# SDR-1000 не сообщает частоту сам, поэтому возвращаем None
return None, None
# ── Quisk API: режим ─────────────────────────────────────────────
def ChangeMode(self, mode):
"""Quisk вызывает при смене режима. mode: 'USB', 'CW', и т.д."""
# Нормализуем DGT-* → USB
# if mode.startswith('DGT'):
# mode = 'USB'
code = self.MODE_MAP.get(mode)
if not code:
log.warning("Неизвестный режим Quisk: %s → по умолчанию USB", mode)
code = '2'
self.cmdbuf += ("MD" + code+';')
self.current_mode = mode
log.debug("Режим → %s (MD%s)", mode, code)
# ── Quisk API: диапазон ─────────────────────────────────────────
def ChangeBand(self, band):
"""Quisk вызывает при смене диапазона. band: '160', '80', …"""
BaseHardware.ChangeBand(self, band)
log.info("Диапазон → %s м", band)
# ── Quisk API: PTT ──────────────────────────────────────────────
def OnButtonPTT(self, event=None):
"""Quisk вызывает при нажатии/отпускании PTT."""
if event:
self.ptt_active = bool(event.GetEventObject().GetValue())
#self._cat("ZZTX%d" % (1 if self.ptt_active else 0))
log.debug("PTT %s", "ON" if self.ptt_active else "OFF")
# ── Quisk API: HeartBeat (~10 Гц) ───────────────────────────────
def HeartBeat(self):
"""Периодический опрос — можно использовать для синхронизации."""
self._cat()
def FastHeartBeat(self): # Called frequently by the GUI thread
pass
# ── Quisk API: TUNE ─────────────────────────────────────────────
def OnBtnFDX(self, is_fdx):
"""Кнопка FDX (полный дуплекс)."""
log.debug("FDX → %s", is_fdx)