/* * QEMU ALSA audio driver * * Copyright (c) 2005 Vassili Karpov (malc) * * Permission is hereby granted, free of charge, to any person obtaining a copy * of this software and associated documentation files (the "Software"), to deal * in the Software without restriction, including without limitation the rights * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the Software is * furnished to do so, subject to the following conditions: * * The above copyright notice and this permission notice shall be included in * all copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN * THE SOFTWARE. */ #include #include "vl.h" #define AUDIO_CAP "alsa" #include "audio_int.h" typedef struct ALSAVoiceOut { HWVoiceOut hw; void *pcm_buf; snd_pcm_t *handle; } ALSAVoiceOut; typedef struct ALSAVoiceIn { HWVoiceIn hw; snd_pcm_t *handle; void *pcm_buf; } ALSAVoiceIn; static struct { int size_in_usec_in; int size_in_usec_out; const char *pcm_name_in; const char *pcm_name_out; unsigned int buffer_size_in; unsigned int period_size_in; unsigned int buffer_size_out; unsigned int period_size_out; unsigned int threshold; int buffer_size_in_overriden; int period_size_in_overriden; int buffer_size_out_overriden; int period_size_out_overriden; int verbose; } conf = { #ifdef HIGH_LATENCY .size_in_usec_in = 1, .size_in_usec_out = 1, #endif .pcm_name_out = "hw:0,0", .pcm_name_in = "hw:0,0", #ifdef HIGH_LATENCY .buffer_size_in = 400000, .period_size_in = 400000 / 4, .buffer_size_out = 400000, .period_size_out = 400000 / 4, #else #define DEFAULT_BUFFER_SIZE 1024 #define DEFAULT_PERIOD_SIZE 256 .buffer_size_in = DEFAULT_BUFFER_SIZE * 4, .period_size_in = DEFAULT_PERIOD_SIZE * 4, .buffer_size_out = DEFAULT_BUFFER_SIZE, .period_size_out = DEFAULT_PERIOD_SIZE, .buffer_size_in_overriden = 0, .buffer_size_out_overriden = 0, .period_size_in_overriden = 0, .period_size_out_overriden = 0, #endif .threshold = 0, .verbose = 0 }; struct alsa_params_req { int freq; audfmt_e fmt; int nchannels; unsigned int buffer_size; unsigned int period_size; }; struct alsa_params_obt { int freq; audfmt_e fmt; int nchannels; snd_pcm_uframes_t samples; }; static void GCC_FMT_ATTR (2, 3) alsa_logerr (int err, const char *fmt, ...) { va_list ap; va_start (ap, fmt); AUD_vlog (AUDIO_CAP, fmt, ap); va_end (ap); AUD_log (AUDIO_CAP, "Reason: %s\n", snd_strerror (err)); } static void GCC_FMT_ATTR (3, 4) alsa_logerr2 ( int err, const char *typ, const char *fmt, ... ) { va_list ap; AUD_log (AUDIO_CAP, "Could not initialize %s\n", typ); va_start (ap, fmt); AUD_vlog (AUDIO_CAP, fmt, ap); va_end (ap); AUD_log (AUDIO_CAP, "Reason: %s\n", snd_strerror (err)); } static void alsa_anal_close (snd_pcm_t **handlep) { int err = snd_pcm_close (*handlep); if (err) { alsa_logerr (err, "Failed to close PCM handle %p\n", *handlep); } *handlep = NULL; } static int alsa_write (SWVoiceOut *sw, void *buf, int len) { return audio_pcm_sw_write (sw, buf, len); } static int aud_to_alsafmt (audfmt_e fmt) { switch (fmt) { case AUD_FMT_S8: return SND_PCM_FORMAT_S8; case AUD_FMT_U8: return SND_PCM_FORMAT_U8; case AUD_FMT_S16: return SND_PCM_FORMAT_S16_LE; case AUD_FMT_U16: return SND_PCM_FORMAT_U16_LE; default: dolog ("Internal logic error: Bad audio format %d\n", fmt); #ifdef DEBUG_AUDIO abort (); #endif return SND_PCM_FORMAT_U8; } } static int alsa_to_audfmt (int alsafmt, audfmt_e *fmt, int *endianness) { switch (alsafmt) { case SND_PCM_FORMAT_S8: *endianness = 0; *fmt = AUD_FMT_S8; break; case SND_PCM_FORMAT_U8: *endianness = 0; *fmt = AUD_FMT_U8; break; case SND_PCM_FORMAT_S16_LE: *endianness = 0; *fmt = AUD_FMT_S16; break; case SND_PCM_FORMAT_U16_LE: *endianness = 0; *fmt = AUD_FMT_U16; break; case SND_PCM_FORMAT_S16_BE: *endianness = 1; *fmt = AUD_FMT_S16; break; case SND_PCM_FORMAT_U16_BE: *endianness = 1; *fmt = AUD_FMT_U16; break; default: dolog ("Unrecognized audio format %d\n", alsafmt); return -1; } return 0; } #if defined DEBUG_MISMATCHES || defined DEBUG static void alsa_dump_info (struct alsa_params_req *req, struct alsa_params_obt *obt) { dolog ("parameter | requested value | obtained value\n"); dolog ("format | %10d | %10d\n", req->fmt, obt->fmt); dolog ("channels | %10d | %10d\n", req->nchannels, obt->nchannels); dolog ("frequency | %10d | %10d\n", req->freq, obt->freq); dolog ("============================================\n"); dolog ("requested: buffer size %d period size %d\n", req->buffer_size, req->period_size); dolog ("obtained: samples %ld\n", obt->samples); } #endif static void alsa_set_threshold (snd_pcm_t *handle, snd_pcm_uframes_t threshold) { int err; snd_pcm_sw_params_t *sw_params; snd_pcm_sw_params_alloca (&sw_params); err = snd_pcm_sw_params_current (handle, sw_params); if (err < 0) { dolog ("Could not fully initialize DAC\n"); alsa_logerr (err, "Failed to get current software parameters\n"); return; } err = snd_pcm_sw_params_set_start_threshold (handle, sw_params, threshold); if (err < 0) { dolog ("Could not fully initialize DAC\n"); alsa_logerr (err, "Failed to set software threshold to %ld\n", threshold); return; } err = snd_pcm_sw_params (handle, sw_params); if (err < 0) { dolog ("Could not fully initialize DAC\n"); alsa_logerr (err, "Failed to set software parameters\n"); return; } } static int alsa_open (int in, struct alsa_params_req *req, struct alsa_params_obt *obt, snd_pcm_t **handlep) { snd_pcm_t *handle; snd_pcm_hw_params_t *hw_params; int err, freq, nchannels; const char *pcm_name = in ? conf.pcm_name_in : conf.pcm_name_out; unsigned int period_size, buffer_size; snd_pcm_uframes_t obt_buffer_size; const char *typ = in ? "ADC" : "DAC"; freq = req->freq; period_size = req->period_size; buffer_size = req->buffer_size; nchannels = req->nchannels; snd_pcm_hw_params_alloca (&hw_params); err = snd_pcm_open ( &handle, pcm_name, in ? SND_PCM_STREAM_CAPTURE : SND_PCM_STREAM_PLAYBACK, SND_PCM_NONBLOCK ); if (err < 0) { alsa_logerr2 (err, typ, "Failed to open `%s':\n", pcm_name); return -1; } err = snd_pcm_hw_params_any (handle, hw_params); if (err < 0) { alsa_logerr2 (err, typ, "Failed to initialize hardware parameters\n"); goto err; } err = snd_pcm_hw_params_set_access ( handle, hw_params, SND_PCM_ACCESS_RW_INTERLEAVED ); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set access type\n"); goto err; } err = snd_pcm_hw_params_set_format (handle, hw_params, req->fmt); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set format %d\n", req->fmt); goto err; } err = snd_pcm_hw_params_set_rate_near (handle, hw_params, &freq, 0); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set frequency %d\n", req->freq); goto err; } err = snd_pcm_hw_params_set_channels_near ( handle, hw_params, &nchannels ); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set number of channels %d\n", req->nchannels); goto err; } if (nchannels != 1 && nchannels != 2) { alsa_logerr2 (err, typ, "Can not handle obtained number of channels %d\n", nchannels); goto err; } if (!((in && conf.size_in_usec_in) || (!in && conf.size_in_usec_out))) { if (!buffer_size) { buffer_size = DEFAULT_BUFFER_SIZE; period_size= DEFAULT_PERIOD_SIZE; } } if (buffer_size) { if ((in && conf.size_in_usec_in) || (!in && conf.size_in_usec_out)) { if (period_size) { err = snd_pcm_hw_params_set_period_time_near ( handle, hw_params, &period_size, 0 ); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set period time %d\n", req->period_size); goto err; } } err = snd_pcm_hw_params_set_buffer_time_near ( handle, hw_params, &buffer_size, 0 ); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set buffer time %d\n", req->buffer_size); goto err; } } else { int dir; snd_pcm_uframes_t minval; if (period_size) { minval = period_size; dir = 0; err = snd_pcm_hw_params_get_period_size_min ( hw_params, &minval, &dir ); if (err < 0) { alsa_logerr ( err, "Could not get minmal period size for %s\n", typ ); } else { if (period_size < minval) { if ((in && conf.period_size_in_overriden) || (!in && conf.period_size_out_overriden)) { dolog ("%s period size(%d) is less " "than minmal period size(%ld)\n", typ, period_size, minval); } period_size = minval; } } err = snd_pcm_hw_params_set_period_size ( handle, hw_params, period_size, 0 ); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set period size %d\n", req->period_size); goto err; } } minval = buffer_size; err = snd_pcm_hw_params_get_buffer_size_min ( hw_params, &minval ); if (err < 0) { alsa_logerr (err, "Could not get minmal buffer size for %s\n", typ); } else { if (buffer_size < minval) { if ((in && conf.buffer_size_in_overriden) || (!in && conf.buffer_size_out_overriden)) { dolog ( "%s buffer size(%d) is less " "than minimal buffer size(%ld)\n", typ, buffer_size, minval ); } buffer_size = minval; } } err = snd_pcm_hw_params_set_buffer_size ( handle, hw_params, buffer_size ); if (err < 0) { alsa_logerr2 (err, typ, "Failed to set buffer size %d\n", req->buffer_size); goto err; } } } else { dolog ("warning: Buffer size is not set\n"); } err = snd_pcm_hw_params (handle, hw_params); if (err < 0) { alsa_logerr2 (err, typ, "Failed to apply audio parameters\n"); goto err; } err = snd_pcm_hw_params_get_buffer_size (hw_params, &obt_buffer_size); if (err < 0) { alsa_logerr2 (err, typ, "Failed to get buffer size\n"); goto err; } err = snd_pcm_prepare (handle); if (err < 0) { alsa_logerr2 (err, typ, "Could not prepare handle %p\n", handle); goto err; } if (!in && conf.threshold) { snd_pcm_uframes_t threshold; int bytes_per_sec; bytes_per_sec = freq << (nchannels == 2) << (req->fmt == AUD_FMT_S16 || req->fmt == AUD_FMT_U16); threshold = (conf.threshold * bytes_per_sec) / 1000; alsa_set_threshold (handle, threshold); } obt->fmt = req->fmt; obt->nchannels = nchannels; obt->freq = freq; obt->samples = obt_buffer_size; *handlep = handle; #if defined DEBUG_MISMATCHES || defined DEBUG if (obt->fmt != req->fmt || obt->nchannels != req->nchannels || obt->freq != req->freq) { dolog ("Audio paramters mismatch for %s\n", typ); alsa_dump_info (req, obt); } #endif #ifdef DEBUG alsa_dump_info (req, obt); #endif return 0; err: alsa_anal_close (&handle); return -1; } static int alsa_recover (snd_pcm_t *handle) { int err = snd_pcm_prepare (handle); if (err < 0) { alsa_logerr (err, "Failed to prepare handle %p\n", handle); return -1; } return 0; } static snd_pcm_sframes_t alsa_get_avail (snd_pcm_t *handle) { snd_pcm_sframes_t avail; avail = snd_pcm_avail_update (handle); if (avail < 0) { if (avail == -EPIPE) { if (!alsa_recover (handle)) { avail = snd_pcm_avail_update (handle); } } if (avail < 0) { alsa_logerr (avail, "Could not obtain number of available frames\n"); return -1; } } return avail; } static int alsa_run_out (HWVoiceOut *hw) { ALSAVoiceOut *alsa = (ALSAVoiceOut *) hw; int rpos, live, decr; int samples; uint8_t *dst; st_sample_t *src; snd_pcm_sframes_t avail; live = audio_pcm_hw_get_live_out (hw); if (!live) { return 0; } a
from six.moves import cStringIO as StringIO
from pathod import language
from pathod.language import http, base
import tutils


def parse_request(s):
    return language.parse_pathoc(s).next()


def test_make_error_response():
    d = StringIO()
    s = http.make_error_response("foo")
    language.serve(s, d, {})


class TestRequest:

    def test_nonascii(self):
        tutils.raises("ascii", parse_request, "get:\xf0")

    def test_err(self):
        tutils.raises(language.ParseException, parse_request, 'GET')

    def test_simple(self):
        r = parse_request('GET:"/foo"')
        assert r.method.string() == "GET"
        assert r.path.string() == "/foo"
        r = parse_request('GET:/foo')
        assert r.path.string() == "/foo"
        r = parse_request('GET:@1k')
        assert len(r.path.string()) == 1024

    def test_multiple(self):
        r = list(language.parse_pathoc("GET:/ PUT:/"))
        assert r[0].method.string() == "GET"
        assert r[1].method.string() == "PUT"
        assert len(r) == 2

        l = """
            GET
            "/foo"
            ir,@1

            PUT

            "/foo



            bar"

            ir,@1
        """
        r = list(language.parse_pathoc(l))
        assert len(r) == 2
        assert r[0].method.string() == "GET"
        assert r[1].method.string() == "PUT"

        l = """
            get:"http://localhost:9999/p/200":ir,@1
            get:"http://localhost:9999/p/200":ir,@2
        """
        r = list(language.parse_pathoc(l))
        assert len(r) == 2
        assert r[0].method.string() == "GET"
        assert r[1].method.string() == "GET"

    def test_nested_response(self):
        l = "get:/p:s'200'"
        r = list(language.parse_pathoc(l))
        assert len(r) == 1
        assert len(r[0].tokens) == 3
        assert isinstance(r[0].tokens[2], http.NestedResponse)
        assert r[0].values({})

    def test_render(self):
        s = StringIO()
        r = parse_request("GET:'/foo'")
        assert language.serve(
            r,
            s,
            language.Settings(request_host="foo.com")
        )

    def test_multiline(self):
        l = """
            GET
            "/foo"
            ir,@1
        """
        r = parse_request(l)
        assert r.method.string() == "GET"
        assert r.path.string() == "/foo"
        assert r.actions

        l = """
            GET

            "/foo



            bar"

            ir,@1
        """
        r = parse_request(l)
        assert r.method.string() == "GET"
        assert r.path.string().endswith("bar")
        assert r.actions

    def test_spec(self):
        def rt(s):
            s = parse_request(s).spec()
            assert parse_request(s).spec() == s
        rt("get:/foo")
        rt("get:/foo:da")

    def test_freeze(self):
        r = parse_request("GET:/:b@100").freeze(language.Settings())
        assert len(r.spec()) > 100

    def test_path_generator(self):
        r = parse_request("GET:@100").freeze(language.Settings())
        assert len(r.spec()) > 100

    def test_websocket(self):
        r = parse_request('ws:/path/')
        res = r.resolve(language.Settings())
        assert res.method.string().lower() == "get"
        assert res.tok(http.Path).value.val == "/path/"
        assert res.tok(http.Method).value.val.lower() == "get"
        assert http.get_header("Upgrade", res.headers).value.val == "websocket"

        r = parse_request('ws:put:/path/')
        res = r.resolve(language.Settings())
        assert r.method.string().lower() == "put"
        assert res.tok(http.Path).value.val == "/path/"
        assert res.tok(http.Method).value.val.lower() == "put"
        assert http.get_header("Upgrade", res.headers).value.val == "websocket"


class TestResponse:

    def dummy_response(self):
        return language.parse_pathod("400'msg'").next()

    def test_response(self):
        r = language.parse_pathod("400:m'msg'").next()
        assert r.status_code.string() == "400"
        assert r.reason.string() == "msg"

        r = language.parse_pathod("400:m'msg':b@100b").next()
        assert r.reason.string() == "msg"
        assert r.body.values({})
        assert str(r)

        r = language.parse_pathod("200").next()
        assert r.status_code.string() == "200"
        assert not r.reason
        assert "OK" in [i[:] for i in r.preamble({})]

    def test_render(self):
        s = StringIO()
        r = language.parse_pathod("400:m'msg'").next()
        assert language.serve(r, s, {})

        r = language.parse_pathod("400:p0,100:dr").next()
        assert "p0" in r.spec()
        s = r.preview_safe()
        assert "p0" not in s.spec()

    def test_raw(self):
        s = StringIO()
        r = language.parse_pathod("400:b'foo'").next()
        language.serve(r, s, {})
        v = s.getvalue()
        assert "Content-Length" in v

        s = StringIO()
        r = language.parse_pathod("400:b'foo':r").next()
        language.serve(r, s, {})
        v = s.getvalue()
        assert "Content-Length" not in v

    def test_length(self):
        def testlen(x):
            s = StringIO()
            x = x.next()
            language.serve(x, s, language.Settings())
            assert x.length(language.Settings()) == len(s.getvalue())
        testlen(language.parse_pathod("400:m'msg':r"))
        testlen(language.parse_pathod("400:m'msg':h'foo'='bar':r"))
        testlen(language.parse_pathod("400:m'msg':h'foo'='bar':b@100b:r"))

    def test_maximum_length(self):
        def testlen(x):
            x = x.next()
            s = StringIO()
            m = x.maximum_length({})
            language.serve(x, s, {})
            assert m >= len(s.getvalue())

        r = language.parse_pathod("400:m'msg':b@100:d0")
        testlen(r)

        r = language.parse_pathod("400:m'msg':b@100:d0:i0,'foo'")
        testlen(r)

        r = language.parse_pathod("400:m'msg':b@100:d0:i0,'foo'")
        testlen(r)

    def test_parse_err(self):
        tutils.raises(
            language.ParseException, language.parse_pathod, "400:msg,b:"
        )
        try:
            language.parse_pathod("400'msg':b:")
        except language.ParseException as v:
            assert v.marked()
            assert str(v)

    def test_nonascii(self):
        tutils.raises("ascii", language.parse_pathod, "foo:b\xf0")

    def test_parse_header(self):
        r = language.parse_pathod('400:h"foo"="bar"').next()
        assert http.get_header("foo", r.headers)

    def test_parse_pause_before(self):
        r = language.parse_pathod("400:p0,10").next()
        assert r.actions[0].spec() == "p0,10"

    def test_parse_pause_after(self):
        r = language.parse_pathod("400:pa,10").next()
        assert r.actions[0].spec() == "pa,10"

    def test_parse_pause_random(self):
        r = language.parse_pathod("400:pr,10").next()
        assert r.actions[0].spec() == "pr,10"

    def test_parse_stress(self):
        # While larger values are known to work on linux, len() technically
        # returns an int and a python 2.7 int on windows has 32bit precision.
        # Therefore, we should keep the body length < 2147483647 bytes in our
        # tests.
        r = language.parse_pathod("400:b@1g").next()
        assert r.length({})

    def test_spec(self):
        def rt(s):
            s = language.parse_pathod(s).next().spec()
            assert language.parse_pathod(s).next().spec() == s
        rt("400:b@100g")
        rt("400")
        rt("400:da")

    def test_websockets(self):
        r = language.parse_pathod("ws").next()
        tutils.raises("no websocket key", r.resolve, language.Settings())
        res = r.resolve(language.Settings(websocket_key="foo"))
        assert res.status_code.string() == "101"


def test_ctype_shortcut():
    e = http.ShortcutContentType.expr()
    v = e.parseString("c'foo'")[0]
    assert v.key.val == "Content-Type"
    assert v.value.val == "foo"

    s = v.spec()
    assert s == e.parseString(s)[0].spec()

    e = http.ShortcutContentType.expr()
    v = e.parseString("c@100")[0]
    v2 = v.freeze({})
    v3 = v2.freeze({})
    assert v2.value.val == v3.value.val


def test_location_shortcut():
    e = http.ShortcutLocation.expr()
    v = e.parseString("l'foo'")[0]
    assert v.key.val == "Location"
    assert v.value.val == "foo"

    s = v.spec()
    assert s == e.parseString(s)[0].spec()

    e = http.ShortcutLocation.expr()
    v = e.parseString("l@100")[0]
    v2 = v.freeze({})
    v3 = v2.freeze({})
    assert v2.value.val == v3.value.val


def test_shortcuts():
    assert language.parse_pathod(
        "400:c'foo'").next().headers[0].key.val == "Content-Type"
    assert language.parse_pathod(
        "400:l'foo'").next().headers[0].key.val == "Location"

    assert "Android" in tutils.render(parse_request("get:/:ua"))
    assert "User-Agent" in tutils.render(parse_request("get:/:ua"))


def test_user_agent():
    e = http.ShortcutUserAgent.expr()
    v = e.parseString("ua")[0]
    assert "Android" in v.string()

    e = http.ShortcutUserAgent.expr()
    v = e.parseString("u'a'")[0]
    assert "Android" not in v.string()

    v = e.parseString("u@100'")[0]
    assert len(str(v.freeze({}).value)) > 100
    v2 = v.freeze({})
    v3 = v2.freeze({})
    assert v2.value.val == v3.value.val


def test_nested_response():
    e = http.NestedResponse.expr()
    v = e.parseString("s'200'")[0]
    assert v.value.val == "200"
    tutils.raises(
        language.ParseException,
        e.parseString,
        "s'foo'"
    )

    v = e.parseString('s"200:b@1"')[0]
    assert "@1" in v.spec()
    f = v.freeze({})
    assert "@1" not in f.spec()


def test_nested_response_freeze():
    e = http.NestedResponse(
        base.TokValueLiteral(
            "200:b'foo':i10,'\\x27'".encode(
                "string_escape"
            )
        )
    )
    assert e.freeze({})
    assert e.values({})


def test_unique_components():
    tutils.raises(
        "multiple body clauses",
        language.parse_pathod,
        "400:b@1:b@1"
    )
generated by cgit v1.2.3 (git 2.25.1) at 2026-02-11 09:51:47 +0000