<commit_before>"""
Loopback tests for pastream.
"""
from __future__ import print_function
import os, sys
import numpy as np
import soundfile as sf
import pytest
import numpy.testing as npt
import time
import tempfile
import platform
import pastream as pas


# Set up the platform specific device
system = platform.system()
if system == 'Windows':
    DEVICE_KWARGS = {'device': 'ASIO4ALL v2, ASIO', 'dtype': 'int24', 'blocksize': 512, 'channels': 8, 'samplerate':48000}
elif system == 'Darwin':
    raise Exception("Currently no support for Mac devices")
else:
    # This is assuming you're using the ALSA device set up by etc/.asoundrc
    DEVICE_KWARGS = {'device': 'aduplex', 'dtype': 'int32', 'blocksize': 512, 'channels': 8, 'samplerate':48000}

if 'SOUNDDEVICE_DEVICE_NAME' in os.environ:
    DEVICE_KWARGS['device'] = os.environ['SOUNDDEVICE_DEVICE_NAME']
if 'SOUNDDEVICE_DEVICE_BLOCKSIZE' in os.environ:
    DEVICE_KWARGS['blocksize'] = os.environ['SOUNDDEVICE_DEVICE_BLOCKSIZE']
if 'SOUNDDEVICE_DEVICE_DTYPE' in os.environ:
    DEVICE_KWARGS['dtype'] = os.environ['SOUNDDEVICE_DEVICE_DTYPE']
if 'SOUNDDEVICE_DEVICE_CHANNELS' in os.environ:
    DEVICE_KWARGS['channels'] = os.environ['SOUNDDEVICE_DEVICE_CHANNELS']
if 'SOUNDDEVICE_DEVICE_SAMPLERATE' in os.environ:
    DEVICE_KWARGS['SAMPLERATE'] = os.environ['SOUNDDEVICE_DEVICE_SAMPLERATE']

PREAMBLE = 0x7FFFFFFF # Value used for the preamble sequence (before appropriate shifting for dtype)

_dtype2elementsize = dict(int32=4,int24=3,int16=2,int8=1)
vhex = np.vectorize('{:#10x}'.format)
tohex = lambda x: vhex(x.view('u4'))

def assert_loopback_equal(inp_fh, preamble, **kwargs):
    inpf2 = sf.SoundFile(inp_fh.name, mode='rb')

    devargs = dict(DEVICE_KWARGS)
    devargs.update(kwargs)

    delay = -1
    found_delay = False
    nframes = mframes = 0
    
    stream = pas.SoundFileStream(inp_fh, **devargs)
    for outframes in stream.blockstream(always_2d=True):
        if not found_delay:
            matches = outframes[:, 0].view('u4') == preamble
            if np.any(matches): 
                found_delay = True
                nonzeros = np.where(matches)[0]
                outframes = outframes[nonzeros[0]:]
                nframes += nonzeros[0]
                delay = nframes
        if found_delay:
            inframes = inpf2.read(len(outframes), dtype='int32', always_2d=True)

            mlen = min(len(inframes), len(outframes))
            inp = inframes[:mlen].view('u4')
            out = outframes[:mlen].view('u4')

            npt.assert_array_equal(inp, out, "Loopback data mismatch")
            mframes += mlen
        nframes += len(outframes)

    assert delay != -1, "Preamble not found or was corrupted"

    print("Matched %d of %d frames; Initial delay of %d frames" % (mframes, nframes, delay))

def gen_random(rdm_fh, nseconds, elementsize):<commit_msg>test_pastream: remove the need to have 2 file handlers for the same file<commit_after>"""
Loopback tests for pastream.
"""
from __future__ import print_function
import os, sys
import numpy as np
import soundfile as sf
import pytest
import numpy.testing as npt
import time
import tempfile
import platform
import pastream as pas
import rtmixer


# Set up the platform specific device
system = platform.system()
if system == 'Windows':
    DEVICE_KWARGS = {'device': 'ASIO4ALL v2, ASIO', 'dtype': 'int24', 'blocksize': 512, 'channels': 8, 'samplerate':48000}
elif system == 'Darwin':
    raise Exception("Currently no support for Mac devices")
else:
    # This is assuming you're using the ALSA device set up by etc/.asoundrc
    DEVICE_KWARGS = {'device': 'aduplex', 'dtype': 'int32', 'blocksize': 512, 'channels': 8, 'samplerate':48000}

if 'SOUNDDEVICE_DEVICE_NAME' in os.environ:
    DEVICE_KWARGS['device'] = os.environ['SOUNDDEVICE_DEVICE_NAME']
if 'SOUNDDEVICE_DEVICE_BLOCKSIZE' in os.environ:
    DEVICE_KWARGS['blocksize'] = os.environ['SOUNDDEVICE_DEVICE_BLOCKSIZE']
if 'SOUNDDEVICE_DEVICE_DTYPE' in os.environ:
    DEVICE_KWARGS['dtype'] = os.environ['SOUNDDEVICE_DEVICE_DTYPE']
if 'SOUNDDEVICE_DEVICE_CHANNELS' in os.environ:
    DEVICE_KWARGS['channels'] = os.environ['SOUNDDEVICE_DEVICE_CHANNELS']
if 'SOUNDDEVICE_DEVICE_SAMPLERATE' in os.environ:
    DEVICE_KWARGS['SAMPLERATE'] = os.environ['SOUNDDEVICE_DEVICE_SAMPLERATE']

PREAMBLE = 0x7FFFFFFF # Value used for the preamble sequence (before appropriate shifting for dtype)

_dtype2elementsize = dict(int32=4,int24=3,int16=2,int8=1)
vhex = lambda x: np.vectorize(('{:#0%dx}'%(x.dtype.itemsize*2+2)).format)
tohex = lambda x: vhex(x.view('u'+x.dtype.name.lstrip('u')))

def assert_loopback_equal(inp_fh, preamble, **kwargs):
    devargs = dict(DEVICE_KWARGS)
    devargs.update(kwargs)

    delay = -1
    found_delay = False
    
    stream = pas.SoundFileStream(inp_fh, **devargs)

    # 'tee' the transmit queue writer so that we can recall any input and match
    # it to the output. We make it larger than usual (4MB) too allow for extra
    # slack
    inpbuff = rtmixer.RingBuffer(stream.framesize[1], 1 << 24)
    stream.txq.__write = stream.txq.write
    def teewrite(buff, size=-1):
        nbuff = inpbuff.write(buff, size=size)
        nframes = stream.txq.__write(buff, size=size)
        assert nbuff == nframes, "Ran out of temporary buffer space. Use a larger qsize"
        return nframes
    stream.txq.write = teewrite

    unsigned_dtype = 'u'+stream.dtype[1]
    nframes = mframes = 0
    inframes = np.zeros((stream.blocksize, stream.channels[1]), dtype=stream.dtype[1])
    for outframes in stream.blockstream(always_2d=True):
        if not found_delay:
            matches = outframes[:, 0].view(unsigned_dtype) == preamble
            if np.any(matches): 
                found_delay = True
                nonzeros = np.where(matches)[0]
                outframes = outframes[nonzeros[0]:]
                nframes += nonzeros[0]
                delay = nframes
        if found_delay:
            readframes = inpbuff.read(inframes, len(outframes))
            inp = inframes[:readframes].view(unsigned_dtype)
            out = outframes[:readframes].view(unsigned_dtype)

            npt.assert_array_equal(inp, out, "Loopback data mismatch")
            mframes += readframes
        nframes += len(outframes)

    assert delay != -1, "Preamble not found or was corrupted"

    print("Matched %d of %d frames; Initial delay of %d frames" % (mframes, nframes, delay))

def gen_random(rdm_fh, nseconds, elementsize):
{'fraction_non_alphanumeric': 0.11622874806800618, 'fraction_numerical': 0.01792890262751159, 'mean_word_length': 3.751101321585903, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6296582', 'n_tokens_mistral': 2211, 'n_tokens_neox': 2128, 'n_words': 544}