GVSDS/Viper_VRTE
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
587d989dec7f3ccb29e162dc08b5956fa08b77c3
Viper_VRTE/zlib/pyzlib.vp
T
TermiNexus 0055ab7ea7 refactor: rawptr[T] -> T* 与 t.CConst 前缀迁移（R14/R16） 
R14：rawptr[X] -> X*（41 文件 / 2037 行改动），X 含模块限定名与嵌套下标（rawptr[GSList[NameEntry]] -> GSList[NameEntry]*）。保留 6 处 rawptr[t.CPtr]（规范例外：存纯裸指针）；# 注释与字符串里的 rawptr[...] 属元语法描述，未动。 R16（保守子集）：t.CConst | str/bytes -> const str/bytes，t.CConst | char | t.CPtr -> const str；typedef 右值位置豁免（const str 在该处无法注册，与 R1 同源）。 验证：TPC 自举三重重编 SUCCESS；TransPyV.exe 全量 77/77 顺序回归通过。
2026-10-05 17:38:21 +08:00

697 lines
25 KiB
Plaintext
Raw Blame History

# 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: 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: zdeflate.zdeflate_stream* = self.stream # implicit type conversion
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: 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 = 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: 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: 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: 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: 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 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: 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: 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: 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: 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: 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: 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: 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))
Reference in new issue View Git Blame Copy Permalink