TermiNexus
27121f56a7
R12（wiki/新规范 L27-39）：把「i: int = 0 与 while i < N: … i += 1」形式的下标循环改写为「for i in range(N):」。保守判定：仅当条件为裸 NAME < EXPR、循环体无 continue、NAME 在体内无其它赋值、无嵌套重绑定、循环后同层不再引用 NAME 时才改写。共 269 处 / 48 文件。 R13：c.Set 是正在退化的路径——c.Set(*p, v) 期望写作 *p = v，c.Set(a, b) 等于 a = b。本次把 includes 内 45 处 c.Set 全部迁移为赋值，includes 已无 c.Set 调用。 验证：bootstrap 三重重编 SUCCESS（无 vast_c 解析回退）；全量 77/77 顺序回归通过；ZlibTest 内部断言 79 passed / 0 failed。
700 lines
25 KiB
Plaintext
700 lines
25 KiB
Plaintext
# from string import memset, memcpy
|
||
# from stdio import malloc, free, calloc, realloc
|
||
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
|
||
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: t.CArray[char, 512] = ""
|
||
pyzlib_error_code_val: int = 0
|
||
|
||
# ============================================================
|
||
# Structures
|
||
# ============================================================
|
||
class Compress:
|
||
pool: rawptr[memhub.MemManager]
|
||
stream: VOIDPTR
|
||
is_initialized: int
|
||
is_finished: int
|
||
level: int
|
||
method: int
|
||
wbits: int
|
||
memLevel: int
|
||
strategy: int
|
||
zdict: BYTEPTR
|
||
zdict_len: sizet
|
||
last_pos: sizet
|
||
input_buf: BYTEPTR
|
||
input_buf_len: sizet
|
||
input_buf_cap: sizet
|
||
header_written: int
|
||
|
||
def compress(self,
|
||
data: BYTEPTR, data_len: sizet,
|
||
out_len: sizet*) -> 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:
|
||
*out_len = 0
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p: BYTEPTR = BYTEPTR(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: BYTEPTR = BYTEPTR(stdlib.realloc(self.input_buf, self.input_buf_cap))
|
||
else:
|
||
new_buf: BYTEPTR = BYTEPTR(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 BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p2: BYTEPTR = BYTEPTR(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 BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p3: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p3, 0, 1)
|
||
return p3
|
||
|
||
def flush(self, mode: int, out_len: sizet*) -> 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: rawptr[zdeflate.zdeflate_stream] = self.stream # 隐式类型转换
|
||
if mode == Z_FINISH:
|
||
if not self.header_written:
|
||
self.header_written = 1
|
||
input_data: BYTEPTR = 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] = (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: sizet = s.writer.total()
|
||
if self.pool == None:
|
||
result: BYTEPTR = BYTEPTR(stdlib.malloc(total))
|
||
else:
|
||
result: BYTEPTR = BYTEPTR(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: BYTEPTR = 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 BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p4: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p4, 0, 1)
|
||
return p4
|
||
|
||
if self.pool == None:
|
||
result: BYTEPTR = BYTEPTR(stdlib.malloc(delta))
|
||
else:
|
||
result: BYTEPTR = BYTEPTR(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 BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p5: BYTEPTR = BYTEPTR(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: rawptr[zdeflate.zdeflate_stream] = self.stream # 隐式类型转换
|
||
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 = BYTEPTR(stdlib.calloc(1, self.input_buf_cap))
|
||
else:
|
||
copy_obj.input_buf = BYTEPTR(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 = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
copy_obj.input_buf = BYTEPTR(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: rawptr[zdeflate.zdeflate_stream] = self.stream # 隐式类型转换
|
||
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: rawptr[memhub.MemManager]
|
||
stream: VOIDPTR
|
||
is_initialized: int
|
||
eof: int
|
||
wbits: int
|
||
zdict: BYTEPTR
|
||
zdict_len: sizet
|
||
_unused_data: BYTEPTR
|
||
_unused_data_len: sizet
|
||
_unused_data_cap: sizet
|
||
_unconsumed_tail: BYTEPTR
|
||
_unconsumed_tail_len: sizet
|
||
_unconsumed_tail_cap: sizet
|
||
input_buf: BYTEPTR
|
||
input_buf_len: sizet
|
||
input_buf_cap: sizet
|
||
|
||
def decompress(self, data: BYTEPTR, data_len: sizet,
|
||
max_length: sizet, out_len: sizet*) -> 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, &self._unused_data, &self._unused_data_len,
|
||
&self._unused_data_cap, data, data_len)
|
||
*out_len = 0
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p6: BYTEPTR = BYTEPTR(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 BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p7: BYTEPTR = BYTEPTR(self.pool.alloc(1))
|
||
string.memset(p7, 0, 1)
|
||
return p7
|
||
s: rawptr[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*) -> 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
|
||
*out_len = 0
|
||
if self.pool == None:
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
p8: BYTEPTR = BYTEPTR(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 = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
copy_obj.input_buf = BYTEPTR(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*) -> BYTEPTR:
|
||
#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*) -> BYTEPTR:
|
||
#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: rawptr[memhub.MemManager], src: BYTEPTR, length: sizet) -> BYTEPTR:
|
||
if not src or length == 0: return None
|
||
if pool == None:
|
||
dst: BYTEPTR = BYTEPTR(stdlib.malloc(length))
|
||
else:
|
||
dst: BYTEPTR = BYTEPTR(pool.alloc(length))
|
||
if dst: string.memcpy(dst, src, length)
|
||
return dst
|
||
|
||
# buf 是"指针的指针"（i8**）：不要写成 `BYTE | t.CPtr[t.CPtr]`——联合里嵌
|
||
# `t.CPtr[...]` 下标时声明侧（StubGen/VpiReader）不认作 CPtr 标记，会坍缩成
|
||
# 首段 BYTE = i8，与定义侧 i8** 不一致（跨模块调用即 ABI 错位，见
|
||
# includes/zlib/zinflate.vp decompress 处说明）。改用"指针别名 | t.CPtr"。
|
||
def append_bytes(pool: rawptr[memhub.MemManager], buf: BYTEPTR*, length: sizet*, cap: sizet*,
|
||
data: BYTEPTR, 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: BYTEPTR = BYTEPTR(stdlib.realloc(*buf, new_cap))
|
||
else:
|
||
new_buf: BYTEPTR = BYTEPTR(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: rawptr[memhub.MemManager], buf: BYTEPTR, length: sizet) -> BYTEPTR:
|
||
if length == 0:
|
||
if pool == None:
|
||
stdlib.free(buf)
|
||
return BYTEPTR(stdlib.calloc(1, 1))
|
||
else:
|
||
pool.free(buf)
|
||
p: BYTEPTR = BYTEPTR(pool.alloc(1))
|
||
string.memset(p, 0, 1)
|
||
return p
|
||
if pool == None:
|
||
result: BYTEPTR = BYTEPTR(stdlib.realloc(buf, length))
|
||
return result if result else buf
|
||
else:
|
||
result: BYTEPTR = BYTEPTR(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: rawptr[memhub.MemManager], data: BYTEPTR, data_len: sizet,
|
||
level: int, wbits: int, out_len: sizet*) -> 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: 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: BYTEPTR = BYTEPTR(stdlib.malloc(gzip_len))
|
||
else:
|
||
gzip_buf: BYTEPTR = BYTEPTR(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] = (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
|
||
*out_len = pos
|
||
return shrink_to_fit(pool, gzip_buf, pos)
|
||
return shrink_to_fit(pool, raw_result, *out_len)
|
||
|
||
|
||
def decompress(pool: rawptr[memhub.MemManager], data: BYTEPTR, data_len: sizet,
|
||
wbits: int, bufsize: sizet, out_len: sizet*) -> 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, *out_len)
|
||
|
||
def compressobj(pool: rawptr[memhub.MemManager], level: int, method: int, wbits: int,
|
||
memLevel: int, strategy: int,
|
||
zdict: BYTEPTR, 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: rawptr[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 = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
obj.input_buf = BYTEPTR(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: rawptr[memhub.MemManager], wbits: int, zdict: BYTEPTR, 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 = BYTEPTR(stdlib.calloc(1, 256))
|
||
else:
|
||
obj.input_buf = BYTEPTR(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: BYTEPTR, length: sizet, value: uint32) -> uint32:
|
||
return zchecksum.zchecksum_adler32(data, length, UINT32(value))
|
||
|
||
def zlib_crc32(data: BYTEPTR, length: sizet, value: uint32) -> uint32:
|
||
return zchecksum.zchecksum_crc32(data, length, UINT32(value))
|