1#!/usr/bin/env python3

2

3 from __future__ import division, print_function

4 import argparse

5 import collections

6 import json

7 import logging

8 import math

9 import re

10 import subprocess

11 import sys

12 import time

13

14 try:

15 import gi

16

17 gi.require_version("GLib", "2.0")

18 gi.require_version("Gst", "1.0")

19 from gi.repository import GObject

20 from gi.repository import Gst

21 from gi.repository import GLib

22

23 Gst.init(None) # This has to be done very early so it can find elements

24 except ImportError:

25 print(

26 "Can't import module: %s. it may not be available for this"

27 "version of Python, which is: " % sys.exc_info()[1],

28 file=sys.stderr,

29 )

30 print((sys.version), file=sys.stderr)

31 sys.exit(127)

32

33

34# Frequency bands for FFT

35 BINS = 256

36# How often to take a sample and do FFT on it.

37 FFT_INTERVAL = 100000000 # In nanoseconds, so this is every 1/10th second

38# Sampling frequency. The effective maximum frequency we can analyze is

39# half of this (see Nyquist's theorem)

40 SAMPLING_FREQUENCY = 44100

41# The default test frequency is in the middle of the band that contains 5000Hz

42# This frequency was determined experimentally to be high enough but more

43# reliable than others we tried.

44 DEFAULT_TEST_FREQUENCY = 5035

45# only sample a signal when peak level is in this range (in dB attenuation,

46# 0 means no attenuation (and horrible clipping).

47 REC_LEVEL_RANGE = (-2.0, -12.0)

48# For our test signal to be considered present, it has to be this much higher

49# than the base level (minimum magnitude). This is in dB.

50 MAGNITUDE_THRESHOLD = 2.5

51# Volume for the sample tone (in %)

52 PLAY_VOLUME = 70

53

54

55 class PIDController(object):

56 """A Proportional-Integrative-Derivative controller (PID) controls a

57 process's output to try to maintain a desired output value (known as

58 'setpoint', by continually adjusting the process's input.

59

60 It does so by calculating the "error" (difference between output and

61 setpoint) and attempting to minimize it manipulating the input.

62

63 The desired change to the input is calculated based on error and three

64 constants (Kp, Ki and Kd). These values can be interpreted in terms of

65 time: P depends on the present error, I on the accumulation of past errors,

66 and D is a prediction of future errors, based on current rate of change.

67 The weighted sum of these three actions is used to adjust the process via a

68 control element.

69

70 In practice, Kp, Ki and Kd are process-dependent and usually have to

71 be tweaked by hand, but once reasonable constants are arrived at, they

72 can apply to a particular process without further modification.

73

74 """

75

76 def __init__(self, Kp, Ki, Kd, setpoint=0):

77 """Creates a PID controller with given constants and setpoint.

78

79 Arguments:

80 Kp, Ki, Kd: PID constants, see class description.

81 setpoint: desired output value; calls to input_change with

82 a process output reading will return a desired change

83 to the input to attempt matching output to this value.

84 """

85 self.setpoint = setpoint

86 self.Kp = Kp

87 self.Ki = Ki

88 self.Kd = Kd

89 self._integral = 0

90 self._previous_error = 0

91 self._change_limit = 0

92

93 def input_change(self, process_feedback, dt):

94 """Calculates desired input value change.

95

96 Based on process feedback and time interval (dt).

97 """

98 error = self.setpoint - process_feedback

99 self._integral = self._integral + (error * dt)

100 derivative = (error - self._previous_error) / dt

101 self._previous_error = error

102 input_change = (

103 (self.Kp * error)

104 + (self.Ki * self._integral)

105 + (self.Kd * derivative)

106 )

107 if self._change_limit and abs(input_change) > abs(self._change_limit):

108 sign = input_change / abs(input_change)

109 input_change = sign * self._change_limit

110 return input_change

111

112 def set_change_limit(self, limit):

113 """Ensures that input value changes are lower than limit.

114

115 Setting limit of zero disables this.

116 """

117 self._change_limit = limit

118

119

120 class PAVolumeController(object):

121 pa_types = {"input": "source", "output": "sink"}

122

123 def __init__(self, type, logger=None):

124 """Initializes the volume controller.

125

126 Arguments:

127 type: either input or output

128

129 """

130 self.type = type

131 self._volume = None

132 self.identifier = None

133 self.logger = logger

134

135 def set_volume(self, volume):

136 if not 0 <= volume <= 100:

137 return False

138 if not self.identifier:

139 return False

140 command = [

141 "pactl",

142 "set-%s-volume" % (self.pa_types[self.type]),

143 self.identifier,

144 str(int(volume)) + "%",

145 ]

146 self._pactl_output(command)

147 self._volume = volume

148 return True

149

150 def get_volume(self):

151 if not self.identifier:

152 return None

153 return self._volume

154

155 def mute(self, mute):

156 mute = str(int(mute))

157 if not self.identifier:

158 return False

159 command = [

160 "pactl",

161 "set-%s-mute" % (self.pa_types[self.type]),

162 self.identifier,

163 mute,

164 ]

165 self._pactl_output(command)

166 return True

167

168 def get_identifier(self):

169 if self.type:

170 self.identifier = self._get_identifier_for(self.type)

171 if self.identifier and self.logger:

172 message = "Using PulseAudio identifier %s for %s" % (

173 self.identifier,

174 self.type,

175 )

176 self.logger.info(message)

177 return self.identifier

178

179 def _get_identifier_for(self, type):

180 """Gets default PulseAudio identifier for given type.

181

182 Arguments:

183 type: either input or output

184

185 Returns:

186 A string: device_name

187

188 """

189

190 # parse `pactl info`, and get "Default Source:" or "Default Sink:" line

191 command = ["pactl", "info"]

192 result = self._pactl_output(command).splitlines()

193 identifier = self.pa_types[type].capitalize()

194 find_string = "Default %s: " % identifier

195

196 line = next(

197 (line for line in result if line.startswith(find_string)), None

198 )

199 if line is None:

200 if self.identifier and self.logger:

201 self.logger.error(

202 'Cannot find "%s" from `pactl info`.' % find_string

203 )

204 return None

205

206 # .removeprefix() but need to support Python 3.8

207 # Refer to https://stackoverflow.com/a/16891418

208 name = line[len(find_string) :].strip()

209

210 if (

211 name.startswith("auto_null")

212 or name.endswith(".monitor")

213 or name.startswith("@DEFAULT_")

214 ):

215 if self.logger:

216 self.logger.error(

217 "Default PulseAudio element for %s is invalid (%s)"

218 % (self.type, name)

219 )

220 return None

221

222 return name

223

224 def _pactl_output(self, command):

225 # This method mainly calls pactl (hence the name). Since pactl may

226 # return a failure if the audio layer is not yet initialized, we will

227 # try running a few times in case of failure. All our invocations of

228 # pactl should be "idempotent" so repeating them should not have

229 # any bad effects.

230 for attempt in range(0, 3):

231 try:

232 return subprocess.check_output(

233 command, universal_newlines=True

234 )

235 except subprocess.CalledProcessError:

236 time.sleep(5)

237 self.logger.error("Fail to execute: {}".format(command))

238 sys.exit(1)

239

240

241 class FileDumper(object):

242 def write_to_file(self, filename, data):

243 try:

244 with open(filename, "w") as f:

245 for i in data:

246 print(i, file=f)

247 return_value = True

248 except (TypeError, IOError):

249 return_value = False

250 return return_value

251

252

253 class SpectrumAnalyzer(object):

254 def __init__(self, points, sampling_frequency=44100, wanted_samples=50):

255 self.spectrum = [0] * points

256 self.number_of_samples = 0

257 self.wanted_samples = wanted_samples

258 self.sampling_frequency = sampling_frequency

259 # Frequencies should contain *real* frequency which is half of

260 # the sampling frequency

261 self.frequencies = [

262 ((sampling_frequency / 2.0) / points) * i for i in range(points)

263 ]

264

265 def _average(self):

266 return sum(self.spectrum) / len(self.spectrum)

267

268 def sample(self, sample):

269 if len(sample) != len(self.spectrum):

270 return

271 self.spectrum = [

272 ((old * self.number_of_samples) + new)

273 / (self.number_of_samples + 1)

274 for old, new in zip(self.spectrum, sample)

275 ]

276 self.number_of_samples += 1

277

278 def frequencies_with_peak_magnitude(self, threshold=1.0):

279 # First establish the base level

280 per_magnitude_bins = collections.defaultdict(int)

281 for magnitude in self.spectrum:

282 per_magnitude_bins[magnitude] += 1

283 base_level = max(

284 per_magnitude_bins, key=lambda x: per_magnitude_bins[x]

285 )

286 # Now return all values that are higher (more positive)

287 # than base_level + threshold

288 peaks = []

289 for i in range(1, len(self.spectrum) - 1):

290 first_index = i - 1

291 last_index = i + 1

292 if (

293 self.spectrum[first_index] < self.spectrum[i]

294 and self.spectrum[last_index] < self.spectrum[i]

295 and self.spectrum[i] > base_level + threshold

296 ):

297 peaks.append(i)

298

299 return peaks

300

301 def frequency_band_for(self, frequency):

302 """Convenience function to tell me which band

303 a frequency is contained in

304 """

305 # Note that actual frequencies are half of what the sampling

306 # frequency would tell us. If SF is 44100 then maximum actual

307 # frequency is 22050, and if I have 10 frequency bins each will

308 # contain only 2205 Hz, not 4410 Hz.

309 max_frequency = self.sampling_frequency / 2

310 if frequency > max_frequency or frequency < 0:

311 return None

312 band = float(frequency) / (max_frequency / len(self.spectrum))

313 return int(math.ceil(band)) - 1

314

315 def frequencies_for_band(self, band):

316 """Convenience function to tell me the delimiting frequencies

317 for a band

318 """

319 if band >= len(self.spectrum) or band < 0:

320 return None

321 lower = self.frequencies[band]

322 upper = lower + ((self.sampling_frequency / 2.0) / len(self.spectrum))

323 return (lower, upper)

324

325 def sampling_complete(self):

326 return self.number_of_samples >= self.wanted_samples

327

328

329 class GStreamerMessageHandler(object):

330 def __init__(

331 self,

332 rec_level_range,

333 logger,

334 volumecontroller,

335 pidcontroller,

336 spectrum_analyzer,

337 ):

338 """Initializes the message handler. It knows how to handle

339 spectrum and level gstreamer messages.

340

341 Arguments:

342 rec_level_range: tuple with acceptable recording level

343 ranges

344 logger: logging object with debug, info, error methods.

345 volumecontroller: an instance of VolumeController to use

346 to adjust RECORDING level

347 pidcontroller: a PID controller instance which helps control

348 volume

349 spectrum_analyzer: instance of SpectrumAnalyzer to collect

350 data from spectrum messages

351

352 """

353 self.current_level = sys.maxsize

354 self.logger = logger

355 self.pid_controller = pidcontroller

356 self.rec_level_range = rec_level_range

357 self.spectrum_analyzer = spectrum_analyzer

358 self.volume_controller = volumecontroller

359

360 def set_quit_method(self, method):

361 """Method that will be called when sampling is complete."""

362 self._quit_method = method

363

364 def bus_message_handler(self, bus, message):

365 if message.type == Gst.MessageType.ELEMENT:

366 message_name = message.get_structure().get_name()

367 if message_name == "spectrum":

368 # TODO: Due to an upstream bug, a structure's get_value method

369 # doesn't work if the value in question is an array (as is the

370 # case with the magnitudes).

371 # https://bugzilla.gnome.org/show_bug.cgi?id=693168

372 # We have to resort to parsing the string representation of the

373 # structure. It's an ugly hack but it works.

374 # Ideally we'd be able to say this to get fft_magnitudes:

375 # message.get_structure.get_value('magnitude').

376 # If an upstream fix ever makes it into gstreamer,

377 # remember to remove this hack and the parse_spectrum method

378 struct_string = message.get_structure().to_string()

379 structure = parse_spectrum_message_structure(struct_string)

380 fft_magnitudes = structure["magnitude"]

381 self.spectrum_method(self.spectrum_analyzer, fft_magnitudes)

382

383 if message_name == "level":

384 # peak_value is our process feedback

385 # It's returned as an array, so I need the first (and only)

386 # element

387 peak_value = message.get_structure().get_value("peak")[0]

388 self.level_method(

389 peak_value, self.pid_controller, self.volume_controller

390 )

391

392 # Adjust recording level

393 def level_method(self, level, pid_controller, volume_controller):

394 # If volume controller doesn't return a valid volume,

395 # we can't control it :(

396 current_volume = volume_controller.get_volume()

397 if current_volume is None:

398 self.logger.error(

399 "Unable to control recording volume. "

400 "Test results may be wrong"

401 )

402 return

403 self.current_level = level

404 change = pid_controller.input_change(level, 0.10)

405 if self.logger:

406 self.logger.debug(

407 "Peak level: %(peak_level).2f, "

408 "volume: %(volume)d%%, Volume change: %(change)f%%"

409 % {

410 "peak_level": level,

411 "change": change,

412 "volume": current_volume,

413 }

414 )

415 volume_controller.set_volume(current_volume + change)

416

417 # Only sample if level is within the threshold

418 def spectrum_method(self, analyzer, spectrum):

419 if (

420 self.rec_level_range[1] <= self.current_level

421 or self.current_level <= self.rec_level_range[0]

422 ):

423 self.logger.debug(

424 "Sampling, recorded %d samples" % analyzer.number_of_samples

425 )

426 analyzer.sample(spectrum)

427 if analyzer.sampling_complete() and self._quit_method:

428 self.logger.info("Sampling complete, ending process")

429 self._quit_method()

430

431

432 class GstAudioObject(object):

433 def __init__(self):

434 self.class_name = self.__class__.__name__

435

436 def _set_state(self, state, description):

437 self.pipeline.set_state(state)

438 message = "%s: %s" % (self.class_name, description)

439 if self.logger:

440 self.logger.info(message)

441

442 def start(self):

443 self._set_state(Gst.State.PLAYING, "Starting")

444

445 def stop(self):

446 self._set_state(Gst.State.NULL, "Stopping")

447

448

449 class Player(GstAudioObject):

450 def __init__(self, frequency=DEFAULT_TEST_FREQUENCY, logger=None):

451 super(Player, self).__init__()

452 self.pipeline_description = (

453 "audiotestsrc wave=sine freq=%s "

454 "! audioconvert "

455 "! audioresample "

456 "! autoaudiosink" % int(frequency)

457 )

458 self.logger = logger

459 if self.logger:

460 self.logger.debug(self.pipeline_description)

461 self.pipeline = Gst.parse_launch(self.pipeline_description)

462

463

464 class Recorder(GstAudioObject):

465 def __init__(

466 self,

467 output_file,

468 bins=BINS,

469 sampling_frequency=SAMPLING_FREQUENCY,

470 fft_interval=FFT_INTERVAL,

471 logger=None,

472 ):

473 super(Recorder, self).__init__()

474 pipeline_description = """autoaudiosrc

475 ! queue

476 ! level message=true

477 ! audioconvert

478 ! audio/x-raw, channels=1, rate=(int)%(rate)s

479 ! audioresample

480 ! spectrum interval=%(fft_interval)s bands = %(bands)s

481 ! wavenc

482 ! filesink location=%(file)s""" % {

483 "bands": bins,

484 "rate": sampling_frequency,

485 "fft_interval": fft_interval,

486 "file": output_file,

487 }

488 self.logger = logger

489 if self.logger:

490 self.logger.debug(pipeline_description)

491 self.pipeline = Gst.parse_launch(pipeline_description)

492

493 def register_message_handler(self, handler_method):

494 if self.logger:

495 message = "Registering message handler: %s" % handler_method

496 self.logger.debug(message)

497 self.bus = self.pipeline.get_bus()

498 self.bus.add_signal_watch()

499 self.bus.connect("message", handler_method)

500

501

502 def parse_spectrum_message_structure(struct_string):

503 # First let's jsonize this

504 # This is the message name, which we don't need

505 text = struct_string.replace("spectrum, ", "")

506 # name/value separator in json is : and not =

507 text = text.replace("=", ": ")

508 # Mutate the {} array notation from the structure to

509 # [] notation for json.

510 text = text.replace("{", "[")

511 text = text.replace("}", "]")

512 # Remove a few stray semicolons that aren't needed

513 text = text.replace(";", "")

514 # Remove the data type fields, as json doesn't need them

515 text = re.sub(r"\(.+?\)", "", text)

516 # double-quote the identifiers

517 text = re.sub(r"([\w-]+):", r'"\1":', text)

518 # Wrap the whole thing in brackets

519 text = "{" + text + "}"

520 # Try to parse and return something sensible here, even if

521 # the data was unparsable.

522 try:

523 return json.loads(text)

524 except ValueError:

525 return None

526

527

528 def process_arguments():

529 description = """

530 Plays a single frequency through the default output, then records on

531 the default input device. Analyzes the recorded signal to test for

532 presence of the played frequency, if present it exits with success.

533 """

534 parser = argparse.ArgumentParser(description=description)

535 parser.add_argument(

536 "-t",

537 "--time",

538 dest="test_duration",

539 action="store",

540 default=30,

541 type=int,

542 help="""Maximum test duration, default %(default)s seconds.

543 It may exit sooner if it determines it has enough data.""",

544 )

545 parser.add_argument(

546 "-a",

547 "--audio",

548 action="store",

549 default="/dev/null",

550 type=str,

551 help="File to save recorded audio in .wav format",

552 )

553 parser.add_argument(

554 "-q",

555 "--quiet",

556 action="store_true",

557 default=False,

558 help="Be quiet, no output unless there's an error.",

559 )

560 parser.add_argument(

561 "-d",

562 "--debug",

563 action="store_true",

564 default=False,

565 help="Debugging output",

566 )

567 parser.add_argument(

568 "-f",

569 "--frequency",

570 action="store",

571 default=DEFAULT_TEST_FREQUENCY,

572 type=int,

573 help="Frequency for test signal, default %(default)s Hz",

574 )

575 parser.add_argument(

576 "-u",

577 "--spectrum",

578 action="store",

579 type=str,

580 help="""File to save spectrum information for plotting

581 (one frequency/magnitude pair per line)""",

582 )

583 return parser.parse_args()

584

585

586#

587 def main():

588 # Get arguments.

589 args = process_arguments()

590

591 # Setup logging

592 level = logging.INFO

593 if args.debug:

594 level = logging.DEBUG

595 if args.quiet:

596 level = logging.ERROR

597 logging.basicConfig(level=level)

598 try:

599 # Launches recording pipeline. I need to hook up into the gst

600 # messages.

601 recorder = Recorder(output_file=args.audio, logger=logging)

602 # Just launches the playing pipeline

603 player = Player(frequency=args.frequency, logger=logging)

604 except GObject.GError as excp:

605 logging.critical("Unable to initialize GStreamer pipelines: %s", excp)

606 sys.exit(127)

607

608 # This just receives a process feedback and tells me how much to change to

609 # achieve the setpoint

610 pidctrl = PIDController(

611 Kp=0.7, Ki=0.01, Kd=0.01, setpoint=REC_LEVEL_RANGE[0]

612 )

613 pidctrl.set_change_limit(5)

614 # This gathers spectrum data.

615 analyzer = SpectrumAnalyzer(

616 points=BINS, sampling_frequency=SAMPLING_FREQUENCY

617 )

618

619 # Volume controllers actually set volumes for their device types.

620 # we should at least issue a warning

621 recorder.volumecontroller = PAVolumeController(

622 type="input", logger=logging

623 )

624 if not recorder.volumecontroller.get_identifier():

625 logging.warning(

626 "Unable to get input volume control identifier. "

627 "Test results will probably be invalid"

628 )

629 recorder.volumecontroller.set_volume(0)

630 recorder.volumecontroller.mute(False)

631

632 player.volumecontroller = PAVolumeController(type="output", logger=logging)

633 if not player.volumecontroller.get_identifier():

634 logging.warning(

635 "Unable to get output volume control identifier. "

636 "Test results will probably be invalid"

637 )

638 player.volumecontroller.set_volume(PLAY_VOLUME)

639 player.volumecontroller.mute(False)

640

641 # This handles the messages from gstreamer and orchestrates

642 # the passed volume controllers, pid controller and spectrum analyzer

643 # accordingly.

644 gmh = GStreamerMessageHandler(

645 rec_level_range=REC_LEVEL_RANGE,

646 logger=logging,

647 volumecontroller=recorder.volumecontroller,

648 pidcontroller=pidctrl,

649 spectrum_analyzer=analyzer,

650 )

651

652 # I need to tell the recorder which method will handle messages.

653 recorder.register_message_handler(gmh.bus_message_handler)

654

655 # Create the loop and add a few triggers

656 loop = GLib.MainLoop()

657 GLib.timeout_add_seconds(0, player.start)

658 GLib.timeout_add_seconds(0, recorder.start)

659 GLib.timeout_add_seconds(args.test_duration, loop.quit)

660

661 # Tell the gmh which method to call when enough samples are collected

662 gmh.set_quit_method(loop.quit)

663

664 loop.run()

665

666 # When the loop ends, set things back to reasonable states

667 player.stop()

668 recorder.stop()

669 player.volumecontroller.set_volume(50)

670 recorder.volumecontroller.set_volume(10)

671

672 # See if data gathering was successful.

673 test_band = analyzer.frequency_band_for(args.frequency)

674 candidate_bands = analyzer.frequencies_with_peak_magnitude(

675 MAGNITUDE_THRESHOLD

676 )

677 for band in candidate_bands:

678 logging.debug(

679 "Band (%.2f,%.2f) contains a magnitude peak"

680 % analyzer.frequencies_for_band(band)

681 )

682 if test_band in candidate_bands:

683 freqs_for_band = analyzer.frequencies_for_band(test_band)

684 logging.info(

685 "PASS: Test frequency of %s in band (%.2f, %.2f) "

686 "which contains a magnitude peak"

687 % ((args.frequency,) + freqs_for_band)

688 )

689 return_value = 0

690 else:

691 logging.info(

692 "FAIL: Test frequency of %s is not in one of the "

693 "bands with magnitude peaks" % args.frequency

694 )

695 return_value = 1

696 # Is the microphone broken?

697 if len(set(analyzer.spectrum)) <= 1:

698 logging.info(

699 "WARNING: Microphone seems broken, didn't even "

700 "record ambient noise"

701 )

702

703 if args.spectrum:

704 logging.info("Saving spectrum data for plotting as %s" % args.spectrum)

705 if not FileDumper().write_to_file(

706 args.spectrum,

707 [

708 "%s,%s" % t

709 for t in zip(analyzer.frequencies, analyzer.spectrum)

710 ],

711 ):

712 logging.error(

713 "Couldn't save spectrum data for plotting", file=sys.stderr

714 )

715

716 return return_value

717

718

719 if __name__ == "__main__":

720 sys.exit(main())