#!/usr/bin/env python3 """ callpipe relay daemon — runs on the Raspberry Pi ("CallPipe Relay"). The Pi is paired to the phone as a Bluetooth HFP hands-free unit, so during a call the phone routes call audio to it over SCO. This daemon is a DUMB AUDIO PIPE between that Bluetooth SCO PCM (via bluez-alsa / arecord+aplay) and the phone's audio TCP server over WiFi: customer <-cellular-> phone <-BT/SCO-> [arecord] -> TCP -> phone -> agent agent -> phone -> TCP -> [aplay] -> BT/SCO -> phone <-cellular-> customer It does no call control — the phone (the hub) tells it when to start/stop via a control frame, because SCO only exists while a call is up and routed to BT. Wire framing on the TCP socket, both directions: [4-byte big-endian length N][1-byte type][payload...] (N = 1 + len(payload)) type 'A' (0x41) = PCM S16LE mono 16 kHz (20 ms = 640 bytes/frame) type 'C' (0x43) = control, payload is UTF-8 JSON Control (phone -> relay): {"cmd":"start"[,"mac":"AA:BB:.."]} begin bridging (mac optional; auto-detected) {"cmd":"stop"} stop bridging Control (relay -> phone): {"event":"hello","mac":"..","rate":16000} {"event":"started"} / {"event":"stopped"} / {"event":"error","msg":".."} Env / config: CALLPIPE_PHONE phone IP (default 10.0.1.239) CALLPIPE_PORT phone audio port (default 9010) CALLPIPE_BT BT MAC override (default: auto-detect from bluealsa) """ import os import re import socket import struct import subprocess import sys import threading import time PHONE_HOST = os.environ.get("CALLPIPE_PHONE", "10.0.1.239") PHONE_PORT = int(os.environ.get("CALLPIPE_PORT", "9010")) BT_OVERRIDE = os.environ.get("CALLPIPE_BT") RATE = 16000 FRAME_MS = 20 FRAME_BYTES = RATE * 2 * FRAME_MS // 1000 # 640 TYPE_AUDIO = 0x41 # 'A' TYPE_CTRL = 0x43 # 'C' import json def log(*a): print("[relay]", *a, file=sys.stderr, flush=True) def detect_mac(): if BT_OVERRIDE: return BT_OVERRIDE try: out = subprocess.check_output(["bluealsa-cli", "list-pcms"], text=True, timeout=5) except Exception as e: log("list-pcms failed:", e) return None for line in out.splitlines(): m = re.search(r"dev_([0-9A-Fa-f_]+)/hfphf", line) if m: return m.group(1).replace("_", ":") return None def alsa_dev(mac): return f"bluealsa:DEV={mac},PROFILE=sco" class Bridge: """Bridges one accepted/connected phone socket <-> Bluetooth SCO PCM.""" def __init__(self, sock): self.sock = sock self.send_lock = threading.Lock() self.arec = None self.aplay = None self.active = False self.mac = None self.cap_thread = None # ---- framing ---- def send(self, ftype, payload=b""): body = bytes([ftype]) + payload with self.send_lock: self.sock.sendall(struct.pack(">I", len(body)) + body) def send_ctrl(self, obj): try: self.send(TYPE_CTRL, json.dumps(obj).encode()) except OSError: pass def _recv_exact(self, n): buf = bytearray() while len(buf) < n: chunk = self.sock.recv(n - len(buf)) if not chunk: return None buf += chunk return bytes(buf) # ---- audio ---- def start_audio(self, mac): if self.active: return self.mac = mac or detect_mac() if not self.mac: self.send_ctrl({"event": "error", "msg": "no HFP device connected"}) return dev = alsa_dev(self.mac) log("start audio on", dev) try: self.aplay = subprocess.Popen( ["aplay", "-q", "-D", dev, "-f", "S16_LE", "-c", "1", "-r", str(RATE), "-t", "raw"], stdin=subprocess.PIPE, ) self.arec = subprocess.Popen( ["arecord", "-q", "-D", dev, "-f", "S16_LE", "-c", "1", "-r", str(RATE), "-t", "raw"], stdout=subprocess.PIPE, ) except Exception as e: self.send_ctrl({"event": "error", "msg": f"spawn failed: {e}"}) self.stop_audio() return self.active = True self.cap_thread = threading.Thread(target=self._capture_loop, daemon=True) self.cap_thread.start() self.send_ctrl({"event": "started"}) def _capture_loop(self): # Reads downlink (customer) PCM from arecord and ships it to the phone. arec = self.arec try: while self.active and arec and arec.poll() is None: data = arec.stdout.read(FRAME_BYTES) if not data: break try: self.send(TYPE_AUDIO, data) except OSError: break except Exception as e: log("capture loop error:", e) log("capture loop ended") def feed_playback(self, pcm): # Writes uplink (agent) PCM to aplay -> Bluetooth mic -> customer. if self.active and self.aplay and self.aplay.stdin: try: self.aplay.stdin.write(pcm) except (BrokenPipeError, OSError): pass def stop_audio(self): self.active = False for p in (self.arec, self.aplay): if p and p.poll() is None: try: p.terminate() except Exception: pass self.arec = None self.aplay = None # ---- main receive loop ---- def run(self): self.send_ctrl({"event": "hello", "mac": detect_mac(), "rate": RATE}) while True: hdr = self._recv_exact(4) if not hdr: break (n,) = struct.unpack(">I", hdr) if n == 0 or n > 65536: log("bad frame len", n) break body = self._recv_exact(n) if body is None: break ftype = body[0] payload = body[1:] if ftype == TYPE_AUDIO: self.feed_playback(payload) elif ftype == TYPE_CTRL: self._on_ctrl(payload) self.stop_audio() def _on_ctrl(self, payload): try: msg = json.loads(payload.decode()) except Exception: return cmd = msg.get("cmd") if cmd == "start": self.start_audio(msg.get("mac")) elif cmd == "stop": self.stop_audio() self.send_ctrl({"event": "stopped"}) def main(): log(f"connecting to phone {PHONE_HOST}:{PHONE_PORT}; frame={FRAME_BYTES}B @ {RATE}Hz") backoff = 1 while True: try: sock = socket.create_connection((PHONE_HOST, PHONE_PORT), timeout=10) sock.settimeout(None) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) log("connected to phone") backoff = 1 Bridge(sock).run() log("phone disconnected") except (ConnectionRefusedError, OSError) as e: log("connect failed:", e) finally: try: sock.close() except Exception: pass time.sleep(backoff) backoff = min(backoff * 2, 15) if __name__ == "__main__": try: main() except KeyboardInterrupt: pass