aifc.py

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

        self._nframes = nframes

    def getnframes(self):
        return self._nframeswritten

    def setcomptype(self, comptype, compname):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'
        if comptype not in ('NONE', 'ULAW', 'ALAW', 'G722'):
            raise Error, 'unsupported compression type'
        self._comptype = comptype
        self._compname = compname

    def getcomptype(self):
        return self._comptype

    def getcompname(self):
        return self._compname

##  def setversion(self, version):
##      if self._nframeswritten:
##          raise Error, 'cannot change parameters after starting to write'
##      self._version = version

    def setparams(self, (nchannels, sampwidth, framerate, nframes, comptype, compname)):
        if self._nframeswritten:
            raise Error, 'cannot change parameters after starting to write'
        if comptype not in ('NONE', 'ULAW', 'ALAW', 'G722'):
            raise Error, 'unsupported compression type'
        self.setnchannels(nchannels)
        self.setsampwidth(sampwidth)
        self.setframerate(framerate)
        self.setnframes(nframes)
        self.setcomptype(comptype, compname)

    def getparams(self):
        if not self._nchannels or not self._sampwidth or not self._framerate:
            raise Error, 'not all parameters set'
        return self._nchannels, self._sampwidth, self._framerate, \
              self._nframes, self._comptype, self._compname

    def setmark(self, id, pos, name):
        if id <= 0:
            raise Error, 'marker ID must be > 0'
        if pos < 0:
            raise Error, 'marker position must be >= 0'
        if type(name) != type(''):
            raise Error, 'marker name must be a string'
        for i in range(len(self._markers)):
            if id == self._markers[i][0]:
                self._markers[i] = id, pos, name
                return
        self._markers.append((id, pos, name))

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

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

    def tell(self):
        return self._nframeswritten

    def writeframesraw(self, data):
        self._ensure_header_written(len(data))
        nframes = len(data) / (self._sampwidth * self._nchannels)
        if self._convert:
            data = self._convert(data)
        self._file.write(data)
        self._nframeswritten = self._nframeswritten + nframes
        self._datawritten = self._datawritten + len(data)

    def writeframes(self, data):
        self.writeframesraw(data)
        if self._nframeswritten != self._nframes or \
              self._datalength != self._datawritten:
            self._patchheader()

    def close(self):
        self._ensure_header_written(0)
        if self._datawritten & 1:
            # quick pad to even size
            self._file.write(chr(0))
            self._datawritten = self._datawritten + 1
        self._writemarkers()
        if self._nframeswritten != self._nframes or \
              self._datalength != self._datawritten or \
              self._marklength:
            self._patchheader()
        if self._comp:
            self._comp.CloseCompressor()
            self._comp = None
        self._file.flush()
        self._file = None

    #
    # Internal methods.
    #

    def _comp_data(self, data):
        import cl
        dum = self._comp.SetParam(cl.FRAME_BUFFER_SIZE, len(data))
        dum = self._comp.SetParam(cl.COMPRESSED_BUFFER_SIZE, len(data))
        return self._comp.Compress(self._nframes, data)

    def _lin2ulaw(self, data):
        import audioop
        return audioop.lin2ulaw(data, 2)

    def _lin2adpcm(self, data):
        import audioop
        if not hasattr(self, '_adpcmstate'):
            self._adpcmstate = None
        data, self._adpcmstate = audioop.lin2adpcm(data, 2,
                               self._adpcmstate)
        return data

    def _ensure_header_written(self, datasize):
        if not self._nframeswritten:
            if self._comptype in ('ULAW', 'ALAW'):
                if not self._sampwidth:
                    self._sampwidth = 2
                if self._sampwidth != 2:
                    raise Error, 'sample width must be 2 when compressing with ULAW or ALAW'
            if self._comptype == 'G722':
                if not self._sampwidth:
                    self._sampwidth = 2
                if self._sampwidth != 2:
                    raise Error, 'sample width must be 2 when compressing with G7.22 (ADPCM)'
            if not self._nchannels:
                raise Error, '# channels not specified'
            if not self._sampwidth:
                raise Error, 'sample width not specified'
            if not self._framerate:
                raise Error, 'sampling rate not specified'
            self._write_header(datasize)

    def _init_compression(self):
        if self._comptype == 'G722':
            import audioop
            self._convert = self._lin2adpcm
            return
        try:
            import cl
        except ImportError:
            if self._comptype == 'ULAW':
                try:
                    import audioop
                    self._convert = self._lin2ulaw
                    return
                except ImportError:
                    pass
            raise Error, 'cannot write compressed AIFF-C files'
        if self._comptype == 'ULAW':
            scheme = cl.G711_ULAW
        elif self._comptype == 'ALAW':
            scheme = cl.G711_ALAW
        else:
            raise Error, 'unsupported compression type'
        self._comp = cl.OpenCompressor(scheme)
        params = [cl.ORIGINAL_FORMAT, 0,
              cl.BITS_PER_COMPONENT, self._sampwidth * 8,
              cl.FRAME_RATE, self._framerate,
              cl.FRAME_BUFFER_SIZE, 100,
              cl.COMPRESSED_BUFFER_SIZE, 100]
        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._comp.SetParams(params)
        # the compressor produces a header which we ignore
        dummy = self._comp.Compress(0, '')
        self._convert = self._comp_data

    def _write_header(self, initlength):
        if self._aifc and self._comptype != 'NONE':
            self._init_compression()
        self._file.write('FORM')
        if not self._nframes:
            self._nframes = initlength / (self._nchannels * self._sampwidth)
        self._datalength = self._nframes * self._nchannels * self._sampwidth
        if self._datalength & 1:
            self._datalength = self._datalength + 1
        if self._aifc:
            if self._comptype in ('ULAW', 'ALAW'):
                self._datalength = self._datalength / 2
                if self._datalength & 1:
                    self._datalength = self._datalength + 1
            elif self._comptype == 'G722':
                self._datalength = (self._datalength + 3) / 4
                if self._datalength & 1:
                    self._datalength = self._datalength + 1
        self._form_length_pos = self._file.tell()
        commlength = self._write_form_length(self._datalength)
        if self._aifc:
            self._file.write('AIFC')
            self._file.write('FVER')
            _write_long(self._file, 4)
            _write_long(self._file, self._version)
        else:
            self._file.write('AIFF')
        self._file.write('COMM')
        _write_long(self._file, commlength)
        _write_short(self._file, self._nchannels)
        self._nframes_pos = self._file.tell()
        _write_long(self._file, self._nframes)
        _write_short(self._file, self._sampwidth * 8)
        _write_float(self._file, self._framerate)
        if self._aifc:
            self._file.write(self._comptype)
            _write_string(self._file, self._compname)
        self._file.write('SSND')
        self._ssnd_length_pos = self._file.tell()
        _write_long(self._file, self._datalength + 8)
        _write_long(self._file, 0)
        _write_long(self._file, 0)

    def _write_form_length(self, datalength):
        if self._aifc:
            commlength = 18 + 5 + len(self._compname)
            if commlength & 1:
                commlength = commlength + 1
            verslength = 12
        else:
            commlength = 18
            verslength = 0
        _write_long(self._file, 4 + verslength + self._marklength + \
                    8 + commlength + 16 + datalength)
        return commlength

    def _patchheader(self):
        curpos = self._file.tell()
        if self._datawritten & 1:
            datalength = self._datawritten + 1
            self._file.write(chr(0))
        else:
            datalength = self._datawritten
        if datalength == self._datalength and \
              self._nframes == self._nframeswritten and \
              self._marklength == 0:
            self._file.seek(curpos, 0)
            return
        self._file.seek(self._form_length_pos, 0)
        dummy = self._write_form_length(datalength)
        self._file.seek(self._nframes_pos, 0)
        _write_long(self._file, self._nframeswritten)
        self._file.seek(self._ssnd_length_pos, 0)
        _write_long(self._file, datalength + 8)
        self._file.seek(curpos, 0)
        self._nframes = self._nframeswritten
        self._datalength = datalength

    def _writemarkers(self):
        if len(self._markers) == 0:
            return
        self._file.write('MARK')
        length = 2
        for marker in self._markers:
            id, pos, name = marker
            length = length + len(name) + 1 + 6
            if len(name) & 1 == 0:
                length = length + 1
        _write_long(self._file, length)
        self._marklength = length + 8
        _write_short(self._file, len(self._markers))
        for marker in self._markers:
            id, pos, name = marker
            _write_short(self._file, id)
            _write_long(self._file, pos)
            _write_string(self._file, name)

def open(f, mode=None):
    if mode is None:
        if hasattr(f, 'mode'):
            mode = f.mode
        else:
            mode = 'rb'
    if mode in ('r', 'rb'):
        return Aifc_read(f)
    elif mode in ('w', 'wb'):
        return Aifc_write(f)
    else:
        raise Error, "mode must be 'r', 'rb', 'w', or 'wb'"

openfp = open # B/W compatibility

if __name__ == '__main__':
    import sys
    if not sys.argv[1:]:
        sys.argv.append('/usr/demos/data/audio/bach.aiff')
    fn = sys.argv[1]
    f = open(fn, 'r')
    print "Reading", fn
    print "nchannels =", f.getnchannels()
    print "nframes   =", f.getnframes()
    print "sampwidth =", f.getsampwidth()
    print "framerate =", f.getframerate()
    print "comptype  =", f.getcomptype()
    print "compname  =", f.getcompname()
    if sys.argv[2:]:
        gn = sys.argv[2]
        print "Writing", gn
        g = open(gn, 'w')
        g.setparams(f.getparams())
        while 1:
            data = f.readframes(1024)
            if not data:
                break
            g.writeframes(data)
        g.close()
        f.close()
        print "Done."