Professional Portfolio

The client sends a structured registration payload instead of only opening a socket. That gives the server enough metadata to track client identity, supported protocol version, tuned channels, monitored channels, loopback state, and encryption readiness.

Important source lines

# udp_client.py
def _send_register(self) -> None:
    data = {
        "nick": self.nick,
        "net": self.net,
        "ssrc": self.ssrc,
        "protocol_version": int(PROTOCOL_VERSION),
        "min_protocol_version": int(PROTOCOL_MIN_VERSION),
        "client_version": self.client_version,
        "device_id": getattr(self, "device_id", None),
        "channels": self.channels,
        "monitor": self.monitor,
        "active_tx": self.active_tx,
    }
    data.update(self._keys_ready_payload())
    msg = pack_hdr(VER, MT_CTRL, self.seq.next(), now_ts48(), self.ssrc) + ctrl_hdr + b
    self.sock.sendto(msg, self.server)

The desktop client keeps a local mirror of the four radio channels, then normalizes and sends channel changes to the server as a dedicated channel-update control packet. Scan state is represented as monitored channels, which lets the server route audio for active and scanned frequencies.

Important source lines

# udp_client.py
self.channels = {
    "A": {"freq": 100.0, "mode": "PTT"},
    "B": {"freq": 101.0, "mode": "PTT"},
    "C": {"freq": 102.0, "mode": "PTT"},
    "D": {"freq": 111.1, "mode": "PTT"},
}
self.active_tx = "A"
self.monitor = {"A": True, "B": False, "C": False, "D": False}

payload_obj = {
    "active": int(active),
    "freqs": freqs_list,
    "scan": bool(any(sc)),
    "scan_channels": sc,
    "monitor_channels": sc,
}
ctrl_hdr = struct.pack("!BH", CTRL_CHAN_UPD, len(payload) & 0xFFFF)

Audio is intentionally gated by push-to-talk before encoding. The transport prefers Opus for voice, falls back to PCM16 when Opus is unavailable, and can wrap the payload with AES-GCM when the selected network requires encrypted audio.

Important source lines

# udp_client.py
def send_audio(self, pcm_float32) -> None:
    if not self._ptt:
        return

    data = self.enc.encode_float32(buf) if self.enc.enabled else b""
    if not data:
        pcm16 = (buf * 32767.0).astype("<i2")
        data = pcm16.tobytes()
        flags |= AUDIO_FLAG_CODEC_PCM | AUDIO_FLAG_PCM_I16

    if self._audio_require_encrypt:
        key_id, key_bytes = self._select_tx_key()
        nonce = os.urandom(_AUDIO_NONCE_LEN)
        ciphertext = crypto.encrypt(nonce, data, _audio_aad(self.ssrc, seq_num))
        data = struct.pack("!I", int(key_id) & 0xFFFFFFFF) + nonce + ciphertext

The client exposes link metrics from the UDP layer and receive-side mixer metrics from the jitter buffer. This makes connection quality visible in the UI and gives diagnostics enough data to explain audio gaps, packet loss, late frames, or queue buildup.

Important source lines

# udp_client.py
def get_link_metrics(self) -> dict:
    total = self._audio_rx_packets + self._audio_rx_est_lost
    loss_pct = (float(self._audio_rx_est_lost) / float(total) * 100.0) if total > 0 else 0.0
    return {
        "rtt_ms": self._rtt_ms,
        "packet_loss_est_pct": loss_pct,
        "late_frames": self._audio_rx_late,
        "tx_packets": self._audio_tx_packets,
        "active_key_net": self.active_key_net(),
    }

# app/rx_audio.py
def snapshot(self) -> dict:
    return {"queue_depth": int(total_depth), "late_frames": int(self._late_frames),
            "active_channels": sorted(set(active_channels))}

The message tab derives current and monitored frequencies, keeps subscription state synchronized with the server, and sends messages as frequency-scoped control payloads. Incoming chat can also feed the overlay toast system so radio messages appear outside the main window.

Important source lines

# app/app.py
udp.send_frequency_chat({
    "action": "sync",
    "frequencies": subs,
    "selected_frequency": selected_for_sync,
})

udp.send_frequency_chat({
    "action": "send",
    "frequency": freq,
    "text": text,
})

# udp_client.py
ctrl_hdr = struct.pack("!BH", CTRL_FREQ_CHAT, len(payload) & 0xFFFF)
msg = pack_hdr(VER, MT_CTRL, self.seq.next(), now_ts48(), self.ssrc) + ctrl_hdr + payload

Security and support tooling are part of the client instead of separate scripts. The client can request channel keys, verify signed update manifests before accepting server offers, and run admin-triggered diagnostics that report microphone levels, playback tests, and short link traces.

Important source lines

# udp_client.py
def send_key_request(self, payload_obj: dict) -> None:
    self._send_ctrl_json(CTRL_KEY_REQUEST, dict(payload_obj))

# app/app.py
ok, err, candidate_payload = verify_signed_update_manifest(signed_manifest, public_key)
if int(payload.get("protocol_version") or PROTOCOL_VERSION) < int(PROTOCOL_MIN_VERSION):
    return False, "protocol_too_old", payload

def _diagnostic_capture_trace(self, payload: dict) -> tuple[str, dict]:
    start = self._diagnostic_runtime_snapshot()
    time.sleep(duration_s)
    end = self._diagnostic_runtime_snapshot()
    details = {"rx_packets_delta": end_link.get("rx_packets", 0) - start_link.get("rx_packets", 0),
               "packet_loss_est_pct": round(float(end_link.get("packet_loss_est_pct", 0.0)), 2)}

The admin app owns the operator workflow. For local restarts it resolves a fresh launch command, starts a new admin/server pair, and keeps the server side hidden from the desktop.

cmd, cwd = self._resolve_admin_launch_command(auto_restart_server=True)
env = os.environ.copy()
env.setdefault("PYTHONUNBUFFERED", "1")
creationflags = int(getattr(subprocess, "CREATE_NO_WINDOW", 0))
subprocess.Popen(cmd, cwd=cwd, env=env, stdin=subprocess.DEVNULL,
                 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                 creationflags=creationflags)

The server parses the compact control header, decodes JSON payloads where needed, and branches by protocol code for registration, channel updates, and other radio control traffic.

code, length = struct.unpack("!BH", payload[:3])
body = payload[3 : 3 + length] if len(payload) >= 3 + length else b""
self._log(f"[CTRL][RX] addr={addr} ssrc={ssrc} code={code} len={len(body)}")

if code == CTRL_REGISTER:
    info = _decode_json(body)
elif code == CTRL_CHAN_UPD:
    # Apply channel/frequency updates from the client.

The server embeds the web gateway in the same process and points it back to loopback UDP, so browser sessions register like radio clients without exposing an extra backend service.

gateway_udp_host = "127.0.0.1"
self.web_gateway = RadioWebGateway(
    WEB_GATEWAY_HOST,
    WEB_GATEWAY_PORT,
    gateway_udp_host,
    self.port,
)
self.web_gateway.start()

Admin commands are wrapped in signed payloads. Update uploads also include a signed manifest with file size, SHA-256, app version, and protocol compatibility fields.

signed = build_signed_admin_control(
    private_key,
    int(code),
    body_obj,
    admin_id=f"{socket.gethostname()}:{os.getpid()}",
)
signed_manifest = build_signed_update_manifest(
    self._load_admin_signing_key(), name=name,
    size=len(data), sha256=sha256_hex(data),
    protocol_version=PROTOCOL_VERSION,
)

Chat is normalized by frequency, held in bounded memory for active users, and inserted into SQLite so frequency history survives server restarts.

bucket = self._chat_history.get(freq)
if bucket is None:
    bucket = deque(maxlen=CHAT_HISTORY_LIMIT)
    self._chat_history[freq] = bucket
record = _normalize_chat_message_record(message, default_frequency=freq)
bucket.append(record)
if not self._chat_store.append_message(record):
    self._increment_counter("chat_store_errors")

INSERT OR REPLACE INTO messages
(id, frequency, ts, sender, sender_kind, sender_client_id,
 sender_ssrc, text, attachment_json)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)

Status snapshots combine server state with the embedded web gateway snapshot, so the admin UI can show uptime, connected rows, gateway clients, and recent operational activity.

def _web_gateway_snapshot(self) -> dict:
    gateway = getattr(self, "web_gateway", None)
    if gateway is None:
        return {"enabled": bool(WEB_GATEWAY_ENABLED), "running": False}
    return gateway.snapshot()

def _admin_status_payload(self, requester_addr=None) -> dict:
    uptime_s = max(0, int(time.time() - float(self._started_at)))
    row_count = len(self.mgr.presence_snapshot())

The emulator is connected through a WebSocket bridge. Browser actions are sent as small JSON messages, and the gateway translates them into the UDP radio protocol.

function sendBridgeMessage(payload) {
  const socket = state.bridge.socket;
  if (!socket || socket.readyState !== WebSocket.OPEN) {
    return false;
  }
  socket.send(JSON.stringify(payload));
  return true;
}

The server-side gateway keeps the browser contract explicit. It registers a web radio session, updates channels, forwards push-to-talk, and bridges browser PCM audio into radio packets.

if msg_type == "connect":
    await self.connect(message)
elif msg_type == "channels":
    self.update_channels(message)
    self.send_channel_update()
elif msg_type == "ptt":
    self.send_ctrl_json(CTRL_PTT, {"ptt": bool(message.get("active"))})
elif msg_type == "audio_pcm16":
    self.send_audio_pcm16(message)

flags = AUDIO_FLAG_PTT | AUDIO_FLAG_CODEC_PCM | AUDIO_FLAG_PCM_I16
body = struct.pack("!BH", flags, len(data) & 0xFFFF) + data
self.send_packet(pack_hdr(VER, MT_AUDIO, self.seq.next(),
                          now_ts48(), self.ssrc) + body)

The page updates channel state immediately, pushes changes to the bridge, and requests microphone audio only while transmitting. Received PCM is decoded in the browser and played through the Web Audio API.

sendBridgeMessage({
  type: "audio_pcm16",
  sampleRate: state.audio.txSampleRate,
  data: bytesToBase64(floatSamplesToPcm16Bytes(frame)),
});

if (state.bridge.realMode) {
  sendBridgeMessage({ type: "ptt", active: state.transmitting });
  if (state.transmitting) {
    startMicStream();
  } else {
    setMicTrackEnabled(false);
    clearTransmitBuffer();
  }
}