linaro
211341cbbb
- init 新增 yolo26/resnet/arcface/lprnet/face-recognition 模板（多模型人脸检测+识别） - 新增 embedding/text 输出类型与模板资源（COCO、字符表等） - 新增 algpkg download-models：从 Gitea LFS 下载 models/BM1688 并做 SHA256 校验 - 补充单元测试与 README
825 lines
36 KiB
Python
825 lines
36 KiB
Python
"""SDK 自测：清单校验 / 打包往返 / 装载策略 / 本地测试台（无需 worker）。
|
||
|
||
有设备的用例标 `@pytest.mark.device`（无 sophon.sail 时自动跳过）：
|
||
|
||
uv run pytest # 静态用例
|
||
uv run pytest -m device # 设备上跑推理
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import http.server
|
||
import importlib.util
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import types
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
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_init_yolo26_template(tmp_path: Path) -> None:
|
||
"""`--algorithm yolo26` 生成可直接运行的检测包（不是 TODO 占位）。"""
|
||
src = tmp_path / "yolo.demo"
|
||
assert cli_main(["init", str(src), "--id", "yolo.demo", "--name", "YOLO26 示例",
|
||
"--algorithm", "yolo26"]) == 0
|
||
manifest = Manifest.parse(json.loads((src / "manifest.json").read_text()))
|
||
assert manifest.id == "yolo.demo" and manifest.kind == "detector"
|
||
assert manifest.default_model().name == "yolo26s_fp16_1b"
|
||
assert manifest.default_model().file == "models/yolo26s_fp16_1b.bmodel"
|
||
assert manifest.output["format"] == "N6"
|
||
|
||
code = (src / manifest.code_package / "__init__.py").read_text()
|
||
compile(code, "alg_yolo_demo/__init__.py", "exec") # 语法可编译（不 import 设备库）
|
||
assert "class Algorithm" in code and "_preprocess" in code and "_postprocess" in code
|
||
assert "TODO" not in code
|
||
|
||
labels = [line for line in (src / "assets" / "labels.txt").read_text().splitlines()
|
||
if line.strip()]
|
||
assert len(labels) == 80 and labels[0] == "person" and labels[-1] == "toothbrush"
|
||
|
||
model = tmp_path / "fake.bmodel"
|
||
model.write_bytes(b"FAKE-BMODEL" * 512)
|
||
out = tmp_path / "yolo.demo-1.0.0.mbank"
|
||
assert cli_main(["build", str(src), "--out", str(out),
|
||
"--model", f"yolo26s_fp16_1b={model}", "--quiet"]) == 0
|
||
assert cli_main(["verify", str(out), "--json"]) == 0
|
||
|
||
|
||
def test_cli_init_yolo26_alias_and_unknown_template(tmp_path: Path) -> None:
|
||
src = tmp_path / "alias.demo"
|
||
assert cli_main(["init", str(src), "--id", "alias.demo", "--name", "别名",
|
||
"--template", "yolov26"]) == 0
|
||
manifest = Manifest.parse(json.loads((src / "manifest.json").read_text()))
|
||
assert manifest.default_model().name == "yolo26s_fp16_1b"
|
||
|
||
bad = tmp_path / "bad.demo"
|
||
assert cli_main(["init", str(bad), "--id", "bad.demo", "--name", "未知",
|
||
"--algorithm", "nope"]) == 2
|
||
assert not (bad / "manifest.json").exists()
|
||
|
||
|
||
def test_cli_init_resnet_template(tmp_path: Path) -> None:
|
||
"""`--algorithm resnet` 生成 ResNet 分类包（ImageNet 归一化 + softmax + 整帧框）。"""
|
||
src = tmp_path / "resnet.demo"
|
||
assert cli_main(["init", str(src), "--id", "resnet.demo", "--name", "ResNet 示例",
|
||
"--algorithm", "resnet"]) == 0
|
||
manifest = Manifest.parse(json.loads((src / "manifest.json").read_text()))
|
||
assert manifest.id == "resnet.demo" and manifest.kind == "classifier"
|
||
assert manifest.default_model().name == "resnet50_fp32_1b"
|
||
assert manifest.default_model().file == "models/resnet50_fp32_1b.bmodel"
|
||
assert manifest.input["mean"] == [0.485, 0.456, 0.406]
|
||
assert manifest.schema_defaults()["conf"] == 0.1
|
||
|
||
code = (src / manifest.code_package / "__init__.py").read_text()
|
||
compile(code, "alg_resnet_demo/__init__.py", "exec")
|
||
assert "class Algorithm" in code and "_preprocess" in code
|
||
assert "_postprocess" in code and "yuv2bgr" in code
|
||
|
||
labels = [line for line in (src / "assets" / "labels.txt").read_text().splitlines()
|
||
if line.strip() and not line.startswith("#")]
|
||
assert len(labels) == 1000 and labels[0] == "class_0" and labels[-1] == "class_999"
|
||
|
||
model = tmp_path / "fake.bmodel"
|
||
model.write_bytes(b"FAKE-BMODEL" * 512)
|
||
out = tmp_path / "resnet.demo-1.0.0.mbank"
|
||
assert cli_main(["build", str(src), "--out", str(out),
|
||
"--model", f"resnet50_fp32_1b={model}", "--quiet"]) == 0
|
||
assert cli_main(["verify", str(out), "--json"]) == 0
|
||
|
||
|
||
def test_resnet_template_infer_with_fake_sail(tmp_path: Path, monkeypatch) -> None:
|
||
"""不依赖设备验证分类后处理：top-1 → 整帧框（score/cls）与阈值过滤。"""
|
||
src = tmp_path / "resnet.infer"
|
||
cli_main(["init", str(src), "--id", "resnet.infer", "--name", "推理示例",
|
||
"--algorithm", "resnet"])
|
||
code_path = src / "alg_resnet_infer" / "__init__.py"
|
||
|
||
# 生成代码只在方法体内使用 sail；空模块足以导入（不需要真实设备）
|
||
monkeypatch.setitem(sys.modules, "sophon", types.ModuleType("sophon"))
|
||
monkeypatch.setitem(sys.modules, "sophon.sail", types.ModuleType("sophon.sail"))
|
||
spec = importlib.util.spec_from_file_location("alg_resnet_infer_under_test", code_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
class Frame:
|
||
def width(self) -> int:
|
||
return 640
|
||
|
||
def height(self) -> int:
|
||
return 480
|
||
|
||
class Tensor:
|
||
def asnumpy(self):
|
||
return [[0.1, 3.0, 0.2]] # logits: top-1 = 类别 1
|
||
|
||
class Net:
|
||
def process(self, *args, **kwargs) -> None:
|
||
return None
|
||
|
||
class Bmcv:
|
||
def bm_image_to_tensor(self, *args, **kwargs) -> None:
|
||
return None
|
||
|
||
algo = module.Algorithm(ctx=None)
|
||
algo._conf = 0.5
|
||
algo._labels = ["cat", "dog", "bird"]
|
||
algo._output_scale = 1.0
|
||
algo._output_tensor = Tensor()
|
||
algo._output_tensors = {0: algo._output_tensor}
|
||
algo._input_tensor = object()
|
||
algo._input_name, algo._graph = "input", "graph"
|
||
algo._net = Net()
|
||
algo._bmcv = Bmcv()
|
||
algo._preprocess = lambda frame: object()
|
||
|
||
result = algo.infer(Frame())
|
||
assert result.boxes.shape == (1, 6)
|
||
assert (float(result.boxes[0, 0]), float(result.boxes[0, 2])) == (0.0, 639.0)
|
||
assert int(result.boxes[0, 5]) == 1
|
||
assert float(result.boxes[0, 4]) > 0.8
|
||
assert result.counts == {"dog": 1} and result.value == 1.0
|
||
assert result.meta["class_id"] == 1
|
||
|
||
algo._conf = 0.99 # 低于阈值 → 空结果，不上报告警
|
||
assert algo.infer(Frame()).boxes.shape == (0, 6)
|
||
|
||
|
||
def test_cli_init_arcface_template(tmp_path: Path) -> None:
|
||
"""`--algorithm arcface` 生成 ArcFace 特征包（embedding 输出、无类别表）。"""
|
||
src = tmp_path / "arcface.demo"
|
||
assert cli_main(["init", str(src), "--id", "arcface.demo", "--name", "ArcFace 示例",
|
||
"--algorithm", "arcface"]) == 0
|
||
manifest = Manifest.parse(json.loads((src / "manifest.json").read_text()))
|
||
assert manifest.id == "arcface.demo" and manifest.kind == "classifier"
|
||
assert manifest.default_model().name == "arcface_resnet50_fp32_1b"
|
||
assert manifest.default_model().file == "models/arcface_resnet50_fp32_1b.bmodel"
|
||
assert manifest.output["type"] == "embedding"
|
||
assert manifest.output["embedding_size"] == 512
|
||
assert manifest.schema_defaults()["interval"] == 0.5
|
||
|
||
code = (src / manifest.code_package / "__init__.py").read_text()
|
||
compile(code, "alg_arcface_demo/__init__.py", "exec")
|
||
assert "class Algorithm" in code and "_preprocess" in code
|
||
assert "meta" in code and "linalg.norm" in code
|
||
|
||
labels = [line for line in (src / "assets" / "labels.txt").read_text().splitlines()
|
||
if line.strip() and not line.startswith("#")]
|
||
assert labels == [] # 特征包不需要类别表
|
||
|
||
model = tmp_path / "fake.bmodel"
|
||
model.write_bytes(b"FAKE-BMODEL" * 512)
|
||
out = tmp_path / "arcface.demo-1.0.0.mbank"
|
||
assert cli_main(["build", str(src), "--out", str(out),
|
||
"--model", f"arcface_resnet50_fp32_1b={model}", "--quiet"]) == 0
|
||
assert cli_main(["verify", str(out), "--json"]) == 0
|
||
|
||
|
||
def test_arcface_template_infer_with_fake_sail(tmp_path: Path, monkeypatch) -> None:
|
||
"""不依赖设备验证特征提取：L2 归一化、512 维 embedding、face 计数。"""
|
||
src = tmp_path / "arcface.infer"
|
||
cli_main(["init", str(src), "--id", "arcface.infer", "--name", "特征示例",
|
||
"--algorithm", "arcface"])
|
||
code_path = src / "alg_arcface_infer" / "__init__.py"
|
||
|
||
monkeypatch.setitem(sys.modules, "sophon", types.ModuleType("sophon"))
|
||
monkeypatch.setitem(sys.modules, "sophon.sail", types.ModuleType("sophon.sail"))
|
||
spec = importlib.util.spec_from_file_location("alg_arcface_infer_under_test", code_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
class Frame:
|
||
def width(self) -> int:
|
||
return 112
|
||
|
||
def height(self) -> int:
|
||
return 112
|
||
|
||
class Tensor:
|
||
def asnumpy(self):
|
||
return [[3.0, 4.0] + [0.0] * 510] # 归一化后前两维 = 0.6, 0.8
|
||
|
||
class Net:
|
||
def process(self, *args, **kwargs) -> None:
|
||
return None
|
||
|
||
class Bmcv:
|
||
def bm_image_to_tensor(self, *args, **kwargs) -> None:
|
||
return None
|
||
|
||
algo = module.Algorithm(ctx=None)
|
||
algo._output_tensor = Tensor()
|
||
algo._input_tensor = object()
|
||
algo._input_name, algo._input_shapes = "input", {}
|
||
algo._output_name, algo._graph = "output", "graph"
|
||
algo._net, algo._bmcv = Net(), Bmcv()
|
||
algo._preprocess = lambda frame: object()
|
||
|
||
result = algo.infer(Frame())
|
||
assert result.boxes is None
|
||
assert result.counts == {"face": 1} and result.value == 1.0
|
||
assert result.meta["dim"] == 512
|
||
assert result.meta["embedding"][0] == pytest.approx(0.6)
|
||
assert result.meta["embedding"][1] == pytest.approx(0.8)
|
||
assert result.meta["norm"] == pytest.approx(1.0)
|
||
|
||
|
||
def test_cli_init_lprnet_template(tmp_path: Path) -> None:
|
||
"""`--algorithm lprnet` 生成车牌识别包（text 输出 + 字符表）。"""
|
||
src = tmp_path / "lprnet.demo"
|
||
assert cli_main(["init", str(src), "--id", "lprnet.demo", "--name", "车牌示例",
|
||
"--algorithm", "lprnet"]) == 0
|
||
manifest = Manifest.parse(json.loads((src / "manifest.json").read_text()))
|
||
assert manifest.id == "lprnet.demo" and manifest.kind == "classifier"
|
||
assert manifest.default_model().name == "lprnet_fp32_1b"
|
||
assert manifest.default_model().file == "models/lprnet_fp32_1b.bmodel"
|
||
assert manifest.default_model().resolution == [94, 24]
|
||
assert manifest.output["type"] == "text"
|
||
assert manifest.output["chars_file"] == "assets/chars.txt"
|
||
assert manifest.schema_defaults()["interval"] == 0.5
|
||
|
||
code = (src / manifest.code_package / "__init__.py").read_text()
|
||
compile(code, "alg_lprnet_demo/__init__.py", "exec")
|
||
assert "_preprocess" in code and "_postprocess" in code and "chars.txt" in code
|
||
|
||
chars = [line for line in (src / "assets" / "chars.txt").read_text().splitlines()
|
||
if line.strip() and not line.startswith("#")]
|
||
assert len(chars) == 68 and chars[0] == "京" and chars[-1] == "-"
|
||
|
||
model = tmp_path / "fake.bmodel"
|
||
model.write_bytes(b"FAKE-BMODEL" * 512)
|
||
out = tmp_path / "lprnet.demo-1.0.0.mbank"
|
||
assert cli_main(["build", str(src), "--out", str(out),
|
||
"--model", f"lprnet_fp32_1b={model}", "--quiet"]) == 0
|
||
assert cli_main(["verify", str(out), "--json"]) == 0
|
||
|
||
|
||
def test_lprnet_template_infer_with_fake_sail(tmp_path: Path, monkeypatch) -> None:
|
||
"""不依赖设备验证 CTC 贪心解码：去重、去 blank、空结果。"""
|
||
src = tmp_path / "lprnet.infer"
|
||
cli_main(["init", str(src), "--id", "lprnet.infer", "--name", "车牌推理",
|
||
"--algorithm", "lprnet"])
|
||
code_path = src / "alg_lprnet_infer" / "__init__.py"
|
||
|
||
monkeypatch.setitem(sys.modules, "sophon", types.ModuleType("sophon"))
|
||
monkeypatch.setitem(sys.modules, "sophon.sail", types.ModuleType("sophon.sail"))
|
||
spec = importlib.util.spec_from_file_location("alg_lprnet_infer_under_test", code_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
class Frame:
|
||
def width(self) -> int:
|
||
return 94
|
||
|
||
def height(self) -> int:
|
||
return 24
|
||
|
||
class Tensor:
|
||
def __init__(self) -> None:
|
||
self.logits = np.full((1, 4, 6), -10.0, dtype=np.float32)
|
||
|
||
def asnumpy(self):
|
||
return self.logits
|
||
|
||
class Net:
|
||
def process(self, *args, **kwargs) -> None:
|
||
return None
|
||
|
||
class Bmcv:
|
||
def bm_image_to_tensor(self, *args, **kwargs) -> None:
|
||
return None
|
||
|
||
tensor = Tensor()
|
||
algo = module.Algorithm(ctx=None)
|
||
algo._chars = ["京", "A", "1", "-"]
|
||
algo._blank = 3
|
||
algo._output_tensor = tensor
|
||
algo._input_tensor = object()
|
||
algo._input_name, algo._input_shapes = "input", {}
|
||
algo._output_name, algo._graph = "output", "graph"
|
||
algo._net, algo._bmcv = Net(), Bmcv()
|
||
algo._preprocess = lambda frame: object()
|
||
|
||
# 模型输出 [1, C=4, T=6]：0,0,3(blank),1,2,2 → CTC 去重后 "京A1"
|
||
for t, cls in enumerate([0, 0, 3, 1, 2, 2]):
|
||
tensor.logits[0, cls, t] = 10.0
|
||
result = algo.infer(Frame())
|
||
assert result.boxes is None
|
||
assert result.meta["text"] == "京A1"
|
||
assert result.meta["chars"] == [0, 1, 2]
|
||
assert result.counts == {"plate": 1} and result.value == 1.0
|
||
|
||
tensor.logits = np.full((1, 4, 6), -10.0, dtype=np.float32) # 全是 blank
|
||
tensor.logits[0, 3, :] = 10.0
|
||
empty = algo.infer(Frame())
|
||
assert empty.meta["text"] == "" and empty.meta["chars"] == []
|
||
assert empty.counts == {} and empty.value == 0.0
|
||
|
||
|
||
def test_cli_init_face_recognition_template(tmp_path: Path, monkeypatch) -> None:
|
||
"""`--algorithm face-recognition` 生成双模型包（检测框 + 512 维 embedding）。"""
|
||
src = tmp_path / "face.demo"
|
||
assert cli_main(["init", str(src), "--id", "face.demo", "--name", "人脸示例",
|
||
"--algorithm", "face-recognition"]) == 0
|
||
manifest = Manifest.parse(json.loads((src / "manifest.json").read_text()))
|
||
assert manifest.id == "face.demo" and manifest.kind == "detector"
|
||
assert [m.name for m in manifest.models] == ["face_detector", "arcface"]
|
||
assert manifest.default_model().name == "face_detector"
|
||
assert manifest.model("arcface").file == "models/arcface_resnet50_fp32_1b.bmodel"
|
||
assert manifest.output["embedding_size"] == 512
|
||
assert manifest.schema_defaults()["nms"] == 0.3
|
||
assert manifest.schema_defaults()["min_face"] == 16
|
||
|
||
code = (src / manifest.code_package / "__init__.py").read_text()
|
||
compile(code, "alg_face_demo/__init__.py", "exec")
|
||
assert "_priorbox" in code and "_detect" in code and "_embed" in code
|
||
assert 'model_bytes("arcface")' in code
|
||
|
||
labels = [line for line in (src / "assets" / "labels.txt").read_text().splitlines()
|
||
if line.strip() and not line.startswith("#")]
|
||
assert labels == ["face"]
|
||
|
||
det = tmp_path / "det.bmodel"
|
||
det.write_bytes(b"DET" * 2048)
|
||
arc = tmp_path / "arc.bmodel"
|
||
arc.write_bytes(b"ARC" * 2048)
|
||
out = tmp_path / "face.demo-1.0.0.mbank"
|
||
assert cli_main(["build", str(src), "--out", str(out),
|
||
"--model", f"face_detector={det}",
|
||
"--model", f"arcface={arc}", "--quiet"]) == 0
|
||
assert cli_main(["verify", str(out), "--json"]) == 0
|
||
|
||
# 不依赖设备验证先验框/解码/NMS 的集成（官方 RetinaFace 口径）
|
||
monkeypatch.setitem(sys.modules, "sophon", types.ModuleType("sophon"))
|
||
monkeypatch.setitem(sys.modules, "sophon.sail", types.ModuleType("sophon.sail"))
|
||
spec = importlib.util.spec_from_file_location(
|
||
"alg_face_detect_under_test", src / "alg_face_demo" / "__init__.py")
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
|
||
priors = module._priorbox(640, 640)
|
||
assert priors.shape == (16800, 4)
|
||
assert np.allclose(priors[0], [4 / 640, 4 / 640, 16 / 640, 16 / 640])
|
||
|
||
algo = module.Algorithm(ctx=None)
|
||
algo._conf, algo._nms, algo._max_faces = 0.5, 0.3, 20
|
||
algo._priors = priors
|
||
algo._det_w = algo._det_h = 640
|
||
loc = np.zeros((1, 16800, 4), dtype=np.float32)
|
||
conf = np.zeros((1, 16800, 2), dtype=np.float32)
|
||
conf[0, :, 1] = 0.01
|
||
conf[0, 0, 1] = 0.9
|
||
dets = algo._detect([loc, conf], 1.0, 640, 640)
|
||
assert dets.shape == (1, 6)
|
||
assert np.allclose(dets[0, :4], [0, 0, 12, 12])
|
||
assert dets[0, 4] == pytest.approx(0.9) and dets[0, 5] == 0
|
||
|
||
|
||
def test_model_download_specs_and_aliases() -> None:
|
||
from algpkg_sdk import models as model_store
|
||
|
||
assert [m.key for m in model_store.resolve_models(["all"])] == list(model_store.MODEL_FILES)
|
||
assert model_store.resolve_models(["yolov26"])[0].key == "yolo26"
|
||
assert model_store.resolve_models(["lpr"])[0].key == "lprnet"
|
||
assert model_store.lfs_batch_url("http://host/repo.git").endswith(
|
||
"/repo.git/info/lfs/objects/batch")
|
||
with pytest.raises(ValueError):
|
||
model_store.resolve_models(["nope"])
|
||
|
||
|
||
def test_cli_download_models_via_lfs(tmp_path: Path, monkeypatch, capsys) -> None:
|
||
"""本地假 LFS 服务验证下载/校验/跳过（不依赖外网与 git-lfs）。"""
|
||
from algpkg_sdk import models as model_store
|
||
|
||
content = b"FAKE-BMODEL-DOWNLOAD" * 4096
|
||
oid = hashlib.sha256(content).hexdigest()
|
||
spec = model_store.ModelFile("demo", "demo_model.bmodel", oid, len(content), "测试模型")
|
||
monkeypatch.setattr(model_store, "MODEL_FILES", {"demo": spec})
|
||
monkeypatch.setattr(model_store, "ALIASES", {"demo": "demo"})
|
||
monkeypatch.setenv("NO_PROXY", "127.0.0.1,localhost")
|
||
|
||
class Handler(http.server.BaseHTTPRequestHandler):
|
||
def do_POST(self) -> None: # LFS batch
|
||
length = int(self.headers.get("Content-Length") or 0)
|
||
payload = json.loads(self.rfile.read(length) or b"{}")
|
||
port = self.server.server_port
|
||
objects = [{
|
||
"oid": item["oid"], "size": item["size"],
|
||
"actions": {"download": {
|
||
"href": f"http://127.0.0.1:{port}/objects/{item['oid']}"}},
|
||
} for item in payload.get("objects", [])]
|
||
body = json.dumps({"objects": objects}).encode()
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "application/vnd.git-lfs+json")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_GET(self) -> None: # LFS object
|
||
if self.path.startswith("/objects/"):
|
||
self.send_response(200)
|
||
self.send_header("Content-Length", str(len(content)))
|
||
self.end_headers()
|
||
self.wfile.write(content)
|
||
else:
|
||
self.send_error(404)
|
||
|
||
def log_message(self, *args) -> None:
|
||
return None
|
||
|
||
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
try:
|
||
dest = tmp_path / "models" / "BM1688"
|
||
url = f"http://127.0.0.1:{server.server_port}/sophon-edge/algpkg-sdk.git"
|
||
assert cli_main(["download-models", "--dest", str(dest),
|
||
"--url", url, "--model", "demo"]) == 0
|
||
assert (dest / "demo_model.bmodel").read_bytes() == content
|
||
assert "下载 1" in capsys.readouterr().out
|
||
|
||
assert cli_main(["download-models", "--dest", str(dest),
|
||
"--url", url, "--model", "demo"]) == 0
|
||
assert "跳过 1" in capsys.readouterr().out
|
||
assert (dest / "demo_model.bmodel").read_bytes() == content
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
thread.join(timeout=5)
|
||
|
||
|
||
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
|