music-lights/alsa_output.py at master · LSDG/music-lights · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import alsaaudio

import ansi

import mainLoop


class SampleOutput(object):
    def __init__(self, sampleGen, config):
        self.sampleGen = sampleGen
        sampleGen.onSongChanged.add(self.onSongChanged)

        self.pcm = None

        self.debug = config.get_def('output', 'debug', 'f').lower() not in ('f', 'false', 'n', 'no', '0', 'off')

    def onSongChanged(self, *args, **kwargs):
        ansi.debug('onSongChanged')

        if self.pcm:
            self.pcm.close()

        self.pcm = alsaaudio.PCM(type=alsaaudio.PCM_PLAYBACK)  # mode=alsaaudio.PCM_NORMAL)

        if self.debug:
            ansi.debug('Song changed; channels: {}; samplerate: {}; period size: {}',
                    self.sampleGen.channels, self.sampleGen.samplerate, self.sampleGen.framesPerChunk)

        self.pcm.setchannels(self.sampleGen.channels)
        self.pcm.setrate(self.sampleGen.samplerate)
        self.pcm.setperiodsize(self.sampleGen.framesPerChunk)
        self.pcm.setformat(alsaaudio.PCM_FORMAT_S16_LE)

    def play(self):
        mainLoop.currentProcess.afterEachLoop = self.queueNextSound

    def queueNextSound(self, event=None):
        if self.debug:
            ansi.stdout(
                    "{cursor.col.0}{clear.line.all}Current time:"
                        " {style.bold}{file.elapsedTime: >7.2f}{style.none} / {file.duration: <7.2f}",
                    file=self.sampleGen,
                    suppressNewline=True
                    )

        chunk = self.sampleGen.nextChunk()
        if self.pcm:
            self.pcm.write(chunk)


class ALSAProcess(mainLoop.QueueHandlerProcess):
    def __init__(self, messageQueue, nice=None):
        super(ALSAProcess, self).__init__(messageQueue, nice)

        self.afterEachLoop = None

    def eachLoop(self):
        super(ALSAProcess, self).eachLoop()

        if self.afterEachLoop:
            self.afterEachLoop()