chore: init SIP research repo (Asterisk & LiveKit)

This commit is contained in:
杨豪
2026-08-31 13:46:57 +08:00
commit 5a54d9fa0c
23 changed files with 1594 additions and 0 deletions
+15
View File
@@ -0,0 +1,15 @@
# LiveKit Server 配置 — 单机双节点共享 Redis 组成集群
port: 7880
bind_addresses: [""]
redis:
address: redis:6379
keys:
devkey: "f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
rtc:
tcp_port: 7881
port_range_start: 50100
port_range_end: 50200
use_external_ip: false
logging:
level: info
json: true
+12
View File
@@ -0,0 +1,12 @@
# livekit-sip 网关配置 — sip1/sip2 共用(各自独立 netns, 端口不冲突)
api_key: devkey
api_secret: "f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
ws_url: ws://livekit:7880
redis:
address: redis:6379
health_port: 8081
prometheus_port: 9091
log_level: info
sip_port: 5060
rtp_port: 10000-10200
use_external_ip: false
+75
View File
@@ -0,0 +1,75 @@
name: livekit-sip
services:
redis:
image: redis:7-alpine
container_name: lk-redis
command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru --save "" --appendonly no
restart: unless-stopped
networks: [lksip]
livekit:
image: livekit/livekit-server:latest
container_name: lk-livekit
command: --config /etc/livekit.yaml
volumes:
- ./conf/livekit.yaml:/etc/livekit.yaml:ro
ports: ["7880:7880"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis]
# 第二个 LiveKit 节点: 同一 Redis 即自动组成多节点集群(水平扩展验证)
livekit2:
image: livekit/livekit-server:latest
container_name: lk-livekit2
command: --config /etc/livekit.yaml
volumes:
- ./conf/livekit.yaml:/etc/livekit.yaml:ro
ports: ["7883:7880"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis]
profiles: [multinode]
# SIP 网关 x2: 无状态, 经 Redis 被随机调度, 天然水平扩展
sip1:
image: livekit/sip:latest
container_name: lk-sip1
command: --config /etc/sip.yaml
volumes:
- ./conf/sip.yaml:/etc/sip.yaml:ro
ports: ["8081:8081", "9091:9091"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis, livekit]
sip2:
image: livekit/sip:latest
container_name: lk-sip2
command: --config /etc/sip.yaml
volumes:
- ./conf/sip.yaml:/etc/sip.yaml:ro
ports: ["8082:8081", "9092:9091"]
restart: unless-stopped
networks: [lksip]
depends_on: [redis, livekit]
# Mock SIP 运营商 (UAS): 应答 livekit-sip 的出站 INVITE
mock-provider:
image: python:3.12-alpine
container_name: lk-mock-provider
working_dir: /app
command: python mock/mock-provider.py
volumes:
- ./mock:/app/mock:ro
environment:
SIP_PORT: "5060"
RTP_PORT: "40000"
RING_DELAY: "1.5"
restart: unless-stopped
networks: [lksip]
networks:
lksip:
driver: bridge
+78
View File
@@ -0,0 +1,78 @@
#!/usr/bin/env python3
"""Mock AI Agent: 加入房间, 发布 440Hz 间歇音(模拟 TTS 输出),
订阅 SIP 参与者音频并统计 RMS(模拟 ASR 输入), 验证双向语音。
依赖: pip install livekit
"""
import argparse, array, asyncio, math, struct, time
from livekit import rtc
from livekit.api import AccessToken, VideoGrants
def rms(data) -> float:
buf = bytes(data)[:4096]
a = array.array("h"); a.frombytes(buf)
if not a: return 0.0
return (sum(x * x for x in a) / len(a)) ** 0.5
async def main():
ap = argparse.ArgumentParser()
ap.add_argument("--url", default="ws://127.0.0.1:7880")
ap.add_argument("--api-key", default="devkey")
ap.add_argument("--api-secret", default="f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04")
ap.add_argument("--room", required=True)
ap.add_argument("--identity", default="mock-agent")
ap.add_argument("--duration", type=float, default=20)
args = ap.parse_args()
token = (AccessToken(args.api_key, args.api_secret)
.with_identity(args.identity).with_name("MockAgent")
.with_grants(VideoGrants(room_join=True, room=args.room,
can_publish=True, can_subscribe=True)))
room = rtc.Room()
stats = {"frames": 0, "rms_sum": 0.0, "src": None}
done = asyncio.Event()
def on_sub(track, pub, participant):
if track.kind != rtc.TrackKind.KIND_AUDIO:
return
print(f"[agent] subscribed: {participant.identity} track={track.sid}", flush=True)
async def consume():
stream = rtc.AudioStream(track)
async for ev in stream:
f = ev.frame
stats["frames"] += 1
stats["rms_sum"] += rms(f.data)
if stats["frames"] % 250 == 0:
print(f"[agent] RX frames={stats['frames']} avg_rms={stats['rms_sum']/stats['frames']:.0f}", flush=True)
stats["src"] = asyncio.create_task(consume())
room.on("track_subscribed", on_sub)
await room.connect(args.url, token.to_jwt())
peers = [p.identity for p in room.remote_participants.values()]
print(f"[agent] joined room={args.room} peers={peers}", flush=True)
src = rtc.AudioSource(48000, 1)
track = rtc.LocalAudioTrack.create_audio_track("agent-tts", src)
await room.local_participant.publish_track(track)
print("[agent] publishing 440Hz tone (500ms on / 500ms off)", flush=True)
async def tone():
t, end = 0.0, time.time() + args.duration
while time.time() < end:
samples = []
on = (t % 1.0) < 0.5
for _ in range(960): # 20ms @ 48kHz
v = int(6000 * math.sin(2 * math.pi * 440 * t)) if on else 0
samples.append(v); t += 1 / 48000
buf = struct.pack("<960h", *samples)
await src.capture_frame(rtc.AudioFrame(buf, 48000, 1, 960))
await asyncio.sleep(0.02)
await tone()
n, r = stats["frames"], (stats["rms_sum"] / stats["frames"]) if stats["frames"] else 0.0
print(f"RESULT room={args.room} rx_frames={n} rx_avg_rms={r:.0f} "
f"bidirectional={'YES' if n > 50 and r > 100 else 'NO'}", flush=True)
await room.disconnect()
if __name__ == "__main__":
asyncio.run(main())
+237
View File
@@ -0,0 +1,237 @@
#!/usr/bin/env python3
"""Mock SIP 运营商 (UAS): 接收 livekit-sip 出站 INVITE, 振铃->应答, 双向收发 PCMU/PCMA RTP.
仅 stdlib, 无依赖。一个 RTP 端口复用所有呼叫(按来源 IP 匹配呼叫)。
用法: python3 mock-provider.py (环境变量 SIP_PORT/RTP_PORT/RING_DELAY)
"""
import asyncio, math, os, random, socket, struct, time
SIP_PORT = int(os.environ.get("SIP_PORT", "5060"))
RTP_PORT = int(os.environ.get("RTP_PORT", "40000"))
RING_DELAY = float(os.environ.get("RING_DELAY", "1.5"))
MY_IP = socket.gethostbyname(socket.gethostname())
def log(*a): print(f"[provider {time.strftime('%H:%M:%S')}]", *a, flush=True)
# ---------- G.711 编码 ----------
def ulaw_encode(s: int) -> int:
s = max(-32635, min(32635, s)); sign = 0x80 if s < 0 else 0
if sign: s = -s
s += 0x84
e = 7
for i in range(7, -1, -1):
if s & (0x40 << i): e = i; break
return (~(sign | (e << 4) | ((s >> (e + 3)) & 0x0F))) & 0xFF
def alaw_encode(s: int) -> int:
s = max(-32635, min(32635, s)); sign = 0x80 if s < 0 else 0
if sign: s = -s
e = 7
for i in range(7, -1, -1):
if s & (0x40 << i): e = i; break
m = (s >> (e + 3)) & 0x0F
return ((sign | (e << 4) | m) ^ 0x55) & 0xFF
def ulaw_decode(u: int) -> int:
u = ~u & 0xFF
sign = u & 0x80; e = (u >> 4) & 0x07; m = u & 0x0F
s = ((m << (e + 3)) + 0x84) << 2
return -s if sign else s
def tone_payload(codec: str, ts: float) -> bytes:
"""20ms / 160 samples @8kHz, 440Hz 500ms-on 500ms-off"""
out = bytearray()
on = (ts % 1.0) < 0.5
for i in range(160):
t = ts + i / 8000.0
v = int(6000 * math.sin(2 * math.pi * 440 * t)) if on else 0
out.append(ulaw_encode(v) if codec == "PCMU" else alaw_encode(v))
return bytes(out)
# ---------- SIP 报文 ----------
def parse_msg(data: bytes):
head, _, body = data.decode("utf-8", "replace").partition("\r\n\r\n")
lines = head.split("\r\n")
headers = []
for ln in lines[1:]:
n, _, v = ln.partition(":")
headers.append((n.strip().lower(), v.strip()))
return lines[0], headers, body
def hget(headers, name):
for n, v in headers:
if n == name: return v
return None
def build_resp(first, headers, code, reason, tag=None, extra=None, body=""):
out = [f"SIP/2.0 {code} {reason}"]
for n, v in headers:
if n == "via": out.append(f"Via: {v}")
out.append(f"From: {hget(headers,'from')}")
to = hget(headers, "to")
if tag and "tag=" not in to: to = f"{to};tag={tag}"
out.append(f"To: {to}")
out.append(f"Call-ID: {hget(headers,'call-id')}")
out.append(f"CSeq: {hget(headers,'cseq')}")
if extra:
out += extra
if body:
out.append(f"Content-Type: application/sdp")
out.append(f"Content-Length: {len(body)}")
return ("\r\n".join(out) + "\r\n\r\n" + body).encode()
def parse_sdp_offer(body: str):
"""返回 (对端媒体IP, 对端媒体端口, payload类型列表)"""
ip, port, pts = None, None, []
for ln in body.splitlines():
if ln.startswith("c=IN IP4 "): ip = ln.split()[2]
elif ln.startswith("m=audio "):
parts = ln.split(); port = int(parts[1]); pts = [p for p in parts[3:] if p.isdigit()]
return ip, port, pts
def sdp_answer(codec_pt: str, codec_name: str) -> str:
sid = random.randint(1, 10**9)
return (f"v=0\r\no=mock {sid} {sid+1} IN IP4 {MY_IP}\r\ns=mock\r\n"
f"c=IN IP4 {MY_IP}\r\nt=0 0\r\n"
f"m=audio {RTP_PORT} RTP/AVP {codec_pt}\r\n"
f"a=rtpmap:{codec_pt} {codec_name}/8000\r\na=sendrecv\r\n")
# ---------- 呼叫对象 ----------
class Call:
def __init__(self, first, headers, addr):
self.id = hget(headers, "call-id")
self.tag = os.urandom(5).hex()
self.first, self.headers, self.addr = first, headers, addr
self.peer_ip = self.peer_port = None
self.codec = "PCMU"; self.pt = "0"
self.ok_raw = None; self.confirmed = False; self.closed = False
self.sent = self.recv = self.recv_bytes = self.recv_amp_sum = 0
self.t0 = time.time()
self.last_resp = None
calls: dict[str, Call] = {}
rtp_sock = None
def pick_codec(pts):
if "0" in pts: return "0", "PCMU"
if "8" in pts: return "8", "PCMA"
return (pts[0], "PCMU") if pts else ("0", "PCMU")
async def ringing_then_answer(call, sip_t):
await asyncio.sleep(RING_DELAY)
if call.closed: return
body = sdp_answer(call.pt, call.codec)
call.ok_raw = build_resp(call.first, call.headers, 200, "OK", tag=call.tag,
extra=[f"Contact: <sip:mock@{MY_IP}:{SIP_PORT}>",
"User-Agent: MockProvider/1.0"], body=body)
call.last_resp = call.ok_raw
sip_t.sendto(call.ok_raw, call.addr)
log(f"ANSWERED call_id={call.id} media={MY_IP}:{RTP_PORT} codec={call.codec} peer={call.peer_ip}:{call.peer_port}")
start_rtp(call)
for _ in range(5): # 重传 200 OK 直到 ACK (UDP 可靠性)
await asyncio.sleep(1.0)
if call.confirmed or call.closed: break
sip_t.sendto(call.ok_raw, call.addr)
def start_rtp(call):
async def sender():
seq = random.randint(0, 65535); ts = random.randint(0, 10**6)
ssrc = random.randint(1, 2**31)
pt = 0 if call.codec == "PCMU" else 8
t0 = time.time()
while not call.closed:
now = time.time() - t0
payload = tone_payload(call.codec, now)
hdr = struct.pack("!BBHII", 0x80, pt, seq & 0xFFFF, ts & 0xFFFFFFFF, ssrc)
try:
rtp_sock.sendto(hdr + payload, (call.peer_ip, call.peer_port))
except OSError:
pass
call.sent += 1; seq += 1; ts += 160
await asyncio.sleep(0.02)
asyncio.create_task(sender())
def close_call(call, reason):
if call.closed: return
call.closed = True
dur = time.time() - call.t0
avg = call.recv_amp_sum / call.recv if call.recv else 0
log(f"CALL_DONE call_id={call.id} reason={reason} dur={dur:.1f}s "
f"sent_rtp={call.sent} recv_rtp={call.recv} recv_bytes={call.recv_bytes} recv_avg_amplitude={avg:.0f}")
calls.pop(call.id, None)
def on_sip(data, addr, sip_t):
first, headers, body = parse_msg(data)
method = first.split()[0] if first else "?"
cid = hget(headers, "call-id")
if method == "INVITE":
call = calls.get(cid)
if call: # 重传: 回最后一条响应
if call.last_resp: sip_t.sendto(call.last_resp, addr)
return
call = Call(first, headers, addr)
call.peer_ip, call.peer_port, pts = parse_sdp_offer(body)
call.pt, call.codec = pick_codec(pts)
calls[cid] = call
log(f"INCOMING_INVITE from={addr[0]}:{addr[1]} uri=\"{first.split()[1]}\" call_id={cid}")
sip_t.sendto(build_resp(first, headers, 100, "Trying"), addr)
sip_t.sendto(build_resp(first, headers, 180, "Ringing"), addr)
call.last_resp = build_resp(first, headers, 180, "Ringing")
asyncio.create_task(ringing_then_answer(call, sip_t))
elif method == "ACK":
call = calls.get(cid)
if call and not call.confirmed:
call.confirmed = True
log(f"ACK call_id={cid} confirmed, RTP bridging")
elif method == "BYE":
call = calls.get(cid)
sip_t.sendto(build_resp(first, headers, 200, "OK", tag=call.tag if call else None), addr)
if call: close_call(call, "bye")
elif method == "CANCEL":
call = calls.get(cid)
sip_t.sendto(build_resp(first, headers, 487, "Request Terminated",
tag=call.tag if call else None), addr)
if call: close_call(call, "cancelled")
elif method == "OPTIONS":
sip_t.sendto(build_resp(first, headers, 200, "OK", tag="mock"), addr)
class RtpProto(asyncio.DatagramProtocol):
def datagram_received(self, data, addr):
if len(data) < 12: return
pt = data[1] & 0x7F
payload = data[12:]
for call in list(calls.values()):
if not call.closed and call.peer_ip == addr[0]:
call.recv += 1; call.recv_bytes += len(data)
if payload and pt in (0, 8):
dec = [ulaw_decode(b) for b in payload[:40]]
call.recv_amp_sum += sum(abs(x) for x in dec) / len(dec)
return
class SipProto(asyncio.DatagramProtocol):
def __init__(self): self.t = None
def connection_made(self, transport): self.t = transport
def datagram_received(self, data, addr):
if self.t: on_sip(data, addr, self.t)
async def main():
global rtp_sock
loop = asyncio.get_running_loop()
sip_t, _ = await loop.create_datagram_endpoint(
lambda: SipProto(), local_addr=("0.0.0.0", SIP_PORT))
rtp_t, _ = await loop.create_datagram_endpoint(
RtpProto, local_addr=("0.0.0.0", RTP_PORT))
rtp_sock = rtp_t
log(f"mock provider up: sip=udp/{SIP_PORT} rtp=udp/{RTP_PORT} ip={MY_IP}")
log("waiting for outbound INVITE from livekit-sip ...")
await asyncio.Event().wait()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
+7
View File
@@ -0,0 +1,7 @@
{
"trunk": {
"name": "mock-provider",
"address": "mock-provider",
"numbers": ["+15105550123"]
}
}
+38
View File
@@ -0,0 +1,38 @@
#!/bin/bash
# 端到端外呼验证: agent 进房 -> CreateSIPParticipant 呼出 -> mock 运营商应答 -> 双向 RTP 统计
# 用法: ./call.sh [房间名] [被叫号码] [时长秒]
set -euo pipefail
cd "$(dirname "$0")/.."
ROOM="${1:-call-$(date +%s)}"
NUMBER="${2:-+15105550123}"
DURATION="${3:-15}"
SECRET="f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
LK="docker run --rm --network host -e LIVEKIT_URL=http://127.0.0.1:7880 \
-e LIVEKIT_API_KEY=devkey -e LIVEKIT_API_SECRET=$SECRET livekit/livekit-cli:latest"
echo "== 1. 确保 outbound trunk 存在 (mock-provider)"
TRUNK_ID=$($LK sip outbound list --json 2>/dev/null | python3 -c '
import json,sys
for t in json.load(sys.stdin).get("items", []):
if t.get("name") == "mock-provider":
print(t.get("sipTrunkId","")); break')
if [ -z "$TRUNK_ID" ]; then
TRUNK_ID=$($LK sip outbound create mock/trunk.json 2>/dev/null | awk '/^SIPTrunkID:/ {print $2}')
echo " created trunk: $TRUNK_ID"
else
echo " reuse trunk: $TRUNK_ID"
fi
echo "== 2. Mock agent 加入房间 $ROOM (发布TTS音, 订阅被叫音频)"
/opt/lk-venv/bin/python mock/mock-agent.py --room "$ROOM" --duration "$DURATION" &
AGENT_PID=$!
sleep 2
echo "== 3. 发起 SIP 外呼 -> $NUMBER"
$LK sip participant create --room "$ROOM" --identity "callee-$RANDOM" \
--trunk "$TRUNK_ID" --number "$NUMBER" --call "$NUMBER" \
--name "MockCallee" --wait --timeout 60s || true
echo "== 4. 等待 agent 结束并输出双向统计"
wait $AGENT_PID
+32
View File
@@ -0,0 +1,32 @@
#!/bin/bash
# 运行状态检测: 各组件健康 + 集群节点 + trunk 配置
set -uo pipefail
SECRET="f4f6c1a9e2b74d5893a0c8e17d2b6a45b7c93f1d6e284a05c9b3d7f2e6a81c04"
ok=0; fail=0
chk() { # name, command...
local name="$1"; shift
if out=$("$@" 2>&1); then echo "PASS $name"; ok=$((ok+1))
else echo "FAIL $name: $(echo "$out" | head -1)"; fail=$((fail+1)); fi
}
echo "== 容器状态"
docker ps --filter "name=lk-" --format "{{.Names}}\t{{.Status}}"
echo "== HTTP 健康"
code() { curl -s -o /dev/null -w "%{http_code}" --max-time 3 "$1"; }
for p in "livekit http://127.0.0.1:7880/" "sip1 http://127.0.0.1:8081/healthz" "sip2 http://127.0.0.1:8082/healthz"; do
set -- $p
c=$(code "$2")
[ "$c" = "200" ] || [ "$c" = "204" ] && { echo "PASS $1 ($c)"; ok=$((ok+1)); } || { echo "FAIL $1 ($c)"; fail=$((fail+1)); }
done
echo "== LiveKit API 功能检查"
LK=(docker run --rm --network host -e LIVEKIT_URL=http://127.0.0.1:7880
-e LIVEKIT_API_KEY=devkey -e LIVEKIT_API_SECRET=$SECRET livekit/livekit-cli:latest)
chk "lk room list" "${LK[@]}" room list
chk "lk sip outbound list" "${LK[@]}" sip outbound list
echo
echo "result: ok=$ok fail=$fail"
[ $fail -eq 0 ]