From c26aec916ea36b4f633ee2e58fd9365ed41fff7a Mon Sep 17 00:00:00 2001 From: "alexei.dolgolyov" Date: Tue, 24 Mar 2026 13:58:04 +0300 Subject: [PATCH] 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. --- server/src/wled_controller/api/__init__.py | 2 + .../src/wled_controller/api/dependencies.py | 7 + .../src/wled_controller/api/routes/backup.py | 1 + .../wled_controller/api/routes/gradients.py | 153 +++++++++++++++ .../wled_controller/api/schemas/gradients.py | 51 +++++ server/src/wled_controller/config.py | 1 + .../src/wled_controller/storage/gradient.py | 83 ++++++++ .../wled_controller/storage/gradient_store.py | 178 ++++++++++++++++++ 8 files changed, 476 insertions(+) create mode 100644 server/src/wled_controller/api/routes/gradients.py create mode 100644 server/src/wled_controller/api/schemas/gradients.py create mode 100644 server/src/wled_controller/storage/gradient.py create mode 100644 server/src/wled_controller/storage/gradient_store.py diff --git a/server/src/wled_controller/api/__init__.py b/server/src/wled_controller/api/__init__.py index da7b7fa..d76cb32 100644 --- a/server/src/wled_controller/api/__init__.py +++ b/server/src/wled_controller/api/__init__.py @@ -23,6 +23,7 @@ from .routes.scene_presets import router as scene_presets_router from .routes.webhooks import router as webhooks_router from .routes.sync_clocks import router as sync_clocks_router from .routes.color_strip_processing import router as cspt_router +from .routes.gradients import router as gradients_router router = APIRouter() router.include_router(system_router) @@ -46,5 +47,6 @@ router.include_router(scene_presets_router) router.include_router(webhooks_router) router.include_router(sync_clocks_router) router.include_router(cspt_router) +router.include_router(gradients_router) __all__ = ["router"] diff --git a/server/src/wled_controller/api/dependencies.py b/server/src/wled_controller/api/dependencies.py index 8c334fb..793440e 100644 --- a/server/src/wled_controller/api/dependencies.py +++ b/server/src/wled_controller/api/dependencies.py @@ -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.sync_clock_store import SyncClockStore 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.backup.auto_backup import AutoBackupEngine 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") +def get_gradient_store() -> GradientStore: + return _get("gradient_store", "Gradient store") + + # ── Event helper ──────────────────────────────────────────────────────── @@ -157,6 +162,7 @@ def init_dependencies( sync_clock_store: SyncClockStore | None = None, sync_clock_manager: SyncClockManager | None = None, cspt_store: ColorStripProcessingTemplateStore | None = None, + gradient_store: GradientStore | None = None, ): """Initialize global dependencies.""" _deps.update({ @@ -178,4 +184,5 @@ def init_dependencies( "sync_clock_store": sync_clock_store, "sync_clock_manager": sync_clock_manager, "cspt_store": cspt_store, + "gradient_store": gradient_store, }) diff --git a/server/src/wled_controller/api/routes/backup.py b/server/src/wled_controller/api/routes/backup.py index 6b5eea7..187e389 100644 --- a/server/src/wled_controller/api/routes/backup.py +++ b/server/src/wled_controller/api/routes/backup.py @@ -54,6 +54,7 @@ STORE_MAP = { "color_strip_processing_templates": "color_strip_processing_templates_file", "automations": "automations_file", "scene_presets": "scene_presets_file", + "gradients": "gradients_file", } _SERVER_DIR = Path(__file__).resolve().parents[4] diff --git a/server/src/wled_controller/api/routes/gradients.py b/server/src/wled_controller/api/routes/gradients.py new file mode 100644 index 0000000..a5a857f --- /dev/null +++ b/server/src/wled_controller/api/routes/gradients.py @@ -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)) diff --git a/server/src/wled_controller/api/schemas/gradients.py b/server/src/wled_controller/api/schemas/gradients.py new file mode 100644 index 0000000..b2104d0 --- /dev/null +++ b/server/src/wled_controller/api/schemas/gradients.py @@ -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") diff --git a/server/src/wled_controller/config.py b/server/src/wled_controller/config.py index 3ba8448..be11681 100644 --- a/server/src/wled_controller/config.py +++ b/server/src/wled_controller/config.py @@ -41,6 +41,7 @@ class StorageConfig(BaseSettings): scene_presets_file: str = "data/scene_presets.json" color_strip_processing_templates_file: str = "data/color_strip_processing_templates.json" sync_clocks_file: str = "data/sync_clocks.json" + gradients_file: str = "data/gradients.json" class MQTTConfig(BaseSettings): diff --git a/server/src/wled_controller/storage/gradient.py b/server/src/wled_controller/storage/gradient.py new file mode 100644 index 0000000..8580fc5 --- /dev/null +++ b/server/src/wled_controller/storage/gradient.py @@ -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"] diff --git a/server/src/wled_controller/storage/gradient_store.py b/server/src/wled_controller/storage/gradient_store.py new file mode 100644 index 0000000..b660ebf --- /dev/null +++ b/server/src/wled_controller/storage/gradient_store.py @@ -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