chore: baseline existing CI-V suite before RepeaterBook integration
This commit is contained in:
@@ -0,0 +1,158 @@
|
||||
"""
|
||||
Serial transport + high-level control for the ICOM ID-5100 over CI-V.
|
||||
|
||||
Talks to the RT Systems USB-29A cable (FTDI) plugged into the radio's [SP2]
|
||||
jack. Uses pure-stdlib termios — no pyserial. Pairs with civ.py (the codec).
|
||||
|
||||
Example:
|
||||
from radio import Radio
|
||||
with Radio("/dev/ttyUSB0", 9600) as r:
|
||||
print(r.get_frequency()) # Hz
|
||||
r.set_frequency(146_520_000)
|
||||
"""
|
||||
import os
|
||||
import time
|
||||
import termios
|
||||
|
||||
import civ
|
||||
|
||||
BAUD_CONST = {
|
||||
1200: termios.B1200, 4800: termios.B4800, 9600: termios.B9600,
|
||||
19200: termios.B19200, 38400: termios.B38400, 115200: termios.B115200,
|
||||
}
|
||||
|
||||
# Mode byte -> name. Analog codes are standard ICOM; confirm DV against the
|
||||
# manual's CI-V table (overridden empirically where the live radio disagrees).
|
||||
MODE_NAMES = {0x00: "LSB", 0x01: "USB", 0x02: "AM", 0x03: "CW",
|
||||
0x05: "FM", 0x17: "DV"}
|
||||
MODE_CODES = {v: k for k, v in MODE_NAMES.items()}
|
||||
|
||||
ACK_OK = 0xFB
|
||||
ACK_NG = 0xFA
|
||||
|
||||
|
||||
class CIVError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class Radio:
|
||||
def __init__(self, port="/dev/ttyUSB0", baud=9600, timeout=0.6):
|
||||
self.port = port
|
||||
self.baud = baud
|
||||
self.timeout = timeout
|
||||
self.fd = None
|
||||
|
||||
# ---- lifecycle -------------------------------------------------------
|
||||
def open(self):
|
||||
speed = BAUD_CONST.get(self.baud)
|
||||
if speed is None:
|
||||
raise ValueError(f"unsupported baud {self.baud}")
|
||||
fd = os.open(self.port, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
|
||||
iflag, oflag, cflag, lflag, ispeed, ospeed, cc = termios.tcgetattr(fd)
|
||||
cflag = termios.CS8 | termios.CREAD | termios.CLOCAL
|
||||
cc = list(cc)
|
||||
cc[termios.VMIN] = 0
|
||||
cc[termios.VTIME] = 0
|
||||
termios.tcsetattr(fd, termios.TCSANOW, [0, 0, cflag, 0, speed, speed, cc])
|
||||
termios.tcflush(fd, termios.TCIOFLUSH)
|
||||
self.fd = fd
|
||||
return self
|
||||
|
||||
def close(self):
|
||||
if self.fd is not None:
|
||||
os.close(self.fd)
|
||||
self.fd = None
|
||||
|
||||
def __enter__(self):
|
||||
return self.open()
|
||||
|
||||
def __exit__(self, *exc):
|
||||
self.close()
|
||||
|
||||
# ---- raw exchange ----------------------------------------------------
|
||||
def _read_for(self, seconds):
|
||||
buf = bytearray()
|
||||
deadline = time.monotonic() + seconds
|
||||
while time.monotonic() < deadline:
|
||||
try:
|
||||
chunk = os.read(self.fd, 256)
|
||||
except BlockingIOError:
|
||||
chunk = b""
|
||||
if chunk:
|
||||
buf.extend(chunk)
|
||||
else:
|
||||
time.sleep(0.005)
|
||||
return bytes(buf)
|
||||
|
||||
def transact(self, cmd, data=b""):
|
||||
"""Send a command, return the radio's reply frame. Returns as soon as a
|
||||
frame from the radio (not our own bus echo) is parsed, so it doesn't
|
||||
burn the full timeout on every call."""
|
||||
if self.fd is None:
|
||||
raise CIVError("port not open")
|
||||
termios.tcflush(self.fd, termios.TCIFLUSH)
|
||||
os.write(self.fd, civ.build_frame(cmd, data))
|
||||
buf = bytearray()
|
||||
deadline = time.monotonic() + self.timeout
|
||||
while time.monotonic() < deadline:
|
||||
try:
|
||||
chunk = os.read(self.fd, 256)
|
||||
except BlockingIOError:
|
||||
chunk = b""
|
||||
if chunk:
|
||||
buf.extend(chunk)
|
||||
for fr in civ.parse_frames(bytes(buf)):
|
||||
if fr.frm == civ.RADIO_ADDR: # radio's reply, not our echo
|
||||
return fr
|
||||
else:
|
||||
time.sleep(0.005)
|
||||
raise CIVError(f"no reply to cmd 0x{cmd:02X} (got {bytes(buf).hex(' ') or 'nothing'})")
|
||||
|
||||
def _command(self, cmd, data):
|
||||
"""Send a set-command and require an OK ack."""
|
||||
fr = self.transact(cmd, data)
|
||||
if fr.cn == ACK_NG:
|
||||
raise CIVError(f"radio rejected cmd 0x{cmd:02X}")
|
||||
if fr.cn not in (ACK_OK, cmd):
|
||||
raise CIVError(f"unexpected ack 0x{fr.cn:02X} for cmd 0x{cmd:02X}")
|
||||
return True
|
||||
|
||||
# ---- frequency -------------------------------------------------------
|
||||
def get_frequency(self):
|
||||
return civ.bcd_to_freq(self.transact(civ.CMD_READ_FREQ).data)
|
||||
|
||||
def set_frequency(self, hz):
|
||||
return self._command(civ.CMD_SET_FREQ, civ.freq_to_bcd(hz))
|
||||
|
||||
# ---- mode ------------------------------------------------------------
|
||||
def get_mode(self):
|
||||
data = self.transact(civ.CMD_READ_MODE).data
|
||||
code = data[0] if data else None
|
||||
return MODE_NAMES.get(code, f"0x{code:02X}" if code is not None else "?"), code
|
||||
|
||||
def set_mode(self, name):
|
||||
code = MODE_CODES.get(name.upper())
|
||||
if code is None:
|
||||
raise ValueError(f"unknown mode {name}; known: {sorted(MODE_CODES)}")
|
||||
return self._command(civ.CMD_SET_MODE, bytes([code]))
|
||||
|
||||
# ---- levels ----------------------------------------------------------
|
||||
def get_volume(self):
|
||||
return civ.bcd_to_level(self.transact(civ.CMD_LEVEL, bytes([civ.SUB_AF])).data[1:])
|
||||
|
||||
def set_volume(self, value):
|
||||
return self._command(civ.CMD_LEVEL, bytes([civ.SUB_AF]) + civ.level_to_bcd(value))
|
||||
|
||||
def get_squelch_level(self):
|
||||
return civ.bcd_to_level(self.transact(civ.CMD_LEVEL, bytes([civ.SUB_SQL_LEVEL])).data[1:])
|
||||
|
||||
def set_squelch_level(self, value):
|
||||
return self._command(civ.CMD_LEVEL, bytes([civ.SUB_SQL_LEVEL]) + civ.level_to_bcd(value))
|
||||
|
||||
# ---- meters ----------------------------------------------------------
|
||||
def get_smeter(self):
|
||||
return civ.bcd_to_level(self.transact(civ.CMD_METER, bytes([civ.SUB_SMETER])).data[1:])
|
||||
|
||||
def squelch_open(self):
|
||||
data = self.transact(civ.CMD_METER, bytes([civ.SUB_SQL_STATUS])).data
|
||||
return bool(data[1]) if len(data) > 1 else False
|
||||
Reference in New Issue
Block a user