feat: add gradient entity with CRUD API and storage

Reusable gradient definitions with built-in presets (rainbow, sunset,
ocean, etc.) and user-created gradients. Includes model, JSON store,
Pydantic schemas, REST routes (list/create/update/clone/delete), and
backup/restore integration.
This commit is contained in:
2026-03-24 13:58:04 +03:00
parent 2c3f08344c
commit c26aec916e
8 changed files with 476 additions and 0 deletions

View File

@@ -23,6 +23,7 @@ from .routes.scene_presets import router as scene_presets_router
from .routes.webhooks import router as webhooks_router from .routes.webhooks import router as webhooks_router
from .routes.sync_clocks import router as sync_clocks_router from .routes.sync_clocks import router as sync_clocks_router
from .routes.color_strip_processing import router as cspt_router from .routes.color_strip_processing import router as cspt_router
from .routes.gradients import router as gradients_router
router = APIRouter() router = APIRouter()
router.include_router(system_router) router.include_router(system_router)
@@ -46,5 +47,6 @@ router.include_router(scene_presets_router)
router.include_router(webhooks_router) router.include_router(webhooks_router)
router.include_router(sync_clocks_router) router.include_router(sync_clocks_router)
router.include_router(cspt_router) router.include_router(cspt_router)
router.include_router(gradients_router)
__all__ = ["router"] __all__ = ["router"]

View File

@@ -21,6 +21,7 @@ from wled_controller.storage.automation_store import AutomationStore
from wled_controller.storage.scene_preset_store import ScenePresetStore from wled_controller.storage.scene_preset_store import ScenePresetStore
from wled_controller.storage.sync_clock_store import SyncClockStore from wled_controller.storage.sync_clock_store import SyncClockStore
from wled_controller.storage.color_strip_processing_template_store import ColorStripProcessingTemplateStore from wled_controller.storage.color_strip_processing_template_store import ColorStripProcessingTemplateStore
from wled_controller.storage.gradient_store import GradientStore
from wled_controller.core.automations.automation_engine import AutomationEngine from wled_controller.core.automations.automation_engine import AutomationEngine
from wled_controller.core.backup.auto_backup import AutoBackupEngine from wled_controller.core.backup.auto_backup import AutoBackupEngine
from wled_controller.core.processing.sync_clock_manager import SyncClockManager from wled_controller.core.processing.sync_clock_manager import SyncClockManager
@@ -114,6 +115,10 @@ def get_cspt_store() -> ColorStripProcessingTemplateStore:
return _get("cspt_store", "Color strip processing template store") return _get("cspt_store", "Color strip processing template store")
def get_gradient_store() -> GradientStore:
return _get("gradient_store", "Gradient store")
# ── Event helper ──────────────────────────────────────────────────────── # ── Event helper ────────────────────────────────────────────────────────
@@ -157,6 +162,7 @@ def init_dependencies(
sync_clock_store: SyncClockStore | None = None, sync_clock_store: SyncClockStore | None = None,
sync_clock_manager: SyncClockManager | None = None, sync_clock_manager: SyncClockManager | None = None,
cspt_store: ColorStripProcessingTemplateStore | None = None, cspt_store: ColorStripProcessingTemplateStore | None = None,
gradient_store: GradientStore | None = None,
): ):
"""Initialize global dependencies.""" """Initialize global dependencies."""
_deps.update({ _deps.update({
@@ -178,4 +184,5 @@ def init_dependencies(
"sync_clock_store": sync_clock_store, "sync_clock_store": sync_clock_store,
"sync_clock_manager": sync_clock_manager, "sync_clock_manager": sync_clock_manager,
"cspt_store": cspt_store, "cspt_store": cspt_store,
"gradient_store": gradient_store,
}) })

View File

@@ -54,6 +54,7 @@ STORE_MAP = {
"color_strip_processing_templates": "color_strip_processing_templates_file", "color_strip_processing_templates": "color_strip_processing_templates_file",
"automations": "automations_file", "automations": "automations_file",
"scene_presets": "scene_presets_file", "scene_presets": "scene_presets_file",
"gradients": "gradients_file",
} }
_SERVER_DIR = Path(__file__).resolve().parents[4] _SERVER_DIR = Path(__file__).resolve().parents[4]

View File

@@ -0,0 +1,153 @@
"""Gradient routes: CRUD for reusable gradient definitions."""
from fastapi import APIRouter, Depends, HTTPException
from wled_controller.api.auth import AuthRequired
from wled_controller.api.dependencies import (
fire_entity_event,
get_color_strip_store,
get_gradient_store,
)
from wled_controller.api.schemas.gradients import (
GradientCreate,
GradientListResponse,
GradientResponse,
GradientUpdate,
)
from wled_controller.storage.gradient import Gradient
from wled_controller.storage.gradient_store import GradientStore
from wled_controller.storage.color_strip_store import ColorStripStore
from wled_controller.storage.base_store import EntityNotFoundError
from wled_controller.utils import get_logger
logger = get_logger(__name__)
router = APIRouter()
def _to_response(gradient: Gradient) -> GradientResponse:
return GradientResponse(
id=gradient.id,
name=gradient.name,
stops=[{"position": s["position"], "color": s["color"]} for s in gradient.stops],
is_builtin=gradient.is_builtin,
description=gradient.description,
tags=gradient.tags,
created_at=gradient.created_at,
updated_at=gradient.updated_at,
)
@router.get("/api/v1/gradients", response_model=GradientListResponse, tags=["Gradients"])
async def list_gradients(
_auth: AuthRequired,
store: GradientStore = Depends(get_gradient_store),
):
"""List all gradients (built-in + user-created)."""
gradients = store.get_all_gradients()
return GradientListResponse(
gradients=[_to_response(g) for g in gradients],
count=len(gradients),
)
@router.post("/api/v1/gradients", response_model=GradientResponse, status_code=201, tags=["Gradients"])
async def create_gradient(
data: GradientCreate,
_auth: AuthRequired,
store: GradientStore = Depends(get_gradient_store),
):
"""Create a new user-defined gradient."""
try:
gradient = store.create_gradient(
name=data.name,
stops=[s.model_dump() for s in data.stops],
description=data.description,
tags=data.tags,
)
fire_entity_event("gradient", "created", gradient.id)
return _to_response(gradient)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
@router.get("/api/v1/gradients/{gradient_id}", response_model=GradientResponse, tags=["Gradients"])
async def get_gradient(
gradient_id: str,
_auth: AuthRequired,
store: GradientStore = Depends(get_gradient_store),
):
"""Get a gradient by ID."""
try:
gradient = store.get_gradient(gradient_id)
return _to_response(gradient)
except (ValueError, EntityNotFoundError) as e:
raise HTTPException(status_code=404, detail=str(e))
@router.put("/api/v1/gradients/{gradient_id}", response_model=GradientResponse, tags=["Gradients"])
async def update_gradient(
gradient_id: str,
data: GradientUpdate,
_auth: AuthRequired,
store: GradientStore = Depends(get_gradient_store),
):
"""Update a gradient (built-in gradients are read-only)."""
try:
stops = [s.model_dump() for s in data.stops] if data.stops is not None else None
gradient = store.update_gradient(
gradient_id=gradient_id,
name=data.name,
stops=stops,
description=data.description,
tags=data.tags,
)
fire_entity_event("gradient", "updated", gradient_id)
return _to_response(gradient)
except (ValueError, EntityNotFoundError) as e:
status = 404 if "not found" in str(e).lower() else 400
raise HTTPException(status_code=status, detail=str(e))
@router.post("/api/v1/gradients/{gradient_id}/clone", response_model=GradientResponse, status_code=201, tags=["Gradients"])
async def clone_gradient(
gradient_id: str,
_auth: AuthRequired,
store: GradientStore = Depends(get_gradient_store),
):
"""Clone a gradient (useful for customizing built-in gradients)."""
try:
original = store.get_gradient(gradient_id)
clone = store.create_gradient(
name=f"{original.name} (copy)",
stops=original.stops,
description=original.description,
tags=original.tags,
)
fire_entity_event("gradient", "created", clone.id)
return _to_response(clone)
except (ValueError, EntityNotFoundError) as e:
status = 404 if "not found" in str(e).lower() else 400
raise HTTPException(status_code=status, detail=str(e))
@router.delete("/api/v1/gradients/{gradient_id}", status_code=204, tags=["Gradients"])
async def delete_gradient(
gradient_id: str,
_auth: AuthRequired,
store: GradientStore = Depends(get_gradient_store),
css_store: ColorStripStore = Depends(get_color_strip_store),
):
"""Delete a gradient (fails if built-in or referenced by sources)."""
try:
# Check references
for source in css_store.get_all_sources():
if getattr(source, "gradient_id", None) == gradient_id:
raise ValueError(
f"Cannot delete: referenced by color strip source '{source.name}'"
)
store.delete_gradient(gradient_id)
fire_entity_event("gradient", "deleted", gradient_id)
except (ValueError, EntityNotFoundError) as e:
status = 404 if "not found" in str(e).lower() else 400
raise HTTPException(status_code=status, detail=str(e))

View File

@@ -0,0 +1,51 @@
"""Gradient schemas (CRUD)."""
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field
class GradientStopSchema(BaseModel):
"""A single gradient color stop."""
position: float = Field(description="Position along gradient (0.0-1.0)", ge=0.0, le=1.0)
color: List[int] = Field(description="RGB color [R, G, B]", min_length=3, max_length=3)
class GradientCreate(BaseModel):
"""Request to create a gradient."""
name: str = Field(description="Gradient name", min_length=1, max_length=100)
stops: List[GradientStopSchema] = Field(description="Color stops", min_length=2)
description: Optional[str] = Field(None, description="Optional description", max_length=500)
tags: List[str] = Field(default_factory=list, description="User-defined tags")
class GradientUpdate(BaseModel):
"""Request to update a gradient."""
name: Optional[str] = Field(None, description="Gradient name", min_length=1, max_length=100)
stops: Optional[List[GradientStopSchema]] = Field(None, description="Color stops", min_length=2)
description: Optional[str] = Field(None, description="Optional description", max_length=500)
tags: Optional[List[str]] = None
class GradientResponse(BaseModel):
"""Gradient response."""
id: str = Field(description="Gradient ID")
name: str = Field(description="Gradient name")
stops: List[GradientStopSchema] = Field(description="Color stops")
is_builtin: bool = Field(description="Whether this is a built-in gradient")
description: Optional[str] = Field(None, description="Description")
tags: List[str] = Field(default_factory=list, description="User-defined tags")
created_at: datetime = Field(description="Creation timestamp")
updated_at: datetime = Field(description="Last update timestamp")
class GradientListResponse(BaseModel):
"""List of gradients."""
gradients: List[GradientResponse] = Field(description="List of gradients")
count: int = Field(description="Number of gradients")

View File

@@ -41,6 +41,7 @@ class StorageConfig(BaseSettings):
scene_presets_file: str = "data/scene_presets.json" scene_presets_file: str = "data/scene_presets.json"
color_strip_processing_templates_file: str = "data/color_strip_processing_templates.json" color_strip_processing_templates_file: str = "data/color_strip_processing_templates.json"
sync_clocks_file: str = "data/sync_clocks.json" sync_clocks_file: str = "data/sync_clocks.json"
gradients_file: str = "data/gradients.json"
class MQTTConfig(BaseSettings): class MQTTConfig(BaseSettings):

View File

@@ -0,0 +1,83 @@
"""Gradient data model.
A Gradient defines a reusable color gradient as a list of color stops.
Gradients are referenced by ID from effect, gradient, and audio color
strip sources. Eight built-in gradients are seeded on first run.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional
@dataclass
class Gradient:
"""Persistent gradient definition with color stops."""
id: str
name: str
stops: list # [{"position": float, "color": [R, G, B]}, ...]
is_builtin: bool
created_at: datetime
updated_at: datetime
description: Optional[str] = None
tags: List[str] = field(default_factory=list)
def to_dict(self) -> dict:
return {
"id": self.id,
"name": self.name,
"stops": self.stops,
"is_builtin": self.is_builtin,
"description": self.description,
"tags": self.tags,
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
}
@staticmethod
def from_dict(data: dict) -> "Gradient":
return Gradient(
id=data["id"],
name=data["name"],
stops=data.get("stops", []),
is_builtin=data.get("is_builtin", False),
description=data.get("description"),
tags=data.get("tags", []),
created_at=datetime.fromisoformat(data["created_at"]),
updated_at=datetime.fromisoformat(data["updated_at"]),
)
@classmethod
def create_from_kwargs(
cls,
*,
id: str,
name: str,
stops: list,
is_builtin: bool = False,
created_at: datetime,
updated_at: datetime,
description: Optional[str] = None,
tags: Optional[List[str]] = None,
) -> "Gradient":
return cls(
id=id,
name=name,
stops=stops,
is_builtin=is_builtin,
created_at=created_at,
updated_at=updated_at,
description=description,
tags=tags or [],
)
def apply_update(self, **kwargs) -> None:
if kwargs.get("name") is not None:
self.name = kwargs["name"]
if kwargs.get("stops") is not None:
self.stops = kwargs["stops"]
if kwargs.get("description") is not None:
self.description = kwargs["description"]
if kwargs.get("tags") is not None:
self.tags = kwargs["tags"]

View File

@@ -0,0 +1,178 @@
"""Gradient storage with built-in seeding.
Provides CRUD for gradient entities. On first run (empty/missing file),
seeds 8 built-in gradients matching the legacy hardcoded palettes.
Built-in gradients are read-only and cannot be deleted or modified.
"""
import uuid
from datetime import datetime, timezone
from typing import List, Optional
from wled_controller.storage.base_store import BaseJsonStore
from wled_controller.storage.gradient import Gradient
from wled_controller.utils import get_logger
logger = get_logger(__name__)
# Built-in gradient definitions — matches the legacy _PALETTE_DEFS.
# Format: (position, R, G, B) tuples converted to stop dicts on seed.
_BUILTIN_DEFS = {
"fire": [(0, 0, 0, 0), (0.33, 200, 24, 0), (0.66, 255, 160, 0), (1.0, 255, 255, 200)],
"ocean": [(0, 0, 0, 32), (0.33, 0, 16, 128), (0.66, 0, 128, 255), (1.0, 128, 224, 255)],
"lava": [(0, 0, 0, 0), (0.25, 128, 0, 0), (0.5, 255, 32, 0), (0.75, 255, 160, 0), (1.0, 255, 255, 128)],
"forest": [(0, 0, 16, 0), (0.33, 0, 80, 0), (0.66, 32, 160, 0), (1.0, 128, 255, 64)],
"rainbow": [
(0, 255, 0, 0), (0.17, 255, 255, 0), (0.33, 0, 255, 0),
(0.5, 0, 255, 255), (0.67, 0, 0, 255), (0.83, 255, 0, 255), (1.0, 255, 0, 0),
],
"aurora": [
(0, 0, 16, 32), (0.2, 0, 80, 64), (0.4, 0, 200, 100),
(0.6, 64, 128, 255), (0.8, 128, 0, 200), (1.0, 0, 16, 32),
],
"sunset": [
(0, 32, 0, 64), (0.25, 128, 0, 128), (0.5, 255, 64, 0),
(0.75, 255, 192, 64), (1.0, 255, 255, 192),
],
"ice": [(0, 0, 0, 64), (0.33, 0, 64, 192), (0.66, 128, 192, 255), (1.0, 240, 248, 255)],
}
def _tuples_to_stops(tuples: list) -> list:
"""Convert [(pos, R, G, B), ...] to [{"position": pos, "color": [R, G, B]}, ...]."""
return [{"position": t[0], "color": [t[1], t[2], t[3]]} for t in tuples]
class GradientStore(BaseJsonStore[Gradient]):
_json_key = "gradients"
_entity_name = "Gradient"
def __init__(self, file_path: str):
super().__init__(file_path, Gradient.from_dict)
if not self._items:
self._seed_builtins()
def _seed_builtins(self) -> None:
"""Create the 8 built-in gradients on first run."""
now = datetime.now(timezone.utc)
for name, tuples in _BUILTIN_DEFS.items():
gid = f"gr_builtin_{name}"
self._items[gid] = Gradient(
id=gid,
name=name.capitalize(),
stops=_tuples_to_stops(tuples),
is_builtin=True,
created_at=now,
updated_at=now,
description=f"Built-in {name} gradient",
)
self._save()
logger.info(f"Seeded {len(_BUILTIN_DEFS)} built-in gradients")
# Aliases
get_all_gradients = BaseJsonStore.get_all
def get_gradient(self, gradient_id: str) -> Gradient:
return self.get(gradient_id)
def get_by_name(self, name: str) -> Optional[Gradient]:
"""Look up a gradient by name (case-insensitive)."""
lower = name.lower()
for g in self._items.values():
if g.name.lower() == lower:
return g
return None
def create_gradient(
self,
name: str,
stops: list,
description: Optional[str] = None,
tags: Optional[List[str]] = None,
) -> Gradient:
self._check_name_unique(name)
gid = f"gr_{uuid.uuid4().hex[:8]}"
now = datetime.now(timezone.utc)
gradient = Gradient(
id=gid,
name=name,
stops=stops,
is_builtin=False,
created_at=now,
updated_at=now,
description=description,
tags=tags or [],
)
self._items[gid] = gradient
self._save()
logger.info(f"Created gradient: {name} ({gid})")
return gradient
def update_gradient(
self,
gradient_id: str,
name: Optional[str] = None,
stops: Optional[list] = None,
description: Optional[str] = None,
tags: Optional[List[str]] = None,
) -> Gradient:
gradient = self.get(gradient_id)
if gradient.is_builtin:
raise ValueError("Built-in gradients are read-only. Clone to edit.")
if name is not None:
self._check_name_unique(name, exclude_id=gradient_id)
gradient.name = name
if stops is not None:
gradient.stops = stops
if description is not None:
gradient.description = description
if tags is not None:
gradient.tags = tags
gradient.updated_at = datetime.now(timezone.utc)
self._save()
logger.info(f"Updated gradient: {gradient_id}")
return gradient
def delete_gradient(self, gradient_id: str) -> None:
gradient = self.get(gradient_id)
if gradient.is_builtin:
raise ValueError("Built-in gradients cannot be deleted")
self.delete(gradient_id)
def get_stops_by_id(self, gradient_id: str) -> Optional[list]:
"""Return stops list for a gradient ID, or None if not found."""
gradient = self._items.get(gradient_id)
if gradient is None:
return None
return gradient.stops
def resolve_stops(self, gradient_id: Optional[str]) -> Optional[list]:
"""Resolve a gradient_id to its stops list, or None if invalid/missing."""
if not gradient_id:
return None
return self.get_stops_by_id(gradient_id)
def migrate_palette_references(self, color_strip_store) -> int:
"""One-way migration: convert legacy palette names to gradient_id.
For each color strip source that has a `palette` field but no `gradient_id`,
sets `gradient_id` to the matching builtin gradient ID.
Returns the number of sources migrated.
"""
migrated = 0
for source in color_strip_store.get_all_sources():
if getattr(source, "gradient_id", None):
continue # already has a gradient_id
palette_name = getattr(source, "palette", None)
if not palette_name:
continue
# Map legacy palette name to builtin gradient ID
builtin_id = f"gr_builtin_{palette_name}"
if builtin_id in self._items:
source.gradient_id = builtin_id
migrated += 1
if migrated:
color_strip_store._save()
logger.info(f"Migrated {migrated} color strip sources from palette names to gradient IDs")
return migrated