#!/usr/bin/env python3 """ webrtc_audio.py — receives TX audio over WebRTC (aiortc). Client (microphone or VB-Cable) -> RTCPeerConnection -> AudioFrame (PCM) -> the same _tx_stream (PyAudio) used by the existing WS pipeline. Signaling (offer/answer/ICE) goes over the existing WebSocket (/ws), so no separate HTTP endpoint is needed. Only one client transmits at a time (IC-746/PTT hardware limitation), so there's no need to mix — the newest connection replaces the previous one. """ import asyncio, errno, fractions, socket, struct, time from aiortc import RTCPeerConnection, RTCSessionDescription, RTCIceCandidate, RTCConfiguration, RTCIceServer from aiortc.contrib.media import MediaBlackhole OPUS_RATE = 48000 _STUN_HOST = "stun.l.google.com" _STUN_PORT = 19302 _stun_ip_cache: str | None = None async def _stun_url() -> str: """ Resolve the STUN server hostname once and cache the IP. Mobile TX opens a brand-new RTCPeerConnection on EVERY PTT press (see mobile_audio.js header — a deliberate choice to avoid holding the mic open). aioice's gather_candidates() does a blocking socket.gethostbyname() on the STUN hostname for every single ICE gathering pass, so without caching, every PTT press pays a fresh DNS round-trip on top of the STUN UDP round-trip. Reported live as noticeable TX start delay. Resolving once and reusing the numeric IP removes that repeated cost; falls back to the hostname if resolution fails so behavior is unchanged in the failure case. Run via asyncio.to_thread (not a plain blocking call) — this shares the process with civ.py's CI-V polling loop on the same event loop, and a blocking DNS lookup right on the loop would stall that polling for however long the lookup takes. """ global _stun_ip_cache if _stun_ip_cache is None: try: _stun_ip_cache = await asyncio.to_thread(socket.gethostbyname, _STUN_HOST) except OSError: _stun_ip_cache = _STUN_HOST return f"stun:{_stun_ip_cache}:{_STUN_PORT}" # EADDRINUSE / EACCES oraz ich odpowiedniki WinSock (10048, 10013). _PORT_BUSY_ERRNOS = {errno.EADDRINUSE, errno.EACCES, 10048, 10013} def install_udp_port_range(spec: str) -> tuple[int, int] | None: """Kieruje gniazda UDP otwierane na losowym porcie (local_addr port 0 - tak robi aioice dla kandydatow ICE) do stalego zakresu `spec` ("lo-hi"), zeby dalo sie go raz przekierowac na routerze. aioice/aiortc nie maja takiej opcji, wiec podmieniamy create_datagram_endpoint na poziomie klasy petli asyncio (dziala dla kazdej petli). Gniazda z jawnym portem (np. odbiornik UDP WSJT-X) i z remote_addr zostaja bez zmian. Porty przydzielane po kolei w kolko - swiezo zwolniony port nie wraca od razu. Zajety port -> nastepny; caly zakres zajety -> losowy port jak dawniej. Zwraca (lo, hi) albo None, gdy wylaczone/bledna wartosc.""" spec = (spec or "").strip() if not spec or spec == "0": return None try: lo_s, _, hi_s = spec.partition("-") lo, hi = int(lo_s), int(hi_s or lo_s) except ValueError: print(f"[webrtc] WEBRTC_UDP_PORTS={spec!r} - bledny format (oczekiwane np. 40000-40019), porty losowe", flush=True) return None if not (1024 <= lo <= hi <= 65535): print(f"[webrtc] WEBRTC_UDP_PORTS={spec!r} - poza zakresem 1024-65535, porty losowe", flush=True) return None import asyncio.base_events as _be current = _be.BaseEventLoop.create_datagram_endpoint orig = getattr(current, "_hamctrl_orig", current) state = {"next": lo} n = hi - lo + 1 async def create_datagram_endpoint(self, protocol_factory, local_addr=None, remote_addr=None, **kw): if local_addr and len(local_addr) >= 2 and local_addr[1] == 0 and remote_addr is None: for k in range(n): port = lo + (state["next"] - lo + k) % n try: res = await orig(self, protocol_factory, local_addr=(local_addr[0], port), **kw) except OSError as e: # Tylko "port zajety" / "port zablokowany" (np. zakres # zarezerwowany przez Windows) -> nastepny port. Inny blad # (np. adres 169.254.x.x, na ktorym nie da sie otworzyc # gniazda wcale) -> zachowanie jak dawniej. if (getattr(e, "winerror", None) or e.errno) in _PORT_BUSY_ERRNOS: continue break state["next"] = lo + (port - lo + 1) % n return res else: print(f"[webrtc] wszystkie porty UDP {lo}-{hi} zajete - port losowy", flush=True) return await orig(self, protocol_factory, local_addr=local_addr, remote_addr=remote_addr, **kw) create_datagram_endpoint._hamctrl_orig = orig _be.BaseEventLoop.create_datagram_endpoint = create_datagram_endpoint return lo, hi async def add_ice_candidate_to_pc(pc: RTCPeerConnection, candidate: dict): """Add a trickle-ICE candidate (from the client) to an existing RTCPeerConnection. Shared by WebRTCAudioReceiver (TX) and WebRTCAudioSender (RX, webrtc_rx_audio.py) - identical parsing either direction, only the target connection differs.""" raw = candidate.get("candidate", "") if not raw: return # empty candidate = end-of-candidates, ignore try: from aioice import Candidate as IceCandidate # candidate string format: "candidate:foundation component protocol priority ip port typ type ..." ice_cand = IceCandidate.from_sdp(raw.replace("candidate:", "", 1)) cand = RTCIceCandidate( component=ice_cand.component, foundation=ice_cand.foundation, ip=ice_cand.host, port=ice_cand.port, priority=ice_cand.priority, protocol=ice_cand.transport, type=ice_cand.type, relatedAddress=ice_cand.related_address, relatedPort=ice_cand.related_port, tcpType=ice_cand.tcptype, sdpMid=candidate.get("sdpMid"), sdpMLineIndex=candidate.get("sdpMLineIndex"), ) await pc.addIceCandidate(cand) print(f"[webrtc] ICE candidate added: {ice_cand.host}:{ice_cand.port} type={ice_cand.type}") except Exception as e: print(f"[webrtc] ICE candidate error: {e} | raw={raw[:80]}") class WebRTCAudioReceiver: """ Manages a single active RTCPeerConnection for TX audio. Received PCM frames (mono, 48kHz, int16) are passed through a callback. """ def __init__(self, on_pcm_frame, on_track_started=None, on_track_ended=None): """ on_pcm_frame(pcm_bytes: bytes) — called for every PCM frame on_track_started() / on_track_ended() — optional hooks (e.g. start_tx/stop_tx) """ self._pc: RTCPeerConnection | None = None self._on_pcm = on_pcm_frame self._on_start = on_track_started self._on_end = on_track_ended self._task: asyncio.Task | None = None self.active = False async def handle_offer(self, sdp: str, type_: str = "offer") -> dict: """ Accept an SDP offer from the client, return an SDP answer. If a previous connection exists — close it (one sender at a time). """ await self.close() self._frame_count = 0 if hasattr(self, "_resampler"): del self._resampler t0 = time.monotonic() config = RTCConfiguration(iceServers=[ RTCIceServer(urls=await _stun_url()) ]) pc = RTCPeerConnection(configuration=config) self._pc = pc @pc.on("track") def on_track(track): if track.kind != "audio": return print(f"[webrtc] Audio track received: {track.kind}") self.active = True if self._on_start: self._on_start() self._task = asyncio.ensure_future(self._consume_track(track)) @pc.on("connectionstatechange") async def on_state_change(): print(f"[webrtc] Connection state: {pc.connectionState}") # 'disconnected' is DELIBERATELY not included - per the WebRTC # spec it's a transient state the ICE transport often recovers # from on its own within a second or two; treating it the same # as 'failed' tears the connection down before it gets the # chance. See the identical fix + longer comment in # webrtc_rx_audio.py (found live via the RX path, which is # long-lived and made this far more visible than TX's naturally # short per-PTT-press connections). if pc.connectionState == "failed": await self.close() @pc.on("iceconnectionstatechange") async def on_ice_state_change(): print(f"[webrtc] ICE state: {pc.iceConnectionState}") @pc.on("icegatheringstatechange") async def on_ice_gathering_change(): print(f"[webrtc] ICE gathering: {pc.iceGatheringState}") offer = RTCSessionDescription(sdp=sdp, type=type_) await pc.setRemoteDescription(offer) t1 = time.monotonic() offer_candidates = sdp.count('a=candidate') print(f"[webrtc] Offer from client: {offer_candidates} ICE candidates") answer = await pc.createAnswer() await pc.setLocalDescription(answer) t2 = time.monotonic() # Wait for the server's ICE gathering to finish (max 3s) for _ in range(30): if pc.iceGatheringState == "complete": break await asyncio.sleep(0.1) t3 = time.monotonic() final_sdp = pc.localDescription.sdp answer_candidates = final_sdp.count('a=candidate') print(f"[webrtc] Server answer: {answer_candidates} ICE candidates, gathering={pc.iceGatheringState} | " f"timing: setRemoteDescription={1000*(t1-t0):.0f}ms createAnswer/setLocal={1000*(t2-t1):.0f}ms " f"iceWait={1000*(t3-t2):.0f}ms total={1000*(t3-t0):.0f}ms") return { "sdp": final_sdp, "type": pc.localDescription.type, } async def add_ice_candidate(self, candidate: dict): """Add an ICE candidate from the client (trickle ICE).""" if not self._pc: return await add_ice_candidate_to_pc(self._pc, candidate) async def _consume_track(self, track): """Loop receiving audio frames and passing PCM to the callback.""" try: while True: frame = await track.recv() # frame is an av.AudioFrame — convert to PCM int16 mono 48kHz pcm = self._frame_to_pcm(frame) if pcm and self._on_pcm: self._on_pcm(pcm) except Exception as e: print(f"[webrtc] Track ended: {e}") finally: self.active = False if self._on_end: self._on_end() def _frame_to_pcm(self, frame) -> bytes: """Convert an av.AudioFrame -> PCM int16 mono @ 48kHz.""" try: n_ch = len(frame.layout.channels) if frame.sample_rate == OPUS_RATE and frame.format.name == "s16": raw = bytes(frame.planes[0]) expected_bytes = frame.samples * n_ch * 2 if len(raw) != expected_bytes: raw = raw[:expected_bytes] if n_ch == 1: pcm = raw elif n_ch == 2: try: import numpy as np stereo = np.frombuffer(raw, dtype=np.int16).reshape(-1, 2) mono = ((stereo[:,0].astype(np.int32) + stereo[:,1].astype(np.int32)) // 2).astype(np.int16) pcm = mono.tobytes() except ImportError: n = len(raw) // 4 samples = struct.unpack(f"<{n*2}h", raw) mono = [(samples[i*2]+samples[i*2+1])//2 for i in range(n)] pcm = struct.pack(f"<{n}h", *mono) else: pcm = raw return pcm # Different sample rate - resampling needed (rare case) from av import AudioResampler if not hasattr(self, "_resampler"): self._resampler = AudioResampler(format="s16", layout="mono", rate=OPUS_RATE) frames = self._resampler.resample(frame) pcm = b"" for f in frames: pcm += bytes(f.planes[0]) return pcm except Exception as e: print(f"[webrtc] frame_to_pcm error: {e}") return b"" async def close(self): """Close the active connection.""" if self._task: self._task.cancel() self._task = None if self._pc: try: await self._pc.close() except Exception: pass self._pc = None self.active = False