"""Ephemeral test peer: generated identity and synthetic media, no saved keys."""
import asyncio, base64, fractions, hashlib, json, time
import aiohttp
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes, serialization
from aiortc import RTCPeerConnection, RTCConfiguration, RTCIceServer, RTCSessionDescription, VideoStreamTrack, AudioStreamTrack
from aiortc.sdp import candidate_from_sdp
from av import VideoFrame

class TestVideo(VideoStreamTrack):
    async def recv(self):
        pts, tb = await self.next_timestamp()
        frame = VideoFrame(320, 240, 'yuv420p')
        for index, plane in enumerate(frame.planes): plane.update(bytes([100 if index == 0 else 128]) * plane.buffer_size)
        frame.pts, frame.time_base = pts, tb
        return frame

async def main():
    async with aiohttp.ClientSession() as http:
        token = None
        async def rpc(op, payload):
            async with http.post(f'http://127.0.0.1:5099/api/ikar/requests/{op}',json={'payload':payload},headers={'Authorization':f'Bearer {token}'} if token else {}) as r:
                data = await r.json()
                if data.get('cmd') != 1: raise RuntimeError(f'RPC {op}: {data.get("payload")}')
                return data['payload']
        key=ec.generate_private_key(ec.SECP256R1())
        public=key.public_key().public_bytes(serialization.Encoding.DER,serialization.PublicFormat.SubjectPublicKeyInfo)
        challenge=await rpc(17,{'phone':f'+999{int(time.time()*1000)}','installationId':hashlib.sha256(public).hexdigest()[:32],'devicePublicKey':base64.b64encode(public).decode()})
        signature=key.sign(base64.b64decode(challenge['signaturePayload']),ec.ECDSA(hashes.SHA256()))
        verified=await rpc(18,{'token':challenge['token'],'deviceSignature':base64.b64encode(signature).decode()})
        token=verified['tokenAttrs']['REGISTER']['token']
        profile=await rpc(23,{'firstName':'IKAR video test','lastName':'Synthetic media'})
        token=profile['token']; user_id=profile['accountId']
        pc=None
        try:
            target=(await rpc(46,{'phone':'+12025550101'}))['contact']['id']
            call=await rpc(78,{'calleeIds':[target],'isVideo':True,'conversationId':f'ikar-rtp-test-{time.time_ns()}'})
            endpoint=json.loads(call['internalCallerParams'])['endpoint']
            # The own test peer uses the local server socket; native APK uses HTTPS.
            endpoint=endpoint.replace('wss://ikar.kusoft.xyz','ws://127.0.0.1:5099')
            async with http.ws_connect(endpoint,heartbeat=10) as socket:
                initial=await socket.receive_json()
                cfg=initial['conversationParams']
                servers=[]
                for kind in ('stun','turn'):
                    s=cfg.get(kind)
                    if s: servers.append(RTCIceServer(s['urls'],username=s.get('username'),credential=s.get('credential')))
                pc=RTCPeerConnection(RTCConfiguration(iceServers=servers))
                pc.addTrack(AudioStreamTrack()); pc.addTrack(TestVideo())
                frames={'audio':0,'video':0}
                readers=[]
                @pc.on('track')
                def on_track(track):
                    async def consume():
                        try:
                            while True:
                                await track.recv(); frames[track.kind]+=1
                        except Exception: pass
                    readers.append(asyncio.create_task(consume()))
                @pc.on('connectionstatechange')
                async def state(): print(json.dumps({'state':pc.connectionState}),flush=True)
                sequence=0
                async def transmit(data):
                    nonlocal sequence
                    sequence+=1
                    await socket.send_json({'command':'transmit-data','sequence':sequence,'data':data,'participantId':target})
                # Wait for the native user to accept before negotiating.
                print(json.dumps({'status':'CALLING','target':target,'testPeer':user_id}),flush=True)
                deadline=time.monotonic()+150
                offered=False
                while time.monotonic()<deadline:
                    try: event=await asyncio.wait_for(socket.receive_json(),timeout=2)
                    except asyncio.TimeoutError: event={}
                    if event.get('notification') in ('accepted-call','participant-joined') and not offered:
                        offered=True
                        await pc.setLocalDescription(await pc.createOffer())
                        await transmit({'sdp':{'type':pc.localDescription.type,'sdp':pc.localDescription.sdp}})
                    if event.get('notification')=='transmitted-data':
                        data=event.get('data',{})
                        if 'sdp' in data:
                            description=data['sdp']
                            await pc.setRemoteDescription(RTCSessionDescription(description['sdp'],description['type']))
                            if description['type']=='offer':
                                await pc.setLocalDescription(await pc.createAnswer())
                                await transmit({'sdp':{'type':pc.localDescription.type,'sdp':pc.localDescription.sdp}})
                        if data.get('candidate',{}).get('candidate'):
                            candidate=data['candidate']; ice=candidate_from_sdp(candidate['candidate'].removeprefix('candidate:'))
                            ice.sdpMid=candidate.get('sdpMid'); ice.sdpMLineIndex=candidate.get('sdpMLineIndex')
                            await pc.addIceCandidate(ice)
                    if frames['audio']>25 and frames['video']>5:
                        stats=await pc.getStats()
                        outbound=[{'kind':s.kind,'packetsSent':s.packetsSent,'bytesSent':s.bytesSent} for s in stats.values() if s.type=='outbound-rtp']
                        report={'passed':True,'connection':pc.connectionState,'receivedFrames':frames,'outbound':outbound,'syntheticMedia':True,'nativeDevice':'Android emulator API35'}
                        print(json.dumps(report),flush=True)
                        await socket.send_json({'command':'hangup','sequence':sequence+1,'reason':'FINISHED'})
                        break
                else: raise RuntimeError(f'Media timed out: {pc.connectionState}, received={frames}')
                for reader in readers: reader.cancel()
        finally:
            if pc: await pc.close()
            await rpc(199,{})

asyncio.run(main())