audio.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

#---------------------------------------------------------------------
# SDL2で正弦波を出力する。引数でバッファサイズを指定
#
# コールバック方式。PythonのGILがどう影響するかの検証
# スレッド間のやりとりにはとりあえず標準の queue を使用
#---------------------------------------------------------------------

import sys
import math
import queue
import ctypes
import atexit

import sdl2

FPS = 60

AUDIO_FREQ     = 44100
AUDIO_FORMAT   = sdl2.AUDIO_S16SYS
AUDIO_CHANNELS = 1

SIN_AMP  = 16000
SIN_FREQ = 440

class SDLError(RuntimeError):
    def __init__(self, msg):
        msg_all = "{}: {}".format(msg, sdl2.SDL_GetError().decode(errors="replace"))
        super().__init__(msg_all)

def is_pow2(n):
    if n == 0: return False
    return (n & (n-1)) == 0

def error(msg):
    sys.exit(msg)

def print_spec(spec, obtained):
    print("""\
freq     : {}
format   : {}
channels : {}
samples  : {}\
""".format(spec.freq, spec.format, spec.channels, spec.samples))

    if obtained:
        print("""\
silence  : {}
size     : {}\
""".format(spec.silence, spec.size))

audio_queue = queue.Queue(32*(AUDIO_FREQ//FPS)) # とりあえず多めで

def audio_pull(userdata, stream, length):
    assert length % 2 == 0

    n_sample = length // 2
    n_pop = 0
    p = ctypes.cast(stream, ctypes.POINTER(ctypes.c_int16))
    for i in range(0, n_sample):
        try:
            sample = audio_queue.get_nowait()
            p[i] = sample
            n_pop += 1
        except queue.Empty:
            break

    if n_pop < n_sample:
        print("UNDERFLOW: {} samples".format(n_sample - n_pop))
        for i in range(n_pop, n_sample):
            p[i] = 0

def usage():
    error("audio <samples>")

def main():
    if len(sys.argv) != 2: usage()
    samples = int(sys.argv[1])
    if samples > 0x8000: error("samples too large")
    if samples < 64: error("samples too small")
    if not is_pow2(samples): error("samples must be power of 2")

    if sdl2.SDL_Init(sdl2.SDL_INIT_AUDIO | sdl2.SDL_INIT_TIMER) != 0:
        raise SDLError("SDL_Init()")
    atexit.register(sdl2.SDL_Quit)

    print("Audio driver: {}".format(sdl2.SDL_GetCurrentAudioDriver()))
    print("")

    want = sdl2.SDL_AudioSpec(
        freq     = AUDIO_FREQ,
        aformat  = AUDIO_FORMAT,
        channels = AUDIO_CHANNELS,
        samples  = samples,
        callback = sdl2.SDL_AudioCallback(audio_pull))
    print("[Desired spec]")
    print_spec(want, False)
    print("")

    have = sdl2.SDL_AudioSpec(0, 0, 0, 0)
    audio = sdl2.SDL_OpenAudioDevice(None, 0, ctypes.byref(want), ctypes.byref(have), 0)
    if not audio: raise SDLError("SDL_OpenAudioDevice()")

    print("[Obtained spec]")
    print_spec(have, True)
    print("")

    if (want.freq     != have.freq or
        want.format   != have.format or
        want.channels != have.channels): error("spec changed")

    sdl2.SDL_PauseAudioDevice(audio, 0)

    t = 0
    samples_per_frame = AUDIO_FREQ // FPS
    while(True):
        n_push = 0
        for i in range(0, samples_per_frame):
            sample = int(SIN_AMP * math.sin(2*math.pi * SIN_FREQ/AUDIO_FREQ * t))
            try:
                audio_queue.put_nowait(sample)
                n_push += 1
                t += 1
            except queue.Full:
                break

        if n_push < samples_per_frame:
            print("OVERFLOW: {} samples".format(samples_per_frame - n_push))

        sdl2.SDL_Delay(1000//FPS - 10)

    sdl2.SDL_CloseAudioDevice(audio)


if __name__ == "__main__": main()