zlib_codec.py (1887B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\stdlib\encodings\zlib_codec.py 2 import codecs 3 import zlib 4 5 def zlib_encode(input, errors = 'strict'): 6 output = zlib.compress(input) 7 return (output, len(input)) 8 9 10 def zlib_decode(input, errors = 'strict'): 11 output = zlib.decompress(input) 12 return (output, len(input)) 13 14 15 class Codec(codecs.Codec): 16 17 def encode(self, input, errors = 'strict'): 18 return zlib_encode(input, errors) 19 20 def decode(self, input, errors = 'strict'): 21 return zlib_decode(input, errors) 22 23 24 class IncrementalEncoder(codecs.IncrementalEncoder): 25 26 def __init__(self, errors = 'strict'): 27 self.errors = errors 28 self.compressobj = zlib.compressobj() 29 30 def encode(self, input, final = False): 31 if final: 32 c = self.compressobj.compress(input) 33 return c + self.compressobj.flush() 34 else: 35 return self.compressobj.compress(input) 36 37 def reset(self): 38 self.compressobj = zlib.compressobj() 39 40 41 class IncrementalDecoder(codecs.IncrementalDecoder): 42 43 def __init__(self, errors = 'strict'): 44 self.errors = errors 45 self.decompressobj = zlib.decompressobj() 46 47 def decode(self, input, final = False): 48 if final: 49 c = self.decompressobj.decompress(input) 50 return c + self.decompressobj.flush() 51 else: 52 return self.decompressobj.decompress(input) 53 54 def reset(self): 55 self.decompressobj = zlib.decompressobj() 56 57 58 class StreamWriter(Codec, codecs.StreamWriter): 59 pass 60 61 62 class StreamReader(Codec, codecs.StreamReader): 63 pass 64 65 66 def getregentry(): 67 return codecs.CodecInfo(name='zlib', encode=zlib_encode, decode=zlib_decode, incrementalencoder=IncrementalEncoder, incrementaldecoder=IncrementalDecoder, streamreader=StreamReader, streamwriter=StreamWriter)