import asyncio import random from time import time from libwavesync import time_machine import pyaudio from collections import deque from time import sleep #PLAY_CALLBACK=True QUEUE_MAX=12 q=deque(maxlen=QUEUE_MAX) # critical for callback variety; recived packet size must be the same as in initialization call glob_chunk_size=367*4 glob_stream_break=0 def snd_callback(in_data, frame_count, time_info, status): # print("callback called",status,buf_chunk_wait,pyaudio.paContinue, pyaudio.paComplete,len(buf_chunk)) # if queue is empty, return artificial silence if glob_stream_break > 0: return (b'\x00'*glob_chunk_size,pyaudio.paComplete) urun=0 while len(q)==0: if glob_stream_break > 0: return (b'\x00'*glob_chunk_size, pyaudio.paComplete) sleep(0.010) urun+=1 if urun == 3 and len(q)==0: print('queue underrun') #if len(q)==0: return (b'\x00'*glob_chunk_size,pyaudio.paContinue) # dirty hack: help empty overly long queue by dropping extra data if len(q)>12: print('qdrop') _=q.popleft() return (q.popleft(), pyaudio.paContinue) class ChunkPlayer: "Play received audio and keep sync" def __init__(self, chunk_queue, receiver, tolerance_ms, buffer_size, device_index, use_callback): # playback mode #global PLAY_CALLBACK self.PLAY_CALLBACK=use_callback # Our data source self.chunk_queue = chunk_queue # Required for showing unified stats. self.receiver = receiver # Configuration self.tolerance_ms = tolerance_ms # Audio state self.audio_config = None self.buffer_size = buffer_size self.device_index = device_index self.stream = None # Generate silence frames (zeroed) of appropriate sizes for chunks self.silence_cache = None # Number of silent frames that need to be inserted to get in sync self.silence_to_insert = 0 # Stats self.stat_time_drops = 0 self.stat_output_delays = 0 self.stat_total_delay = 0 # Stream self.pyaudio = None # Used to quit main loop self.stop = False # Calculated sizes self.frame_size = None self.chunk_frames = None # CMD_AUDIO = 1 # CMD_DROPS = 2 # CMD_CFG = 3 self.CMD_ADJUST=128 def clear_state(self): "Clear player queue" self.silence_to_insert = 0 # Clear the chunk list, but preserve CFG commands cfg = None for cmd, item in self.chunk_queue.chunk_list: if cmd == self.chunk_queue.CMD_CFG: cfg = item break self.chunk_queue.chunk_list.clear() if cfg is not None: self.chunk_queue.chunk_list.append((self.chunk_queue.CMD_CFG, cfg)) self.chunk_queue.do_recovery() def get_silent_chunk(self): "Generate and cache silent chunks" if self.silence_cache is not None: return self.silence_cache silent_chunk = b'\x00' * self.audio_config.chunk_size self.silence_cache = silent_chunk return silent_chunk def _open_stream(self): import pyaudio self.clear_state() # Open stream cfg = self.audio_config global glob_chunk_size glob_chunk_size=cfg.chunk_size # for callback: MUST BE ACTUALLY WORKING NUMBER - otherwise the stream will fail with both stream.is_active() and stream.is_stopped() as false if self.PLAY_CALLBACK: frames_per_buffer = int(cfg.chunk_size/cfg.frame_size) else: frames_per_buffer = self.buffer_size #AudioConfig 44100Hz 16bits 2ch latency=4000ms sink=-250ms size chunk=1468 frame=4 if self.device_index == -1: # We are tested. Don't open stream, but calculate chunk_frames. pass else: assert self.stream is None self.pyaudio = pyaudio.PyAudio() if self.device_index is None: config = self.pyaudio.get_host_api_info_by_index(0) device_index = config['defaultOutputDevice'] print('CONFIG:',config) print("Using default output device index", device_index) else: device_index = self.device_index audio_format = ( pyaudio.paInt24 if cfg.sample == 24 else pyaudio.paInt16 ) if self.PLAY_CALLBACK: stream = self.pyaudio.open(output=True, channels=cfg.channels, rate=cfg.rate, format=audio_format, frames_per_buffer=frames_per_buffer, output_device_index=device_index, stream_callback=snd_callback) else: stream = self.pyaudio.open(output=True, channels=cfg.channels, rate=cfg.rate, format=audio_format, frames_per_buffer=frames_per_buffer, output_device_index=device_index) self.stream = stream print('audio config:',self.audio_config) self.chunk_frames = self.audio_config.chunk_size / cfg.frame_size def _close_stream(self): self.stream.stop_stream() self.stream.close() self.pyaudio.terminate() self.stream = None async def chunk_player(self): "Reads asynchronously chunks from the list and plays them" cnt = 0 # Chunk/s stat recent_start = time() recent = 0 mark=0 mid_tolerance_s = self.tolerance_ms / 2 / 1000 one_ms = 1/1000.0 if self.PLAY_CALLBACK: wait_if_greater=one_ms*10 else: wait_if_greater=one_ms # max_delay = 5 max_delay = 85 global glob_stream_break # get initial NTP sync, thrice to be more accurate time_machine.ntp_getoffset() time_machine.ntp_getoffset() time_machine.ntp_runtimer() try: while not self.stop: if not self.chunk_queue.chunk_list: if self.audio_config is not None: print("Queue empty - waiting") # reopen stream - cheat for resyncing the audio queue on termux/android; will it work? if self.PLAY_CALLBACK: q.clear() print('RESYNC - stream reopen'); glob_stream_break=1 await asyncio.sleep(0.030) self.stream.stop_stream() print('RESYNC - stream wait'); await asyncio.sleep(0.005) print('RESYNC - stream restarting'); glob_stream_break=0 self.stream.start_stream() print('RESYNC - stream restarted'); # if self.stream: # print('closing audio device') # self._close_stream() # print('closed audio device') # await asyncio.sleep(10) # print('opening audio device') # self._open_stream() self.chunk_queue.chunk_available.clear() await self.chunk_queue.chunk_available.wait() recent_start = time() recent = 0 if self.audio_config is not None: await asyncio.sleep(self.audio_config.latency_ms / 1000 / 4) print("Got stream flowing. q_len=%d" % len(self.chunk_queue.chunk_list)) continue # for i in self.chunk_queue.chunk_list: # if i[0]==1: print(i[1][0]) cmd, item = self.chunk_queue.chunk_list.popleft() # CMD_AUDIO = 1 # CMD_DROPS = 2 # CMD_CFG = 3 if cmd == self.chunk_queue.CMD_CFG: print('CMDCFG:',item) print("Got new configuration - opening audio stream") self.clear_state() self.audio_config = item if self.stream: self._close_stream() self._open_stream() # Calculate maximum sensible delay in given configuration, in seconds max_delay = (2000 + self.audio_config.sink_latency_ms + self.audio_config.latency_ms) / 1000 #print("max_delay calculated:",max_delay) # Shad's mod #max_delay=60 # Shad's mod for testing on lousy-sync android #max_delay=15 # Shad's mod for testing on lousy-sync android print("Assuming maximum chunk delay of %.2fms in this setup" % (max_delay * 1000)) continue elif cmd == self.chunk_queue.CMD_DROPS: print('CMDDROPS:',item) if item > 200: print("Recovering after a huge packet loss of %d packets" % item) self.clear_state() else: # Just slowly resync self.silence_to_insert += item continue elif cmd == self.CMD_ADJUST: print('CMDADJUST:',item) # CMD_AUDIO if self.stream is None: # No output, no playing. continue # for packet time difference oldmark=mark mark, chunk = item # check to make packets with wrong timing visible; times valid for 44kHz/16bit/2ch if mark-oldmark < 0.0075 or mark-oldmark > 0.0095: print('out-of-spec td: {:.4f}'.format(mark-oldmark)) # throw out misordered packet if mark < oldmark: print('bad timing ordering, packet skipped') continue desired_time = mark - (self.audio_config.sink_latency_ms/1000.0) # 0) We got the next chunk to be played now = time_machine.now() # Negative when we're lagging behind. delay = desired_time - now self.stat_total_delay += delay recent += 1 cnt += 1 # Probabilistic drop of lagging chunks to get back on track. # Probability of drop is higher, the more chunk lags behind current # time. Similar to the RED algorithm in TCP congestion. if delay < -mid_tolerance_s: over = -delay - mid_tolerance_s prob = over / mid_tolerance_s if random.random() < prob: s = "Drop chunk: q_len=%2d delay=%.1fms < 0. tolerance=%.1fms: P=%.2f" s = s % (len(self.chunk_queue.chunk_list), delay * 1000, self.tolerance_ms, prob) print(s) self.stat_time_drops += 1 continue elif delay > max_delay: # Probably we hanged for so long time that the time recovering # mechanism rolled over. Recover print("Huge recovery - delay of %.2f exceeds the max delay of %.2f" % ( delay, max_delay)) self.clear_state() continue # If chunk is in the future - wait until it's within the tolerance else: desired_time=desired_time-0.003*len(q) delay = desired_time - now if delay > one_ms: to_wait = max(one_ms, delay - one_ms) await asyncio.sleep(to_wait) # Wait until we can write chunk into output buffer. This might # delay us too much - the probabilistic dropping mechanism will kick # in. times = 0 buffer_space = 0 if not self.PLAY_CALLBACK: while True: # this is for some reason always zero on android/termux!!! buffer_space = self.stream.get_write_available() if buffer_space < self.chunk_frames: self.stat_output_delays += 1 await asyncio.sleep(one_ms) times += 1 if times > 200: print("Hey, the output is STUCK! try --callback if persistent") await asyncio.sleep(1) break continue self.stream.write(chunk) break # callback play, probabilistic dropping of excess chunks to not overload the queue else: ql=len(q) if ql>2: await asyncio.sleep(0.002) if ql>6: prob = ql/10+0.2 if random.random() > prob: s = "DropQ chunk: q_len=%2d delay=%.1fms < 0. tolerance=%.1fms: P=%.2f" s = s % (len(self.chunk_queue.chunk_list), delay * 1000, self.tolerance_ms, prob) print(s) self.stat_time_drops += 1 continue q.append(chunk) # guard against stream stops; doesn't seem to be really necessary but better be sure if not self.stream.is_active(): self.stream.stop_stream() self.stream.start_stream() print('restarting stream') # Main status line if recent > 200: frames_in_chunk = len(chunk) / self.audio_config.frame_size took = time() - recent_start chunks_per_s = recent / took if self.receiver is not None: network_latency = self.receiver.stat_network_latency network_drops = self.receiver.stat_network_drops else: network_latency = 0 network_drops = 0 s = ("STAT: chunks: q_len=%-3d bs=%4.1f " "ch/s=%5.1f " "net lat: %-5.1fms " "avg_delay=%-5.2f drops: time=%d net=%d out_delay=%d cqlen=%d") s = s % ( len(self.chunk_queue.chunk_list), buffer_space / frames_in_chunk, chunks_per_s, 1000.0 * network_latency, 1000.0 * self.stat_total_delay/cnt, self.stat_time_drops, network_drops, self.stat_output_delays, len(q) ) print(s) recent = 0 recent_start = time() # Warnings if self.receiver is not None: if self.receiver.stat_network_latency > 1: print("WARNING: Your network latency seems HUGE. " "Are the clocks synchronised?") elif self.receiver.stat_network_latency <= -0.05: print("WARNING: You either exceeded the speed of " "light or have unsynchronised clocks") finally: print("- Finishing chunk player") glob_stream_break=1 await asyncio.sleep(0.05) if self.stream: self._close_stream()