aifc.py

来自「mallet是自然语言处理、机器学习领域的一个开源项目。」· Python 代码 · 共 962 行 · 第 1/3 页

            raise Error, 'COMM chunk and/or SSND chunk missing'
        if self._aifc and self._decomp:
            import cl
            params = [cl.ORIGINAL_FORMAT, 0,
                  cl.BITS_PER_COMPONENT, self._sampwidth * 8,
                  cl.FRAME_RATE, self._framerate]
            if self._nchannels == 1:
                params[1] = cl.MONO
            elif self._nchannels == 2:
                params[1] = cl.STEREO_INTERLEAVED
            else:
                raise Error, 'cannot compress more than 2 channels'
            self._decomp.SetParams(params)

    def __init__(self, f):
        if type(f) == type(''):
            f = __builtin__.open(f, 'rb')
        # else, assume it is an open file object already
        self.initfp(f)

    #
    # User visible methods.
    #
    def getfp(self):
        return self._file

    def rewind(self):
        self._ssnd_seek_needed = 1
        self._soundpos = 0

    def close(self):
        if self._decomp:
            self._decomp.CloseDecompressor()
            self._decomp = None
        self._file = None

    def tell(self):
        return self._soundpos

    def getnchannels(self):
        return self._nchannels

    def getnframes(self):
        return self._nframes

    def getsampwidth(self):
        return self._sampwidth

    def getframerate(self):
        return self._framerate

    def getcomptype(self):
        return self._comptype

    def getcompname(self):
        return self._compname

##  def getversion(self):
##      return self._version

    def getparams(self):
        return self.getnchannels(), self.getsampwidth(), \
              self.getframerate(), self.getnframes(), \
              self.getcomptype(), self.getcompname()

    def getmarkers(self):
        if len(self._markers) == 0:
            return None
        return self._markers

    def getmark(self, id):
        for marker in self._markers:
            if id == marker[0]:
                return marker
        raise Error, 'marker ' + `id` + ' does not exist'

    def setpos(self, pos):
        if pos < 0 or pos > self._nframes:
            raise Error, 'position not in range'
        self._soundpos = pos
        self._ssnd_seek_needed = 1

    def readframes(self, nframes):
        if self._ssnd_seek_needed:
            self._ssnd_chunk.seek(0)
            dummy = self._ssnd_chunk.read(8)
            pos = self._soundpos * self._framesize
            if pos:
                self._ssnd_chunk.seek(pos + 8)
            self._ssnd_seek_needed = 0
        if nframes == 0:
            return ''
        data = self._ssnd_chunk.read(nframes * self._framesize)
        if self._convert and data:
            data = self._convert(data)
        self._soundpos = self._soundpos + len(data) / (self._nchannels * self._sampwidth)
        return data

    #
    # Internal methods.
    #

    def _decomp_data(self, data):
        import cl
        dummy = self._decomp.SetParam(cl.FRAME_BUFFER_SIZE,
                          len(data) * 2)
        return self._decomp.Decompress(len(data) / self._nchannels,
                           data)

    def _ulaw2lin(self, data):
        import audioop
        return audioop.ulaw2lin(data, 2)

    def _adpcm2lin(self, data):
        import audioop
        if not hasattr(self, '_adpcmstate'):
            # first time
            self._adpcmstate = None
        data, self._adpcmstate = audioop.adpcm2lin(data, 2,
                               self._adpcmstate)
        return data

    def _read_comm_chunk(self, chunk):
        self._nchannels = _read_short(chunk)
        self._nframes = _read_long(chunk)
        self._sampwidth = (_read_short(chunk) + 7) / 8
        self._framerate = int(_read_float(chunk))
        self._framesize = self._nchannels * self._sampwidth
        if self._aifc:
            #DEBUG: SGI's soundeditor produces a bad size :-(
            kludge = 0
            if chunk.chunksize == 18:
                kludge = 1
                print 'Warning: bad COMM chunk size'
                chunk.chunksize = 23
            #DEBUG end
            self._comptype = chunk.read(4)
            #DEBUG start
            if kludge:
                length = ord(chunk.file.read(1))
                if length & 1 == 0:
                    length = length + 1
                chunk.chunksize = chunk.chunksize + length
                chunk.file.seek(-1, 1)
            #DEBUG end
            self._compname = _read_string(chunk)
            if self._comptype != 'NONE':
                if self._comptype == 'G722':
                    try:
                        import audioop
                    except ImportError:
                        pass
                    else:
                        self._convert = self._adpcm2lin
                        self._framesize = self._framesize / 4
                        return
                # for ULAW and ALAW try Compression Library
                try:
                    import cl
                except ImportError:
                    if self._comptype == 'ULAW':
                        try:
                            import audioop
                            self._convert = self._ulaw2lin
                            self._framesize = self._framesize / 2
                            return
                        except ImportError:
                            pass
                    raise Error, 'cannot read compressed AIFF-C files'
                if self._comptype == 'ULAW':
                    scheme = cl.G711_ULAW
                    self._framesize = self._framesize / 2
                elif self._comptype == 'ALAW':
                    scheme = cl.G711_ALAW
                    self._framesize = self._framesize / 2
                else:
                    raise Error, 'unsupported compression type'
                self._decomp = cl.OpenDecompressor(scheme)
                self._convert = self._decomp_data
        else:
            self._comptype = 'NONE'
            self._compname = 'not compressed'

    def _readmark(self, chunk):
        nmarkers = _read_short(chunk)
        # Some files appear to contain invalid counts.
        # Cope with this by testing for EOF.
        try:
            for i in range(nmarkers):
                id = _read_short(chunk)
                pos = _read_long(chunk)
                name = _read_string(chunk)
                if pos or name:
                    # some files appear to have
                    # dummy markers consisting of
                    # a position 0 and name ''
                    self._markers.append((id, pos, name))
        except EOFError:
            print 'Warning: MARK chunk contains only',
            print len(self._markers),
            if len(self._markers) == 1: print 'marker',
            else: print 'markers',
            print 'instead of', nmarkers

class Aifc_write:
    # Variables used in this class:
    #
    # These variables are user settable through appropriate methods
    # of this class:
    # _file -- the open file with methods write(), close(), tell(), seek()
    #       set through the __init__() method
    # _comptype -- the AIFF-C compression type ('NONE' in AIFF)
    #       set through the setcomptype() or setparams() method
    # _compname -- the human-readable AIFF-C compression type
    #       set through the setcomptype() or setparams() method
    # _nchannels -- the number of audio channels
    #       set through the setnchannels() or setparams() method
    # _sampwidth -- the number of bytes per audio sample
    #       set through the setsampwidth() or setparams() method
    # _framerate -- the sampling frequency
    #       set through the setframerate() or setparams() method
    # _nframes -- the number of audio frames written to the header
    #       set through the setnframes() or setparams() method
    # _aifc -- whether we're writing an AIFF-C file or an AIFF file
    #       set through the aifc() method, reset through the
    #       aiff() method
    #
    # These variables are used internally only:
    # _version -- the AIFF-C version number
    # _comp -- the compressor from builtin module cl
    # _nframeswritten -- the number of audio frames actually written
    # _datalength -- the size of the audio samples written to the header
    # _datawritten -- the size of the audio samples actually written

    def __init__(self, f):
        if type(f) == type(''):
            filename = f
            f = __builtin__.open(f, 'wb')
        else:
            # else, assume it is an open file object already
            filename = '???'
        self.initfp(f)
        if filename[-5:] == '.aiff':
            self._aifc = 0
        else:
            self._aifc = 1

    def initfp(self, file):
        self._file = file
        self._version = _AIFC_version
        self._comptype = 'NONE'
        self._compname = 'not compressed'
        self._comp = None
        self._convert = None
        self._nchannels = 0
        self._sampwidth = 0
        self._framerate = 0
        self._nframes = 0
        self._nframeswritten = 0
        self._datawritten = 0
        self._datalength = 0
        self._markers = []
        self._marklength = 0
        self._aifc = 1      # AIFF-C is default

    def __del__(self):
        if self._file:
            self.close()

    #
    # User visible methods.
    #
    def aiff(self):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'
        self._aifc = 0

    def aifc(self):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'
        self._aifc = 1

    def setnchannels(self, nchannels):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'
        if nchannels < 1:
            raise Error, 'bad # of channels'
        self._nchannels = nchannels

    def getnchannels(self):
        if not self._nchannels:
            raise Error, 'number of channels not set'
        return self._nchannels

    def setsampwidth(self, sampwidth):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'
        if sampwidth < 1 or sampwidth > 4:
            raise Error, 'bad sample width'
        self._sampwidth = sampwidth

    def getsampwidth(self):
        if not self._sampwidth:
            raise Error, 'sample width not set'
        return self._sampwidth

    def setframerate(self, framerate):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'
        if framerate <= 0:
            raise Error, 'bad frame rate'
        self._framerate = framerate

    def getframerate(self):
        if not self._framerate:
            raise Error, 'frame rate not set'
        return self._framerate

    def setnframes(self, nframes):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'