Skip to content

pixano_inference.plugins

Discovery of custom model implementations shipped as installable packages.

A third-party model is packaged like any Python distribution and advertises itself through a pixano_inference.models entry point. Because the package is installed in the same environment as the server, it is importable both in the ingress process and in every Ray Serve worker — no source directory needs to be shipped to workers.

Example plugin pyproject.toml::

[project.entry-points."pixano_inference.models"]
my_detector = "my_pkg.model"        # importing the module runs @register_model
# or point directly at the class:
my_detector = "my_pkg.model:MyDetector"

At startup the server loads all such entry points (importing them registers their models). The core ships no model implementations and depends on no ML framework: every model -- including the first-party ones (SAM2, CLIP, Grounding DINO, ...) -- is a separate package discovered this way.

describe_plugins()

Summarise the last discovery: which model plugins loaded, and which failed and why.

Returns:

Type Description
str

A one-line description for logs and error messages.
Source code in pixano_inference/plugins.py 
77
78
79
80
81
82
83
84
85
86
def describe_plugins() -> str:
    """Summarise the last discovery: which model plugins loaded, and which failed and why.

    Returns:
        A one-line description for logs and error messages.
    """
    summary = f"model plugins loaded: {', '.join(sorted(_DISCOVERED)) or 'none'}"
    if _FAILED:
        summary += "; failed to load: " + "; ".join(f"{name} ({error})" for name, error in sorted(_FAILED.items()))
    return summary

ensure_models_loaded(force=False)

Register all installed models (entry-point plugins).

Idempotent — safe to call from every entry point that needs the registry populated (config resolution, app creation, replica init).

Parameters:

Name Type Description Default
force bool

Re-run discovery even if it already ran in this process.

 False
Source code in pixano_inference/plugins.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def ensure_models_loaded(force: bool = False) -> None:
    """Register all installed models (entry-point plugins).

    Idempotent — safe to call from every entry point that needs the registry populated
    (config resolution, app creation, replica init).

    Args:
        force: Re-run discovery even if it already ran in this process.
    """
    global _LOADED
    if _LOADED and not force:
        return
    load_plugin_models()
    _LOADED = True
    logger.info(
        "%s; registered model classes: %s",
        describe_plugins(),
        ", ".join(sorted(ModelClassRegistry.list_all())) or "none",
    )

load_plugin_models()

Discover and import all pixano_inference.models entry-point plugins.

Each entry point is loaded (importing its module, which runs the @register_model decorators); if it resolves to an :class:InferenceModel subclass, it is also registered defensively. A failing plugin is logged and skipped rather than aborting startup.

Returns:

Type Description
``{"loaded"

[...], "failed": [...]}`` entry-point names.
Source code in pixano_inference/plugins.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def load_plugin_models() -> dict[str, list[str]]:
    """Discover and import all ``pixano_inference.models`` entry-point plugins.

    Each entry point is loaded (importing its module, which runs the ``@register_model``
    decorators); if it resolves to an :class:`InferenceModel` subclass, it is also
    registered defensively. A failing plugin is logged and skipped rather than aborting
    startup.

    Returns:
        ``{"loaded": [...], "failed": [...]}`` entry-point names.
    """
    loaded: list[str] = []
    failed: list[str] = []
    _DISCOVERED.clear()
    _FAILED.clear()
    for entry_point in entry_points(group=ENTRY_POINT_GROUP):
        try:
            obj = entry_point.load()
            if isinstance(obj, type) and issubclass(obj, InferenceModel):
                ModelClassRegistry.ensure_registered(obj)
            loaded.append(entry_point.name)
            _DISCOVERED[entry_point.name] = entry_point.value
            logger.debug("Loaded model plugin '%s' (%s)", entry_point.name, entry_point.value)
        except Exception as exc:
            logger.warning("Failed to load model plugin '%s' (%s): %s", entry_point.name, entry_point.value, exc)
            failed.append(entry_point.name)
            _FAILED[entry_point.name] = f"{type(exc).__name__}: {exc}"
    return {"loaded": loaded, "failed": failed}