id5100/radio.py

"""
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

    # ---- repeater config (verified on live ID-5100 2026-06-14) ------------
    # Offset *amount* is not CI-V-settable on this radio (cmd 0x0D -> NG); the
    # radio applies its configured auto-offset when a duplex direction is set.
    def set_duplex(self, direction):
        """direction: '+', '-', or 'simplex'. Radio uses its configured offset."""
        code = {"+": civ.DUP_PLUS, "-": civ.DUP_MINUS,
                "simplex": civ.DUP_SIMPLEX}.get(direction)
        if code is None:
            raise ValueError(f"bad duplex {direction!r}")
        return self._command(civ.CMD_DUPLEX, bytes([code]))

    def get_duplex(self):
        code = self.transact(civ.CMD_DUPLEX).data[0]
        return {civ.DUP_PLUS: "+", civ.DUP_MINUS: "-",
                civ.DUP_SIMPLEX: "simplex"}.get(code, f"0x{code:02X}")

    def set_tone(self, hz):
        # wire format: sub(0x00) + 0x00 pad + 2-byte BCD (00 00 08 85 = 88.5 Hz)
        return self._command(civ.CMD_TONE_PARAM,
                             bytes([civ.SUB_RPT_TONE, 0x00]) + civ.tone_to_bcd(hz))

    def get_tone(self):
        # reply data = sub + pad + 2-byte BCD; take the last two bytes
        return civ.bcd_to_tone(self.transact(civ.CMD_TONE_PARAM,
                                             bytes([civ.SUB_RPT_TONE])).data[2:])

    def set_tone_on(self, on):
        return self._command(civ.CMD_TONE_SWITCH,
                             bytes([civ.SUB_TONE_ENC, 0x01 if on else 0x00]))

    def tune_repeater(self, rec):
        """Apply a normalized record: freq -> duplex direction -> tone.
        Offset amount comes from the radio's own config (not CI-V-settable).
        Returns (steps, errors): each step True/False; errors keyed by step."""
        steps, errors = {}, {}

        def attempt(name, fn):
            try:
                fn(); steps[name] = True
            except (CIVError, ValueError, TypeError, OSError) as e:
                steps[name] = False; errors[name] = str(e)

        attempt("freq", lambda: self.set_frequency(int(round(rec["output_mhz"] * 1e6))))
        attempt("duplex", lambda: self.set_duplex(rec.get("duplex", "simplex")))
        if rec.get("tone_mode") == "ctcss" and rec.get("tone_hz"):
            attempt("tone_freq", lambda: self.set_tone(rec["tone_hz"]))
            attempt("tone_on", lambda: self.set_tone_on(True))
        return steps, errors