linaro
170e4a10a4
本仓库原先内嵌在 sophon-edge 的 `algpkg-sdk/`（该仓库历史上的三次相关提交： e8101ba「算法开发能力独立成项目」、090c406「契约名落地为真实模块 + 多模型 ctx」、 b666161「sail 存根 + 编写风格修正」）。现拆出独立建库，便于算法作者单独 clone/迭代、 也便于单独发版（打成 wheel 分发给第三方或进 worker 交付包的 vendor/）。 - 内容：`src/algpkg_sdk/`（api/manifest/loader/builder/testkit/cli）+ 契约名模块 `src/video_algpkg_api/`、`examples/person-count`（完整示例）、`tests/`（pytest，无设备自动跳过）、 `README.md`（已改成独立仓库口径：命令在项目根执行，配套 worker 用路径依赖/`ALGPKG_SDK_DIR`）、 `pyproject.toml`（console script `algpkg`）、`uv.lock`； - 不含 `.venv/`（`.gitignore` 已覆盖，clone 后 `uv sync` 重建）。 配套 worker 仓库（sophon-edge）： `workers/video/pyproject.toml` 以路径依赖指向本仓库（`path = "../../../algpkg-sdk"`，可用 `ALGPKG_SDK_DIR` 覆盖）；交付打包 `deploy/pack_worker.py` 会把本仓库打成 wheel 放进 worker 包 `vendor/`。
389 lines
16 KiB
Python
389 lines
16 KiB
Python
"""SDK 自测：清单校验 / 打包往返 / 装载策略 / 本地测试台（无需 worker）。
|
||
|
||
有设备的用例标 `@pytest.mark.device`（无 sophon.sail 时自动跳过）：
|
||
|
||
uv run pytest # 静态用例
|
||
uv run pytest -m device # 设备上跑推理
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from algpkg_sdk import (AlgoBuilder, AlgoError, AlgoPackage, AlgoPackageError, Manifest,
|
||
ManifestError, install_sdk_alias, sha256_file)
|
||
from algpkg_sdk import testkit
|
||
from algpkg_sdk.cli import main as cli_main
|
||
|
||
EXAMPLES = Path(__file__).resolve().parents[1] / "examples"
|
||
|
||
|
||
def _manifest(**overrides) -> dict:
|
||
payload = {
|
||
"manifest_format": 1, "id": "t.demo", "name": "测试包", "version": "1.0.0",
|
||
"api_version": 1, "kind": "detector", "entry": "Algorithm",
|
||
"code_package": "alg_t_demo",
|
||
"models": [{"name": "default", "file": "models/t.bmodel", "precision": "fp16",
|
||
"resolution": [640, 640], "batch": 1, "default": True}],
|
||
"params_schema": [{"key": "conf", "type": "float", "default": 0.25,
|
||
"min": 0.01, "max": 0.99, "label": "置信度"}],
|
||
"defaults": {"conf": 0.25},
|
||
}
|
||
payload.update(overrides)
|
||
return payload
|
||
|
||
|
||
# ---------------------------------------------------------------------- 清单
|
||
|
||
def test_manifest_required_fields() -> None:
|
||
Manifest.parse(_manifest())
|
||
for missing in ("id", "version", "models"):
|
||
bad = _manifest()
|
||
bad.pop(missing)
|
||
with pytest.raises(ManifestError):
|
||
Manifest.parse(bad)
|
||
|
||
|
||
def test_manifest_rejects_bad_values() -> None:
|
||
with pytest.raises(ManifestError): # id 形态
|
||
Manifest.parse(_manifest(id="Bad Id!"))
|
||
with pytest.raises(ManifestError): # 模型路径越界
|
||
Manifest.parse(_manifest(models=[{"name": "d", "file": "../x.bmodel"}]))
|
||
with pytest.raises(ManifestError): # api_version 过高
|
||
Manifest.parse(_manifest(api_version=99))
|
||
with pytest.raises(ManifestError): # params_schema 重复键
|
||
Manifest.parse(_manifest(params_schema=[
|
||
{"key": "conf", "type": "float"}, {"key": "conf", "type": "float"}]))
|
||
|
||
|
||
def test_manifest_default_model_pick() -> None:
|
||
manifest = Manifest.parse(_manifest(models=[
|
||
{"name": "fp16", "file": "models/a.bmodel"},
|
||
{"name": "int8", "file": "models/b.bmodel", "default": True}]))
|
||
assert manifest.default_model().name == "int8"
|
||
assert manifest.model("fp16").file == "models/a.bmodel"
|
||
with pytest.raises(ManifestError):
|
||
manifest.model("nope")
|
||
|
||
|
||
def test_manifest_check_params_semantics() -> None:
|
||
manifest = Manifest.parse(_manifest())
|
||
merged = manifest.check_params({"conf": 0.5})
|
||
assert merged["conf"] == 0.5
|
||
assert manifest.check_params({})["conf"] == 0.25 # 补默认
|
||
with pytest.raises(ManifestError): # 越界拒绝（不静默截断）
|
||
manifest.check_params({"conf": 2.0})
|
||
with pytest.raises(ManifestError): # 类型不符
|
||
manifest.check_params({"conf": "abc"})
|
||
|
||
|
||
def test_manifest_legacy_and_entry_helpers() -> None:
|
||
manifest = Manifest.parse(_manifest(entry="sub:Alg", legacy={"algorithm": "yolo"}))
|
||
assert manifest.entry_module == "alg_t_demo.sub"
|
||
assert manifest.entry_class == "Alg"
|
||
assert manifest.legacy["algorithm"] == "yolo"
|
||
|
||
|
||
# ---------------------------------------------------------------------- 打包往返
|
||
|
||
def _make_source(root: Path, *, model_bytes: bytes = b"FAKE-BMODEL" * 512) -> Path:
|
||
src = root / "t.demo"
|
||
(src / "alg_t_demo").mkdir(parents=True)
|
||
(src / "alg_t_demo" / "__init__.py").write_text(
|
||
"from video_algpkg_api import Algorithm as Base, AlgoResult\n"
|
||
"class Algorithm(Base):\n"
|
||
" def load(self):\n"
|
||
" self._data = self.ctx.model_bytes()\n"
|
||
" self.ctx.logger.info('loaded %s bytes', len(self._data))\n"
|
||
" def infer(self, frame):\n"
|
||
" return AlgoResult(boxes=None, value=len(self._data))\n",
|
||
encoding="utf-8",
|
||
)
|
||
(src / "models").mkdir()
|
||
(src / "models" / "t.bmodel").write_bytes(model_bytes)
|
||
(src / "assets").mkdir()
|
||
(src / "assets" / "labels.txt").write_text("person\ncar\n", encoding="utf-8")
|
||
(src / "manifest.json").write_text(json.dumps(_manifest()), encoding="utf-8")
|
||
return src
|
||
|
||
|
||
def test_build_roundtrip_and_bytes_fidelity() -> None:
|
||
install_sdk_alias()
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
src = _make_source(root)
|
||
out = root / "t.demo-1.0.0.mbank"
|
||
summary = AlgoBuilder(src, verbose=False).build(out)
|
||
|
||
assert out.is_file() and summary["algo_id"] == "t.demo"
|
||
assert summary["sha256"] == sha256_file(out)
|
||
|
||
pkg = AlgoPackage(out)
|
||
pkg.open()
|
||
manifest = pkg.require_manifest()
|
||
assert manifest.version == "1.0.0"
|
||
data, meta = pkg.read_model("default")
|
||
assert data == (src / "models" / "t.bmodel").read_bytes() # 包内模型与源文件逐字节相同
|
||
assert meta["precision"] == "fp16"
|
||
assert pkg.read_asset("assets/labels.txt").decode().splitlines() == ["person", "car"]
|
||
files = {item["path"] for item in pkg.list_files()}
|
||
assert {"manifest.json", "assets/labels.txt", "models/t.bmodel"} <= files
|
||
pkg.close()
|
||
|
||
|
||
def test_build_creates_version_snapshot_and_signature() -> None:
|
||
from sqlite_vfs.core import SQLiteVFS
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
src = _make_source(root)
|
||
out = root / "signed.mbank"
|
||
AlgoBuilder(src, verbose=False).build(out, sign_key="KEY-1", sign_key_id="ci")
|
||
vfs = SQLiteVFS(str(out), readonly=True, check_same_thread=False)
|
||
try:
|
||
assert [s["name"] for s in vfs.list_snapshots()] == ["1.0.0"]
|
||
report = vfs.verify_signature("KEY-1")
|
||
assert report["ok"] and report["key_id"] == "ci"
|
||
assert not vfs.verify_signature("KEY-WRONG")["ok"]
|
||
finally:
|
||
vfs.close()
|
||
|
||
|
||
def test_build_reports_missing_model_and_code() -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
src = _make_source(root)
|
||
(src / "models" / "t.bmodel").unlink()
|
||
with pytest.raises(Exception) as exc:
|
||
AlgoBuilder(src, verbose=False).build(root / "x.mbank")
|
||
assert "模型" in str(exc.value)
|
||
|
||
src2 = _make_source(root / "other")
|
||
import shutil
|
||
shutil.rmtree(src2 / "alg_t_demo")
|
||
with pytest.raises(Exception) as exc2:
|
||
AlgoBuilder(src2, verbose=False).build(root / "y.mbank")
|
||
assert "代码目录" in str(exc2.value)
|
||
|
||
|
||
def test_encrypt_removes_plain_source() -> None:
|
||
from sqlite_vfs.core import SQLiteVFS
|
||
|
||
key = "uA9mYq0T2wGZ0dX1vJ8kL3pR5sN7bC4eF6hK2mP0qT8="
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
src = _make_source(root)
|
||
out = root / "enc.mbank"
|
||
AlgoBuilder(src, key=key, verbose=False).build(out)
|
||
vfs = SQLiteVFS(str(out), readonly=True, check_same_thread=False)
|
||
try:
|
||
files = [e["path"] for e in vfs.all_entries() if not e["is_directory"]]
|
||
finally:
|
||
vfs.close()
|
||
assert not any(f.endswith(".py") for f in files) # 明文源码不得留在包里
|
||
# 不带 key 打开结构没问题；装载（import）会因缺密钥失败
|
||
pkg = AlgoPackage(out)
|
||
pkg.open()
|
||
with pytest.raises(AlgoPackageError) as exc:
|
||
pkg.mount()
|
||
assert "密钥" in str(exc.value)
|
||
pkg.close()
|
||
|
||
|
||
def test_bad_file_rejected() -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
bad = Path(tmp) / "bad.mbank"
|
||
bad.write_bytes(b"not a database at all")
|
||
with pytest.raises(AlgoPackageError):
|
||
AlgoPackage(bad).open()
|
||
|
||
|
||
# ---------------------------------------------------------------------- 多模型包
|
||
|
||
def _make_multi_model_source(root: Path) -> Path:
|
||
"""一个包带两份模型（检测 + 二次判定）：代码按名字取第二份（多模型包的标准写法）。"""
|
||
src = root / "t.multi"
|
||
(src / "alg_t_multi").mkdir(parents=True)
|
||
(src / "alg_t_multi" / "__init__.py").write_text(
|
||
"from video_algpkg_api import Algorithm as Base, AlgoResult\n"
|
||
"class Algorithm(Base):\n"
|
||
" def load(self):\n"
|
||
" self._det = self.ctx.model_bytes() # 选中模型（默认检测器）\n"
|
||
" self._cls = self.ctx.model_bytes('fp_filter') # 按名字取第二份\n"
|
||
" def infer(self, frame):\n"
|
||
" return AlgoResult(boxes=None, value=len(self._det) + len(self._cls))\n",
|
||
encoding="utf-8",
|
||
)
|
||
(src / "models").mkdir()
|
||
(src / "models" / "det.bmodel").write_bytes(b"DET" * 256)
|
||
(src / "models" / "fp.bmodel").write_bytes(b"FPFILTER" * 128)
|
||
(src / "manifest.json").write_text(json.dumps(_manifest(
|
||
id="t.multi", name="多模型包", code_package="alg_t_multi",
|
||
models=[{"name": "detector", "file": "models/det.bmodel", "default": True},
|
||
{"name": "fp_filter", "file": "models/fp.bmodel", "precision": "int8"}])),
|
||
encoding="utf-8")
|
||
return src
|
||
|
||
|
||
def test_multi_model_package_context_access() -> None:
|
||
"""`ctx.model_bytes(name)` 按 manifest.models[] 取包内任意模型；不给名 = 选中模型（旧写法兼容）。"""
|
||
install_sdk_alias()
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp)
|
||
src = _make_multi_model_source(root)
|
||
out = root / "multi.mbank"
|
||
AlgoBuilder(src, verbose=False).build(out)
|
||
|
||
pkg = AlgoPackage(out)
|
||
pkg.open()
|
||
pkg.mount()
|
||
try:
|
||
instance = pkg.new_instance("detector", {})
|
||
ctx = instance.ctx
|
||
det = (src / "models" / "det.bmodel").read_bytes()
|
||
fp = (src / "models" / "fp.bmodel").read_bytes()
|
||
assert ctx.model_name == "detector"
|
||
assert ctx.model_bytes() == det # 选中模型（向后兼容）
|
||
assert ctx.model_bytes("fp_filter") == fp # 按 name 取第二份
|
||
assert ctx.model_bytes("models/fp.bmodel") == fp # 按 file 路径也认
|
||
assert [m["name"] for m in ctx.models] == ["detector", "fp_filter"]
|
||
assert ctx.model_spec("fp_filter")["precision"] == "int8"
|
||
assert ctx.model_spec()["name"] == "detector"
|
||
with pytest.raises(AlgoError) as exc:
|
||
ctx.model_bytes("nope")
|
||
assert "fp_filter" in str(exc.value) # 报错带可选清单
|
||
assert instance.infer(None).value == len(det) + len(fp) # 包内代码真读到两份不同字节
|
||
finally:
|
||
pkg.close()
|
||
|
||
|
||
# ---------------------------------------------------------------------- 契约名（video_algpkg_api）
|
||
|
||
def test_contract_module_importable_without_prior_registration() -> None:
|
||
"""`video_algpkg_api` 是随 SDK 分发的**真实模块**，不只是运行时注册的别名。
|
||
|
||
新解释器里**先** import 它也要成功（本地开发/IDE/类型检查的常见姿势），
|
||
且与 `algpkg_sdk.api` 是同一批对象（重导出，非副本）——否则 worker 侧
|
||
`isinstance(instance, Algorithm)` 会出现两个基类（"本地过、设备挂"）。
|
||
"""
|
||
script = (
|
||
"import video_algpkg_api as contract\n"
|
||
"assert contract.__file__.endswith('video_algpkg_api/__init__.py'), contract.__file__\n"
|
||
"import algpkg_sdk.api as api\n"
|
||
"for name in contract.__all__:\n"
|
||
" assert getattr(contract, name) is getattr(api, name), name\n"
|
||
"print('contract-ok')\n"
|
||
)
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
proc = subprocess.run([sys.executable, "-c", script], capture_output=True,
|
||
text=True, timeout=30, cwd=tmp)
|
||
assert proc.returncode == 0, proc.stderr
|
||
assert "contract-ok" in proc.stdout
|
||
|
||
|
||
def test_contract_alias_still_registered_to_api_module() -> None:
|
||
"""worker 侧注册语义不变：导入 `algpkg_sdk` 后，`video_algpkg_api` 解析到 api 模块本身。
|
||
|
||
（`install_sdk_alias` 用 setdefault：设备上 worker 先注册，真实模块被惰性遮蔽，
|
||
两条路径拿到的是同一批对象。）
|
||
"""
|
||
import video_algpkg_api as resolved
|
||
|
||
from algpkg_sdk import api
|
||
|
||
assert resolved is api
|
||
|
||
|
||
# ---------------------------------------------------------------------- CLI
|
||
|
||
def test_cli_init_build_inspect_check(tmp_path: Path) -> None:
|
||
src = tmp_path / "cli.demo"
|
||
assert cli_main(["init", str(src), "--id", "cli.demo", "--name", "CLI 示例"]) == 0
|
||
assert (src / "manifest.json").is_file()
|
||
code_pkg = next((src.glob("alg_cli_demo")), None)
|
||
assert code_pkg is not None and (code_pkg / "__init__.py").is_file()
|
||
|
||
model = tmp_path / "fake.bmodel"
|
||
model.write_bytes(b"FAKE" * 1024)
|
||
out = tmp_path / "cli.demo-1.0.0.mbank"
|
||
assert cli_main(["build", str(src), "--out", str(out), "--model", f"default={model}",
|
||
"--quiet"]) == 0
|
||
assert cli_main(["inspect", str(out), "--json"]) == 0
|
||
assert cli_main(["verify", str(out), "--json"]) == 0
|
||
assert cli_main(["list", str(tmp_path)]) == 0
|
||
|
||
|
||
def test_cli_check_reports_static_results(tmp_path: Path) -> None:
|
||
"""check 在无设备时也要给出"清单/结构 OK、试加载跳过"的诚实结论。"""
|
||
src = tmp_path / "chk.demo"
|
||
cli_main(["init", str(src), "--id", "chk.demo", "--name", "检查示例"])
|
||
model = tmp_path / "fake.bmodel"
|
||
model.write_bytes(b"FAKE" * 1024)
|
||
out = tmp_path / "chk.mbank"
|
||
cli_main(["build", str(src), "--out", str(out), "--model", f"default={model}", "--quiet"])
|
||
report = testkit.check(out)
|
||
names = {item["name"]: item for item in report["checks"]}
|
||
assert names["manifest 合法"]["ok"]
|
||
assert names["模型文件在包内"]["ok"]
|
||
if not testkit.has_device():
|
||
assert not names["代码挂载（试加载）"]["ok"] and "跳过" in names["代码挂载（试加载）"]["detail"]
|
||
|
||
|
||
def test_module_entry_point_help() -> None:
|
||
proc = subprocess.run([sys.executable, "-m", "algpkg_sdk.cli", "--help"],
|
||
capture_output=True, text=True, timeout=30)
|
||
assert proc.returncode == 0
|
||
assert "check" in proc.stdout and "run" in proc.stdout and "compare" in proc.stdout
|
||
|
||
|
||
# ---------------------------------------------------------------------- 示例包（真实结构）
|
||
|
||
def test_example_person_count_manifest() -> None:
|
||
example = EXAMPLES / "person-count"
|
||
manifest = Manifest.parse(json.loads((example / "manifest.json").read_text()))
|
||
assert manifest.id == "demo.person-count" and manifest.kind == "counter"
|
||
assert manifest.output["type"] == "counter"
|
||
assert (example / manifest.code_package).is_dir()
|
||
# 模型文件由打包时 --model 指定（示例目录不带 bmodel），这里只校验声明形态
|
||
assert manifest.default_model().file.startswith("models/")
|
||
|
||
|
||
# ---------------------------------------------------------------------- 设备用例
|
||
|
||
@pytest.mark.device
|
||
def test_check_deep_on_device() -> None:
|
||
"""设备：入口类可实例化（模型真加载）。
|
||
|
||
默认用**自包含**示例包（examples/person-count 打出来的包）；内置包（依赖 worker 内部模块）
|
||
属"特权包"，需额外传 `extra_path=[worker 工程目录]`。
|
||
"""
|
||
import os
|
||
|
||
pkg_path = os.environ.get("ALGPKG_DEVICE_PACKAGE")
|
||
if not pkg_path or not Path(pkg_path).is_file():
|
||
pytest.skip("未提供 ALGPKG_DEVICE_PACKAGE（指向含真实模型的 .mbank）")
|
||
extra = [p for p in (os.environ.get("ALGPKG_EXTRA_PATH") or "").split(":") if p]
|
||
report = testkit.check(pkg_path, deep=True, extra_path=extra)
|
||
assert report["ok"], report["checks"]
|
||
|
||
|
||
@pytest.mark.device
|
||
def test_run_on_device() -> None:
|
||
"""设备：跑一段视频，拿到逐帧检出与单帧耗时。"""
|
||
import os
|
||
|
||
pkg_path = os.environ.get("ALGPKG_DEVICE_PACKAGE")
|
||
clip = os.environ.get("ALGPKG_DEVICE_CLIP")
|
||
if not pkg_path or not clip:
|
||
pytest.skip("未提供 ALGPKG_DEVICE_PACKAGE / ALGPKG_DEVICE_CLIP")
|
||
extra = [p for p in (os.environ.get("ALGPKG_EXTRA_PATH") or "").split(":") if p]
|
||
report = testkit.run(pkg_path, source=clip, frames=3, warmup=1, extra_path=extra)
|
||
summary = report.summary()
|
||
assert summary["frames"] >= 1
|
||
assert summary["infer_ms"]["avg"] > 0
|