home *** CD-ROM | disk | FTP | other *** search
- """ Python 'bz2_codec' Codec - bz2 compression encoding
-
- Unlike most of the other codecs which target Unicode, this codec
- will return Python string objects for both encode and decode.
-
- Adapted by Raymond Hettinger from zlib_codec.py which was written
- by Marc-Andre Lemburg (mal@lemburg.com).
-
- """
- import codecs
- import bz2 # this codec needs the optional bz2 module !
-
- ### Codec APIs
-
- def bz2_encode(input,errors='strict'):
-
- """ Encodes the object input and returns a tuple (output
- object, length consumed).
-
- errors defines the error handling to apply. It defaults to
- 'strict' handling which is the only currently supported
- error handling for this codec.
-
- """
- assert errors == 'strict'
- output = bz2.compress(input)
- return (output, len(input))
-
- def bz2_decode(input,errors='strict'):
-
- """ Decodes the object input and returns a tuple (output
- object, length consumed).
-
- input must be an object which provides the bf_getreadbuf
- buffer slot. Python strings, buffer objects and memory
- mapped files are examples of objects providing this slot.
-
- errors defines the error handling to apply. It defaults to
- 'strict' handling which is the only currently supported
- error handling for this codec.
-
- """
- assert errors == 'strict'
- output = bz2.decompress(input)
- return (output, len(input))
-
- class Codec(codecs.Codec):
-
- def encode(self, input, errors='strict'):
- return bz2_encode(input, errors)
- def decode(self, input, errors='strict'):
- return bz2_decode(input, errors)
-
- class IncrementalEncoder(codecs.IncrementalEncoder):
- def __init__(self, errors='strict'):
- assert errors == 'strict'
- self.errors = errors
- self.compressobj = bz2.BZ2Compressor()
-
- def encode(self, input, final=False):
- if final:
- c = self.compressobj.compress(input)
- return c + self.compressobj.flush()
- else:
- return self.compressobj.compress(input)
-
- def reset(self):
- self.compressobj = bz2.BZ2Compressor()
-
- class IncrementalDecoder(codecs.IncrementalDecoder):
- def __init__(self, errors='strict'):
- assert errors == 'strict'
- self.errors = errors
- self.decompressobj = bz2.BZ2Decompressor()
-
- def decode(self, input, final=False):
- try:
- return self.decompressobj.decompress(input)
- except EOFError:
- return ''
-
- def reset(self):
- self.decompressobj = bz2.BZ2Decompressor()
-
- class StreamWriter(Codec,codecs.StreamWriter):
- pass
-
- class StreamReader(Codec,codecs.StreamReader):
- pass
-
- ### encodings module API
-
- def getregentry():
- return codecs.CodecInfo(
- name="bz2",
- encode=bz2_encode,
- decode=bz2_decode,
- incrementalencoder=IncrementalEncoder,
- incrementaldecoder=IncrementalDecoder,
- streamwriter=StreamWriter,
- streamreader=StreamReader,
- )
-