Mercurial > repos > rliterman > csp2

comparison CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/encodings/bz2_codec.py @ 69:33d812a61356

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit 2e9511a184a1ca667c7be0c6321a36dc4e3d116d
author jpayne
date Tue, 18 Mar 2025 17:55:14 -0400
parents
children
comparison
equal deleted inserted replaced
67:0e9998148a16 69:33d812a61356
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 Codec(codecs.Codec):
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 IncrementalEncoder(codecs.IncrementalEncoder):
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 IncrementalDecoder(codecs.IncrementalDecoder):
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 StreamWriter(Codec, codecs.StreamWriter):
61 charbuffertype = bytes
62
63 class StreamReader(Codec, codecs.StreamReader):
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 )