Source code for app.adapters.stub_backend

"""Stub full backend used by tests + ephemeral mode.

sfmapi ships no real SfM engine; the wire surface and orchestration
shell live in this repo, while engine implementations (pycolmap,
OpenSfM, hloc, custom forks) live in separate packages. This module
provides a no-op stub that satisfies the protocol structurally —
useful for:

* unit / integration / contract tests that exercise routing,
  scheduling, snapshot writing, etc., without needing a real engine,
* the ``SFMAPI_EPHEMERAL=true`` self-contained demo runtime, which
  registers it on lifespan startup,
* SDK live tests that boot the app in a subprocess.

Most operations raise :class:`CapabilityUnavailableError` — only
``capabilities()`` and trivial probes return real values. Production
deployments register a real backend via
:func:`app.adapters.registry.register_backend`.
"""

from __future__ import annotations

from collections.abc import Iterator
from pathlib import Path
from typing import Any

from app.core.errors import CapabilityUnavailableError


[docs] class StubBackend: """No-op stub satisfying the full ``SfmBackend`` union.""" name = "stub" version = "0.0.1" vendor = "test"
[docs] def capabilities(self) -> set[str]: # Stub advertises no capabilities. Tests that need a method # to succeed should subclass + override, or register their # own backend with the desired capability set. return set()
[docs] def extract_features( self, *, database_path: Path, image_root: Path, image_list: list[str], options: dict, ) -> dict: raise CapabilityUnavailableError(capability="features.extract")
[docs] def match(self, *, database_path: Path, mode: str, options: dict) -> dict: raise CapabilityUnavailableError(capability=f"pairs.{mode}")
[docs] def verify_matches(self, *, database_path: Path, options: dict) -> dict: raise CapabilityUnavailableError(capability="matches.verify")
[docs] def read_keypoints(self, **_: Any) -> tuple[list[list[float]], bytes, int]: raise CapabilityUnavailableError(capability="features.extract")
[docs] def iter_two_view_geometries(self, *, database_path: Path) -> Iterator: return iter([])
[docs] def iter_correspondences(self, *, database_path: Path) -> Iterator: return iter([])
[docs] def run_mapping(self, **_: Any) -> tuple[list[dict], list[Any]]: raise CapabilityUnavailableError(capability="map.incremental")
[docs] def bundle_adjustment(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="ba.standard")
[docs] def triangulate(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="triangulate.retri")
[docs] def relocalize(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="relocalize.images")
[docs] def pose_graph_optimize(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="pgo.optimize")
[docs] def export(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="export.ply")
[docs] def convert_spherical_to_cubemap(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="projection.cubemap_rig")
[docs] def project_images(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="projection.equirectangular_to_cubemap")
[docs] def render_spherical_cubemap_images(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="projection.equirectangular_to_cubemap")
[docs] def build_vlad_index(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="similarity.vlad")
[docs] def convert_artifact(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="artifacts.convert")
[docs] def localize_from_memory(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="localize.from_memory")
[docs] def apply_sim3(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="georegister.sim3")
[docs] def merge_reconstructions(self, **_: Any) -> dict: raise CapabilityUnavailableError(capability="recon.merge")
[docs] def localize_batch(self, **_: Any) -> list: raise CapabilityUnavailableError(capability="localize.batch")
[docs] def read_reconstruction(self, path: Path) -> Any: raise CapabilityUnavailableError(capability="features.extract")
[docs] def runtime_versions(self) -> dict[str, str]: return {"stub_version": "0.0.1"}