1  """Python 'bz2_codec' Codec - bz2 compression encoding.
       2  
       3  This codec de/encodes from bytes to bytes and is therefore usable with
       4  bytes.transform() and bytes.untransform().
       5  
       6  Adapted by Raymond Hettinger from zlib_codec.py which was written
       7  by Marc-Andre Lemburg (mal@lemburg.com).
       8  """
       9  
      10  import codecs
      11  import bz2 # this codec needs the optional bz2 module !
      12  
      13  ### Codec APIs
      14  
      15  def bz2_encode(input, errors='strict'):
      16      assert errors == 'strict'
      17      return (bz2.compress(input), len(input))
      18  
      19  def bz2_decode(input, errors='strict'):
      20      assert errors == 'strict'
      21      return (bz2.decompress(input), len(input))
      22  
      23  class ESC[4;38;5;81mCodec(ESC[4;38;5;149mcodecsESC[4;38;5;149m.ESC[4;38;5;149mCodec):
      24      def encode(self, input, errors='strict'):
      25          return bz2_encode(input, errors)
      26      def decode(self, input, errors='strict'):
      27          return bz2_decode(input, errors)
      28  
      29  class ESC[4;38;5;81mIncrementalEncoder(ESC[4;38;5;149mcodecsESC[4;38;5;149m.ESC[4;38;5;149mIncrementalEncoder):
      30      def __init__(self, errors='strict'):
      31          assert errors == 'strict'
      32          self.errors = errors
      33          self.compressobj = bz2.BZ2Compressor()
      34  
      35      def encode(self, input, final=False):
      36          if final:
      37              c = self.compressobj.compress(input)
      38              return c + self.compressobj.flush()
      39          else:
      40              return self.compressobj.compress(input)
      41  
      42      def reset(self):
      43          self.compressobj = bz2.BZ2Compressor()
      44  
      45  class ESC[4;38;5;81mIncrementalDecoder(ESC[4;38;5;149mcodecsESC[4;38;5;149m.ESC[4;38;5;149mIncrementalDecoder):
      46      def __init__(self, errors='strict'):
      47          assert errors == 'strict'
      48          self.errors = errors
      49          self.decompressobj = bz2.BZ2Decompressor()
      50  
      51      def decode(self, input, final=False):
      52          try:
      53              return self.decompressobj.decompress(input)
      54          except EOFError:
      55              return ''
      56  
      57      def reset(self):
      58          self.decompressobj = bz2.BZ2Decompressor()
      59  
      60  class ESC[4;38;5;81mStreamWriter(ESC[4;38;5;149mCodec, ESC[4;38;5;149mcodecsESC[4;38;5;149m.ESC[4;38;5;149mStreamWriter):
      61      charbuffertype = bytes
      62  
      63  class ESC[4;38;5;81mStreamReader(ESC[4;38;5;149mCodec, ESC[4;38;5;149mcodecsESC[4;38;5;149m.ESC[4;38;5;149mStreamReader):
      64      charbuffertype = bytes
      65  
      66  ### encodings module API
      67  
      68  def getregentry():
      69      return codecs.CodecInfo(
      70          name="bz2",
      71          encode=bz2_encode,
      72          decode=bz2_decode,
      73          incrementalencoder=IncrementalEncoder,
      74          incrementaldecoder=IncrementalDecoder,
      75          streamwriter=StreamWriter,
      76          streamreader=StreamReader,
      77          _is_text_encoding=False,
      78      )