Merge pull request #73 from jkoelker/oneiro-zmt.9

refactor(lora): consolidate LoRA/embedding

Author: jkoelker

src/oneiro/discord/commands.py

@@ -10,7 +10,7 @@
 from oneiro.discord.handlers import DreamContext, create_dream_callbacks
 from oneiro.pipelines import SCHEDULER_CHOICES
 from oneiro.pipelines.civitai_checkpoint import CivitaiCheckpointPipeline
-from oneiro.pipelines.lora import is_lora_compatible
+from oneiro.pipelines.lora import is_resource_compatible
 from oneiro.queue import QueueStatus
 from oneiro.services.generation import (
     MAX_GUIDANCE_SCALE,
@@ -633,7 +633,7 @@ async def fetch_command(
             # Check compatibility and prepare warning
             compatibility_warning = ""
             if model_type == "LORA" and pipeline_type and version.base_model:
-                if not is_lora_compatible(pipeline_type, version.base_model):
+                if not is_resource_compatible(pipeline_type, version.base_model):
                     compatibility_warning = (
                         f"\n⚠️ **Note**: This LoRA (base: {version.base_model}) may not be "
                         f"compatible with the current model ({pipeline_type})"

src/oneiro/pipelines/embedding.py

@@ -6,7 +6,7 @@
 from typing import Any
 
 from oneiro.civitai import CivitaiClient, parse_civitai_url
-from oneiro.pipelines.lora import PIPELINE_BASE_MODEL_MAP
+from oneiro.pipelines.lora import is_resource_compatible
 
 
 class EmbeddingSource(str, Enum):
@@ -281,34 +281,6 @@ def parse_embeddings_from_config(
     return embeddings
 
 
-def is_embedding_compatible(pipeline_type: str, civitai_base_model: str | None) -> bool:
-    """Check if a Civitai embedding is compatible with a pipeline type.
-
-    Args:
-        pipeline_type: Pipeline type (flux2, zimage, qwen, etc.)
-        civitai_base_model: Base model string from Civitai API
-
-    Returns:
-        True if compatible, False otherwise
-    """
-    if civitai_base_model is None:
-        # Can't verify, assume compatible
-        return True
-
-    compatible_bases = PIPELINE_BASE_MODEL_MAP.get(pipeline_type, [])
-    if not compatible_bases:
-        # Unknown pipeline type, assume compatible
-        return True
-
-    # Check if any compatible base model matches (case-insensitive substring)
-    civitai_lower = civitai_base_model.lower()
-    for base in compatible_bases:
-        if base.lower() in civitai_lower or civitai_lower in base.lower():
-            return True
-
-    return False
-
-
 class EmbeddingIncompatibleError(Exception):
     """Raised when an embedding is incompatible with the pipeline type."""
 
@@ -382,7 +354,7 @@ async def resolve_embedding_path(
 
         # Validate compatibility
         if validate_compatibility and pipeline_type:
-            if not is_embedding_compatible(pipeline_type, version.base_model):
+            if not is_resource_compatible(pipeline_type, version.base_model):
                 raise EmbeddingIncompatibleError(
                     embedding.name,
                     pipeline_type,

src/oneiro/pipelines/lora.py

@@ -468,8 +468,8 @@ def parse_loras_from_config(
 }
 
 
-def is_lora_compatible(pipeline_type: str, civitai_base_model: str | None) -> bool:
-    """Check if a Civitai LoRA is compatible with a pipeline type.
+def is_resource_compatible(pipeline_type: str, civitai_base_model: str | None) -> bool:
+    """Check if a Civitai resource (LoRA, embedding, etc.) is compatible with a pipeline type.
 
     Args:
         pipeline_type: Pipeline type (flux2, zimage, qwen, etc.)
@@ -569,7 +569,7 @@ async def resolve_lora_path(
 
         # Validate compatibility
         if validate_compatibility and pipeline_type:
-            if not is_lora_compatible(pipeline_type, version.base_model):
+            if not is_resource_compatible(pipeline_type, version.base_model):
                 raise LoraIncompatibleError(
                     lora.adapter_name or f"civitai_{lora.civitai_id}",
                     pipeline_type,

src/oneiro/services/generation.py

@@ -4,7 +4,7 @@
 from typing import TYPE_CHECKING, Any
 
 from oneiro.pipelines import LoraConfig, LoraSource
-from oneiro.pipelines.lora import is_lora_compatible
+from oneiro.pipelines.lora import is_resource_compatible
 
 if TYPE_CHECKING:
     from oneiro.civitai import CivitaiClient
@@ -184,7 +184,9 @@ async def resolve_loras(
                     try:
                         model_info = await civitai_client.get_model(civitai_id)
                         version = model_info.latest_version
-                        if version and not is_lora_compatible(pipeline_type, version.base_model):
+                        if version and not is_resource_compatible(
+                            pipeline_type, version.base_model
+                        ):
                             result.warnings.append(
                                 f"⚠️ LoRA `{model_info.name}` (base: {version.base_model}) "
                                 f"may not be compatible with current model ({pipeline_type})"
@@ -219,7 +221,7 @@ async def resolve_loras(
                             try:
                                 model_info = await civitai_client.get_model(civitai_id)
                                 version = model_info.latest_version
-                                if version and not is_lora_compatible(
+                                if version and not is_resource_compatible(
                                     pipeline_type, version.base_model
                                 ):
                                     result.warnings.append(

tests/test_embedding.py

@@ -6,14 +6,13 @@
 import pytest
 
 from oneiro.pipelines.embedding import (
-    PIPELINE_BASE_MODEL_MAP,
     EmbeddingConfig,
     EmbeddingIncompatibleError,
     EmbeddingSource,
-    is_embedding_compatible,
     parse_embedding_config,
     parse_embeddings_from_config,
 )
+from oneiro.pipelines.lora import PIPELINE_BASE_MODEL_MAP, is_resource_compatible
 
 
 class TestEmbeddingConfig:
@@ -341,47 +340,47 @@ def test_no_embeddings(self):
 
 
 class TestIsEmbeddingCompatible:
-    """Tests for is_embedding_compatible function."""
+    """Tests for is_resource_compatible function."""
 
     def test_flux1_compatible(self):
         """Flux.1 embeddings compatible with flux1 pipeline."""
-        assert is_embedding_compatible("flux1", "Flux.1 Dev")
-        assert is_embedding_compatible("flux1", "Flux.1 Schnell")
-        assert is_embedding_compatible("flux1", "Flux.1 D")
+        assert is_resource_compatible("flux1", "Flux.1 Dev")
+        assert is_resource_compatible("flux1", "Flux.1 Schnell")
+        assert is_resource_compatible("flux1", "Flux.1 D")
 
     def test_flux2_compatible(self):
         """Flux.2 embeddings compatible with flux2 pipeline."""
-        assert is_embedding_compatible("flux2", "Flux.2")
+        assert is_resource_compatible("flux2", "Flux.2")
 
     def test_flux1_flux2_incompatible(self):
         """Flux.1 and Flux.2 embeddings are NOT cross-compatible."""
-        assert not is_embedding_compatible("flux1", "Flux.2")
-        assert not is_embedding_compatible("flux2", "Flux.1 Dev")
+        assert not is_resource_compatible("flux1", "Flux.2")
+        assert not is_resource_compatible("flux2", "Flux.1 Dev")
 
     def test_sdxl_compatible(self):
         """SDXL embeddings compatible with sdxl pipeline."""
-        assert is_embedding_compatible("sdxl", "SDXL 1.0")
-        assert is_embedding_compatible("sdxl", "Pony")
-        assert is_embedding_compatible("sdxl", "Illustrious")
+        assert is_resource_compatible("sdxl", "SDXL 1.0")
+        assert is_resource_compatible("sdxl", "Pony")
+        assert is_resource_compatible("sdxl", "Illustrious")
 
     def test_incompatible_base_model(self):
         """Incompatible base model returns False."""
-        assert not is_embedding_compatible("flux2", "SDXL 1.0")
-        assert not is_embedding_compatible("sdxl", "Flux.1 Dev")
-        assert not is_embedding_compatible("flux2", "SD 1.5")
+        assert not is_resource_compatible("flux2", "SDXL 1.0")
+        assert not is_resource_compatible("sdxl", "Flux.1 Dev")
+        assert not is_resource_compatible("flux2", "SD 1.5")
 
     def test_none_base_model_is_compatible(self):
         """None base model assumed compatible."""
-        assert is_embedding_compatible("flux2", None)
+        assert is_resource_compatible("flux2", None)
 
     def test_unknown_pipeline_is_compatible(self):
         """Unknown pipeline type assumed compatible."""
-        assert is_embedding_compatible("unknown", "SDXL 1.0")
+        assert is_resource_compatible("unknown", "SDXL 1.0")
 
     def test_case_insensitive(self):
         """Comparison is case-insensitive."""
-        assert is_embedding_compatible("flux1", "flux.1 dev")
-        assert is_embedding_compatible("flux1", "FLUX.1 DEV")
+        assert is_resource_compatible("flux1", "flux.1 dev")
+        assert is_resource_compatible("flux1", "FLUX.1 DEV")
 
 
 class TestEmbeddingIncompatibleError:

tests/test_lora.py

@@ -10,7 +10,7 @@
     LoraConfig,
     LoraIncompatibleError,
     LoraSource,
-    is_lora_compatible,
+    is_resource_compatible,
     parse_lora_config,
     parse_loras_from_config,
     parse_loras_from_model_config,
@@ -574,40 +574,40 @@ def test_mixed_string_and_dict_refs(self):
 
 
 class TestIsLoraCompatible:
-    """Tests for is_lora_compatible function."""
+    """Tests for is_resource_compatible function."""
 
     def test_flux_compatible(self):
         """Flux.1 LoRAs compatible with flux1 pipeline (not flux2)."""
-        assert is_lora_compatible("flux1", "Flux.1 Dev")
-        assert is_lora_compatible("flux1", "Flux.1 Schnell")
-        assert is_lora_compatible("flux1", "Flux.1 D")
-        assert is_lora_compatible("flux2", "Flux.2")
-        assert not is_lora_compatible("flux2", "Flux.1 Dev")
+        assert is_resource_compatible("flux1", "Flux.1 Dev")
+        assert is_resource_compatible("flux1", "Flux.1 Schnell")
+        assert is_resource_compatible("flux1", "Flux.1 D")
+        assert is_resource_compatible("flux2", "Flux.2")
+        assert not is_resource_compatible("flux2", "Flux.1 Dev")
 
     def test_sdxl_compatible(self):
         """SDXL LoRAs compatible with sdxl pipeline."""
-        assert is_lora_compatible("sdxl", "SDXL 1.0")
-        assert is_lora_compatible("sdxl", "Pony")
-        assert is_lora_compatible("sdxl", "Illustrious")
+        assert is_resource_compatible("sdxl", "SDXL 1.0")
+        assert is_resource_compatible("sdxl", "Pony")
+        assert is_resource_compatible("sdxl", "Illustrious")
 
     def test_incompatible_base_model(self):
         """Incompatible base model returns False."""
-        assert not is_lora_compatible("flux2", "SDXL 1.0")
-        assert not is_lora_compatible("sdxl", "Flux.1 Dev")
-        assert not is_lora_compatible("flux2", "SD 1.5")
+        assert not is_resource_compatible("flux2", "SDXL 1.0")
+        assert not is_resource_compatible("sdxl", "Flux.1 Dev")
+        assert not is_resource_compatible("flux2", "SD 1.5")
 
     def test_none_base_model_is_compatible(self):
         """None base model assumed compatible."""
-        assert is_lora_compatible("flux2", None)
+        assert is_resource_compatible("flux2", None)
 
     def test_unknown_pipeline_is_compatible(self):
         """Unknown pipeline type assumed compatible."""
-        assert is_lora_compatible("unknown", "SDXL 1.0")
+        assert is_resource_compatible("unknown", "SDXL 1.0")
 
     def test_case_insensitive(self):
         """Comparison is case-insensitive."""
-        assert is_lora_compatible("flux1", "flux.1 dev")
-        assert is_lora_compatible("flux1", "FLUX.1 DEV")
+        assert is_resource_compatible("flux1", "flux.1 dev")
+        assert is_resource_compatible("flux1", "FLUX.1 DEV")
 
 
 class TestLoraIncompatibleError: