-
Notifications
You must be signed in to change notification settings - Fork 868
Description
this is my code:
import asyncio
import json
import uuid
from aiortc import RTCPeerConnection, RTCSessionDescription
import requests as r
from aiortc import RTCConfiguration, RTCIceServer
from aiortc.contrib.media import MediaPlayer
VOL_URL = "https://huoshanpush.xinkongan.com/live/adminai.sdp?expire=1749198033&sign=82974154c31d14a71df3b859b555b91c"
def get_answer_sdp(offer: str):
headers = {
"Content-Type": "text/plain; charset=utf-8"
}
session_id = str(uuid.uuid4())
vol_url = f"{VOL_URL}"
body = {
"localSdp": {
"sdp": offer,
"type": "offer"
},
"sessionId": session_id
}
response = r.post(vol_url, headers=headers, data=json.dumps(body).encode('utf-8'))
result = json.loads(response.text)
return result['remoteSdp']['sdp']
async def run_webrtc_push():
config = RTCConfiguration([
RTCIceServer(urls=["stun:stun.music.aliyun.com:19302"])
])
pc = RTCPeerConnection(configuration=config)
pc.addTransceiver("video", direction="sendonly")
# 这里若需要主动发送视频流需要手动拿出 PlayerStreamTrack 对象进行发送
media_source = MediaPlayer("DJI_20240502135112_0004_S.MP4")
pc.addTrack(media_source.video)
# 生成 offer
offer = await pc.createOffer()
await pc.setLocalDescription(offer)
print("[Local Offer SDP]\n", pc.localDescription.sdp)
# 获取 answer
answer_sdp = get_answer_sdp(pc.localDescription.sdp)
answer = RTCSessionDescription(sdp=answer_sdp, type="answer")
await pc.setRemoteDescription(answer)
print("[Remote SDP Answer Received]")
print(f"answer sdp \n{answer_sdp}")
# 保持连接
print("[WebRTC] Start streaming...")
try:
while True:
await asyncio.sleep(300)
except KeyboardInterrupt:
pass
finally:
await pc.close()
print("[WebRTC] Closed")
if name == "main":
asyncio.run(run_webrtc_push())
