orisys-sdk/build/common/verify/wheel.py

"""Isolated venv smoke tests for packaged orisys wheels.

Three levels of confidence, all in a throwaway virtualenv so the active
environment is never touched:

* :func:`verify_built_wheel` - the public API imports (fast, original check);
* :func:`verify_wheel_assets` - bundled configs/sensors resolve and load via
  ``importlib.resources``;
* :func:`verify_wheel_runtime` - the compiled wheel actually processes real
  frames through the public pipeline with no camera attached.

Every level can be bootstrapped from another interpreter so each supported
CPython ABI can be verified against its own wheel.
"""

from __future__ import annotations

import os
import subprocess
import sys
import tempfile
import venv
from contextlib import contextmanager
from pathlib import Path

# Public SDK surface that must import from a released wheel.
_SMOKE_IMPORT_SCRIPT = """
import orisys
from orisys import Sensor, list_config, export_config, util, finger
from orisys.finger import UVMeshViewer, build_uv_surface_mesh, FingerRuntime
from orisys.finger.mesh import UVSurfaceMesh
from orisys.finger.viewer import init_uv_mesh_viewer
from orisys.hhd import nHHD
from orisys.tracker import Tracker
from orisys import capture
from orisys.Stitcher import Stitcher

assert orisys.__version__
assert Sensor is not None
assert callable(list_config)
assert callable(export_config)
assert util is not None
assert finger is not None
assert UVMeshViewer is not None
assert build_uv_surface_mesh is not None
assert FingerRuntime is not None
assert UVSurfaceMesh is not None
assert init_uv_mesh_viewer is not None
assert nHHD is not None
assert Tracker is not None
assert capture is not None
assert Stitcher is not None
print(f"orisys {orisys.__version__} public API imports OK")
"""

# Bundled data that must be resolvable and loadable from inside site-packages.
# The wheel ships the compiled package, so these paths only exist if the bundle
# step actually ran for this build.
_ASSET_SCRIPT = r"""
import sys
from importlib.metadata import distribution
from pathlib import Path

import numpy as np
import orisys

# Locate bundled data through the installed distribution, NOT through
# orisys.__file__ / importlib.resources. Nuitka compiles the package into one
# extension module whose __init__ installs nuitka_module_loader, which reports a
# synthetic __file__ (e.g. ...\site-packages\orisys\orisys\__init__.py) and a
# synthetic __path__ ending in '\not_existing'. Neither points at the real data
# directory, so importlib.resources silently finds nothing in a Nuitka wheel.
dist = distribution("orisys")
dist_root = Path(dist.locate_file("")).resolve()
package_root = Path(dist.locate_file("orisys")).resolve()

_diagnosis = (
    f"orisys distribution resolved to {dist_root} (sys.prefix={sys.prefix!r}, "
    f"sys.executable={sys.executable!r}), outside the verification venv. A stale "
    "orisys in the parent environment is visible through --system-site-packages; "
    "run `pip uninstall -y orisys` there, or re-run without --offline."
)
assert str(Path(sys.prefix).resolve()) in str(dist_root), _diagnosis
assert package_root.is_dir(), f"package directory not found: {package_root}"

required = [
    "configs/box.json",
    "configs/finger.json",
    "sensors/box/force_calibration.json",
    "sensors/box/sensor.json",
    "sensors/finger/manifest.json",
    "sensors/finger/sensor.json",
    "sensors/mini/sensor.json",
    "sensors/finger/assets/30fps-0703/mesh.npz",
    "sensors/finger/assets/30fps-0703/camera_intrinsics.npz",
    "sensors/finger/assets/30fps-0703/world_to_cam.npy",
    "sensors/finger/assets/visualization/mesh.npz",
    "sensors/finger/assets/visualization/world_to_cam.npy",
]

missing = [name for name in required if not (package_root / name).is_file()]
assert not missing, (
    f"bundled assets missing from the wheel: {missing}\n"
    f"package root: {package_root}\ninstalled tree: "
    f"{sorted(p.name for p in package_root.iterdir()) if package_root.is_dir() else '<missing>'}"
)

for name in required:
    path = package_root / name
    if name.endswith(".npz"):
        data = np.load(str(path))
        assert len(data.files) > 0, f"{name} is an empty npz"
    elif name.endswith(".npy"):
        arr = np.load(str(path))
        assert arr.size > 0, f"{name} is an empty npy"

configs = orisys.list_config()
assert configs, "orisys.list_config() returned nothing"

print(f"bundled assets OK ({len(required)} files, {len(configs)} configs) "
      f"from {package_root}")
"""

# End-to-end pipeline run against a video file: no camera, no GUI, no imshow.
_RUNTIME_SCRIPT = r"""
import json
import os
import sys
from importlib.metadata import distribution
from pathlib import Path

import numpy as np

os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")

import orisys

# Provenance via the installed distribution: a Nuitka-compiled package reports a
# synthetic orisys.__file__ (see the asset check) and a stale orisys visible
# through --system-site-packages would otherwise satisfy this check.
dist = distribution("orisys")
dist_root = Path(dist.locate_file("")).resolve()
assert str(Path(sys.prefix).resolve()) in str(dist_root), (
    f"the installed orisys came from {dist_root}, outside the verification venv "
    f"{sys.prefix!r}; run `pip uninstall -y orisys` in the parent environment or "
    "re-run without --offline."
)

video, config_name, frames = sys.argv[1], sys.argv[2], int(sys.argv[3])

sensor = orisys.Sensor(
    vid_src=video,
    config_name=config_name,
    backend="cpu",
    verbose=False,
)
processed = 0
try:
    for _ in range(frames):
        img = sensor.get_img()
        assert img is not None, "get_img returned None"
        assert isinstance(img, np.ndarray) and img.ndim in (2, 3)

        sensor.compute_deformation(check_motion=False)
        flow, vnormal, vshear = sensor.read_info(
            sensor.info.VRAW, sensor.info.VNORMAL, sensor.info.VSHEAR
        )
        for name, array in (("flow", flow), ("vnormal", vnormal), ("vshear", vshear)):
            assert np.isfinite(array).all(), f"{name} contains NaN/Inf"

        contact = sensor.compute_contact()
        assert isinstance(contact, dict) and "centroid_found" in contact

        fnormal, fsx, fsy = sensor.read_info(
            sensor.info.FNORMAL, sensor.info.FSHEARX, sensor.info.FSHEARY
        )
        assert np.isfinite([fnormal, fsx, fsy]).all(), "force output contains NaN/Inf"
        processed += 1
finally:
    sensor.disconnect()

assert processed == frames, f"only processed {processed}/{frames} frames"
print(json.dumps({
    "version": orisys.__version__,
    "frames": processed,
    "flow_shape": list(flow.shape),
    "video": video,
    "config": config_name,
    "package_root": str(Path(dist.locate_file("orisys")).resolve()),
}))
"""


# Default runtime-verification inputs. Overridable by CLI flags or environment.
DEFAULT_RUNTIME_CONFIG = os.getenv("ORISYS_VERIFY_CONFIG", "box")
# Clips tracked in the repository, smallest first.
DEFAULT_VIDEO_CANDIDATES = (
    Path("data") / "sample" / "s1_480p.mp4",
    Path("data") / "sample" / "s1_720p.mp4",
    Path("tests") / "idle_1min.mp4",
)


def default_verification_video(repo_root: Path) -> Path | None:
    """Resolve the clip used for the end-to-end runtime check, if any."""
    env_video = os.getenv("ORISYS_VERIFY_VIDEO")
    repo_root = Path(repo_root)
    if env_video:
        candidate = Path(env_video)
        return candidate if candidate.is_absolute() else repo_root / candidate
    for relative in DEFAULT_VIDEO_CANDIDATES:
        candidate = repo_root / relative
        if candidate.is_file():
            return candidate
    return None


def _venv_python(venv_dir: Path) -> Path:
    if sys.platform == "win32":
        return venv_dir / "Scripts" / "python.exe"
    return venv_dir / "bin" / "python"


def _hermetic_env() -> dict:
    """Environment for the verification venv with PYTHONPATH removed.

    An inherited PYTHONPATH=src would let the checkout satisfy the checks
    instead of the installed wheel, which is precisely the failure mode the
    provenance assertions in the scripts exist to catch.
    """
    return {key: value for key, value in os.environ.items() if key != "PYTHONPATH"}


def _create_venv(
    venv_dir: Path,
    *,
    python_executable: Path | None = None,
    system_site_packages: bool = False,
) -> Path:
    """Create a throwaway venv, optionally from a different interpreter."""
    if python_executable is None:
        venv.create(venv_dir, with_pip=True, system_site_packages=system_site_packages)
    else:
        command = [str(python_executable), "-m", "venv"]
        if system_site_packages:
            command.append("--system-site-packages")
        command.append(str(venv_dir))
        subprocess.check_call(command, env=_hermetic_env())
    return _venv_python(venv_dir)


def _install_wheel(
    python: Path, wheel_path: Path, *, offline: bool = False, env: dict | None = None
) -> None:
    """Install the wheel under test, guaranteeing it lands in the venv.

    ``--force-reinstall --no-deps`` is not optional: pip reports a same-version
    requirement as "already satisfied" and skips the file, which is fatal here
    because ``--offline`` also exposes the parent environment's site-packages.
    Without it the venv check can silently validate a *different* orisys than
    the one passed in (the scripts additionally assert provenance).
    """
    if offline:
        # Dependencies come from the parent environment via --system-site-packages.
        commands = [
            [
                str(python),
                "-m",
                "pip",
                "install",
                "-q",
                "--no-index",
                "--no-deps",
                "--force-reinstall",
                str(wheel_path),
            ]
        ]
    else:
        commands = [
            [str(python), "-m", "pip", "install", "-q", "--upgrade", "pip"],
            [str(python), "-m", "pip", "install", "-q", str(wheel_path)],
            [
                str(python),
                "-m",
                "pip",
                "install",
                "-q",
                "--force-reinstall",
                "--no-deps",
                str(wheel_path),
            ],
        ]

    for command in commands:
        subprocess.check_call(command, stdout=subprocess.DEVNULL, env=env)


class WheelVerificationError(RuntimeError):
    """A verification step failed inside the throwaway venv.

    Carries the child's stdout/stderr so the real traceback is visible instead of
    a bare non-zero exit code.
    """


def _run_script(
    python: Path, script: str, args: list[str] | None = None, *, env: dict
) -> str:
    """Run a verification script in an already-prepared venv."""
    completed = subprocess.run(
        [str(python), "-c", script, *(args or [])],
        capture_output=True,
        text=True,
        env=env,
        check=False,
    )
    if completed.returncode != 0:
        command = " ".join(str(part) for part in (args or []))
        raise WheelVerificationError(
            f"verification script failed in the venv (exit {completed.returncode}"
            + (f", args: {command}" if command else "")
            + ").\n--- stdout ---\n"
            + (completed.stdout.strip() or "<empty>")
            + "\n--- stderr ---\n"
            + (completed.stderr.strip() or "<empty>")
        )
    return completed.stdout.strip()


@contextmanager
def _installed_venv(
    wheel_path: Path,
    *,
    python_executable: Path | None = None,
    offline: bool = False,
    prefix: str = "orisys-wheel-verify-",
):
    """Create a venv with the wheel installed, for one or more checks.

    Creating a venv and installing the runtime dependencies is by far the most
    expensive part of verification (tens of seconds plus a large download when
    not using ``--offline``), so callers that run several checks must share one
    venv instead of building one per check.
    """
    wheel_path = wheel_path.resolve()
    env = _hermetic_env()
    with tempfile.TemporaryDirectory(prefix=prefix) as tmp:
        venv_dir = Path(tmp) / "venv"
        interpreter = python_executable or Path(sys.executable)
        print(
            f"  [{wheel_path.name}] creating venv from {interpreter.name}"
            + ("" if offline else " and installing dependencies (first run downloads packages)")
            + " ...",
            flush=True,
        )
        python = _create_venv(
            venv_dir,
            python_executable=python_executable,
            system_site_packages=offline,
        )
        _install_wheel(python, wheel_path, offline=offline, env=env)
        yield python, env


def verify_built_wheel(wheel_path: Path, *, python_executable: Path | None = None) -> None:
    """Install the wheel in a temporary venv and smoke-test public imports."""
    with _installed_venv(wheel_path, python_executable=python_executable) as (python, env):
        _run_script(python, _SMOKE_IMPORT_SCRIPT, env=env)


def verify_wheel_assets(wheel_path: Path, *, python_executable: Path | None = None) -> str:
    """Assert the wheel carries loadable bundled configs and sensor assets."""
    with _installed_venv(wheel_path, python_executable=python_executable) as (python, env):
        return _run_script(python, _ASSET_SCRIPT, env=env)


def verify_wheel_runtime(
    wheel_path: Path,
    *,
    video: Path,
    config_name: str,
    frames: int = 40,
    python_executable: Path | None = None,
    offline: bool = False,
) -> dict:
    """Run real frames through the installed wheel with no camera attached.

    This is the only check that proves the compiled extension plus the bundled
    sensor assets actually produce finite flow/force output end to end.
    """
    with _installed_venv(
        wheel_path, python_executable=python_executable, offline=offline
    ) as (python, env):
        return _runtime_check(python, env, video=video, config_name=config_name, frames=frames)


def _runtime_check(
    python: Path, env: dict, *, video: Path, config_name: str, frames: int
) -> dict:
    video = Path(video).resolve()
    if not video.is_file():
        raise FileNotFoundError(f"verification video not found: {video}")
    if frames < 1:
        raise ValueError("frames must be >= 1")

    print(f"  processing {frames} frame(s) with config {config_name!r} ...", flush=True)
    output = _run_script(
        python, _RUNTIME_SCRIPT, [str(video), config_name, str(frames)], env=env
    )
    import json

    return json.loads(output.splitlines()[-1])


def verify_wheel(
    wheel_path: Path,
    *,
    video: Path | None = None,
    config_name: str = DEFAULT_RUNTIME_CONFIG,
    frames: int = 40,
    python_executable: Path | None = None,
    offline: bool = False,
    check_assets: bool = True,
) -> dict:
    """Run every isolated-check level for one wheel inside a single venv.

    Use this instead of calling the individual ``verify_wheel_*`` helpers: it
    pays the venv/dependency cost once and returns the runtime record plus the
    asset and import summaries.
    """
    with _installed_venv(
        wheel_path, python_executable=python_executable, offline=offline
    ) as (python, env):
        imports = _run_script(python, _SMOKE_IMPORT_SCRIPT, env=env)
        record: dict = {"wheel": Path(wheel_path).name, "imports": imports}
        if check_assets:
            record["assets"] = _run_script(python, _ASSET_SCRIPT, env=env)
        else:
            record["assets"] = "asset check skipped (--no-assets)"
        if video is not None:
            record.update(
                _runtime_check(
                    python, env, video=video, config_name=config_name, frames=frames
                )
            )
        return record