3d7bd70355
- 新增 Alembic 初始迁移、领域明细模型及归档持久化与重试链路\n- 拆分视频会话注册表、分类处理、推理时间窗聚合与流处理\n- 消耗日志：TSV/Markdown 含 top2/top3；item_id 优先产品编码；待确认记「待确认」行，语音确认后落正式行并更新汇总\n- 待确认时内存/DB 明细为占位行，确认后替换；拒绝时移除占位\n- 分类 probs 先 detach/cpu 再转 NumPy，修复 MPS/CUDA 上推理被静默跳过\n- 补充集成测试、归档与设备张量等单测 Made-with: Cursor
37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
"""_probs_data_to_numpy1d：CPU / CUDA / MPS 上均能离设备再转 NumPy。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
torch = pytest.importorskip("torch")
|
||
|
||
from app.services.consumable_vision_algorithm import _probs_data_to_numpy1d
|
||
|
||
|
||
def test_probs_numpy_cpu_tensor() -> None:
|
||
t = torch.tensor([0.1, 0.3, 0.6], dtype=torch.float32)
|
||
arr = _probs_data_to_numpy1d(t)
|
||
assert arr.dtype == np.float64
|
||
np.testing.assert_allclose(arr, [0.1, 0.3, 0.6], rtol=1e-5)
|
||
|
||
|
||
@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA 不可用，跳过设备张量用例")
|
||
def test_probs_numpy_cuda_tensor() -> None:
|
||
t = torch.tensor([0.0, 1.0], dtype=torch.float32, device="cuda")
|
||
arr = _probs_data_to_numpy1d(t)
|
||
assert arr.dtype == np.float64
|
||
np.testing.assert_allclose(arr, [0.0, 1.0], rtol=1e-5)
|
||
|
||
|
||
@pytest.mark.skipif(
|
||
not hasattr(torch.backends, "mps") or not torch.backends.mps.is_available(),
|
||
reason="MPS 不可用，跳过设备张量用例",
|
||
)
|
||
def test_probs_numpy_mps_tensor() -> None:
|
||
t = torch.tensor([0.25, 0.75], dtype=torch.float32, device="mps")
|
||
arr = _probs_data_to_numpy1d(t)
|
||
assert arr.dtype == np.float64
|
||
np.testing.assert_allclose(arr, [0.25, 0.75], rtol=1e-5)
|