Adding Custom Pipelines
Developer guide for creating and integrating new AI processing pipelines
Last updated: 2026-01-03
Adding Custom Pipelines
Learn how to extend 360 Hextile V2 with custom AI processing pipelines. The modular architecture makes it easy to add new models.
Overview
Adding a new pipeline requires: 1. Create processor class (Python) 2. Define UI schema (JSON in Python) 3. Register processor (one line) 4. Done! System handles the rest automatically
Time Estimate: 2-4 hours for basic processor
Quick Example
Here's a complete minimal processor:
# backend/processors/tiled/my_model.py
from backend.processors.base import TiledProcessor
from PIL import Image
import torch
class MyModelProcessor(TiledProcessor):
processor_name = "my_model"
processor_display_name = "My Custom Model"
def __init__(self):
self.model = None
def get_ui_schema(self):
return {
"mode_id": "my_model",
"display_name": "My Custom Model",
"type": "tiled",
"description": "My custom AI processor",
"icon": "Sparkles",
"capabilities": {
"supports_lora": False,
"requires_template": True,
"min_resolution": [512, 256],
"max_resolution": [8192, 4096]
},
"tabs": [
{
"id": "prompts",
"label": "Prompts",
"icon": "TextT",
"fields": [
{
"id": "prompt",
"type": "textarea",
"label": "Prompt",
"required": True,
"default": ""
}
]
}
]
}
def initialize(self, config):
# Load your model here
self.model = load_my_model(config.get("model_path"))
self._initialized = True
def shutdown(self):
del self.model
self._cleanup_vram()
self._initialized = False
def process_tile(self, tile_image_path, mask_path, output_path,
tile_data, config):
# Process one tile
image = Image.open(tile_image_path)
prompt = config.get("prompt", "")
# Your model inference here
result = self.model.generate(image, prompt)
result.save(output_path)
return {"status": "success"}
Register it:
# backend/processors/__init__.py
from backend.processors.tiled.my_model import MyModelProcessor
ProcessorRegistry.register(MyModelProcessor)
That's it! Your pipeline now appears in the dropdown.
Step-by-Step Guide
Step 1: Choose Processor Type
TiledProcessor - For models using hexagonal tiling: - Diffusion models (Stable Diffusion, Flux, etc.) - Models that benefit from tile-based processing - Need cardinal prompt interpolation - Require hextile templates
DirectProcessor - For models processing full images: - Upscalers (Real-ESRGAN, etc.) - Models that handle any resolution - Don't need tiling or masking - Faster, simpler processing
Step 2: Create Processor File
Create file in appropriate directory:
backend/processors/
├── tiled/
│ └── my_model.py ← For tiled processors
└── direct/
└── my_upscaler.py ← For direct processors
Step 3: Implement Base Class
For Tiled Processor:
from backend.processors.base import TiledProcessor
from typing import Dict, Any, Tuple, List
from PIL import Image
import torch
class MyTiledProcessor(TiledProcessor):
# Required class attributes
processor_name = "my_tiled" # Unique ID (used in API/configs)
processor_display_name = "My Tiled Processor"
def __init__(self):
super().__init__()
self.model = None
self.lora_models = {} # If you support LoRA
def get_ui_schema(self) -> Dict[str, Any]:
"""Define UI structure (see Step 4)"""
return {...}
def initialize(self, config: Dict[str, Any]):
"""Load model and setup"""
model_path = config.get("model", "default/path")
# Load your model
self.model = YourModel.from_pretrained(
model_path,
torch_dtype=torch.float16
).to("cuda")
# Optional: Enable optimizations
self.model.enable_xformers_memory_efficient_attention()
self._initialized = True
def shutdown(self):
"""Cleanup and free VRAM"""
if self.model:
del self.model
self.model = None
# Clear LoRA if used
self.lora_models.clear()
# Aggressive VRAM cleanup
self._cleanup_vram()
self._initialized = False
def process_tile(self,
tile_image_path: str,
mask_image_path: str,
output_path: str,
tile_data: Dict,
config: Dict) -> Dict:
"""Process one hexagonal tile"""
# Load tile and mask
image = Image.open(tile_image_path).convert("RGB")
mask = Image.open(mask_image_path).convert("L")
# Get parameters from config
prompt = config.get("prompt", "")
strength = config.get("strength", 0.75)
steps = config.get("num_inference_steps", 50)
seed = config.get("seed", -1)
# Optional: Build prompt with cardinal interpolation
if config.get("interpolation") == "6CardinalDirections":
prompt = self._build_cardinal_prompt(tile_data, config)
# Generate with your model
generator = torch.Generator("cuda")
if seed != -1:
generator.manual_seed(seed)
result = self.model(
prompt=prompt,
image=image,
mask_image=mask,
strength=strength,
num_inference_steps=steps,
generator=generator
)
# Save result
result.images[0].save(output_path)
return {
"status": "success",
"tile_index": tile_data.get("index"),
"prompt_used": prompt
}
def _build_cardinal_prompt(self, tile_data: Dict, config: Dict) -> str:
"""Build prompt with cardinal interpolation"""
from backend.core.prompt_utils import interpolate_cardinal_prompt
global_prompt = config.get("global_prompt", "")
cardinal_prompts = config.get("cardinal_prompts", {})
# Get tile's cardinal vector from tile_data
view_vector = tile_data.get("view_direction", [0, 0, -1])
return interpolate_cardinal_prompt(
global_prompt,
cardinal_prompts,
view_vector
)
# Optional: LoRA support
def supports_lora(self) -> bool:
return True
def _load_lora(self, lora_path: str, strength: float):
"""Load LoRA adapter"""
self.model.load_lora_weights(lora_path)
self.lora_models[lora_path] = strength
def _unload_all_loras(self):
"""Unload all LoRAs"""
for lora_path in list(self.lora_models.keys()):
self.model.unload_lora_weights()
self.lora_models.clear()
# Optional: Config validation
def validate_config(self, config: Dict) -> Tuple[bool, List[str]]:
"""Validate configuration"""
errors = []
if not config.get("prompt"):
errors.append("Prompt is required")
steps = config.get("num_inference_steps", 50)
if not (10 <= steps <= 150):
errors.append("Steps must be between 10 and 150")
return (len(errors) == 0, errors)
For Direct Processor:
from backend.processors.base import DirectProcessor
from PIL import Image
import numpy as np
class MyDirectProcessor(DirectProcessor):
processor_name = "my_direct"
processor_display_name = "My Direct Processor"
def __init__(self):
super().__init__()
self.model = None
def get_ui_schema(self) -> Dict[str, Any]:
"""Define UI structure"""
return {...}
def initialize(self, config: Dict[str, Any]):
"""Load model"""
from my_model_library import MyModel
self.model = MyModel(
model_path=config.get("model"),
gpu_id=config.get("gpu_id", 0)
)
self._initialized = True
def shutdown(self):
"""Cleanup"""
del self.model
self.model = None
self._cleanup_vram()
self._initialized = False
def process_image(self,
input_path: str,
output_path: str,
config: Dict) -> Dict:
"""Process full image without tiling"""
# Load image
image = Image.open(input_path).convert("RGB")
image_array = np.array(image)
# Get parameters
scale = config.get("scale", 4)
# Process with your model
result_array = self.model.process(
image_array,
scale=scale
)
# Save result
result_image = Image.fromarray(result_array)
result_image.save(output_path)
return {
"status": "success",
"input_size": f"{image.width}x{image.height}",
"output_size": f"{result_image.width}x{result_image.height}"
}
def get_output_dimensions(self,
input_width: int,
input_height: int,
config: Dict) -> Tuple[int, int]:
"""Calculate output dimensions"""
scale = config.get("scale", 4)
return (input_width * scale, input_height * scale)
Step 4: Define UI Schema
The UI schema tells the frontend how to render the configuration form:
def get_ui_schema(self) -> Dict[str, Any]:
return {
# Basic info
"mode_id": "my_model",
"display_name": "My Custom Model",
"type": "tiled", # or "direct"
"description": "Description shown in dropdown",
"icon": "Sparkles", # Phosphor icon name
# Capabilities
"capabilities": {
"supports_lora": True, # Show LoRA tab?
"supports_controlnet": False,
"supports_sequence": False, # Video support?
"requires_template": True, # Need hextile template?
"min_resolution": [512, 256],
"max_resolution": [16384, 8192],
"output_scale_factor": 1 # For upscalers
},
# UI Tabs
"tabs": [
{
"id": "input",
"label": "Input",
"icon": "FileImage",
"fields": [
{
"id": "input_file",
"type": "file",
"label": "Input Image",
"required": True,
"accept": "image/*",
"description": "360° equirectangular image"
}
]
},
{
"id": "template",
"label": "Template",
"icon": "GridFour",
"fields": [
{
"id": "template",
"type": "template_selector",
"label": "Hextile Template",
"required": True
}
]
},
{
"id": "prompts",
"label": "Prompts",
"icon": "TextT",
"fields": [
{
"id": "global_prompt",
"type": "textarea",
"label": "Prompt",
"required": True,
"placeholder": "Describe your desired output...",
"description": "Main prompt applied to all tiles"
},
{
"id": "negative_prompt",
"type": "textarea",
"label": "Negative Prompt",
"placeholder": "Things to avoid...",
"description": "Optional negative guidance"
},
{
"id": "interpolation",
"type": "select",
"label": "Interpolation Mode",
"default": "6CardinalDirections",
"options": [
{"value": "none", "label": "None (Global only)"},
{"value": "6CardinalDirections", "label": "6 Cardinal Directions"}
]
},
{
"id": "cardinal_prompts",
"type": "cardinal_grid",
"label": "Cardinal Prompts",
"visible_if": {
"interpolation": "6CardinalDirections"
},
"directions": {
"front": {"label": "Front", "placeholder": "Looking forward..."},
"right": {"label": "Right", "placeholder": "To the right..."},
"back": {"label": "Back", "placeholder": "Behind..."},
"left": {"label": "Left", "placeholder": "To the left..."},
"above": {"label": "Above", "placeholder": "Looking up..."},
"below": {"label": "Below", "placeholder": "Looking down..."}
}
}
]
},
{
"id": "generation",
"label": "Generation",
"icon": "Sliders",
"fields": [
{
"id": "model",
"type": "text",
"label": "Model Path",
"default": "my-org/my-model",
"description": "HuggingFace model ID or local path"
},
{
"id": "strength",
"type": "slider",
"label": "Strength",
"default": 0.75,
"min": 0.0,
"max": 1.0,
"step": 0.05,
"description": "How much to transform (0=original, 1=full generation)"
},
{
"id": "guidance_scale",
"type": "slider",
"label": "Guidance Scale",
"default": 7.5,
"min": 1.0,
"max": 20.0,
"step": 0.5
},
{
"id": "num_inference_steps",
"type": "number",
"label": "Inference Steps",
"default": 50,
"min": 10,
"max": 150,
"step": 1
},
{
"id": "seed",
"type": "number",
"label": "Seed",
"default": -1,
"description": "Random seed (-1 for random)"
}
]
},
{
"id": "lora",
"label": "LoRA",
"icon": "Stack",
"fields": [
{
"id": "lora",
"type": "lora_selector",
"label": "LoRA Models",
"description": "Fine-tune with LoRA adapters"
}
]
},
{
"id": "output",
"label": "Output",
"icon": "Export",
"fields": [
{
"id": "output_width",
"type": "number",
"label": "Width",
"default": 8192,
"min": 512,
"max": 16384,
"step": 64
},
{
"id": "output_height",
"type": "number",
"label": "Height",
"default": 4096,
"min": 256,
"max": 8192,
"step": 64
},
{
"id": "file_format",
"type": "select",
"label": "Format",
"default": "PNG",
"options": [
{"value": "PNG", "label": "PNG (Lossless)"},
{"value": "JPG", "label": "JPG (Compressed)"}
]
},
{
"id": "jpg_quality",
"type": "slider",
"label": "JPG Quality",
"default": 95,
"min": 50,
"max": 100,
"visible_if": {
"file_format": "JPG"
}
}
]
}
]
}
Field Types Available:
- text: Single line text input
- textarea: Multi-line text input
- number: Number input with min/max/step
- slider: Range slider with live value
- checkbox: Boolean toggle
- select: Dropdown with options
- file: File upload
- template_selector: Special template picker
- lora_selector: Special LoRA manager
- cardinal_grid: Special 360° prompt grid
Conditional Visibility:
Use visible_if to show/hide fields based on other values:
{
"id": "jpg_quality",
"visible_if": {
"file_format": "JPG" # Only show if format is JPG
}
}
Step 5: Register Processor
Add one line to the registry:
# backend/processors/__init__.py
from backend.processors.tiled.my_model import MyTiledProcessor
from backend.processors.direct.my_upscaler import MyDirectProcessor
# Register
ProcessorRegistry.register(MyTiledProcessor)
ProcessorRegistry.register(MyDirectProcessor)
Step 6: Test
- Start backend:
cd backend
python -m uvicorn api.main:app --reload
- Check registration:
curl http://localhost:8000/api/processors
# Should see your processor in list
- Check schema:
curl http://localhost:8000/api/processors/my_model/schema
# Should return your UI schema
- Test in UI:
- Start frontend:
cd frontend && npm run dev - Open dropdown - see your processor
- Click it - form renders from schema
- Configure and test render
Advanced Features
LoRA Support
def supports_lora(self) -> bool:
return True
def _load_lora(self, lora_path: str, strength: float):
"""Load LoRA weights"""
self.model.load_lora_weights(lora_path)
self.current_loras[lora_path] = strength
def _unload_all_loras(self):
"""Unload all LoRAs"""
for lora in list(self.current_loras.keys()):
self.model.unload_lora_weights()
self.current_loras.clear()
def process_tile(self, ...):
# Load LoRAs before generation
lora_config = config.get("lora", {})
if lora_config.get("enabled"):
for lora in lora_config.get("models", []):
self._load_lora(lora["path"], lora["strength"])
# Generate...
# Unload after
self._unload_all_loras()
Cardinal Prompt Interpolation
from backend.core.prompt_utils import interpolate_cardinal_prompt
def _build_cardinal_prompt(self, tile_data: Dict, config: Dict) -> str:
"""Build prompt for tile based on its viewing direction"""
global_prompt = config.get("global_prompt", "")
cardinal_prompts = config.get("cardinal_prompts", {})
# Get tile's viewing direction (from template metadata)
view_vector = tile_data.get("view_direction", [0, 0, -1])
# Interpolate based on direction
return interpolate_cardinal_prompt(
global_prompt,
cardinal_prompts,
view_vector,
temperature=config.get("prompt_interpolation", {}).get("temperature", 0.67)
)
Config Validation
def validate_config(self, config: Dict) -> Tuple[bool, List[str]]:
"""Validate before processing"""
errors = []
# Check required fields
if not config.get("prompt"):
errors.append("Prompt is required")
# Check ranges
steps = config.get("num_inference_steps", 50)
if not (10 <= steps <= 150):
errors.append("Steps must be between 10 and 150")
strength = config.get("strength", 0.75)
if not (0.0 <= strength <= 1.0):
errors.append("Strength must be between 0.0 and 1.0")
# Check model path
model_path = config.get("model")
if not model_path or not os.path.exists(model_path):
errors.append(f"Model not found: {model_path}")
return (len(errors) == 0, errors)
Progress Callbacks
def process_tile(self, tile_image_path, mask_path, output_path,
tile_data, config):
"""Process tile with progress updates"""
# Get callback if provided
progress_callback = config.get("progress_callback")
if progress_callback:
progress_callback(0.0, "Loading tile...")
image = Image.open(tile_image_path)
if progress_callback:
progress_callback(0.2, "Generating...")
result = self.model(image, ...)
if progress_callback:
progress_callback(0.9, "Saving...")
result.save(output_path)
if progress_callback:
progress_callback(1.0, "Complete")
return {"status": "success"}
Best Practices
Memory Management
Always cleanup in shutdown():
def shutdown(self):
# Delete model
if self.model:
del self.model
self.model = None
# Clear caches
self.lora_models.clear()
# Aggressive VRAM cleanup
self._cleanup_vram()
self._initialized = False
Error Handling
def process_tile(self, ...):
try:
# Your processing code
result = self.model.generate(...)
result.save(output_path)
return {"status": "success"}
except torch.cuda.OutOfMemoryError:
return {
"status": "error",
"error": "CUDA out of memory. Try sequential mode or lower resolution."
}
except Exception as e:
return {
"status": "error",
"error": str(e)
}
Logging
import logging
logger = logging.getLogger(__name__)
def initialize(self, config):
logger.info(f"Initializing {self.processor_display_name}")
logger.debug(f"Config: {config}")
self.model = load_model(...)
logger.info(f"Model loaded successfully")
self._initialized = True
Testing
Create a test file:
# tests/test_my_processor.py
import pytest
from backend.processors.tiled.my_model import MyTiledProcessor
def test_processor_registration():
from backend.processors.registry import ProcessorRegistry
processors = ProcessorRegistry.list_pipelines()
names = [p["mode_id"] for p in processors]
assert "my_model" in names
def test_ui_schema():
processor = MyTiledProcessor()
schema = processor.get_ui_schema()
assert schema["mode_id"] == "my_model"
assert schema["type"] == "tiled"
assert len(schema["tabs"]) > 0
def test_config_validation():
processor = MyTiledProcessor()
# Valid config
valid, errors = processor.validate_config({
"prompt": "test",
"strength": 0.75,
"num_inference_steps": 50
})
assert valid
assert len(errors) == 0
# Invalid config
valid, errors = processor.validate_config({
"prompt": "", # Empty prompt
"num_inference_steps": 200 # Out of range
})
assert not valid
assert len(errors) > 0
Common Patterns
Model Loading Patterns
HuggingFace Diffusers:
from diffusers import StableDiffusionPipeline
def initialize(self, config):
self.pipe = StableDiffusionPipeline.from_pretrained(
config.get("model"),
torch_dtype=torch.float16,
use_safetensors=True
).to("cuda")
self.pipe.enable_xformers_memory_efficient_attention()
Custom PyTorch Model:
def initialize(self, config):
self.model = MyCustomModel()
self.model.load_state_dict(
torch.load(config.get("model_path"))
)
self.model.to("cuda")
self.model.eval()
Non-PyTorch Model:
from realesrgan_ncnn_py import Realesrgan
def initialize(self, config):
self.upscaler = Realesrgan(
gpuid=0,
model=config.get("model"),
scale=4
)
Deployment
Model Weights
Option 1: HuggingFace Auto-Download
- Model downloads on first use
- Cached in ~/.cache/huggingface/
- Requires internet connection
Option 2: Local Models
- Place models in resources/models/
- Configure path in schema default
- No internet needed
Option 3: Hybrid - Allow both HuggingFace ID and local path - Check if path exists, else download
Dependencies
Add to requirements.txt:
# Your processor dependencies
diffusers>=0.21.0 # If using diffusers
transformers>=4.30.0
accelerate>=0.20.0
my-custom-library>=1.0.0
Examples
See existing processors for reference:
- SDXL: backend/processors/tiled/sdxl_inpaint.py
- Flux: backend/processors/tiled/flux_inpaint.py
- Real-ESRGAN: backend/processors/direct/realesrgan.py
Troubleshooting
Processor Not Appearing
Check:
1. Registered in __init__.py?
2. No syntax errors in file?
3. Backend restarted?
4. Check logs: python -m uvicorn api.main:app
Schema Not Rendering
Check:
1. get_ui_schema() returns valid JSON?
2. All required fields present?
3. Field types are valid?
4. Check browser console for errors
VRAM Issues
Solutions:
- Implement proper shutdown()
- Call _cleanup_vram()
- Test with sequential mode first
- Monitor GPU memory: nvidia-smi
Next Steps
- Architecture - Understand the system
- API Reference - Endpoint documentation
- Contributing - Submit your processor
Questions? Open an issue on GitHub or check existing processors for examples!
Last Updated: 2026-01-03 | V2 Developer Guide
Esc
Searching...
No results found.
Type to search the documentation
↑↓ Navigate
↵ Select
Esc Close