693 lines
25 KiB
Plaintext
693 lines
25 KiB
Plaintext
# from string import memset, memcpy
|
|
# from stdio import malloc, free, calloc, realloc
|
|
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
|
|
|
|
|
|
# ============================================================
|
|
# Constants - matching Python zlib module names and values
|
|
# ============================================================
|
|
|
|
# Compression levels
|
|
define Z_NO_COMPRESSION = 0
|
|
define Z_BEST_SPEED = 1
|
|
define Z_BEST_COMPRESSION = 9
|
|
define Z_DEFAULT_COMPRESSION = (-1)
|
|
|
|
# Compression methods
|
|
define DEFLATED = 8
|
|
|
|
# Flush modes
|
|
define Z_NO_FLUSH = 0
|
|
define Z_PARTIAL_FLUSH = 1
|
|
define Z_SYNC_FLUSH = 2
|
|
define Z_FULL_FLUSH = 3
|
|
define Z_FINISH = 4
|
|
define Z_BLOCK = 5
|
|
define Z_TREES = 6
|
|
|
|
# Strategies
|
|
define Z_DEFAULT_STRATEGY = 0
|
|
define Z_FILTERED = 1
|
|
define Z_HUFFMAN_ONLY = 2
|
|
define Z_RLE = 3
|
|
define Z_FIXED = 4
|
|
|
|
# Return codes
|
|
define Z_OK = 0
|
|
define Z_STREAM_END = 1
|
|
define Z_NEED_DICT = 2
|
|
define Z_ERRNO = (-1)
|
|
define Z_STREAM_ERROR = (-2)
|
|
define Z_DATA_ERROR = (-3)
|
|
define Z_MEM_ERROR = (-4)
|
|
define Z_BUF_ERROR = (-5)
|
|
define Z_VERSION_ERROR = (-6)
|
|
|
|
# Window / Buffer constants
|
|
define MAX_WBITS = 15
|
|
define DEF_BUF_SIZE = 16384
|
|
define DEF_MEM_LEVEL = 8
|
|
|
|
# Version
|
|
define ZLIB_VERSION = "1.3.2"
|
|
|
|
pyzlib_error_msg: [512]char = ""
|
|
pyzlib_error_code_val: int = 0
|
|
|
|
# ============================================================
|
|
# Structures
|
|
# ============================================================
|
|
class Compress:
|
|
pool: memhub.MemManager*
|
|
stream: rawptr
|
|
is_initialized: int
|
|
is_finished: int
|
|
level: int
|
|
method: int
|
|
wbits: int
|
|
memLevel: int
|
|
strategy: int
|
|
zdict: bytes
|
|
zdict_len: sizet
|
|
last_pos: sizet
|
|
input_buf: bytes
|
|
input_buf_len: sizet
|
|
input_buf_cap: sizet
|
|
header_written: int
|
|
|
|
def compress(self, data: bytes, data_len: sizet, out_len: sizet*) -> bytes:
|
|
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:
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p: bytes = bytes(self.pool.alloc(1))
|
|
string.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: bytes = bytes(stdlib.realloc(self.input_buf, self.input_buf_cap))
|
|
else:
|
|
new_buf: bytes = bytes(self.pool.alloc(self.input_buf_cap))
|
|
if new_buf:
|
|
string.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")
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p2: bytes = bytes(self.pool.alloc(1))
|
|
string.memset(p2, 0, 1)
|
|
return p2
|
|
self.input_buf = new_buf
|
|
string.memcpy(self.input_buf + self.input_buf_len, data, data_len)
|
|
self.input_buf_len += data_len
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p3: bytes = bytes(self.pool.alloc(1))
|
|
string.memset(p3, 0, 1)
|
|
return p3
|
|
|
|
def flush(self, mode: int, out_len: sizet*) -> bytes:
|
|
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* = self.stream # implicit type conversion
|
|
if mode == Z_FINISH:
|
|
if not self.header_written:
|
|
self.header_written = 1
|
|
input_data: bytes = self.input_buf
|
|
input_len: sizet = 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, &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: uint32 = s.adler
|
|
s.writer.buf[s.writer.byte_pos + 0] = uint8((adler >> 24) & 0xFF)
|
|
s.writer.buf[s.writer.byte_pos + 1] = uint8((adler >> 16) & 0xFF)
|
|
s.writer.buf[s.writer.byte_pos + 2] = uint8((adler >> 8) & 0xFF)
|
|
s.writer.buf[s.writer.byte_pos + 3] = uint8((adler >> 0) & 0xFF)
|
|
s.writer.byte_pos += 4
|
|
self.is_finished = 1
|
|
total: sizet = s.writer.total()
|
|
if self.pool == None:
|
|
result: bytes = bytes(stdlib.malloc(total))
|
|
else:
|
|
result: bytes = bytes(self.pool.alloc(total))
|
|
if result: string.memcpy(result, s.writer.buf, total)
|
|
*out_len = total
|
|
if self.pool == None:
|
|
stdlib.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: bytes = self.input_buf
|
|
input_len: sizet = 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: sizet = self.last_pos
|
|
total: sizet = s.writer.total()
|
|
delta: sizet = total - prev_pos
|
|
self.last_pos = s.writer.byte_pos
|
|
if delta == 0:
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p4: bytes = bytes(self.pool.alloc(1))
|
|
string.memset(p4, 0, 1)
|
|
return p4
|
|
|
|
if self.pool == None:
|
|
result: bytes = bytes(stdlib.malloc(delta))
|
|
else:
|
|
result: bytes = bytes(self.pool.alloc(delta))
|
|
if result: string.memcpy(result, s.writer.buf + prev_pos, delta)
|
|
*out_len = delta
|
|
return result
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p5: bytes = bytes(self.pool.alloc(1))
|
|
string.memset(p5, 0, 1)
|
|
return p5
|
|
|
|
def copy(self) -> Compress*:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "compress object not initialized")
|
|
return None
|
|
if self.pool == None:
|
|
copy_obj: Compress* = stdlib.calloc(1, Compress.__sizeof__())
|
|
else:
|
|
copy_obj: Compress* = self.pool.alloc(Compress.__sizeof__())
|
|
string.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* = self.stream # implicit type conversion
|
|
copy_obj.stream = s.copy()
|
|
if not copy_obj.stream:
|
|
if self.pool == None:
|
|
stdlib.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 = bytes(stdlib.calloc(1, self.input_buf_cap))
|
|
else:
|
|
copy_obj.input_buf = bytes(self.pool.alloc(self.input_buf_cap))
|
|
string.memset(copy_obj.input_buf, 0, self.input_buf_cap)
|
|
if copy_obj.input_buf:
|
|
string.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 = bytes(stdlib.calloc(1, 256))
|
|
else:
|
|
copy_obj.input_buf = bytes(self.pool.alloc(256))
|
|
string.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* = self.stream # implicit type conversion
|
|
s.destroy()
|
|
if self.pool == None:
|
|
stdlib.free(self.zdict)
|
|
stdlib.free(self.input_buf)
|
|
stdlib.free(self)
|
|
else:
|
|
self.pool.free(self.zdict)
|
|
self.pool.free(self.input_buf)
|
|
self.pool.free(self)
|
|
|
|
def __del__(self):
|
|
self.delete()
|
|
|
|
|
|
class Decompress:
|
|
pool: memhub.MemManager*
|
|
stream: rawptr
|
|
is_initialized: int
|
|
eof: int
|
|
wbits: int
|
|
zdict: bytes
|
|
zdict_len: sizet
|
|
_unused_data: bytes
|
|
_unused_data_len: sizet
|
|
_unused_data_cap: sizet
|
|
_unconsumed_tail: bytes
|
|
_unconsumed_tail_len: sizet
|
|
_unconsumed_tail_cap: sizet
|
|
input_buf: bytes
|
|
input_buf_len: sizet
|
|
input_buf_cap: sizet
|
|
|
|
def decompress(self, data: bytes, data_len: sizet,
|
|
max_length: sizet, out_len: sizet*) -> bytes:
|
|
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, &self._unused_data, &self._unused_data_len,
|
|
&self._unused_data_cap, data, data_len)
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p6: bytes = bytes(self.pool.alloc(1))
|
|
string.memset(p6, 0, 1)
|
|
return p6
|
|
self._unconsumed_tail_len = 0
|
|
if data and data_len > 0:
|
|
append_bytes(self.pool, &self.input_buf, &self.input_buf_len,
|
|
&self.input_buf_cap, data, data_len)
|
|
if self.input_buf_len == 0:
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p7: bytes = bytes(self.pool.alloc(1))
|
|
string.memset(p7, 0, 1)
|
|
return p7
|
|
s: zinflate.zinflate_stream* = 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: uint8* = None
|
|
err: int = s.decompress(self.input_buf, self.input_buf_len,
|
|
max_length, &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, &self._unused_data, &self._unused_data_len,
|
|
&self._unused_data_cap,
|
|
s.input_data + s.input_pos,
|
|
s.input_len - s.input_pos)
|
|
self.input_buf_len = 0
|
|
else:
|
|
consumed: sizet = s.input_pos
|
|
if consumed < self.input_buf_len:
|
|
remaining: sizet = self.input_buf_len - consumed
|
|
string.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: sizet, out_len: sizet*) -> bytes:
|
|
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
|
|
*out_len = 0
|
|
if self.pool == None:
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
p8: bytes = bytes(self.pool.alloc(1))
|
|
string.memset(p8, 0, 1)
|
|
return p8
|
|
|
|
def copy(self) -> Decompress*:
|
|
if not self.is_initialized:
|
|
set_error(Z_STREAM_ERROR, "decompress object not initialized")
|
|
return None
|
|
if self.pool == None:
|
|
copy_obj: Decompress* = stdlib.calloc(1, Decompress.__sizeof__())
|
|
else:
|
|
copy_obj: Decompress* = self.pool.alloc(Decompress.__sizeof__())
|
|
string.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 = bytes(stdlib.calloc(1, 256))
|
|
else:
|
|
copy_obj.input_buf = bytes(self.pool.alloc(256))
|
|
string.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:
|
|
stdlib.free(self.zdict)
|
|
stdlib.free(self._unused_data)
|
|
stdlib.free(self._unconsumed_tail)
|
|
stdlib.free(self.input_buf)
|
|
stdlib.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: sizet*) -> bytes:
|
|
#if not self:
|
|
# if length:
|
|
# c.Set(c.Deref(length), 0)
|
|
# return None
|
|
if length:
|
|
*length = self._unused_data_len
|
|
return self._unused_data
|
|
|
|
def unconsumed_tail(self, length: sizet*) -> bytes:
|
|
#if not self:
|
|
# if length:
|
|
# c.Set(c.Deref(length), 0)
|
|
# return None
|
|
if length:
|
|
*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:
|
|
string.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*, src: bytes, length: sizet) -> bytes:
|
|
if not src or length == 0: return None
|
|
if pool == None:
|
|
dst: bytes = bytes(stdlib.malloc(length))
|
|
else:
|
|
dst: bytes = bytes(pool.alloc(length))
|
|
if dst: string.memcpy(dst, src, length)
|
|
return dst
|
|
|
|
# buf is a "pointer to pointer" (i8**). Do NOT write `BYTE | t.CPtr[t.CPtr]`: with an embedded `t.CPtr[...]` subscript the declare side (StubGen/VpiReader) does not recognize it as a CPtr marker and collapses it to BYTE = i8, mismatching the definition side's i8** (cross-module ABI mismatch; see includes/zlib/zinflate.vp decompress). Use "pointer alias | t.CPtr" instead.
|
|
def append_bytes(pool: memhub.MemManager*, buf: bytes*, length: sizet*, cap: sizet*,
|
|
data: bytes, data_len: sizet) -> int:
|
|
if not data or data_len == 0: return 0
|
|
needed: sizet = *length + data_len
|
|
if needed > *cap:
|
|
new_cap: sizet = *cap * 2
|
|
if new_cap < needed: new_cap = needed
|
|
if pool == None:
|
|
new_buf: bytes = bytes(stdlib.realloc(*buf, new_cap))
|
|
else:
|
|
new_buf: bytes = bytes(pool.alloc(new_cap))
|
|
if new_buf:
|
|
string.memcpy(new_buf, *buf, *length)
|
|
pool.free(*buf)
|
|
if not new_buf: return -1
|
|
*buf = new_buf
|
|
*cap = new_cap
|
|
string.memcpy(*buf + *length, data, data_len)
|
|
*length = *length + data_len
|
|
return 0
|
|
|
|
|
|
def shrink_to_fit(pool: memhub.MemManager*, buf: bytes, length: sizet) -> bytes:
|
|
if length == 0:
|
|
if pool == None:
|
|
stdlib.free(buf)
|
|
return bytes(stdlib.calloc(1, 1))
|
|
else:
|
|
pool.free(buf)
|
|
p: bytes = bytes(pool.alloc(1))
|
|
string.memset(p, 0, 1)
|
|
return p
|
|
if pool == None:
|
|
result: bytes = bytes(stdlib.realloc(buf, length))
|
|
return result if result else buf
|
|
else:
|
|
result: bytes = bytes(pool.alloc(length))
|
|
if result:
|
|
string.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() -> int:
|
|
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*, data: bytes, data_len: sizet,
|
|
level: int, wbits: int, out_len: sizet*) -> bytes:
|
|
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: int = 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: sizet = 10 + *out_len + 8
|
|
if pool == None:
|
|
gzip_buf: bytes = bytes(stdlib.malloc(gzip_len))
|
|
else:
|
|
gzip_buf: bytes = bytes(pool.alloc(gzip_len))
|
|
if not gzip_buf:
|
|
if pool == None:
|
|
stdlib.free(raw_result)
|
|
else:
|
|
pool.free(raw_result)
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
pos: sizet = 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
|
|
|
|
string.memcpy(gzip_buf + pos, raw_result, *out_len)
|
|
pos += *out_len
|
|
if pool == None:
|
|
stdlib.free(raw_result)
|
|
else:
|
|
pool.free(raw_result)
|
|
|
|
crc: uint32 = zchecksum.zchecksum_crc32(data, data_len, 0)
|
|
gzip_buf[pos + 0] = uint8((crc) & 0xFF)
|
|
gzip_buf[pos + 1] = uint8((crc >> 8) & 0xFF)
|
|
gzip_buf[pos + 2] = uint8((crc >> 16) & 0xFF)
|
|
gzip_buf[pos + 3] = uint8((crc >> 24) & 0xFF)
|
|
gzip_buf[pos + 4] = uint8((data_len) & 0xFF)
|
|
gzip_buf[pos + 5] = uint8((data_len >> 8) & 0xFF)
|
|
gzip_buf[pos + 6] = uint8((data_len >> 16) & 0xFF)
|
|
gzip_buf[pos + 7] = uint8((data_len >> 24) & 0xFF)
|
|
pos += 8
|
|
*out_len = pos
|
|
return shrink_to_fit(pool, gzip_buf, pos)
|
|
return shrink_to_fit(pool, raw_result, *out_len)
|
|
|
|
|
|
def decompress(pool: memhub.MemManager*, data: bytes, data_len: sizet,
|
|
wbits: int, bufsize: sizet, out_len: sizet*) -> bytes:
|
|
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, *out_len)
|
|
|
|
def compressobj(pool: memhub.MemManager*, level: int, method: int, wbits: int,
|
|
memLevel: int, strategy: int,
|
|
zdict: bytes, zdict_len: sizet) -> Compress*:
|
|
if pool == None:
|
|
obj: Compress* = stdlib.calloc(1, Compress.__sizeof__())
|
|
else:
|
|
obj: Compress* = pool.alloc(Compress.__sizeof__())
|
|
string.memset(obj, 0, Compress.__sizeof__())
|
|
if not obj:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
obj.pool = pool
|
|
actual_wbits: int = wbits
|
|
if wbits > MAX_WBITS: actual_wbits = wbits - 16
|
|
s: zdeflate.zdeflate_stream* = zdeflate.zdeflate_create(pool, level, actual_wbits, memLevel, strategy)
|
|
if not s:
|
|
if pool == None:
|
|
stdlib.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 = bytes(stdlib.calloc(1, 256))
|
|
else:
|
|
obj.input_buf = bytes(pool.alloc(256))
|
|
string.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*, wbits: int, zdict: bytes, zdict_len: sizet) -> Decompress*:
|
|
if pool == None:
|
|
obj: Decompress* = stdlib.calloc(1, Decompress.__sizeof__())
|
|
else:
|
|
obj: Decompress* = pool.alloc(Decompress.__sizeof__())
|
|
string.memset(obj, 0, Decompress.__sizeof__())
|
|
if not obj:
|
|
set_error(Z_MEM_ERROR, "out of memory")
|
|
return None
|
|
obj.pool = pool
|
|
actual_wbits: int = 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 = bytes(stdlib.calloc(1, 256))
|
|
else:
|
|
obj.input_buf = bytes(pool.alloc(256))
|
|
string.memset(obj.input_buf, 0, 256)
|
|
obj.input_buf_cap = 256
|
|
obj.input_buf_len = 0
|
|
return obj
|
|
|
|
def zlib_adler32(data: bytes, length: sizet, value: uint32) -> uint32:
|
|
return zchecksum.zchecksum_adler32(data, length, UINT32(value))
|
|
|
|
def zlib_crc32(data: bytes, length: sizet, value: uint32) -> uint32:
|
|
return zchecksum.zchecksum_crc32(data, length, UINT32(value))
|