bugfix

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())
+

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:
+class RTCSignaling:
-		peer.l_candidates = []
+	/*
-		peer.sdp_id = 0
+	{
-
+		signaling_url:
-	for t in its:
+		info:
-		for c in  t._connection.local_candidates:
+		connect_opts:
-			if c not in peer.l_candidates:
+		onclose:
-				# print(f'{c=}, {dir(c)}')
+		onlogin:
-				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()
+	def __init__(self, opts):
-
+		self.signaling_url = opts.signaling_url
-	async def ws_send(self, s):
+		self.info = opts.info
-		await self.ws.send(s)
+		self.connect_opts = opts.connect_opts
-
+		self.peers = []
-	def get_pc(self, data):
+		self.sessions = DictObject()
-		return self.peers[data['from'].id].pc
+		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=}')
+				debug(f'ws recv(): {d.type=}')
-				func = self.handlers.get(d.type)
+				await self.signaling_recvdata(d)
-				if func:
+			await self.socket.close()
-					f = partial(func, self)
+
-					await f(d)
+	def reconnect(self):
-			self.ws.close()
+		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({
+		debug(f'login send {self.heartbeat_period=}')
-			'type':'login',
+		d = DictObject(
-			'info':self.info
+			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:
+	async def logout(self):
-			candi = {
+		d = DictObject(
-				'candidate':'candidate:' + candidate.to_sdp(),
+			type='logout'
-				'sdpMid':candidate.sdpMid,
+		)
-				'type': candidate.type
+		await self.send_data(d)
+	
+	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 socket_send(self, s):
+		await self.socket.send(s)
+
+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
 	}
-			# print('***********on_icecandidate()', candi)
+	*/
-			await self.ws_send(json.dumps({
-				"type":"iceCandidate",
-				"to":to,
-				"candidate":candi
-			}))
 
-	async def save_onlineList(self, data):
+	def __init__(self, signaling, session, opts):
-		# print(f'{self}, {type(self)}')
+		self.id = bricks.uuid()
-		self.onlineList = data.onlineList
+		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)
 
-	async def vad_voiceend(self, peer, audio):
+	def add_handler(self, typ, f):
-		if audio is not None:
+		self.signal_handlers[typ] = f
-			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):
+	def get_handler(self, typ):
-		opts = DictObject(iceServers=self.iceServers)
+		return self.signal_handlers[typ]
-		pc = RTCPeerConnection(opts)
+	
-		player = MyMediaPlayer('./1.mp4')
+
-		llmtrack = LLMAudioStreamTrack(self.omni_infer)
+	async def p2pconnect(self, peer):
-		self.peers[data['from'].id] = DictObject(**{
+		await self.getLocalStream()
-			'info':data['from'],
+		p = self.peers.get(peer.id)
-			'llmtrack':llmtrack,
+		if not p:
-			'player':player,
+			await self.createPeerConnection(peer)
-			'pc':pc
+		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
 			})
-		pc.addTrack(llmtrack)
+			await self.signaling_send(d)
-		# pc.addTrack(MyAudioStreamTrack(player))
+			self.requester = true
-		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):
+	async def h_callrequest(self, data):
-		peer = self.peers[peerid]
+		if self.opts.auto_callaccept or true):
-		pc = peer.pc
+			await self.p2pconnect(data.msgfrom, 'responser')
-		if track.kind == 'audio':
+			d = DictObject(**{
-			f = partial(self.vad_voiceend, peer)
+				"type":'callaccepted',
-			vadtrack = AudioTrackVad(track, stage=3, onvoiceend=f)
+				"msgto":data.msgfrom
-			peer.vadtrack = vadtrack
+			})
-			vadtrack.start_vad()
+			await self.signaling_send(d)
-
-	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):
+	async def h_callaccepted(self, data):
-		pc = self.get_pc(data)
+		self.createDataChannel(data.msgfrom)
-		peer = self.peers[data['from'].id]
+		await self.send_offer(data.msgfrom, true)
-		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_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({
+		d = DictObject(**{
-			'type':'answer',
+			type:'answer',
-			'answer':{'type':pc.localDescription.type, 'sdp':pc.localDescription.sdp},
+			answer:pc.localDescription,
-			'to':data['from']
+			msgto:data.msgfrom
-		}))
+		})
-		cands = await pc.get_local_candidates(peer)
+		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 h_icecandidate(self, data):
+		candidate = RTCIceCandidate(data.candidate)
+		pc = self.peers[data.msgfrom.id].pc
+		await pc.addIceCandidate(candidate)
+	
+	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)
-		await self.ws_send(json.dumps({
+        d = DictObject(**{
-			'type':'offer', 
+            "type":'offer',
-			'offer': {'type':pc.localDescription.type, 'sdp':pc.localDescription.sdp},
+            "offer":pc.localDescription,
-			'to':data['from']
+            "msgto":peer
-		}))
+        })
-		"""
+		await self.signaling_send(d)
 	
-	async def accept_answer(self, data):
+	async def send_candidate(self, peer, event):
-		pc = self.get_pc(data)
+		debug(f'send_candidate called, {peer=},{event=}') 
-		answer = RTCSessionDescription(data.answer);
+        if (event.candidate) {
-		pc.setRemoteDescription(answer)
+            candidate = event.candidate
-	
+            self.signaling_send({ 
-	async def accept_iceCandidate(self, data):
+                type: 'icecandidate', 
-		pc = self.get_pc(data)
+                msgto:peer,
-		candidate = data.candidate
+                candidate: 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():
+	signaling_send(d){
-	hf_socks5proxy()
+		d.session = self.session
-	agent = RTCLLM(ws_url='wss://sage.open-computing.cn/wss/ws/rtc_signaling.ws',
+		self.signaling.send_data(d)
-					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__':
+	async recvdata_handler(data){
-	asyncio.run(main())
+		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)
+		}
+	}
+}