修复了一些错误

This commit is contained in:
TermiNexus committed 2026-09-18 20:01:14 +08:00
1 parent 61acfc2759
commit 0d77b68cc4
26 files changed
+116 -116

No files matched your search

+12 -12
View File
@@ -42,14 +42,14 @@ class dict:
__values__: t.CPtr
__count__: t.CSizeT = 0
__capacity__: t.CSizeT = 8
__mbuddy__: memhub.MemBuddy | t.CPtr
__mbuddy__: rawptr[memhub.MemBuddy]
__iter_index__: t.CSizeT = 0
def __new__(self, mb: memhub.MemBuddy | t.CPtr) -> t.CPtr:
def __new__(self, mb: rawptr[memhub.MemBuddy]) -> t.CPtr:
# Heap allocation for dict object
return mb.alloc(dict.__sizeof__())
def __init__(self, mb: memhub.MemBuddy | t.CPtr):
def __init__(self, mb: rawptr[memhub.MemBuddy]):
self.__mbuddy__ = mb
self.__keys__ = mb.alloc(self.__capacity__ * 8)
self.__values__ = mb.alloc(self.__capacity__ * _VARIANT_SIZE)
@@ -194,10 +194,10 @@ class dict:
# JSON utilities: serialization / deserialization via includes/json library
# ============================================================
def to_json(self, pool: memhub.MemManager | t.CPtr) -> json.JsonValue | t.CPtr:
def to_json(self, pool: rawptr[memhub.MemManager]) -> rawptr[json.JsonValue]:
"""Convert dict instance to JsonValue object (JSON_OBJECT).
Automatically select corresponding JSON value type according to Variant tag."""
obj: json.JsonValue | t.CPtr = json.object(pool)
obj: rawptr[json.JsonValue] = json.object(pool)
i: t.CSizeT = 0
while i < self.__count__:
key_ptr: str | t.CPtr = t.CPtr(t.CUInt64T(self.__keys__) + i * 8)
@@ -205,7 +205,7 @@ class dict:
val_ptr: str | t.CPtr = t.CPtr(t.CUInt64T(self.__values__) + i * _VARIANT_SIZE + 8)
vt: t.CInt = vtype_ptr[0]
vp: t.CPtr = val_ptr[0]
jv: json.JsonValue | t.CPtr = json.null(pool)
jv: rawptr[json.JsonValue] = json.null(pool)
if jv != None:
if vt == VARIANT_INT:
jv.vtype = 2 # JSON_INT
@@ -231,14 +231,14 @@ class dict:
i += 1
return obj
def from_json(self, pool: memhub.MemManager | t.CPtr, root: json.JsonValue | t.CPtr):
def from_json(self, pool: rawptr[memhub.MemManager], root: rawptr[json.JsonValue]):
"""Populate dict from JsonValue object.
Extract values and store them as matching Variant types based on JSON type tag."""
if root == None:
return
if root.vtype != 6: # JSON_OBJECT
return
cur: json.JsonValue | t.CPtr = root.child
cur: rawptr[json.JsonValue] = root.child
while cur != None:
if cur.key != None:
if cur.vtype == 2: # JSON_INT
@@ -260,12 +260,12 @@ class dict:
self._set_variant_raw(cur.key, VARIANT_FLOAT, storage)
cur = cur.next
def dumps(self, pool: memhub.MemManager | t.CPtr) -> str:
def dumps(self, pool: rawptr[memhub.MemManager]) -> str:
"""Serialize dict to JSON string."""
obj: json.JsonValue | t.CPtr = self.to_json(pool)
obj: rawptr[json.JsonValue] = self.to_json(pool)
return json.write(pool, obj, False)
def loads(self, pool: memhub.MemManager | t.CPtr, json_str: str):
def loads(self, pool: rawptr[memhub.MemManager], json_str: str):
"""Parse JSON string and populate dict instance."""
root: json.JsonValue | t.CPtr = json.parse(pool, json_str)
root: rawptr[json.JsonValue] = json.parse(pool, json_str)
self.from_json(pool, root)
+1 -1
View File
@@ -13,7 +13,7 @@ class _str:
# (optional; no error will be raised if the field cannot be located)
__requires__: list[str] = ['__mbuddy__']
__data__: str
__mbuddy__: memhub.MemBuddy | t.CPtr
__mbuddy__: rawptr[memhub.MemBuddy]
def upper(self) -> str:
src: str = self.__data__
+3 -3
View File
@@ -24,15 +24,15 @@ class list[T]:
__data__: t.CPtr
__count__: t.CSizeT
__capacity__: t.CSizeT
__pool__: memhub.MemManager | t.CPtr
__pool__: rawptr[memhub.MemManager]
__elem_size__: t.CSizeT
__iter_index__: t.CSizeT
def __new__(self, pool: memhub.MemManager | t.CPtr, elem_size: t.CSizeT = 0):
def __new__(self, pool: rawptr[memhub.MemManager], elem_size: t.CSizeT = 0):
ret = pool.alloc(48)
return ret
def __init__(self, pool: memhub.MemManager | t.CPtr, elem_size: t.CSizeT = 0):
def __init__(self, pool: rawptr[memhub.MemManager], elem_size: t.CSizeT = 0):
self.__pool__ = pool
self.__count__ = 0
self.__capacity__ = 8
+11 -11
View File
@@ -43,7 +43,7 @@ COUNT: t.CDefine = 3
MAX_ARGS: t.CDefine = 32
# Module-level membuddy pointer (assigned inside main by user)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# ============================================================
@@ -68,13 +68,13 @@ class Argument:
# ============================================================
class ParsedArgs:
_ht: hashtable.HashTable | t.CPtr
__mbuddy__: memhub.MemBuddy | t.CPtr
_ht: rawptr[hashtable.HashTable]
__mbuddy__: rawptr[memhub.MemBuddy]
def __new__(self, mb: memhub.MemBuddy | t.CPtr) -> t.CPtr:
def __new__(self, mb: rawptr[memhub.MemBuddy]) -> t.CPtr:
return mb.alloc(16)
def __init__(self, mb: memhub.MemBuddy | t.CPtr):
def __init__(self, mb: rawptr[memhub.MemBuddy]):
self.__mbuddy__ = mb
self._ht = hashtable.HashTable(mb)
@@ -125,19 +125,19 @@ class ArgumentParser:
_description: str
_args: Argument | t.CPtr
_arg_count: int
__mbuddy__: memhub.MemBuddy | t.CPtr
__mbuddy__: rawptr[memhub.MemBuddy]
def __new__(self, prog: str = "", description: str = "",
pool: memhub.MemBuddy | t.CPtr = None) -> t.CPtr:
mb: memhub.MemBuddy | t.CPtr = pool
pool: rawptr[memhub.MemBuddy] = None) -> t.CPtr:
mb: rawptr[memhub.MemBuddy] = pool
if mb is None:
mb = _mbuddy
ret: t.CPtr = mb.alloc(40)
return ret
def __init__(self, prog: str = "", description: str = "",
pool: memhub.MemBuddy | t.CPtr = None):
mb: memhub.MemBuddy | t.CPtr = pool
pool: rawptr[memhub.MemBuddy] = None):
mb: rawptr[memhub.MemBuddy] = pool
if mb is None:
mb = _mbuddy
self.__mbuddy__ = mb
@@ -453,7 +453,7 @@ def release(args: ParsedArgs | t.CPtr):
# 用临时变量拆分链式属性访问,避免 args._ht.__slots__ 跨模块类型推断失败
# 注意:类型注解必须用 hashtable.HashTable(带模块前缀),
# 否则编译器无法解析 HashTable 类型,导致 ht.__slots__ 属性访问失败
ht: hashtable.HashTable | t.CPtr = args._ht
ht: rawptr[hashtable.HashTable] = args._ht
if ht.__slots__ is not None:
args.__mbuddy__.free(ht.__slots__)
args.__mbuddy__.free(t.CPtr(t.CUInt64T(ht)))
+2 -2
View File
@@ -6,7 +6,7 @@
# 内部位流/Huffman 助手为模块级私有函数。
#
# 用法:
# gz: gzip.GzipFile | t.CPtr = gzip.GzipFile("train-images-idx3-ubyte.gz")
# gz: rawptr[gzip.GzipFile] = gzip.GzipFile("train-images-idx3-ubyte.gz")
# if gz is None or gz.ok == 0: ...
# buf: bytes = gz.read_all() # 内部 malloc, read_all 返回即解压完成
# n: t.CSizeT = gz.dst_len # 原始长度(由对象持有, close 时释放)
@@ -44,7 +44,7 @@ class GzipFile:
if path is None:
self.err = -1
return
f: fileio.File | t.CPtr = fileio.File(path, fileio.MODE.R)
f: rawptr[fileio.File] = fileio.File(path, fileio.MODE.R)
if f.closed:
self.err = -2 # 打不开文件
return
+3 -3
View File
@@ -51,14 +51,14 @@ class HashTable:
__count__: t.CSizeT = 0 # Number of occupied entries
__capacity__: t.CSizeT = 16 # Total slot count (power of two)
__tombstones__: t.CSizeT = 0 # Tombstone entry count
__mbuddy__: memhub.MemBuddy | t.CPtr
__mbuddy__: rawptr[memhub.MemBuddy]
__iter_index__: t.CSizeT = 0
def __new__(self, mb: memhub.MemBuddy | t.CPtr) -> t.CPtr:
def __new__(self, mb: rawptr[memhub.MemBuddy]) -> t.CPtr:
# Heap-allocate HashTable object (6 fields × 8 bytes = 48 bytes)
return mb.alloc(48)
def __init__(self, mb: memhub.MemBuddy | t.CPtr):
def __init__(self, mb: rawptr[memhub.MemBuddy]):
self.__mbuddy__ = mb
self.__capacity__ = _HT_INIT_CAP
self.__count__ = 0
+2 -2
View File
@@ -211,8 +211,8 @@ class MemManager:
def __exit__(self):
self.reset()
def alloc_buf(self, capacity: t.CSizeT) -> viperio.Buf | t.CPtr:
buf: viperio.Buf | t.CPtr = self.alloc(viperio.Buf.__sizeof__())
def alloc_buf(self, capacity: t.CSizeT) -> rawptr[viperio.Buf]:
buf: rawptr[viperio.Buf] = self.alloc(viperio.Buf.__sizeof__())
if buf is None:
return None
data_ptr: t.CPtr = self.alloc(capacity)
+18 -18
View File
@@ -38,9 +38,9 @@ class ndarray[T]:
ndim: t.CInt
size: t.CSizeT
owns_data: t.CInt
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
def __new__(self, pool: memhub.MemManager | t.CPtr, n: t.CSizeT):
def __new__(self, pool: rawptr[memhub.MemManager], n: t.CSizeT):
self.pool = pool
self.data = pool.alloc(n * T.__sizeof__())
if self.data:
@@ -295,7 +295,7 @@ Uint8Array: t.CTypedef = ndarray[t.CUnsignedChar]
# Internal helpers (specialized for float64 by default)
# Use _alloc_ndarray_t[T] / _empty_like_t[T] for other types
# ============================================================
def _alloc_ndarray(pool: memhub.MemManager | t.CPtr) -> ndarray[t.CDouble] | t.CPtr:
def _alloc_ndarray(pool: rawptr[memhub.MemManager]) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = pool.alloc(ndarray[t.CDouble].__sizeof__())
if a:
string.memset(a, 0, ndarray[t.CDouble].__sizeof__())
@@ -330,7 +330,7 @@ def _empty_like(a: ndarray[t.CDouble] | t.CPtr) -> ndarray[t.CDouble] | t.CPtr:
# ============================================================
# Array creation functions (float64 default)
# ============================================================
def array(pool: memhub.MemManager | t.CPtr, data: t.CDouble | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def array(pool: rawptr[memhub.MemManager], data: t.CDouble | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = _alloc_ndarray(pool)
if not a: return None
a.data = pool.alloc(n * t.CDouble.__sizeof__())
@@ -346,7 +346,7 @@ def array(pool: memhub.MemManager | t.CPtr, data: t.CDouble | t.CPtr, n: t.CSize
a.owns_data = 1
return a
def zeros(pool: memhub.MemManager | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def zeros(pool: rawptr[memhub.MemManager], n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = _alloc_ndarray(pool)
if not a: return None
a.data = pool.alloc(n * t.CDouble.__sizeof__())
@@ -362,19 +362,19 @@ def zeros(pool: memhub.MemManager | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] |
a.owns_data = 1
return a
def ones(pool: memhub.MemManager | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def ones(pool: rawptr[memhub.MemManager], n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = zeros(pool, n)
if not a: return None
a.fill(t.CDouble(1.0))
return a
def full(pool: memhub.MemManager | t.CPtr, n: t.CSizeT, val: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
def full(pool: rawptr[memhub.MemManager], n: t.CSizeT, val: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = zeros(pool, n)
if not a: return None
a.fill(val)
return a
def arange(pool: memhub.MemManager | t.CPtr, start: t.CDouble, stop: t.CDouble, step: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
def arange(pool: rawptr[memhub.MemManager], start: t.CDouble, stop: t.CDouble, step: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
if step == t.CDouble(0.0): return None
n: t.CSizeT = 0
if step > t.CDouble(0.0):
@@ -398,7 +398,7 @@ def arange(pool: memhub.MemManager | t.CPtr, start: t.CDouble, stop: t.CDouble,
i += 1
return a
def linspace(pool: memhub.MemManager | t.CPtr, start: t.CDouble, stop: t.CDouble, num: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def linspace(pool: rawptr[memhub.MemManager], start: t.CDouble, stop: t.CDouble, num: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
if num == 0: return None
a: ndarray[t.CDouble] | t.CPtr = zeros(pool, num)
if not a: return None
@@ -412,7 +412,7 @@ def linspace(pool: memhub.MemManager | t.CPtr, start: t.CDouble, stop: t.CDouble
i += 1
return a
def empty2d(pool: memhub.MemManager | t.CPtr, rows: t.CSizeT, cols: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def empty2d(pool: rawptr[memhub.MemManager], rows: t.CSizeT, cols: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = _alloc_ndarray(pool)
if not a: return None
total: t.CSizeT = rows * cols
@@ -430,16 +430,16 @@ def empty2d(pool: memhub.MemManager | t.CPtr, rows: t.CSizeT, cols: t.CSizeT) ->
_compute_strides(a)
return a
def zeros2d(pool: memhub.MemManager | t.CPtr, rows: t.CSizeT, cols: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def zeros2d(pool: rawptr[memhub.MemManager], rows: t.CSizeT, cols: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
return empty2d(pool, rows, cols)
def ones2d(pool: memhub.MemManager | t.CPtr, rows: t.CSizeT, cols: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def ones2d(pool: rawptr[memhub.MemManager], rows: t.CSizeT, cols: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = empty2d(pool, rows, cols)
if not a: return None
a.fill(t.CDouble(1.0))
return a
def eye(pool: memhub.MemManager | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def eye(pool: rawptr[memhub.MemManager], n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = empty2d(pool, n, n)
if not a: return None
i: t.CSizeT = 0
@@ -448,7 +448,7 @@ def eye(pool: memhub.MemManager | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] | t
i += 1
return a
def diag(pool: memhub.MemManager | t.CPtr, vals: ndarray[t.CDouble] | t.CPtr) -> ndarray[t.CDouble] | t.CPtr:
def diag(pool: rawptr[memhub.MemManager], vals: ndarray[t.CDouble] | t.CPtr) -> ndarray[t.CDouble] | t.CPtr:
n: t.CSizeT = vals.size
a: ndarray[t.CDouble] | t.CPtr = empty2d(pool, n, n)
if not a: return None
@@ -1033,10 +1033,10 @@ def np_greater_equal(a: ndarray[t.CDouble] | t.CPtr, b: ndarray[t.CDouble] | t.C
# ============================================================
# Additional creation functions (float64 default)
# ============================================================
def empty(pool: memhub.MemManager | t.CPtr, n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
def empty(pool: rawptr[memhub.MemManager], n: t.CSizeT) -> ndarray[t.CDouble] | t.CPtr:
return zeros(pool, n)
def full2d(pool: memhub.MemManager | t.CPtr, rows: t.CSizeT, cols: t.CSizeT, val: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
def full2d(pool: rawptr[memhub.MemManager], rows: t.CSizeT, cols: t.CSizeT, val: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
a: ndarray[t.CDouble] | t.CPtr = empty2d(pool, rows, cols)
if not a: return None
a.fill(val)
@@ -1066,10 +1066,10 @@ def ones_like(a: ndarray[t.CDouble] | t.CPtr) -> ndarray[t.CDouble] | t.CPtr:
result.fill(t.CDouble(1.0))
return result
def arange1(pool: memhub.MemManager | t.CPtr, stop: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
def arange1(pool: rawptr[memhub.MemManager], stop: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
return arange(pool, t.CDouble(0.0), stop, t.CDouble(1.0))
def linspace2(pool: memhub.MemManager | t.CPtr, start: t.CDouble, stop: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
def linspace2(pool: rawptr[memhub.MemManager], start: t.CDouble, stop: t.CDouble) -> ndarray[t.CDouble] | t.CPtr:
return linspace(pool, start, stop, 50)
def meshgrid(x: ndarray[t.CDouble] | t.CPtr, y: ndarray[t.CDouble] | t.CPtr) -> ndarray[t.CDouble] | t.CPtr:
+1 -1
View File
@@ -12,7 +12,7 @@ import os._posix
# os.path is for cross-platform pure string operations, can be used directly
# Module-level global MBuddy pointer (initialized by the user in main)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# =============================================================================
+3 -3
View File
@@ -9,14 +9,14 @@ import stdlib
MAX_PATH_LEN: t.CDefine = 4096
# 模块级全局 MBuddy 指针(由用户在 main 中赋值初始化)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# =============================================================================
# 文件/目录属性查询
# =============================================================================
def _stat(path: str) -> posix.Stat | t.CPtr:
def _stat(path: str) -> rawptr[posix.Stat]:
"""调用 stat 并返回 Stat 结构体指针。失败返回 None。
注意:Stat 结构体通过 _mbuddy 分配,调用者负责释放。"""
st: posix.Stat = posix.Stat()
@@ -149,7 +149,7 @@ def listdir(path: str, out_names: str, max_count: ULONG) -> ULONG:
count: ULONG = 0
while count < max_count:
entry: posix.Dirent | t.CPtr = posix.readdir(dirp)
entry: rawptr[posix.Dirent] = posix.readdir(dirp)
if entry == None:
break
+1 -1
View File
@@ -13,7 +13,7 @@ MAX_PATH_LEN: t.CDefine = 260
INVALID_FILE_ATTRIBUTES: t.CDefine = 0xFFFFFFFF
# 模块级全局 MBuddy 指针(由用户在 main 中赋值初始化)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# =============================================================================
+1 -1
View File
@@ -6,7 +6,7 @@ import platmacro
# 模块级全局 MBuddy 指针(由用户在 main 中赋值初始化)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# =============================================================================
+1 -1
View File
@@ -25,7 +25,7 @@ import zlib.pyzlib as pyzlib
# ============================================================
# Module-level global MBuddy pointer (assigned by the user in main)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# Socket initialization flag
_socket_ready: bool = False
+1 -1
View File
@@ -380,7 +380,7 @@ def download(url: str, path: str, threads: INT = 1,
job.status = 0
ctxs[i] = t.CPtr(t.CUInt64T(job))
th: vt.Thread | t.CPtr = vt.Thread(_dl_worker, ctxs[i])
th: rawptr[vt.Thread] = vt.Thread(_dl_worker, ctxs[i])
if th is None:
stdlib.free(job.buf)
stdlib.free(job.path)
+14 -14
View File
@@ -224,7 +224,7 @@ def _TlsErrText():
# 核心请求
# ============================================================
def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str, auth: str = None) -> requests.Response | t.CPtr:
def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str, auth: str = None) -> rawptr[requests.Response]:
"""HTTPS GET/POST 核心实现。
verify=False 时跳过证书链/主机名校验(不推荐);verify=True 时
@@ -237,7 +237,7 @@ def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str,
_last_verify_flag = 1 if verify else 0
resp: requests.Response | t.CPtr = requests.create_response()
resp: rawptr[requests.Response] = requests.create_response()
if resp is None: return None
if _EnsureReady() != 0:
@@ -487,7 +487,7 @@ def _HttpsRequest(method: str, url: str, verify: bool, ca_file: str, data: str,
# 公共 API
# ============================================================
def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str, auth: str = None) -> requests.Response | t.CPtr:
def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data: str, content_type: str, auth: str = None) -> rawptr[requests.Response]:
"""HTTPS GET/POST + 重定向跟随(跨 http/https,最多 5 跳)。
auth(可选)原样透传给 _HttpsRequest;None 时行为与原来完全一致。
@@ -505,7 +505,7 @@ def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data:
hop: INT = 0
while hop < 5:
resp: requests.Response | t.CPtr = _HttpsRequest(method, cur, verify, ca_file, data, content_type, auth)
resp: rawptr[requests.Response] = _HttpsRequest(method, cur, verify, ca_file, data, content_type, auth)
if resp is None:
requests._mbuddy.free(cur)
return None
@@ -542,7 +542,7 @@ def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data:
if string.strncmp(nxt, "https://", 8) != 0:
requests._mbuddy.free(loc)
requests._mbuddy.free(cur)
http_resp: requests.Response | t.CPtr = requests.get(nxt)
http_resp: rawptr[requests.Response] = requests.get(nxt)
requests._mbuddy.free(nxt)
return http_resp
@@ -556,7 +556,7 @@ def _HttpsRedirectLoop(method: str, url: str, verify: bool, ca_file: str, data:
return None
def get(url: str, verify: bool = True, ca_file: str = None) -> requests.Response | t.CPtr:
def get(url: str, verify: bool = True, ca_file: str = None) -> rawptr[requests.Response]:
"""HTTPS GET request(自动跟随 3xx 重定向,最多 5 跳)。
Args:
@@ -570,7 +570,7 @@ def get(url: str, verify: bool = True, ca_file: str = None) -> requests.Response
def post(url: str, data: str = None, content_type: str = "application/x-www-form-urlencoded",
verify: bool = True, ca_file: str = None, auth: str = None) -> requests.Response | t.CPtr:
verify: bool = True, ca_file: str = None, auth: str = None) -> rawptr[requests.Response]:
"""HTTPS POST request(自动跟随 3xx 重定向,最多 5 跳)。
Args:
@@ -599,20 +599,20 @@ def post(url: str, data: str = None, content_type: str = "application/x-www-form
# (_st_state=2,失败原因见 stream_error_text()/last_error())。
# ============================================================
def _SseDeltaContent(root: json.JsonValue | t.CPtr) -> str:
def _SseDeltaContent(root: rawptr[json.JsonValue]) -> str:
"""从流式 data 事件取 choices[0].delta.content;缺失返回 None。"""
if root is None or not root.is_object():
return None
cv: json.JsonValue | t.CPtr = root["choices"]
cv: rawptr[json.JsonValue] = root["choices"]
if cv is None or not cv.is_array():
return None
m0: json.JsonValue | t.CPtr = cv.get_item(0)
m0: rawptr[json.JsonValue] = cv.get_item(0)
if m0 is None or not m0.is_object():
return None
dv: json.JsonValue | t.CPtr = m0["delta"]
dv: rawptr[json.JsonValue] = m0["delta"]
if dv is None or not dv.is_object():
return None
content_v: json.JsonValue | t.CPtr = dv["content"]
content_v: rawptr[json.JsonValue] = dv["content"]
if content_v is None or not content_v.is_string():
return None
return content_v.as_string()
@@ -820,7 +820,7 @@ def post_stream(url: str, data: str, content_type: str = "application/json",
sock.close()
requests._mbuddy.free(errmsg)
return -1
jpool: memhub.MemPool | t.CPtr = memhub.MemPool(pmem, 524288)
jpool: rawptr[memhub.MemPool] = memhub.MemPool(pmem, 524288)
leftover: t.CArray[t.CChar, 16384]
llen: INT = 0
@@ -880,7 +880,7 @@ def post_stream(url: str, data: str, content_type: str = "application/json",
if pl[0] != 0:
if string.strncmp(pl, "[DONE]", 6) != 0 or pl[6] != 0:
jpool.reset()
ev: json.JsonValue | t.CPtr = json.parse(jpool, pl)
ev: rawptr[json.JsonValue] = json.parse(jpool, pl)
dtx: str = _SseDeltaContent(ev)
if dtx is not None:
_StreamAppendContent(dtx)
+2 -2
View File
@@ -16,7 +16,7 @@ import posix
COPY_BUF_SIZE: t.CDefine = 65536 # 2^16
# Module-level global MBuddy pointer (initialized by the user in main)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# =============================================================================
@@ -193,7 +193,7 @@ def _rmtree_posix(path: str) -> int:
path_len: t.CSizeT = len(path)
result: int = 0
entry: posix.Dirent | t.CPtr = posix.readdir(dirp)
entry: rawptr[posix.Dirent] = posix.readdir(dirp)
while entry != None:
name: str = entry.d_name
# Skip . and ..
+1 -1
View File
@@ -123,7 +123,7 @@ def _MakeAddr(host: str, port: INT) -> SocketAddr:
# DNS resolution — gethostbyname returns struct hostent*
# h_addr_list is char** → h_addr_list[0] is char* → *(uint32_t*)h_addr_list[0] is IP
if platmacro.IS_WINDOWS:
he_raw: w32.winsock2.WSAHostEnt | t.CPtr = w32.winsock2.gethostbyname(host)
he_raw: rawptr[w32.winsock2.WSAHostEnt] = w32.winsock2.gethostbyname(host)
if he_raw is not None:
# he_raw.h_addr_list → he_raw->h_addr_list (char**)
# c.Deref(t.CUInt64T(he_raw.h_addr_list, t.CPtr)) → *(uint64_t*)(he_raw->h_addr_list) → h_addr_list[0]
+1 -1
View File
@@ -18,7 +18,7 @@ import posix
OUTPUT_BUF_SIZE: t.CDefine = 65536
# Module global MBuddy pointer (initialized in main)
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# =============================================================================
+1 -1
View File
@@ -13,7 +13,7 @@ import w32.win32process
import posix
# Module global MBuddy pointer (initialized in user main), Plans to ban
_mbuddy: memhub.MemBuddy | t.CPtr
_mbuddy: rawptr[memhub.MemBuddy]
# =============================================================================
+2 -2
View File
@@ -8,9 +8,9 @@ class Vector[T]():
length: t.CSizeT
capacity: t.CSizeT
elem_size: t.CSizeT
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
def __init__(self, pool: memhub.MemManager | t.CPtr, capacity: t.CSizeT, _hint: T):
def __init__(self, pool: rawptr[memhub.MemManager], capacity: t.CSizeT, _hint: T):
self.elem_size = T.__sizeof__()
self.capacity = capacity
self.length = 0
+2 -2
View File
@@ -11,11 +11,11 @@
# 成员名为原始 UTF-8 字节(写入时置 flag bit 11,读取时不自动解码)。
#
# 用法(对齐 Python 语义,只读模式自动把整个 zip 读入内存):
# w: zipfile.ZipFile | t.CPtr = zipfile.ZipFile("out.zip", "w")
# w: rawptr[zipfile.ZipFile] = zipfile.ZipFile("out.zip", "w")
# w.write("a.txt", buf, buf_len) # 默认 ZIP_DEFLATED
# w.close()
#
# r: zipfile.ZipFile | t.CPtr = zipfile.ZipFile("out.zip", "r")
# r: rawptr[zipfile.ZipFile] = zipfile.ZipFile("out.zip", "r")
# olen: t.CSizeT = 0
# data: bytes = r.read("a.txt", c.Addr(olen)) # malloc 缓冲,用后 free
# err = zipfile.error_code() / zipfile.error()
+14 -14
View File
@@ -68,7 +68,7 @@ pyzlib_error_code_val: t.CInt = 0
# ============================================================
@t.Object
class Compress:
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
stream: VOIDPTR
is_initialized: t.CInt
is_finished: t.CInt
@@ -141,7 +141,7 @@ class Compress:
if not out_len:
set_error(Z_STREAM_ERROR, "out_len is None")
return None
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
s: rawptr[zdeflate.zdeflate_stream] = self.stream # 隐式类型转换
if mode == Z_FINISH:
if not self.header_written:
self.header_written = 1
@@ -232,7 +232,7 @@ class Compress:
set_error(Z_MEM_ERROR, "out of memory")
return None
copy_obj.pool = self.pool
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
s: rawptr[zdeflate.zdeflate_stream] = self.stream # 隐式类型转换
copy_obj.stream = s.copy()
if not copy_obj.stream:
if self.pool == None:
@@ -275,7 +275,7 @@ class Compress:
def delete(self):
if self.is_initialized and self.stream:
s: zdeflate.zdeflate_stream | t.CPtr = self.stream # 隐式类型转换
s: rawptr[zdeflate.zdeflate_stream] = self.stream # 隐式类型转换
s.destroy()
if self.pool == None:
stdlib.free(self.zdict)
@@ -292,7 +292,7 @@ class Compress:
@t.Object
class Decompress:
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
stream: VOIDPTR
is_initialized: t.CInt
eof: t.CInt
@@ -340,7 +340,7 @@ class Decompress:
p7: BYTEPTR = BYTEPTR(self.pool.alloc(1))
string.memset(p7, 0, 1)
return p7
s: zinflate.zinflate_stream | t.CPtr = zinflate.zinflate_create(self.pool, self.wbits)
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
@@ -475,7 +475,7 @@ def set_error(code: int, msg: str):
else:
pyzlib_error_msg[0] = '\0'
def clone_bytes(pool: memhub.MemManager | t.CPtr, src: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
def clone_bytes(pool: rawptr[memhub.MemManager], src: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
if not src or length == 0: return None
if pool == None:
dst: BYTEPTR = BYTEPTR(stdlib.malloc(length))
@@ -488,7 +488,7 @@ def clone_bytes(pool: memhub.MemManager | t.CPtr, src: BYTEPTR, length: t.CSizeT
# `t.CPtr[...]` 下标时声明侧(StubGen/VpiReader)不认作 CPtr 标记,会坍缩成
# 首段 BYTE = i8,与定义侧 i8** 不一致(跨模块调用即 ABI 错位,见
# includes/zlib/zinflate.vp decompress 处说明)。改用"指针别名 | t.CPtr"。
def append_bytes(pool: memhub.MemManager | t.CPtr, buf: BYTEPTR | t.CPtr, length: t.CSizeT | t.CPtr, cap: t.CSizeT | t.CPtr,
def append_bytes(pool: rawptr[memhub.MemManager], buf: BYTEPTR | 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
@@ -510,7 +510,7 @@ def append_bytes(pool: memhub.MemManager | t.CPtr, buf: BYTEPTR | t.CPtr, length
return 0
def shrink_to_fit(pool: memhub.MemManager | t.CPtr, buf: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
def shrink_to_fit(pool: rawptr[memhub.MemManager], buf: BYTEPTR, length: t.CSizeT) -> BYTEPTR:
if length == 0:
if pool == None:
stdlib.free(buf)
@@ -546,7 +546,7 @@ def clear_error():
pyzlib_error_msg[0] = '\0'
pyzlib_error_code_val = 0
def compress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSizeT,
def compress(pool: rawptr[memhub.MemManager], 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")
@@ -609,7 +609,7 @@ def compress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSizeT
return shrink_to_fit(pool, raw_result, c.Deref(out_len))
def decompress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSizeT,
def decompress(pool: rawptr[memhub.MemManager], 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")
@@ -623,7 +623,7 @@ def decompress(pool: memhub.MemManager | t.CPtr, data: BYTEPTR, data_len: t.CSiz
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,
def compressobj(pool: rawptr[memhub.MemManager], 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:
@@ -637,7 +637,7 @@ def compressobj(pool: memhub.MemManager | t.CPtr, level: t.CInt, method: t.CInt,
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)
s: rawptr[zdeflate.zdeflate_stream] = zdeflate.zdeflate_create(pool, level, actual_wbits, memLevel, strategy)
if not s:
if pool == None:
stdlib.free(obj)
@@ -667,7 +667,7 @@ def compressobj(pool: memhub.MemManager | t.CPtr, level: t.CInt, method: t.CInt,
return obj
def decompressobj(pool: memhub.MemManager | t.CPtr, wbits: t.CInt, zdict: BYTEPTR, zdict_len: t.CSizeT) -> Decompress | t.CPtr:
def decompressobj(pool: rawptr[memhub.MemManager], wbits: t.CInt, zdict: BYTEPTR, zdict_len: t.CSizeT) -> Decompress | t.CPtr:
if pool == None:
obj: Decompress | t.CPtr = stdlib.calloc(1, Decompress.__sizeof__())
else:
+4 -4
View File
@@ -96,13 +96,13 @@ zdeflate_codelen_order: t.CArray[int, 19] = [
# ============================================================
@t.Object
class zbit_writer:
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
buf: BYTE | t.CPtr
cap: t.CSizeT
byte_pos: t.CSizeT
bit_pos: t.CInt
def __init__(self, pool: memhub.MemManager | t.CPtr):
def __init__(self, pool: rawptr[memhub.MemManager]):
self.pool = pool
self.cap = 4096
if pool == None:
@@ -277,10 +277,10 @@ class zbit_reader:
# Memory allocation helper for bare metal
# Can be replaced with custom allocator
# ============================================================
def zdef_alloc(pool: memhub.MemManager | t.CPtr, n: t.CSizeT) -> VOIDPTR:
def zdef_alloc(pool: rawptr[memhub.MemManager], n: t.CSizeT) -> VOIDPTR:
if pool == None: return stdlib.malloc(n)
return pool.alloc(n)
def zdef_free(pool: memhub.MemManager | t.CPtr, p: VOIDPTR):
def zdef_free(pool: rawptr[memhub.MemManager], p: VOIDPTR):
if pool == None:
stdlib.free(p)
return
+5 -5
View File
@@ -9,7 +9,7 @@ import t, c
@t.Object
class zdeflate_stream:
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
writer: zdef.zbit_writer
window: BYTE | t.CPtr
window_pos: t.CInt
@@ -95,7 +95,7 @@ class zdeflate_stream:
lit_tree.encode_symbol(256, c.Addr(self.writer))
def encode_block_data(self, data: UINT8PTR, length: t.CSizeT,
lit_tree: zhuff.zhuff_tree | t.CPtr, dist_tree: zhuff.zhuff_tree | t.CPtr):
lit_tree: rawptr[zhuff.zhuff_tree], dist_tree: rawptr[zhuff.zhuff_tree]):
pos: t.CSizeT = 0
while pos < length:
best_dist: t.CInt = 0
@@ -340,7 +340,7 @@ def zdeflate_count_cl_freqs(all_lengths: INTPTR, total: t.CInt, cl_freqs: INTPTR
# 按字面量(符号 val)逐个统计,保证总覆盖数恰为 total
def zdeflate_write_cl_encoded(all_lengths: INTPTR, total: t.CInt,
w: zdef.zbit_writer | t.CPtr, cl_tree: zhuff.zhuff_tree | t.CPtr):
w: rawptr[zdef.zbit_writer], cl_tree: rawptr[zhuff.zhuff_tree]):
i: t.CInt = 0
while i < total:
if all_lengths[i] == 0:
@@ -381,7 +381,7 @@ def zdeflate_write_cl_encoded(all_lengths: INTPTR, total: t.CInt,
def zdeflate_create(pool: memhub.MemManager | t.CPtr, level: t.CInt, wbits: t.CInt, mem_level: t.CInt, strategy: t.CInt) -> zdeflate_stream | t.CPtr:
def zdeflate_create(pool: rawptr[memhub.MemManager], level: t.CInt, wbits: t.CInt, mem_level: t.CInt, strategy: t.CInt) -> zdeflate_stream | t.CPtr:
n: t.CSizeT = zdeflate_stream.__sizeof__()
s: zdeflate_stream | t.CPtr = zdef.zdef_alloc(pool, n)
if not s: return None
@@ -408,7 +408,7 @@ def zdeflate_create(pool: memhub.MemManager | t.CPtr, level: t.CInt, wbits: t.CI
return s
def zdeflate_one_shot(pool: memhub.MemManager | t.CPtr, data: UINT8PTR, length: t.CSizeT,
def zdeflate_one_shot(pool: rawptr[memhub.MemManager], data: UINT8PTR, length: t.CSizeT,
level: t.CInt, wbits: t.CInt, out_len: t.CSizeT | t.CPtr) -> t.CUInt8T | t.CPtr:
s: zdeflate_stream | t.CPtr = zdeflate_create(pool, level, wbits, 8, 0)
if not s: return None
+4 -4
View File
@@ -15,7 +15,7 @@ class zhuff_code:
@t.Object
class zhuff_tree:
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
codes: t.CArray[zhuff_code, ZHUFF_MAX_CODES]
count: t.CInt
max_bits: t.CInt
@@ -95,7 +95,7 @@ class zhuff_tree:
else:
self.codes[i].code = 0
def encode_symbol(self, symbol: int, writer: zdef.zbit_writer | t.CPtr):
def encode_symbol(self, symbol: int, writer: rawptr[zdef.zbit_writer]):
if symbol < 0 or symbol >= self.count:
return
if self.codes[symbol].bits == 0:
@@ -305,7 +305,7 @@ class zhuff_decode_node:
@t.Object
class zhuff_decode_tree:
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
nodes: t.CArray[zhuff_decode_node, 2 * ZHUFF_MAX_CODES]
node_count: t.CInt
root: t.CInt
@@ -346,7 +346,7 @@ class zhuff_decode_tree:
else:
self.nodes[self.nodes[node].children[dir]].symbol = i
def decode_symbol(self, reader: zdef.zbit_reader | t.CPtr) -> t.CInt:
def decode_symbol(self, reader: rawptr[zdef.zbit_reader]) -> t.CInt:
node: int = self.root
while self.nodes[node].symbol == -1:
bit: t.CUnsignedInt
+6 -6
View File
@@ -9,7 +9,7 @@ import t, c
@t.Object
class zinflate_stream:
pool: memhub.MemManager | t.CPtr
pool: rawptr[memhub.MemManager]
reader: zdef.zbit_reader
window: BYTEPTR
@@ -109,7 +109,7 @@ class zinflate_stream:
return 0
def read_huffman(self, dt: zhuff.zhuff_decode_tree | t.CPtr) -> t.CInt:
def read_huffman(self, dt: rawptr[zhuff.zhuff_decode_tree]) -> t.CInt:
node: int = dt.root
while dt.nodes[node].symbol == -1:
bit: UINT
@@ -142,8 +142,8 @@ class zinflate_stream:
lit_tree.build_fixed_lit_tree()
dist_tree.build_fixed_dist_tree()
lit_dt: zhuff.zhuff_decode_tree | t.CPtr = zdef.zdef_alloc(self.pool, zhuff.zhuff_decode_tree.__sizeof__())
dist_dt: zhuff.zhuff_decode_tree | t.CPtr = zdef.zdef_alloc(self.pool, zhuff.zhuff_decode_tree.__sizeof__())
lit_dt: rawptr[zhuff.zhuff_decode_tree] = zdef.zdef_alloc(self.pool, zhuff.zhuff_decode_tree.__sizeof__())
dist_dt: rawptr[zhuff.zhuff_decode_tree] = zdef.zdef_alloc(self.pool, zhuff.zhuff_decode_tree.__sizeof__())
if not lit_dt or not dist_dt:
if lit_dt: zdef.zdef_free(self.pool, lit_dt)
if dist_dt: zdef.zdef_free(self.pool, dist_dt)
@@ -435,7 +435,7 @@ class zinflate_stream:
zdef.zdef_free(self.pool, self)
def zinflate_one_shot(pool: memhub.MemManager | t.CPtr, data: UINT8PTR, length: t.CSizeT,
def zinflate_one_shot(pool: rawptr[memhub.MemManager], data: UINT8PTR, length: t.CSizeT,
wbits: t.CInt, bufsize: t.CSizeT, out_len: t.CSizeT | t.CPtr) -> t.CUInt8T | t.CPtr:
s: zinflate_stream | t.CPtr = zinflate_create(pool, wbits)
if not s: return None
@@ -446,7 +446,7 @@ def zinflate_one_shot(pool: memhub.MemManager | t.CPtr, data: UINT8PTR, length:
return out
def zinflate_create(pool: memhub.MemManager | t.CPtr, wbits: t.CInt) -> zinflate_stream | t.CPtr:
def zinflate_create(pool: rawptr[memhub.MemManager], wbits: t.CInt) -> zinflate_stream | t.CPtr:
s: zinflate_stream | t.CPtr = zdef.zdef_alloc(pool, zinflate_stream.__sizeof__())
if not s: return None
string.memset(s, 0, zinflate_stream.__sizeof__())