Source code for lalandre_core.embedding_presets

"""Helpers for embedding preset resolution across services."""

from __future__ import annotations

import os
from typing import Optional

from .config import EmbeddingPresetConfig, get_config




def list_embedding_presets(
    *,
    enabled_only: bool = False,
    indexing_only: bool = False,
) -> list[EmbeddingPresetConfig]:
    """Return configured embedding presets."""
    config = get_config()
    presets = config.embedding_presets
    if enabled_only:
        presets = [preset for preset in presets if preset.enabled]
    if indexing_only:
        presets = [preset for preset in presets if preset.indexing_enabled]
    return [preset.model_copy(deep=True) for preset in presets]





def get_embedding_preset(
    preset_id: str | None,
    *,
    enabled_only: bool = False,
    indexing_only: bool = False,
) -> Optional[EmbeddingPresetConfig]:
    """Resolve a preset by ID."""
    if preset_id is None:
        return None
    config = get_config()
    preset = config.get_embedding_preset(preset_id)
    if preset is None:
        return None
    if enabled_only and not preset.enabled:
        return None
    if indexing_only and not preset.indexing_enabled:
        return None
    return preset.model_copy(deep=True)





def get_default_embedding_preset() -> EmbeddingPresetConfig:
    """Return the configured default embedding preset."""
    return get_config().get_default_embedding_preset().model_copy(deep=True)





def get_default_embedding_preset_id() -> str:
    """Return the ID of the default embedding preset."""
    return get_default_embedding_preset().preset_id





def resolve_embedding_preset_or_default(preset_id: str | None) -> EmbeddingPresetConfig:
    """Resolve a preset, falling back to the configured default when missing/invalid."""
    preset = get_embedding_preset(preset_id, enabled_only=True)
    if preset is not None:
        return preset
    return get_default_embedding_preset()





def resolve_embed_queue_name(preset_id: str | None = None) -> str:
    """Return the queue name for a preset, defaulting to the configured default preset."""
    preset = resolve_embedding_preset_or_default(preset_id)
    return preset.resolved_queue_name()





def resolve_worker_embedding_preset(
    preset_id: str | None = None,
    *,
    env_var: str = "EMBEDDING_PRESET_ID",
) -> EmbeddingPresetConfig:
    """Resolve the preset bound to the current embedding worker."""
    effective_preset_id = preset_id or os.getenv(env_var)
    return resolve_embedding_preset_or_default(effective_preset_id)