diff --git a/rtcllm/rtc.old.py b/rtcllm/rtc.old.py new file mode 100644 index 0000000..b96ccd3 --- /dev/null +++ b/rtcllm/rtc.old.py @@ -0,0 +1,246 @@ +import os +import sys +sys.path.append('./mini_omni') + +import asyncio +import random +import json + +from functools import partial + +from appPublic.dictObject import DictObject +from appPublic.hf import hf_socks5proxy +from appPublic.worker import awaitify +from aiortc import MediaStreamTrack, VideoStreamTrack, AudioStreamTrack, RTCPeerConnection, RTCSessionDescription, RTCIceCandidate +from aiortc.sdp import candidate_from_sdp, candidate_to_sdp +from aiortc.contrib.media import MediaBlackhole, MediaPlayer, MediaRecorder, MediaRelay + +# from websockets.asyncio.client import connect +# from websockets.asyncio.client import connect +from websockets.client import connect +import websockets +from stt import asr +from vad import AudioTrackVad +from a2a import LLMAudioStreamTrack +from aav import MyMediaPlayer, MyAudioStreamTrack, MyVideoStreamTrack + +from mini_omni.inference import OmniInference + +videos = ['./1.mp4', './2.mp4'] + +async def pc_get_local_candidates(pc, peer): + its = pc.__dict__.get('_RTCPeerConnection__iceTransports') + coros = map(lambda t: t.iceGatherer.gather(), its ) + await asyncio.gather(*coros) + ss = pc.__dict__.get('_RTCPeerConnection__iceTransport') + + if not peer.l_candidates: + peer.l_candidates = [] + peer.sdp_id = 0 + + for t in its: + for c in t._connection.local_candidates: + if c not in peer.l_candidates: + # print(f'{c=}, {dir(c)}') + c.sdpMid = str(peer.sdp_id) + peer.sdp_id += 1 + peer.l_candidates.append(c) + pc.emit('icecandidate', c) + +RTCPeerConnection.get_local_candidates = pc_get_local_candidates + +class RTCLLM: + def __init__(self, ws_url, iceServers): + self.ws_url = ws_url + self.omni_infer = OmniInference(ckpt_dir='/d/models/mini-omni') + self.iceServers = iceServers + self.peers = DictObject() + self.vid = 0 + self.info = { + 'id':'rtcllm_agent', + 'name':'rtcllm_agent' + } + self.dc = None + self.loop = asyncio.get_event_loop() + + async def ws_send(self, s): + await self.ws.send(s) + + def get_pc(self, data): + return self.peers[data['from'].id].pc + + async def run(self): + async with connect(self.ws_url) as self.ws: + await self.login() + while True: + msg = await self.ws.recv() + data = DictObject(**json.loads(msg)) + d = data.data + print(f'ws recv(): {d.type=}') + func = self.handlers.get(d.type) + if func: + f = partial(func, self) + await f(d) + self.ws.close() + + async def login(self): + await self.ws_send(json.dumps({ + 'type':'login', + 'info':self.info + })) + + async def on_icecandidate(self, pc, to, candidate): + if candidate: + candi = { + 'candidate':'candidate:' + candidate.to_sdp(), + 'sdpMid':candidate.sdpMid, + 'type': candidate.type + } + # print('***********on_icecandidate()', candi) + await self.ws_send(json.dumps({ + "type":"iceCandidate", + "to":to, + "candidate":candi + })) + + async def save_onlineList(self, data): + # print(f'{self}, {type(self)}') + self.onlineList = data.onlineList + + async def vad_voiceend(self, peer, audio): + if audio is not None: + feed = awaitify(peer.llmtrack._feed) + ret = await feed(audio) + print(f'self.feed("{audio}") return {ret}') + # os.remove(audio) + + async def auto_accept_call(self, data): + opts = DictObject(iceServers=self.iceServers) + pc = RTCPeerConnection(opts) + player = MyMediaPlayer('./1.mp4') + llmtrack = LLMAudioStreamTrack(self.omni_infer) + self.peers[data['from'].id] = DictObject(**{ + 'info':data['from'], + 'llmtrack':llmtrack, + 'player':player, + 'pc':pc + }) + pc.addTrack(llmtrack) + # pc.addTrack(MyAudioStreamTrack(player)) + pc.addTrack(MyVideoStreamTrack(player)) + # pc.addTrack(LoopingVideoTrack('./1.mp4')) + # pc.addTrack(player.video) + await self.ws_send(json.dumps({'type':'callAccepted', 'to':data['from']})) + print('auto_accept_call() end') + + async def pc_track(self, peerid, track): + peer = self.peers[peerid] + pc = peer.pc + if track.kind == 'audio': + f = partial(self.vad_voiceend, peer) + vadtrack = AudioTrackVad(track, stage=3, onvoiceend=f) + peer.vadtrack = vadtrack + vadtrack.start_vad() + + async def pc_connectionState_changed(self, peerid): + peer = self.peers[peerid] + pc = peer.pc + print(f'conn_state={pc.connectionState} ...........') + if pc.connectionState == 'connected': + peer.dc = pc.createDataChannel(peer.info.name) + return + if pc.connectionState == 'closed': + await pc.close() + if peer.dc: + await peer.dc.close() + await self.ws_send(json.dumps({ + 'type':'disconnect', + 'to': peer.info + })) + peers = { + k:v for k,v in self.peers.items() if k != peerid + } + self.peers = peers + + async def response_offer(self, data): + pc = self.get_pc(data) + peer = self.peers[data['from'].id] + if pc is None: + print(f'{self.peers=}, {data=}') + return + pc.on("connectionstatechange", partial(self.pc_connectionState_changed, data['from'].id)) + pc.on('track', partial(self.pc_track, data['from'].id)) + pc.on('icecandidate', partial(self.on_icecandidate, pc, data['from'])) + + offer = RTCSessionDescription(** data.offer) + await pc.setRemoteDescription(offer) + answer = await pc.createAnswer() + await pc.setLocalDescription(answer) + await self.ws_send(json.dumps({ + 'type':'answer', + 'answer':{'type':pc.localDescription.type, 'sdp':pc.localDescription.sdp}, + 'to':data['from'] + })) + cands = await pc.get_local_candidates(peer) + + """ + offer = await pc.createOffer() + await pc.setLocalDescription(offer) + await self.ws_send(json.dumps({ + 'type':'offer', + 'offer': {'type':pc.localDescription.type, 'sdp':pc.localDescription.sdp}, + 'to':data['from'] + })) + """ + + async def accept_answer(self, data): + pc = self.get_pc(data) + answer = RTCSessionDescription(data.answer); + pc.setRemoteDescription(answer) + + async def accept_iceCandidate(self, data): + pc = self.get_pc(data) + candidate = data.candidate + # print('accepted candidate=', candidate) + """ + rtc_candidate = RTCIceCandidate( + ip=ip, + port=port, + protocol=protocol, + priority=priority, + foundation=foundation, + component=component, + type=type, + sdpMid=candidate['sdpMid'], + sdpMLineIndex=candidate['sdpMLineIndex'] + ) + """ + rtc_candidate = candidate_from_sdp(candidate['candidate'].split(":", 1)[1]) + rtc_candidate.sdpMid = candidate['sdpMid'] + rtc_candidate.sdpMLineIndex = candidate['sdpMLineIndex'] + await pc.addIceCandidate(rtc_candidate) + # print('addIceCandidate ok') + + handlers = { + 'onlineList':save_onlineList, + 'callRequest':auto_accept_call, + 'offer':response_offer, + 'answer':accept_answer, + 'iceCandidate':accept_iceCandidate, + } + +async def main(): + hf_socks5proxy() + agent = RTCLLM(ws_url='wss://sage.open-computing.cn/wss/ws/rtc_signaling.ws', + iceServers=[{ + 'urls':'stun:stun.open-computing.cn:13478'},{ + 'urls':'turn:stun.open-computing.cn:13479', + 'username':'turn', + 'credential':'server' + }]) + print('running ...') + await agent.run() + +if __name__ == '__main__': + asyncio.run(main()) + diff --git a/rtcllm/rtc.py b/rtcllm/rtc.py index b96ccd3..b4ac9f5 100644 --- a/rtcllm/rtc.py +++ b/rtcllm/rtc.py @@ -11,6 +11,7 @@ from functools import partial from appPublic.dictObject import DictObject from appPublic.hf import hf_socks5proxy from appPublic.worker import awaitify +from appPublic.log import debug, error, exception, critical from aiortc import MediaStreamTrack, VideoStreamTrack, AudioStreamTrack, RTCPeerConnection, RTCSessionDescription, RTCIceCandidate from aiortc.sdp import candidate_from_sdp, candidate_to_sdp from aiortc.contrib.media import MediaBlackhole, MediaPlayer, MediaRecorder, MediaRelay @@ -26,221 +27,400 @@ from aav import MyMediaPlayer, MyAudioStreamTrack, MyVideoStreamTrack from mini_omni.inference import OmniInference -videos = ['./1.mp4', './2.mp4'] -async def pc_get_local_candidates(pc, peer): - its = pc.__dict__.get('_RTCPeerConnection__iceTransports') - coros = map(lambda t: t.iceGatherer.gather(), its ) - await asyncio.gather(*coros) - ss = pc.__dict__.get('_RTCPeerConnection__iceTransport') - - if not peer.l_candidates: - peer.l_candidates = [] - peer.sdp_id = 0 - - for t in its: - for c in t._connection.local_candidates: - if c not in peer.l_candidates: - # print(f'{c=}, {dir(c)}') - c.sdpMid = str(peer.sdp_id) - peer.sdp_id += 1 - peer.l_candidates.append(c) - pc.emit('icecandidate', c) - -RTCPeerConnection.get_local_candidates = pc_get_local_candidates - -class RTCLLM: - def __init__(self, ws_url, iceServers): - self.ws_url = ws_url - self.omni_infer = OmniInference(ckpt_dir='/d/models/mini-omni') - self.iceServers = iceServers - self.peers = DictObject() - self.vid = 0 - self.info = { - 'id':'rtcllm_agent', - 'name':'rtcllm_agent' - } - self.dc = None - self.loop = asyncio.get_event_loop() - - async def ws_send(self, s): - await self.ws.send(s) - - def get_pc(self, data): - return self.peers[data['from'].id].pc +class RTCSignaling: + /* + { + signaling_url: + info: + connect_opts: + onclose: + onlogin: + } + */ + def __init__(self, opts): + self.signaling_url = opts.signaling_url + self.info = opts.info + self.connect_opts = opts.connect_opts + self.peers = [] + self.sessions = DictObject() + self.handlers = DictObject() + self.sessionhandlers = DictObject() + self.init_websocket() + self.hb_task = None + self.socket = None + self.heartbeat_period = opts.heartbeat_period + self.onclose = opts.onclose + self.onlogin = opts.onlogin + self.onopen = opts.onopen + if not self.heartbeat_period: + self.heartbeat_period = 0 + async def run(self): - async with connect(self.ws_url) as self.ws: + async with connect(self.ws_url) as self.socket: await self.login() while True: - msg = await self.ws.recv() + msg = await self.socket.recv() data = DictObject(**json.loads(msg)) d = data.data - print(f'ws recv(): {d.type=}') - func = self.handlers.get(d.type) - if func: - f = partial(func, self) - await f(d) - self.ws.close() + debug(f'ws recv(): {d.type=}') + await self.signaling_recvdata(d) + await self.socket.close() + + def reconnect(self): + debug('eror happened') + if self.hb_task: + self.hb_task.cancel() + self.hb_task = None + + if self.onclose: + self.onclose() + + return + + def del_handler(self, sessionid): + handlers = { k:v \ + for k,v in self.handlers.items() if k!=sessionid} + self.handlers = handlers + + async def signaling_recvdata(self, data): + debug(f'ws recv data={data}') + if data.session: + sessionid = data.session.sessionid + sessiontype = data.session.sessiontype + handler = self.handlers[sessionid] + if not handler: + k = self.sessionhandlers[sessiontype] + if not k: + e = Exception('recvdata_handler() exception(' + sessiontype + ')') + raise e + + h = k(self, data.session, self.connect_opts) + handler = h.recvdata_handler.bind(h) + self.add_handler(sessionid, handler) + + await handler(data) + else: + await self.recvdata_handler(data) + + async def add_handler(self, key, handler): + self.handlers[key] = handler + + async def add_sessionhandler(self, sessiontype, handler): + self.sessionhandlers[sessiontype] = handler + + async def recvdata_handler(self, data): + debug(f'recv data={data}') + if data.type == 'online'): + for p in data.online: + d = self.peers[p.id] + if (!d) d = DictOject() + d.update(p) + self.peers[p.id] = d + + if self.onlogin: + self.onlogin(data.online) + + return + + debug(f'recv data= {data} NOT HANDLED') + + async def new_session(self, sessiontype, peer): + k = self.sessionhandlers[sessiontype] + if not k: + e = Exception('new_session() exception(' + sessiontype + ')') + raise e + sessionid = getID() + session = DictObject(**{ + sessiontype:sessiontype, + sessionid:sessionid + }) + d = DictObject(**{ + "type":'new_session', + "session": session + }) + await self.send_data(d) + opts = self.connect_options.clone() + opts.peer_info = peer + h = k(self, session, opts) + self.add_handler(sessionid, h.recvdata_handler) + return h async def login(self): - await self.ws_send(json.dumps({ - 'type':'login', - 'info':self.info - })) + debug(f'login send {self.heartbeat_period=}') + d = DictObject( + type='login' + ) + self.send_data(d) + if self.heartbeat_period > 0): + debug(f'call login again in {self.heartbeat_period} seconds') + self.hb_task = schedule_once(self.login.bind(self), self.heartbeat_period) + - async def on_icecandidate(self, pc, to, candidate): - if candidate: - candi = { - 'candidate':'candidate:' + candidate.to_sdp(), - 'sdpMid':candidate.sdpMid, - 'type': candidate.type - } - # print('***********on_icecandidate()', candi) - await self.ws_send(json.dumps({ - "type":"iceCandidate", - "to":to, - "candidate":candi - })) - - async def save_onlineList(self, data): - # print(f'{self}, {type(self)}') - self.onlineList = data.onlineList + async def logout(self): + d = DictObject( + type='logout' + ) + await self.send_data(d) - async def vad_voiceend(self, peer, audio): - if audio is not None: - feed = awaitify(peer.llmtrack._feed) - ret = await feed(audio) - print(f'self.feed("{audio}") return {ret}') - # os.remove(audio) - - async def auto_accept_call(self, data): - opts = DictObject(iceServers=self.iceServers) - pc = RTCPeerConnection(opts) - player = MyMediaPlayer('./1.mp4') - llmtrack = LLMAudioStreamTrack(self.omni_infer) - self.peers[data['from'].id] = DictObject(**{ - 'info':data['from'], - 'llmtrack':llmtrack, - 'player':player, - 'pc':pc - }) - pc.addTrack(llmtrack) - # pc.addTrack(MyAudioStreamTrack(player)) - pc.addTrack(MyVideoStreamTrack(player)) - # pc.addTrack(LoopingVideoTrack('./1.mp4')) - # pc.addTrack(player.video) - await self.ws_send(json.dumps({'type':'callAccepted', 'to':data['from']})) - print('auto_accept_call() end') + await send_data(self, d): + d.msgfrom = self.info + s = json.dumps(d) + debug(f'send_data() {s=}') + await self.socket.send(s) - async def pc_track(self, peerid, track): - peer = self.peers[peerid] - pc = peer.pc - if track.kind == 'audio': - f = partial(self.vad_voiceend, peer) - vadtrack = AudioTrackVad(track, stage=3, onvoiceend=f) - peer.vadtrack = vadtrack - vadtrack.start_vad() + async def socket_send(self, s): + await self.socket.send(s) - async def pc_connectionState_changed(self, peerid): - peer = self.peers[peerid] - pc = peer.pc - print(f'conn_state={pc.connectionState} ...........') - if pc.connectionState == 'connected': - peer.dc = pc.createDataChannel(peer.info.name) +class RTCP2PConnect: + /* + opts:{ + ice_servers: + peer_info: + auto_callaccept: true or false + media_options: { video:trur or false, audio:true or false } + data_connect: true or false + } + */ + + def __init__(self, signaling, session, opts): + self.id = bricks.uuid() + self.signaling = signaling + self.session = session + self.requester = false + self.opts = opts + self.peers = DictObject() + self.signal_handlers = DictObject() + self.local_stream = None + self.localVideo = None + self.add_handler('sessioncreated', self.h_sessioncreated) + self.add_handler('callrequest', self.h_callrequest) + self.add_handler('callaccepted', self.h_callaccepted) + self.add_handler('offer', self.h_offer) + self.add_handler('answer', self.h_answer) + self.add_handler('icecandidate', self.h_icecandidate) + self.add_handler('sessionquit', self.h_sessionquit) + + def add_handler(self, typ, f): + self.signal_handlers[typ] = f + + def get_handler(self, typ): + return self.signal_handlers[typ] + + + async def p2pconnect(self, peer): + await self.getLocalStream() + p = self.peers.get(peer.id) + if not p: + await self.createPeerConnection(peer) + else: + debug(f'{peer=}, connect exists {self=}') + + debug(f'p2pconnect() called {self=} {peer=}') + + async def h_sessioncreated(self, data): + await self.p2pconnect(self.opts.peer_info, 'requester') + if self.opts.peer_info: + d = DictObject(**{ + "type":'callrequest', + "msgto":self.opts.peer_info + }) + await self.signaling_send(d) + self.requester = true + + async def h_callrequest(self, data): + if self.opts.auto_callaccept or true): + await self.p2pconnect(data.msgfrom, 'responser') + d = DictObject(**{ + "type":'callaccepted', + "msgto":data.msgfrom + }) + await self.signaling_send(d) return - if pc.connectionState == 'closed': - await pc.close() - if peer.dc: - await peer.dc.close() - await self.ws_send(json.dumps({ - 'type':'disconnect', - 'to': peer.info - })) - peers = { - k:v for k,v in self.peers.items() if k != peerid - } - self.peers = peers - async def response_offer(self, data): - pc = self.get_pc(data) - peer = self.peers[data['from'].id] - if pc is None: - print(f'{self.peers=}, {data=}') - return - pc.on("connectionstatechange", partial(self.pc_connectionState_changed, data['from'].id)) - pc.on('track', partial(self.pc_track, data['from'].id)) - pc.on('icecandidate', partial(self.on_icecandidate, pc, data['from'])) - - offer = RTCSessionDescription(** data.offer) + async def h_callaccepted(self, data): + self.createDataChannel(data.msgfrom) + await self.send_offer(data.msgfrom, true) + + async def h_offer(self, data): + debug(f'h_offer(), {self=}, peer={data.msgfrom}') + pc = self.peers[data.msgfrom.id].pc + offer = RTCSessionDescription(data.offer) await pc.setRemoteDescription(offer) answer = await pc.createAnswer() await pc.setLocalDescription(answer) - await self.ws_send(json.dumps({ - 'type':'answer', - 'answer':{'type':pc.localDescription.type, 'sdp':pc.localDescription.sdp}, - 'to':data['from'] - })) - cands = await pc.get_local_candidates(peer) - - """ - offer = await pc.createOffer() - await pc.setLocalDescription(offer) - await self.ws_send(json.dumps({ - 'type':'offer', - 'offer': {'type':pc.localDescription.type, 'sdp':pc.localDescription.sdp}, - 'to':data['from'] - })) - """ + d = DictObject(**{ + type:'answer', + answer:pc.localDescription, + msgto:data.msgfrom + }) + await self.signaling_send(d) + if self.peers[data.msgfrom.id].role != 'requester': + await self.send_offer(data.msgfrom) + + async def h_answer(self, data): + desc = RTCSessionDescription(data.answer) + pc = self.peers[data.msgfrom.id].pc + await pc.setRemoteDescription(desc) - async def accept_answer(self, data): - pc = self.get_pc(data) - answer = RTCSessionDescription(data.answer); - pc.setRemoteDescription(answer) + async def h_icecandidate(self, data): + candidate = RTCIceCandidate(data.candidate) + pc = self.peers[data.msgfrom.id].pc + await pc.addIceCandidate(candidate) - async def accept_iceCandidate(self, data): - pc = self.get_pc(data) - candidate = data.candidate - # print('accepted candidate=', candidate) - """ - rtc_candidate = RTCIceCandidate( - ip=ip, - port=port, - protocol=protocol, - priority=priority, - foundation=foundation, - component=component, - type=type, - sdpMid=candidate['sdpMid'], - sdpMLineIndex=candidate['sdpMLineIndex'] - ) - """ - rtc_candidate = candidate_from_sdp(candidate['candidate'].split(":", 1)[1]) - rtc_candidate.sdpMid = candidate['sdpMid'] - rtc_candidate.sdpMLineIndex = candidate['sdpMLineIndex'] - await pc.addIceCandidate(rtc_candidate) - # print('addIceCandidate ok') - - handlers = { - 'onlineList':save_onlineList, - 'callRequest':auto_accept_call, - 'offer':response_offer, - 'answer':accept_answer, - 'iceCandidate':accept_iceCandidate, + async def h_sessionquit(self, data): + pc = self.peers[data.msgfrom.id].pc + self.peer_close(data.msgfrom) + + async send_offer(self, peer, initial){ + debug(f'send_offer(), peers={self.peers}, {peer=}') + pc = self.peers[peer.id].pc + if initial: + self.peers[peer.id].role = 'requester' + else: + self.peers[peer.id].role = 'responser' + + offer = await pc.createOffer() + await pc.setLocalDescription(offer) + d = DictObject(**{ + "type":'offer', + "offer":pc.localDescription, + "msgto":peer + }) + await self.signaling_send(d) + + async def send_candidate(self, peer, event): + debug(f'send_candidate called, {peer=},{event=}') + if (event.candidate) { + candidate = event.candidate + self.signaling_send({ + type: 'icecandidate', + msgto:peer, + candidate: candidate + }) + } } -async def main(): - hf_socks5proxy() - agent = RTCLLM(ws_url='wss://sage.open-computing.cn/wss/ws/rtc_signaling.ws', - iceServers=[{ - 'urls':'stun:stun.open-computing.cn:13478'},{ - 'urls':'turn:stun.open-computing.cn:13479', - 'username':'turn', - 'credential':'server' - }]) - print('running ...') - await agent.run() + signaling_send(d){ + d.session = self.session + self.signaling.send_data(d) + } -if __name__ == '__main__': - asyncio.run(main()) + async recvdata_handler(data){ + debug('recvdata=', data, self.signal_handlers) + f = self.get_handler(data.type) + if (f) { + await f(data) + return + } + debug('recvdata=', data, 'NOT HANDLED') + } + + async ice_statechange(peer, event){ + pc = self.peers[peer.id].pc + debug(`oniceconnectionstatechange, pc.iceConnectionState is ${pc.iceConnectionState}.`) + } + + async connection_statechange(peer, event){ + pc = self.peers[peer.id].pc + debug(`${peer.id} state changed. new state=${pc.connectionState}`) + debug('state=', pc.connectionState, typeof(pc.connectionState)) + if (pc.connectionState == 'disconnected'){ + self.peer_close(peer) + if (self.opts.on_pc_disconnected){ + self.opts.on_pc_disconnected(peer) + } + return + } + if (pc.connectionState == 'connected'){ + debug('state is connected, data_connect=', + self.opts.data_connect) + if(self.opts.on_pc_connected){ + self.opts.on_pc_connected(peer) + } + } + } + + async dc_accepted(peer, event){ + debug('accept datachannel ....') + self.peers[peer.id].dc = event.channel + await self.dc_created(peer, self.peers[peer.id].dc) + } + async dc_created(self, peer, dc): + debug('dc_created.....', dc) + dc.onmessage = self.datachannel_message.bind(peer) + dc.onopen = self.datachannel_open(peer) + dc.onclose = self.datachannel_close(peer) + + async datachannel_message(self, peer, event): + debug('datachannel_message():', self, arguments) + dc = self.peers[peer.id].dc + if self.opts.on_dc_messaage: + await self.opts.on_dc_message(dc, event.data) + + async def datachannel_open(self, peer): + debug('datachannel_open():', self, arguments) + dc = self.peers[peer.id].dc + if self.opts.on_dc_open: + await self.opts.on_dc_open(dc) + + async def datachannel_close(self, peer): + debug('datachannel_close():', self, arguments) + dc = self.peers[peer.id].dc + if self.opts.on_dc_close: + await self.opts.on_dc_close(dc) + + async def createDataChannel(self, peer): + pc = self.peers[peer.id].pc + self.peers[peer.id].dc = pc.createDataChannel('chat', {ordered:true}) + dc = self.peers[peer.id].dc + await self.dc_created(peer, self.peers[peer.id].dc) + debug('dc created', self.peers[peer.id].dc) + + async def createPeerConnection(self, peer): + configuration = DictObject(**{ + "iceServers":self.opts.ice_servers + }) + debug(f'RTCPC {configuration=}') + pc = RTCPeerConnection(configuration) + if self.local_stream: + for track in self.local_stream.getTracks(): + pc.addTrack(track, self.local_stream) + self.peers[peer.id] = peer + self.peers[peer.id].pc = pc + self.peers[peer.id].role = '' + pc.onicecandidate = partial(self.send_candidate, peer) + pc.oniceconnectionstatechange = \ + partial(self.ice_statechange, peer) + pc.onconnectionstatechange = \ + partial(self.connection_statechange, peer) + pc.ondatachanel = partial(self.dc_accepted, peer) + pc.ontrack = partial(self.onpctrack, peer) + + def peer_close(self, peer){ + pc = self.peers[peer.id].pc + video = self.peers[peer.id].video + for r in pc.getReceivers(): + r.track.stop() + + for s in pc.getSenders(): + s.track.stop() + + dc = self.peers[peer.id].dc + if (dc){ + dc.close() + } + pc.close() + self.peers = {k:v for k,v in self.peers.items() if k!=peer.id} + cnt = len(self.peers.keys()) + if (keys.length == 0){ + self.localVideo.get_stream() + .getTracks().forEach(track => track.stop()) + self.local_stream.getTracks().forEach(track => track.stop()) + self.local_screan.getTracks().forEach(track => track.stop()) + self.signaling.del_handler(self.session.sessionid) + } + } +}