#!/usr/bin/python
# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Command line tool to analyze wave file and detect artifacts."""
import argparse
import logging
import math
import numpy
import pprint
import wave
import common
from autotest_lib.client.cros.audio import audio_analysis
from autotest_lib.client.cros.audio import audio_data
from autotest_lib.client.cros.audio import audio_quality_measurement
def add_args(parser):
"""Adds command line arguments."""
parser.add_argument('filename', metavar='WAV_FILE', type=str,
help='The wave file to check.')
parser.add_argument('--debug', action='store_true', default=False,
help='Show debug message.')
parser.add_argument('--spectral', action='store_true', default=False,
help='Spectral analysis on each channel.')
parser.add_argument('--quality', action='store_true', default=False,
help='Quality analysis on each channel. Implies '
'--spectral')
def parse_args(parser):
"""Parses args."""
args = parser.parse_args()
# Quality is checked within spectral analysis.
if args.quality:
args.spectral = True
return args
class WaveFileException(Exception):
"""Error in WaveFile."""
pass
class WaveFile(object):
"""Class which handles wave file reading.
Properties:
raw_data: audio_data.AudioRawData object for data in wave file.
rate: sampling rate.
"""
def __init__(self, filename):
"""Inits a wave file.
@param filename: file name of the wave file.
"""
self.raw_data = None
self.rate = None
self._filename = filename
self._wave_reader = None
self._n_channels = None
self._sample_width_bits = None
self._n_frames = None
self._binary = None
self._read_wave_file()
def _read_wave_file(self):
"""Reads wave file header and samples.
@raises:
WaveFileException: Wave format is not supported.
"""
try:
self._wave_reader = wave.open(self._filename, 'r')
self._read_wave_header()
self._read_wave_binary()
except wave.Error as e:
if 'unknown format: 65534' in str(e):
raise WaveFileException(
'WAVE_FORMAT_EXTENSIBLE is not supproted. '
'Try command "sox in.wav -t wavpcm out.wav" to convert '
'the file to WAVE_FORMAT_PCM format.')
finally:
if self._wave_reader:
self._wave_reader.close()
def _read_wave_header(self):
"""Reads wave file header.
@raises WaveFileException: wave file is compressed.
"""
# Header is a tuple of
# (nchannels, sampwidth, framerate, nframes, comptype, compname).
header = self._wave_reader.getparams()
logging.debug('Wave header: %s', header)
self._n_channels = header[0]
self._sample_width_bits = header[1] * 8
self.rate = header[2]
self._n_frames = header[3]
comptype = header[4]
compname = header[5]
if comptype != 'NONE' or compname != 'not compressed':
raise WaveFileException('Can not support compressed wav file.')
def _read_wave_binary(self):
"""Reads in samples in wave file."""
self._binary = self._wave_reader.readframes(self._n_frames)
format_str = 'S%d_LE' % self._sample_width_bits
self.raw_data = audio_data.AudioRawData(
binary=self._binary,
channel=self._n_channels,
sample_format=format_str)
class QualityCheckerError(Exception):
"""Error in QualityChecker."""
pass
class QualityChecker(object):
"""Quality checker controls the flow of checking quality of raw data."""
def __init__(self, raw_data, rate):
"""Inits a quality checker.
@param raw_data: An audio_data.AudioRawData object.
@param rate: Sampling rate.
"""
self._raw_data = raw_data
self._rate = rate
def do_spectral_analysis(self, check_quality=False):
"""Gets the spectral_analysis result.
@param check_quality: Check quality of each channel.
"""
self.has_data()
for channel_idx in xrange(self._raw_data.channel):
signal = self._raw_data.channel_data[channel_idx]
max_abs = max(numpy.abs(signal))
logging.debug('Channel %d max abs signal: %f', channel_idx, max_abs)
if max_abs == 0:
logging.info('No data on channel %d, skip this channel',
channel_idx)
continue
saturate_value = audio_data.get_maximum_value_from_sample_format(
self._raw_data.sample_format)
normalized_signal = audio_analysis.normalize_signal(
signal, saturate_value)
logging.debug('saturate_value: %f', saturate_value)
logging.debug('max signal after normalized: %f', max(normalized_signal))
spectral = audio_analysis.spectral_analysis(
normalized_signal, self._rate)
logging.info('Channel %d spectral:\n%s', channel_idx,
pprint.pformat(spectral))
if check_quality:
quality = audio_quality_measurement.quality_measurement(
signal=normalized_signal,
rate=self._rate,
dominant_frequency=spectral[0][0])
logging.info('Channel %d quality:\n%s', channel_idx,
pprint.pformat(quality))
def has_data(self):
"""Checks if data has been set.
@raises QualityCheckerError: if data or rate is not set yet.
"""
if not self._raw_data or not self._rate:
raise QualityCheckerError('Data and rate is not set yet')
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description='Check signal quality of a wave file. Each channel should'
' either be all zeros, or sine wave of a fixed frequency.')
add_args(parser)
args = parse_args(parser)
level = logging.DEBUG if args.debug else logging.INFO
format = '%(asctime)-15s:%(levelname)s:%(pathname)s:%(lineno)d: %(message)s'
logging.basicConfig(format=format, level=level)
wavefile = WaveFile(args.filename)
checker = QualityChecker(wavefile.raw_data, wavefile.rate)
if args.spectral:
checker.do_spectral_analysis(check_quality=args.quality)