1#!/usr/bin/env python3

2# Copyright 2016 Canonical Ltd.

3# All rights reserved.

4#

5# Written by:

6# Authors: Maciej Kisielewski <maciej.kisielewski@canonical.com>

7

8

9 import alsaaudio

10 import argparse

11 import cmath

12 import contextlib

13 import math

14 import os

15 import struct

16 import sys

17 import threading

18

19 RATE = 44100

20 PERIOD = 441 # chunk size (in samples) that will be used when talking to alsa

21

22

23 def fft(x):

24 N = len(x)

25 if N <= 1:

26 return x

27 even = fft(x[0::2])

28 odd = fft(x[1::2])

29 T = [cmath.exp(-2j * cmath.pi * k / N) * odd[k] for k in range(N // 2)]

30 return [even[k] + T[k] for k in range(N // 2)] + [

31 even[k] - T[k] for k in range(N // 2)

32 ]

33

34

35 def sine(freq, length, period_len, amplitude=0.5):

36 """

37 Generate `period_len` samples of sine of `freq` frequency and `amplitude`.

38 It generates at least `length` samples, totalling at next multiple of

39 period_len. E.g. sine(440, 15, 10) will generate two 10-item lists of

40 samples. Returns list of floats in the range of [-1.0 .. 1.0].

41 """

42 t = 0

43 while t < length:

44 sample = []

45 for i in range(period_len):

46 sample.append(

47 amplitude * math.sin(2 * math.pi * ((t + i) / (RATE / freq)))

48 )

49 yield sample

50 t += period_len

51

52

53 class Player:

54 def __init__(self, device=None):

55 if not device:

56 available_pcms = alsaaudio.pcms(alsaaudio.PCM_PLAYBACK)

57 if not available_pcms:

58 raise SystemExit("No PCMs detected")

59 self.pcm = alsaaudio.PCM(device=available_pcms[0])

60 else:

61 self.pcm = alsaaudio.PCM(device=device)

62 self.pcm.setchannels(1)

63 self.pcm.setformat(alsaaudio.PCM_FORMAT_FLOAT_LE)

64 self.pcm.setperiodsize(PERIOD)

65

66 def play(self, chunk):

67 assert len(chunk) == PERIOD

68 # alsa expects bytes, so we need to repack the list of floats into a

69 # bytes sequence

70 buff = b"".join([struct.pack("<f", x) for x in chunk])

71 self.pcm.write(buff)

72

73 @contextlib.contextmanager

74 def changed_volume(self):

75 """Change volume to 50% and unmute the output while in the context"""

76 available_mixers = alsaaudio.mixers()

77 if not available_mixers:

78 # no mixers available - silently ignore change_volume request

79 yield

80 return

81 try:

82 # get default mixer - this may fail on some systems

83 mixer = alsaaudio.Mixer()

84 except alsaaudio.ALSAAudioError:

85 # pick the first mixer available

86 mixer = alsaaudio.Mixer(available_mixers[0])

87 stored_mute = mixer.getmute()

88 stored_volume = mixer.getvolume()

89 mixer.setmute(0)

90 mixer.setvolume(50)

91 yield

92 # the getters returned lists of volumes/mutes per channel, setters

93 # require scalars so we need to set it one by one

94 for ch, mute in enumerate(stored_mute):

95 mixer.setmute(mute, ch)

96 for ch, vol in enumerate(stored_volume):

97 mixer.setvolume(vol, ch)

98

99

100 class Recorder:

101 def __init__(self):

102 self.pcm = alsaaudio.PCM(alsaaudio.PCM_CAPTURE)

103 self.pcm.setchannels(1)

104 self.pcm.setformat(alsaaudio.PCM_FORMAT_FLOAT_LE)

105 self.pcm.setperiodsize(PERIOD)

106

107 def record(self, length):

108 samples = []

109 while len(samples) < length:

110 raw = self.pcm.read()

111 samples += [x[0] for x in struct.iter_unpack("<f", raw[1])]

112

113 return samples

114

115

116 def playback_test(seconds, device):

117 player = Player(device)

118 with player.changed_volume():

119 for chunk in sine(440, seconds * RATE, PERIOD):

120 player.play(chunk)

121

122

123 def loopback_test(seconds, device, freq=455.5):

124 def generator():

125 player = Player(device)

126 with player.changed_volume():

127 for chunk in sine(freq, seconds * RATE, PERIOD):

128 player.play(chunk)

129

130 plr_thread = threading.Thread(target=generator)

131 plr_thread.daemon = True

132 plr_thread.start()

133

134 rec = Recorder()

135 samples = rec.record(seconds * RATE)

136

137 # fft requires len(samples) to be a power of 2.

138 # let's trim samples to match that

139 real_len = 2 ** math.floor(math.floor(math.log2(len(samples))))

140 real_seconds = seconds * (real_len / len(samples))

141 samples = samples[0:real_len]

142

143 Y = fft(samples)

144 freqs = [abs(y) for y in Y[: int(RATE / 2)]]

145 dominant = freqs.index(max(freqs)) / real_seconds

146 print("Dominant frequency is {}, expected {}".format(dominant, freq))

147

148 epsilon = 1.0

149 if abs(dominant - freq) < epsilon:

150 return 0

151 else:

152 return 1

153

154

155 def main():

156 actions = {

157 "playback": playback_test,

158 "loopback": loopback_test,

159 }

160 parser = argparse.ArgumentParser(description="Sound testing using ALSA")

161 parser.add_argument("action", metavar="ACTION", choices=actions.keys())

162 parser.add_argument("--duration", type=int, default=5)

163 parser.add_argument("--device", type=str)

164 args = parser.parse_args()

165 if args.device:

166 device = args.device

167 elif "ALSADEVICE" in os.environ:

168 device = os.environ["ALSADEVICE"]

169 else:

170 device = None

171 return actions[args.action](args.duration, device)

172

173

174 if __name__ == "__main__":

175 sys.exit(main())