696 lines
26 KiB
Python
696 lines
26 KiB
Python
from stdint import *
|
|
import zlib.zdeflate as zdeflate
|
|
import zlib.zinflate as zinflate
|
|
import zlib.zchecksum as zchecksum
|
|
import zlib.zdef as zdef
|
|
import zlib.zhuff as zhuff
|
|
import stdlib
|
|
import string
|
|
import memhub
|
|
import t, c
|
|
|
|
|
|
# ============================================================
|
|
# Constants - matching Python zlib module names and values
|
|
# ============================================================
|
|
|
|
# Compression levels
|
|
Z_NO_COMPRESSION: t.CDefine = 0
|
|
Z_BEST_SPEED: t.CDefine = 1
|
|
Z_BEST_COMPRESSION: t.CDefine = 9
|
|
Z_DEFAULT_COMPRESSION: t.CDefine = (-1)
|
|
|
|
# Compression methods
|
|
DEFLATED: t.CDefine = 8
|
|
|
|
# Flush modes
|
|
Z_NO_FLUSH: t.CDefine = 0
|
|
Z_PARTIAL_FLUSH: t.CDefine = 1
|
|
Z_SYNC_FLUSH: t.CDefine = 2
|
|
Z_FULL_FLUSH: t.CDefine = 3
|
|
Z_FINISH: t.CDefine = 4
|
|
Z_BLOCK: t.CDefine = 5
|
|
Z_TREES: t.CDefine = 6
|
|
|
|
# Strategies
|
|
Z_DEFAULT_STRATEGY: t.CDefine = 0
|
|
Z_FILTERED: t.CDefine = 1
|
|
Z_HUFFMAN_ONLY: t.CDefine = 2
|
|
Z_RLE: t.CDefine = 3
|
|
Z_FIXED: t.CDefine = 4
|
|
|
|
# Return codes
|
|
Z_OK: t.CDefine = 0
|
|
Z_STREAM_END: t.CDefine = 1
|
|
Z_NEED_DICT: t.CDefine = 2
|
|
Z_ERRNO: t.CDefine = (-1)
|
|
Z_STREAM_ERROR: t.CDefine = (-2)
|
|
Z_DATA_ERROR: t.CDefine = (-3)
|
|
Z_MEM_ERROR: t.CDefine = (-4)
|
|
Z_BUF_ERROR: t.CDefine = (-5)
|
|
Z_VERSION_ERROR: t.CDefine = (-6)
|
|
|
|
# Window / Buffer constants
|
|
MAX_WBITS: t.CDefine = 15
|
|
DEF_BUF_SIZE: t.CDefine = 16384
|
|
DEF_MEM_LEVEL: t.CDefine = 8
|
|
|
|
# Version
|
|
ZLIB_VERSION: t.CDefine = "1.3.2"
|
|
|
|
pyzlib_error_msg: t.CArray[t.CChar, 512] = ""
|
|
pyzlib_error_code_val: t.CInt = 0
|
|
|
|
# ============================================================
|
|
# Structures
|
|
# ============================================================
|
|
@t.Object
|
|
class Compress:
|
|
pool: memhub.MemManager | t.CPtr
|
|
stream: VOIDPTR
|
|
is_initialized: t.CInt
|
|
is_finished: t.CInt
|
|
level: t.CInt
|
|
method: t.CInt
|
|
wbits: t.CInt
|
|
memLevel: t.CInt
|
|
strategy: t.CInt
|
|
zdict: BYTEPTR
|
|
zdict_len: t.CSizeT
|
|
last_pos: t.CSizeT
|
|
input_buf: BYTEPTR
|
|
input_buf_len: t.CSizeT
|
|
input_buf_cap: t.CSizeT
|
|
header_written: t.CInt
|
|
|
|
def compress(self,
|
|
data: BYTEPTR, data_len: t.CSizeT,
|
|
out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "compress object not initialized")
|
|
return None
|
|
if self.is_finished:
|
|
set_error(Z_STREAM_ERROR, "compress object already finished")
|
|
return None
|
|
if not out_len:
|
|
set_error(Z_STREAM_ERROR, "out_len is None")
|
|
return None
|
|
if data_len == 0:
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p, 0, 1)
|
|
return p
|
|
while self.input_buf_len + data_len > self.input_buf_cap:
|
|
self.input_buf_cap *= 2
|
|
if self.pool == None:
|
|
new_buf: BYTEPTR = BYTEPTR(realloc(self.input_buf, self.input_buf_cap))
|
|
else:
|
|
new_buf: BYTEPTR = BYTEPTR(self.pool.alloc(self.input_buf_cap))
|
|
if new_buf:
|
|
memcpy(new_buf, self.input_buf, self.input_buf_len)
|
|
self.pool.free(self.input_buf)
|
|
if not new_buf:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p2: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p2, 0, 1)
|
|
return p2
|
|
self.input_buf = new_buf
|
|
memcpy(self.input_buf + self.input_buf_len, data, data_len)
|
|
self.input_buf_len += data_len
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p3: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p3, 0, 1)
|
|
return p3
|
|
|
|
def flush(self, mode: t.CInt, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "compress object not initialized")
|
|
return None
|
|
if not out_len:
|
|
set_error(Z_STREAM_ERROR, "out_len is None")
|
|
return None
|
|
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
|
|
if mode == Z_FINISH:
|
|
if not self.header_written:
|
|
self.header_written = 1
|
|
input_data: BYTEPTR = self.input_buf
|
|
input_len: t.CSizeT = self.input_buf_len
|
|
if input_len == 0:
|
|
s.writer.write_bits(1, 1)
|
|
s.writer.write_bits(1, 2)
|
|
lit_tree = zhuff.zhuff_tree()
|
|
lit_tree.build_fixed_lit_tree()
|
|
lit_tree.encode_symbol(256, c.Addr(s.writer))
|
|
else:
|
|
s.adler = zchecksum.zchecksum_adler32(input_data, input_len, 1)
|
|
s.compress_block(input_data, input_len, 1)
|
|
s.is_finished = 1
|
|
if s.wbits >= 0:
|
|
s.writer.align()
|
|
adler: t.CUInt32T = s.adler
|
|
s.writer.buf[s.writer.byte_pos + 0] = (adler >> 24) & 0xFF
|
|
s.writer.buf[s.writer.byte_pos + 1] = (adler >> 16) & 0xFF
|
|
s.writer.buf[s.writer.byte_pos + 2] = (adler >> 8) & 0xFF
|
|
s.writer.buf[s.writer.byte_pos + 3] = (adler >> 0) & 0xFF
|
|
s.writer.byte_pos += 4
|
|
self.is_finished = 1
|
|
total: t.CSizeT = s.writer.total()
|
|
if self.pool == None:
|
|
result: BYTEPTR = BYTEPTR(malloc(total))
|
|
else:
|
|
result: BYTEPTR = BYTEPTR(self.pool.alloc(total))
|
|
if result: memcpy(result, s.writer.buf, total)
|
|
c.Set(c.Deref(out_len), total)
|
|
if self.pool == None:
|
|
free(self.input_buf)
|
|
else:
|
|
self.pool.free(self.input_buf)
|
|
self.input_buf = None
|
|
self.input_buf_len = 0
|
|
self.input_buf_cap = 0
|
|
return result
|
|
elif mode == Z_SYNC_FLUSH or mode == Z_FULL_FLUSH:
|
|
input_data: BYTEPTR = self.input_buf
|
|
input_len: t.CSizeT = self.input_buf_len
|
|
if input_len > 0:
|
|
s.adler = zchecksum.zchecksum_adler32(input_data, input_len, s.adler)
|
|
s.compress_block(input_data, input_len, 0)
|
|
self.input_buf_len = 0
|
|
s.writer.write_bits(0, 1)
|
|
s.writer.write_bits(0, 2)
|
|
s.writer.align()
|
|
prev_pos: t.CSizeT = self.last_pos
|
|
total: t.CSizeT = s.writer.total()
|
|
delta: t.CSizeT = total - prev_pos
|
|
self.last_pos = s.writer.byte_pos
|
|
if delta == 0:
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p4: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p4, 0, 1)
|
|
return p4
|
|
|
|
if self.pool == None:
|
|
result: BYTEPTR = BYTEPTR(malloc(delta))
|
|
else:
|
|
result: BYTEPTR = BYTEPTR(self.pool.alloc(delta))
|
|
if result: memcpy(result, s.writer.buf + prev_pos, delta)
|
|
c.Set(c.Deref(out_len), delta)
|
|
return result
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p5: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p5, 0, 1)
|
|
return p5
|
|
|
|
def copy(self) -> Compress | t.CPtr:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "compress object not initialized")
|
|
return None
|
|
if self.pool == None:
|
|
copy_obj: Compress | t.CPtr = calloc(1, Compress.__sizeof__())
|
|
else:
|
|
copy_obj: Compress | t.CPtr = self.pool.alloc(Compress.__sizeof__())
|
|
memset(copy_obj, 0, Compress.__sizeof__())
|
|
if not copy_obj:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
copy_obj.pool = self.pool
|
|
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
|
|
copy_obj.stream = s.copy()
|
|
if not copy_obj.stream:
|
|
if self.pool == None:
|
|
free(copy_obj)
|
|
else:
|
|
self.pool.free(copy_obj)
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
copy_obj.is_initialized = 1
|
|
copy_obj.is_finished = self.is_finished
|
|
copy_obj.level = self.level
|
|
copy_obj.method = self.method
|
|
copy_obj.wbits = self.wbits
|
|
copy_obj.memLevel = self.memLevel
|
|
copy_obj.strategy = self.strategy
|
|
copy_obj.last_pos = self.last_pos
|
|
copy_obj.header_written = self.header_written
|
|
if self.input_buf and self.input_buf_len > 0:
|
|
if self.pool == None:
|
|
copy_obj.input_buf = BYTEPTR(calloc(1, self.input_buf_cap))
|
|
else:
|
|
copy_obj.input_buf = BYTEPTR(self.pool.alloc(self.input_buf_cap))
|
|
memset(copy_obj.input_buf, 0, self.input_buf_cap)
|
|
if copy_obj.input_buf:
|
|
memcpy(copy_obj.input_buf, self.input_buf, self.input_buf_len)
|
|
copy_obj.input_buf_cap = self.input_buf_cap
|
|
copy_obj.input_buf_len = self.input_buf_len
|
|
else:
|
|
if self.pool == None:
|
|
copy_obj.input_buf = BYTEPTR(calloc(1, 256))
|
|
else:
|
|
copy_obj.input_buf = BYTEPTR(self.pool.alloc(256))
|
|
memset(copy_obj.input_buf, 0, 256)
|
|
copy_obj.input_buf_cap = 256
|
|
copy_obj.input_buf_len = 0
|
|
if self.zdict and self.zdict_len > 0:
|
|
copy_obj.zdict = clone_bytes(self.pool, self.zdict, self.zdict_len)
|
|
copy_obj.zdict_len = self.zdict_len
|
|
return copy_obj
|
|
|
|
def delete(self):
|
|
if self.is_initialized and self.stream:
|
|
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
|
|
s.destroy()
|
|
if self.pool == None:
|
|
free(self.zdict)
|
|
free(self.input_buf)
|
|
free(self)
|
|
else:
|
|
self.pool.free(self.zdict)
|
|
self.pool.free(self.input_buf)
|
|
self.pool.free(self)
|
|
|
|
def __del__(self):
|
|
self.delete()
|
|
|
|
|
|
@t.Object
|
|
class Decompress:
|
|
pool: memhub.MemManager | t.CPtr
|
|
stream: VOIDPTR
|
|
is_initialized: t.CInt
|
|
eof: t.CInt
|
|
wbits: t.CInt
|
|
zdict: BYTEPTR
|
|
zdict_len: t.CSizeT
|
|
_unused_data: BYTEPTR
|
|
_unused_data_len: t.CSizeT
|
|
_unused_data_cap: t.CSizeT
|
|
_unconsumed_tail: BYTEPTR
|
|
_unconsumed_tail_len: t.CSizeT
|
|
_unconsumed_tail_cap: t.CSizeT
|
|
input_buf: BYTEPTR
|
|
input_buf_len: t.CSizeT
|
|
input_buf_cap: t.CSizeT
|
|
|
|
def decompress(self, data: BYTEPTR, data_len: t.CSizeT,
|
|
max_length: t.CSizeT, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "decompress object not initialized")
|
|
return None
|
|
if not out_len:
|
|
set_error(Z_STREAM_ERROR, "out_len is None")
|
|
return None
|
|
if self.eof:
|
|
if data and data_len > 0:
|
|
append_bytes(self.pool, c.Addr(self._unused_data), c.Addr(self._unused_data_len),
|
|
c.Addr(self._unused_data_cap), data, data_len)
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p6: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p6, 0, 1)
|
|
return p6
|
|
self._unconsumed_tail_len = 0
|
|
if data and data_len > 0:
|
|
append_bytes(self.pool, c.Addr(self.input_buf), c.Addr(self.input_buf_len),
|
|
c.Addr(self.input_buf_cap), data, data_len)
|
|
if self.input_buf_len == 0:
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p7: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p7, 0, 1)
|
|
return p7
|
|
s: zinflate.zinflate_stream | t.CPtr = zinflate.zinflate_create(self.pool, self.wbits)
|
|
if not s:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
if self.zdict and self.zdict_len > 0:
|
|
s.set_dictionary(self.zdict, self.zdict_len)
|
|
out: t.CUInt8T | t.CPtr = None
|
|
err: t.CInt = s.decompress(self.input_buf, self.input_buf_len,
|
|
max_length, c.Addr(out), out_len)
|
|
if err != 0:
|
|
s.destroy()
|
|
set_error(Z_DATA_ERROR, "decompression failed")
|
|
return None
|
|
if s.is_finished:
|
|
self.eof = 1
|
|
if s.input_pos < s.input_len:
|
|
append_bytes(self.pool, c.Addr(self._unused_data), c.Addr(self._unused_data_len),
|
|
c.Addr(self._unused_data_cap),
|
|
s.input_data + s.input_pos,
|
|
s.input_len - s.input_pos)
|
|
self.input_buf_len = 0
|
|
else:
|
|
consumed: t.CSizeT = s.input_pos
|
|
if consumed < self.input_buf_len:
|
|
remaining: t.CSizeT = self.input_buf_len - consumed
|
|
memmove(self.input_buf, self.input_buf + consumed, remaining)
|
|
self.input_buf_len = remaining
|
|
else:
|
|
self.input_buf_len = 0
|
|
s.destroy()
|
|
return out
|
|
|
|
def flush(self, length: t.CSizeT, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "decompress object not initialized")
|
|
return None
|
|
if not out_len:
|
|
set_error(Z_STREAM_ERROR, "out_len is None")
|
|
return None
|
|
c.Set(c.Deref(out_len), 0)
|
|
if self.pool == None:
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
p8: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
|
memset(p8, 0, 1)
|
|
return p8
|
|
|
|
def copy(self) -> Decompress | t.CPtr:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "decompress object not initialized")
|
|
return None
|
|
if self.pool == None:
|
|
copy_obj: Decompress | t.CPtr = calloc(1, Decompress.__sizeof__())
|
|
else:
|
|
copy_obj: Decompress | t.CPtr = self.pool.alloc(Decompress.__sizeof__())
|
|
memset(copy_obj, 0, Decompress.__sizeof__())
|
|
if not copy_obj:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
copy_obj.pool = self.pool
|
|
copy_obj.is_initialized = 1
|
|
copy_obj.eof = self.eof
|
|
copy_obj.wbits = self.wbits
|
|
if self.zdict and self.zdict_len > 0:
|
|
copy_obj.zdict = clone_bytes(self.pool, self.zdict, self.zdict_len)
|
|
copy_obj.zdict_len = self.zdict_len
|
|
if self._unused_data and self._unused_data_len > 0:
|
|
copy_obj._unused_data = clone_bytes(self.pool, self._unused_data, self._unused_data_len)
|
|
copy_obj._unused_data_len = self._unused_data_len
|
|
copy_obj._unused_data_cap = self._unused_data_len
|
|
if self._unconsumed_tail and self._unconsumed_tail_len > 0:
|
|
copy_obj._unconsumed_tail = clone_bytes(self.pool, self._unconsumed_tail, self._unconsumed_tail_len)
|
|
copy_obj._unconsumed_tail_len = self._unconsumed_tail_len
|
|
copy_obj._unconsumed_tail_cap = self._unconsumed_tail_len
|
|
if self.input_buf and self.input_buf_len > 0:
|
|
copy_obj.input_buf = clone_bytes(self.pool, self.input_buf, self.input_buf_len)
|
|
copy_obj.input_buf_len = self.input_buf_len
|
|
copy_obj.input_buf_cap = self.input_buf_cap
|
|
else:
|
|
if self.pool == None:
|
|
copy_obj.input_buf = BYTEPTR(calloc(1, 256))
|
|
else:
|
|
copy_obj.input_buf = BYTEPTR(self.pool.alloc(256))
|
|
memset(copy_obj.input_buf, 0, 256)
|
|
copy_obj.input_buf_cap = 256
|
|
copy_obj.input_buf_len = 0
|
|
return copy_obj
|
|
|
|
def delete(self):
|
|
if self.pool == None:
|
|
free(self.zdict)
|
|
free(self._unused_data)
|
|
free(self._unconsumed_tail)
|
|
free(self.input_buf)
|
|
free(self)
|
|
else:
|
|
self.pool.free(self.zdict)
|
|
self.pool.free(self._unused_data)
|
|
self.pool.free(self._unconsumed_tail)
|
|
self.pool.free(self.input_buf)
|
|
self.pool.free(self)
|
|
|
|
def unused_data(self, length: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
#if not self:
|
|
# if length:
|
|
# c.Set(c.Deref(length), 0)
|
|
# return None
|
|
if length:
|
|
c.Set(c.Deref(length), self._unused_data_len)
|
|
return self._unused_data
|
|
|
|
def unconsumed_tail(self, length: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
#if not self:
|
|
# if length:
|
|
# c.Set(c.Deref(length), 0)
|
|
# return None
|
|
if length:
|
|
c.Set(c.Deref(length), self._unconsumed_tail_len)
|
|
return self._unconsumed_tail
|
|
|
|
# def eof(self) -> t.CInt:
|
|
# # if not obj: return 0
|
|
# return self.eof
|
|
|
|
|
|
|
|
def set_error(code: int, msg: str):
|
|
global pyzlib_error_code_val, pyzlib_error_msg
|
|
pyzlib_error_code_val = code
|
|
if msg:
|
|
strncpy(pyzlib_error_msg, msg, pyzlib_error_msg.__sizeof__() - 1)
|
|
pyzlib_error_msg[pyzlib_error_msg.__sizeof__() - 1] = '\0'
|
|
else:
|
|
pyzlib_error_msg[0] = '\0'
|
|
|
|
def clone_bytes(pool: memhub.MemManager | t.CPtr, src: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
|
|
if not src or length == 0: return None
|
|
if pool == None:
|
|
dst: BYTEPTR = BYTEPTR(malloc(length))
|
|
else:
|
|
dst: BYTEPTR = BYTEPTR(pool.alloc(length))
|
|
if dst: memcpy(dst, src, length)
|
|
return dst
|
|
|
|
def append_bytes(pool: memhub.MemManager | t.CPtr, buf: BYTE | t.CPtr[t.CPtr], length: t.CSizeT | t.CPtr, cap: t.CSizeT | t.CPtr,
|
|
data: BYTEPTR, data_len: t.CSizeT) -> t.CInt:
|
|
if not data or data_len == 0: return 0
|
|
needed: t.CSizeT = c.Deref(length) + data_len
|
|
if needed > c.Deref(cap):
|
|
new_cap: t.CSizeT = c.Deref(cap) * 2
|
|
if new_cap < needed: new_cap = needed
|
|
if pool == None:
|
|
new_buf: BYTEPTR = BYTEPTR(realloc(c.Deref(buf), new_cap))
|
|
else:
|
|
new_buf: BYTEPTR = BYTEPTR(pool.alloc(new_cap))
|
|
if new_buf:
|
|
memcpy(new_buf, c.Deref(buf), c.Deref(length))
|
|
pool.free(c.Deref(buf))
|
|
if not new_buf: return -1
|
|
c.Set(c.Deref(buf), new_buf)
|
|
c.Set(c.Deref(cap), new_cap)
|
|
memcpy(c.Deref(buf) + c.Deref(length), data, data_len)
|
|
c.Set(c.Deref(length), c.Deref(length) + data_len)
|
|
return 0
|
|
|
|
|
|
def shrink_to_fit(pool: memhub.MemManager | t.CPtr, buf: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
|
|
if length == 0:
|
|
if pool == None:
|
|
free(buf)
|
|
return BYTEPTR(calloc(1, 1))
|
|
else:
|
|
pool.free(buf)
|
|
p: BYTEPTR = BYTEPTR(pool.alloc(1))
|
|
memset(p, 0, 1)
|
|
return p
|
|
if pool == None:
|
|
result: BYTEPTR = BYTEPTR(realloc(buf, length))
|
|
return result if result else buf
|
|
else:
|
|
result: BYTEPTR = BYTEPTR(pool.alloc(length))
|
|
if result:
|
|
memcpy(result, buf, length)
|
|
pool.free(buf)
|
|
return result
|
|
return buf
|
|
|
|
|
|
def runtime_version() -> str:
|
|
return ZLIB_VERSION
|
|
|
|
def get_error() -> str:
|
|
return pyzlib_error_msg
|
|
|
|
def get_error_code() -> t.CInt:
|
|
return pyzlib_error_code_val
|
|
|
|
def clear_error():
|
|
global pyzlib_error_msg, pyzlib_error_code_val
|
|
pyzlib_error_msg[0] = '\0'
|
|
pyzlib_error_code_val = 0
|
|
|
|
def compress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSizeT,
|
|
level: t.CInt, wbits: t.CInt, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
if not data and data_len > 0:
|
|
set_error(Z_STREAM_ERROR, "data is None but data_len > 0")
|
|
return None
|
|
if not out_len:
|
|
set_error(Z_STREAM_ERROR, "out_len is None")
|
|
return None
|
|
actual_wbits: t.CInt = wbits
|
|
if wbits > MAX_WBITS:
|
|
actual_wbits = -MAX_WBITS
|
|
raw_result: UINT8PTR = zdeflate.zdeflate_one_shot(pool, data, data_len, level, actual_wbits, out_len)
|
|
if not raw_result:
|
|
set_error(Z_DATA_ERROR, "compression failed")
|
|
return None
|
|
if wbits > MAX_WBITS:
|
|
gzip_len: t.CSizeT = 10 + c.Deref(out_len) + 8
|
|
if pool == None:
|
|
gzip_buf: BYTEPTR = BYTEPTR(malloc(gzip_len))
|
|
else:
|
|
gzip_buf: BYTEPTR = BYTEPTR(pool.alloc(gzip_len))
|
|
if not gzip_buf:
|
|
if pool == None:
|
|
free(raw_result)
|
|
else:
|
|
pool.free(raw_result)
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
pos: t.CSizeT = 0
|
|
gzip_buf[pos + 0] = 0x1F
|
|
gzip_buf[pos + 1] = 0x8B
|
|
gzip_buf[pos + 2] = 0x08
|
|
gzip_buf[pos + 3] = 0x00
|
|
gzip_buf[pos + 4] = 0x00
|
|
gzip_buf[pos + 5] = 0x00
|
|
gzip_buf[pos + 6] = 0x00
|
|
gzip_buf[pos + 7] = 0x00
|
|
gzip_buf[pos + 8] = 0x00
|
|
gzip_buf[pos + 9] = 0xFF
|
|
pos += 10
|
|
|
|
memcpy(gzip_buf + pos, raw_result, c.Deref(out_len))
|
|
pos += c.Deref(out_len)
|
|
if pool == None:
|
|
free(raw_result)
|
|
else:
|
|
pool.free(raw_result)
|
|
|
|
crc: t.CUInt32T = zchecksum.zchecksum_crc32(data, data_len, 0)
|
|
gzip_buf[pos + 0] = (crc) & 0xFF
|
|
gzip_buf[pos + 1] = (crc >> 8) & 0xFF
|
|
gzip_buf[pos + 2] = (crc >> 16) & 0xFF
|
|
gzip_buf[pos + 3] = (crc >> 24) & 0xFF
|
|
gzip_buf[pos + 4] = (data_len) & 0xFF
|
|
gzip_buf[pos + 5] = (data_len >> 8) & 0xFF
|
|
gzip_buf[pos + 6] = (data_len >> 16) & 0xFF
|
|
gzip_buf[pos + 7] = (data_len >> 24) & 0xFF
|
|
pos += 8
|
|
c.Set(c.Deref(out_len), pos)
|
|
return shrink_to_fit(pool, gzip_buf, pos)
|
|
return shrink_to_fit(pool, raw_result, c.Deref(out_len))
|
|
|
|
|
|
def decompress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSizeT,
|
|
wbits: t.CInt, bufsize: t.CSizeT, out_len: t.CSizeT | t.CPtr) -> BYTEPTR:
|
|
if not data and data_len > 0:
|
|
set_error(Z_STREAM_ERROR, "data is None but data_len > 0")
|
|
return None
|
|
if not out_len:
|
|
set_error(Z_STREAM_ERROR, "out_len is None")
|
|
return None
|
|
result: UINT8PTR = zinflate.zinflate_one_shot(pool, data, data_len, wbits, bufsize, out_len)
|
|
if not result:
|
|
set_error(Z_DATA_ERROR, "decompression failed")
|
|
return None
|
|
return shrink_to_fit(pool, result, c.Deref(out_len))
|
|
|
|
def compressobj(pool: memhub.MemManager | t.CPtr, level: t.CInt, method: t.CInt, wbits: t.CInt,
|
|
memLevel: t.CInt, strategy: t.CInt,
|
|
zdict: BYTEPTR, zdict_len: t.CSizeT) -> Compress | t.CPtr:
|
|
if pool == None:
|
|
obj: Compress | t.CPtr = calloc(1, Compress.__sizeof__())
|
|
else:
|
|
obj: Compress | t.CPtr = pool.alloc(Compress.__sizeof__())
|
|
memset(obj, 0, Compress.__sizeof__())
|
|
if not obj:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
obj.pool = pool
|
|
actual_wbits: t.CInt = wbits
|
|
if wbits > MAX_WBITS: actual_wbits = wbits - 16
|
|
s: zdeflate.zdeflate_stream | t.CPtr = zdeflate.zdeflate_create(pool, level, actual_wbits, memLevel, strategy)
|
|
if not s:
|
|
if pool == None:
|
|
free(obj)
|
|
else:
|
|
pool.free(obj)
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
obj.stream = s
|
|
obj.is_initialized = 1
|
|
obj.level = level
|
|
obj.method = method
|
|
obj.wbits = wbits
|
|
obj.memLevel = memLevel
|
|
obj.strategy = strategy
|
|
if pool == None:
|
|
obj.input_buf = BYTEPTR(calloc(1, 256))
|
|
else:
|
|
obj.input_buf = BYTEPTR(pool.alloc(256))
|
|
memset(obj.input_buf, 0, 256)
|
|
obj.input_buf_cap = 256
|
|
obj.input_buf_len = 0
|
|
obj.header_written = 0
|
|
if zdict and zdict_len > 0:
|
|
s.set_dictionary(zdict, zdict_len)
|
|
obj.zdict = clone_bytes(pool, zdict, zdict_len)
|
|
obj.zdict_len = zdict_len
|
|
return obj
|
|
|
|
|
|
def decompressobj(pool: memhub.MemManager | t.CPtr, wbits: t.CInt, zdict: BYTEPTR, zdict_len: t.CSizeT) -> Decompress | t.CPtr:
|
|
if pool == None:
|
|
obj: Decompress | t.CPtr = calloc(1, Decompress.__sizeof__())
|
|
else:
|
|
obj: Decompress | t.CPtr = pool.alloc(Decompress.__sizeof__())
|
|
memset(obj, 0, Decompress.__sizeof__())
|
|
if not obj:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
obj.pool = pool
|
|
actual_wbits: t.CInt = wbits
|
|
if wbits > MAX_WBITS: actual_wbits = wbits - 16
|
|
obj.stream = None
|
|
obj.is_initialized = 1
|
|
obj.wbits = actual_wbits
|
|
if zdict and zdict_len > 0:
|
|
obj.zdict = clone_bytes(pool, zdict, zdict_len)
|
|
obj.zdict_len = zdict_len
|
|
if pool == None:
|
|
obj.input_buf = BYTEPTR(calloc(1, 256))
|
|
else:
|
|
obj.input_buf = BYTEPTR(pool.alloc(256))
|
|
memset(obj.input_buf, 0, 256)
|
|
obj.input_buf_cap = 256
|
|
obj.input_buf_len = 0
|
|
return obj
|
|
|
|
def zlib_adler32(data: BYTEPTR, length: t.CSizeT, value: t.CUnsignedLong) -> t.CUnsignedLong:
|
|
return zchecksum.zchecksum_adler32(data, length, UINT32(value))
|
|
|
|
def zlib_crc32(data: BYTEPTR, length: t.CSizeT, value: t.CUnsignedLong) -> t.CUnsignedLong:
|
|
return zchecksum.zchecksum_crc32(data, length, UINT32(value))
|