Stable Diffusion XL on macOS: Efficient Testing with BDD and safetensors

1from pathlib import Path
2from typing import Optional, Union
3from PIL import Image
4import logging
5from diffusers import StableDiffusionXLPipeline, EulerAncestralDiscreteScheduler
6import torch
7
8
9class SDXLPipeline:
10    def __init__(self, model_path: Union[str, Path]):
11        """Initialize SDXL pipeline.
12        
13        Args:
14            model_path: Path to the local SDXL model file (.safetensors)
15        """
16        self.pipeline = None
17        self.device = None
18        self.dtype = None
19        
20        self.model_path = Path(model_path)
21        if not self.model_path.exists():
22            raise FileNotFoundError(f"Model file not found: {self.model_path}")
23        
24        self._setup_device()
25
26    def _setup_device(self) -> None:
27        """Configure device settings for optimal performance.
28        
29        Automatically selects MPS if available, otherwise falls back to CPU.
30        """
31        # Metal Performance Shaders devices auto-detection
32        self.device = "mps" if torch.backends.mps.is_available() else "cpu"
33        self.dtype = torch.float32  # Optimal for MacOS
34        logger.info(f"Using device: {self.device}")
35
36    def load(self) -> bool:
37        """Load the SDXL model with optimal settings."""
38        try:
39            # Use safetensors for efficient loading
40            self.pipeline = StableDiffusionXLPipeline.from_single_file(
41                str(self.model_path),
42                torch_dtype=self.dtype,
43                use_safetensors=True
44            )
45            self.pipeline.to(self.device)
46            self.pipeline.enable_attention_slicing()  # Memory optimization
47            
48            # SDXL Turbo specific settings
49            self.pipeline.scheduler = EulerAncestralDiscreteScheduler.from_config(
50                self.pipeline.scheduler.config,
51                timestep_spacing="trailing"  # Optimal for Turbo models
52            )
53            return True
54        except Exception as e:
55            logger.error(f"Error loading pipeline: {str(e)}")
56            return False

1from sdxl_mac_diffusers_guide import SDXLPipeline
2from pathlib import Path
3
4def main():
5    # Initialize pipeline with model path
6    model_path = "models/sdxl_turbo.safetensors"
7    pipeline = SDXLPipeline(model_path)
8    
9    if not pipeline.load():
10        print("Failed to load pipeline")
11        return
12    
13    # Generate image with optimal settings for SDXL Turbo
14    prompt = "A photorealistic image of a white cat in a garden"
15    image = pipeline.generate(
16        prompt,
17        num_inference_steps=1,    # Single step for Turbo
18        guidance_scale=0.0,      # Optimal for Turbo
19        width=512,               # Optimal resolution
20        height=512
21    )
22    
23    # Save the generated image
24    if image:
25        output_dir = Path("output")
26        output_dir.mkdir(exist_ok=True)
27        image.save(output_dir / "generated_cat.png")

1Feature: Pipeline Initialization
2  As an AI developer
3  I want to initialize the SDXL pipeline correctly
4  So that I can ensure stable model execution
5
6  Background:
7    Given the test model file exists
8
9  @initialization
10  Scenario: Basic pipeline initialization
11    When I create a new pipeline instance
12    Then the pipeline should be configured correctly
13    And the device settings should match the system
14
15  @error @initialization
16  Scenario: Handle missing model file
17    Given the model file does not exist
18    When I try to create a pipeline instance
19    Then it should raise a FileNotFoundError
20    And provide a clear error message

1from pathlib import Path
2import pytest
3from pytest_bdd import scenarios, given, when, then, parsers
4from diffusers import StableDiffusionXLPipeline
5
6# シナリオの読み込み
7scenarios('pipeline/initialization.feature')
8
9# Background steps
10@given('the test model file exists', target_fixture='model_path')
11def model_file_exists(tmp_path):
12    path = tmp_path / 'test_model.safetensors'
13    path.touch()
14    return path
15
16# Scenario: Basic pipeline initialization
17@when('I create a new pipeline instance')
18def create_pipeline(model_path):
19    return SDXLPipeline(model_path)
20
21@then('the pipeline should be configured correctly')
22def verify_pipeline_configuration(pipeline):
23    assert isinstance(pipeline, SDXLPipeline)
24    assert pipeline.model_path is not None
25
26@then('the device settings should match the system')
27def verify_device_settings(pipeline):
28    import torch
29    expected_device = 'mps' if torch.backends.mps.is_available() else 'cpu'
30    assert pipeline.device == expected_device
31
32# Scenario: Handle missing model file
33@given('the model file does not exist')
34def no_model_file(tmp_path):
35    return tmp_path / 'nonexistent.safetensors'
36
37@when('I try to create a pipeline instance')
38def try_create_pipeline(model_path):
39    with pytest.raises(FileNotFoundError) as exc_info:
40        SDXLPipeline(model_path)
41    return exc_info
42
43@then('it should raise a FileNotFoundError')
44def verify_error_type(exc_info):
45    assert exc_info.type == FileNotFoundError
46
47@then('provide a clear error message')
48def verify_error_message(exc_info):
49    assert 'Model file not found' in str(exc_info.value)

1Feature: SDXL Pipeline Management
2  As an AI application developer
3  I want to manage SDXL pipeline efficiently
4  So that I can provide stable image generation service
5
6  Background:
7    Given the SDXL Turbo model is available
8
9  Scenario: Initialize pipeline with optimal settings
10    When I create a pipeline with default settings
11    Then the pipeline should be configured for MacOS
12    And MPS acceleration should be enabled if available
13
14  Scenario: Handle memory efficiently
15    Given I have an initialized pipeline
16    When I load the model with attention slicing
17    Then the memory usage should be optimized
18    And the model should be loaded successfully
19
20  Scenario: Generate images with optimal settings
21    Given I have a loaded pipeline
22    When I generate an image with the following settings:
23      | Parameter         | Value |
24      | num_inference_steps | 1     |
25      | guidance_scale    | 0.0   |
26    Then the generation should complete successfully
27    And the output image should be 512x512 pixels

1from pathlib import Path
2import pytest
3from pytest_bdd import scenario, given, when, then, parsers
4
5@scenario('pipeline.feature', 'Initialize pipeline with optimal settings')
6def test_pipeline_initialization():
7    pass
8
9@given('the SDXL Turbo model is available', target_fixture='model_path')
10def model_available(tmp_path):
11    model_path = tmp_path / 'sdxl_turbo.safetensors'
12    model_path.touch()
13    return model_path
14
15@when('I create a pipeline with default settings')
16def create_pipeline(model_path):
17    pipeline = SDXLPipeline(model_path)
18    return pipeline
19
20@then('the pipeline should be configured for MacOS')
21def verify_macos_config(pipeline):
22    assert pipeline.device in ['mps', 'cpu']
23    assert pipeline.dtype == torch.float32
24
25@then('MPS acceleration should be enabled if available')
26def verify_mps_acceleration(pipeline):
27    if torch.backends.mps.is_available():
28        assert pipeline.device == 'mps'
29    else:
30        assert pipeline.device == 'cpu'

1Feature: Pipeline Optimization
2  As a system developer
3  I want to optimize the pipeline performance
4  So that it runs efficiently on MacOS
5
6  Background:
7    Given I have an initialized pipeline
8
9  @memory
10  Scenario Outline: Memory optimization settings
11    When I enable <optimization> optimization
12    Then the memory usage should be less than <limit> GB
13    And the model should still function correctly
14
15    Examples:
16      | optimization      | limit |
17      | attention_slicing | 8     |
18      | fp16             | 6     |

1Feature: Batch Image Generation
2  As a service provider
3  I want to generate multiple images efficiently
4  So that I can handle concurrent requests
5
6  Background:
7    Given the pipeline is initialized with optimal settings
8
9  @performance
10  Scenario Outline: Generate images in batch
11    When I generate <count> images in parallel
12    Then all images should be generated successfully
13    And memory usage should stay below <memory_limit> GB
14    And total processing time should be less than <time_limit> seconds
15
16    Examples:
17      | count | memory_limit | time_limit |
18      | 2     | 12          | 5          |
19      | 4     | 16          | 10         |
20
21  @error-handling
22  Scenario: Handle resource exhaustion
23    Given system memory is limited to 4GB
24    When I attempt to generate multiple images
25    Then it should fail gracefully
26    And release all allocated resources
27    And provide appropriate error information

1import pytest
2from pytest_bdd import scenario, given, when, then, parsers
3from concurrent.futures import ThreadPoolExecutor
4import psutil
5import time
6
7@scenario('generation/batch_processing.feature', 'Generate images in batch')
8def test_batch_generation():
9    pass
10
11@given('the pipeline is initialized with optimal settings')
12def optimized_pipeline(model_path):
13    pipeline = SDXLPipeline(model_path)
14    pipeline.enable_attention_slicing()
15    pipeline.enable_sequential_cpu_offload()
16    return pipeline
17
18@when(parsers.parse('I generate {count:d} images in parallel'))
19def generate_batch(pipeline, count):
20    prompts = [f'Test prompt {i}' for i in range(count)]
21    start_time = time.time()
22    
23    with ThreadPoolExecutor(max_workers=count) as executor:
24        futures = [
25            executor.submit(pipeline.generate, prompt)
26            for prompt in prompts
27        ]
28        results = [f.result() for f in futures]
29    
30    return {
31        'results': results,
32        'time_taken': time.time() - start_time
33    }
34
35@then('all images should be generated successfully')
36def verify_generation(results):
37    assert all(img is not None for img in results['results'])
38    for img in results['results']:
39        assert img.size == (512, 512)
40
41@then(parsers.parse('memory usage should stay below {memory_limit:d} GB'))
42def verify_memory_usage(memory_limit):
43    memory_gb = psutil.Process().memory_info().rss / (1024 * 1024 * 1024)
44    assert memory_gb < memory_limit
45
46@then(parsers.parse('total processing time should be less than {time_limit:d} seconds'))
47def verify_processing_time(results, time_limit):
48    assert results['time_taken'] < time_limit