lasp/scripts/play_sine

#!/usr/bin/python3
import argparse
import numpy as np
import sys, logging, os, argparse
logging.basicConfig(level=logging.DEBUG)
import multiprocessing
from lasp.lasp_multiprocessingpatch import apply_patch
from lasp.lasp_avstream import StreamManager
from lasp.lasp_common import AvType
from lasp.lasp_siggen import Siggen, SignalType, SiggenData
from lasp.device import DaqConfigurations


if __name__ == '__main__':
    multiprocessing.set_start_method('forkserver', force=True)
    parser = argparse.ArgumentParser(
        description='Play a sine wave'
    )
    device_help = 'DAQ Device to play to'
    parser.add_argument('--device', '-d', help=device_help, type=str,
                        default='Default')

    args = parser.parse_args()


    configs = DaqConfigurations.loadConfigs()

    config_keys = [key for key in configs.keys()]
    for i, key in enumerate(config_keys):
        print(f'{i:2} : {key}')

    choosen_index = input('Number of configuration to use: ')
    try:
        daqindex = int(choosen_index)
    except:
        sys.exit(0)


    choosen_key = config_keys[daqindex]
    daqconfig = configs[choosen_key].output_config

    print(f'Choosen configuration: {choosen_key}')

    try:
        streammgr = StreamManager()
        outq = streammgr.getOutputQueue()

        siggendata = SiggenData(
            fs=48e3,
            nframes_per_block=2048,
            dtype=np.dtype(np.int16),
            eqdata=None,
            level_dB=-20,
            signaltype=SignalType.Periodic,
            signaltypedata=(1000.,)
        )
        siggen = Siggen(outq, siggendata)

        streammgr.activateSiggen()

        streammgr.startStream(AvType.audio_output, daqconfig)

        input('Press any key to stop...')
        streammgr.stopStream(AvType.audio_output)
    finally:
        siggen.cleanup()
        streammgr.cleanup()