Add noise gate, palette quantization filters and drag-and-drop filter ordering
- Add noise gate filter: suppresses per-pixel color flicker below threshold using stateful frame comparison with pre-allocated int16 buffers - Add palette quantization filter: maps pixels to nearest color in preset or custom hex palette, using chunked processing for memory efficiency - Add "string" option type to filter schema system (base, API, frontend) - Replace up/down buttons with pointer-event drag-and-drop in PP template filter list, with clone/placeholder feedback and modal auto-scroll - Add frame_interpolation locale keys (was missing from all 3 locales) - Update TODO.md: mark completed processing pipeline items Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -20,6 +20,8 @@ import wled_controller.core.filters.flip # noqa: F401
|
||||
import wled_controller.core.filters.color_correction # noqa: F401
|
||||
import wled_controller.core.filters.frame_interpolation # noqa: F401
|
||||
import wled_controller.core.filters.filter_template # noqa: F401
|
||||
import wled_controller.core.filters.noise_gate # noqa: F401
|
||||
import wled_controller.core.filters.palette_quantization # noqa: F401
|
||||
|
||||
__all__ = [
|
||||
"FilterOptionDef",
|
||||
|
||||
@@ -13,12 +13,13 @@ class FilterOptionDef:
|
||||
|
||||
key: str
|
||||
label: str
|
||||
option_type: str # "float" | "int" | "bool" | "select"
|
||||
option_type: str # "float" | "int" | "bool" | "select" | "string"
|
||||
default: Any
|
||||
min_value: Any
|
||||
max_value: Any
|
||||
step: Any
|
||||
choices: Optional[List[Dict[str, str]]] = None # for "select": [{value, label}]
|
||||
max_length: Optional[int] = None # for "string" type
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = {
|
||||
@@ -32,6 +33,8 @@ class FilterOptionDef:
|
||||
}
|
||||
if self.choices is not None:
|
||||
d["choices"] = self.choices
|
||||
if self.max_length is not None:
|
||||
d["max_length"] = self.max_length
|
||||
return d
|
||||
|
||||
|
||||
@@ -86,10 +89,12 @@ class PostprocessingFilter(ABC):
|
||||
val = bool(raw) if not isinstance(raw, bool) else raw
|
||||
elif opt_def.option_type == "select":
|
||||
val = str(raw) if raw is not None else opt_def.default
|
||||
elif opt_def.option_type == "string":
|
||||
val = str(raw) if raw is not None else opt_def.default
|
||||
else:
|
||||
val = raw
|
||||
# Clamp to range (skip for bools and selects)
|
||||
if opt_def.option_type not in ("bool", "select"):
|
||||
# Clamp to range (skip for non-numeric types)
|
||||
if opt_def.option_type not in ("bool", "select", "string"):
|
||||
if opt_def.min_value is not None and val < opt_def.min_value:
|
||||
val = opt_def.min_value
|
||||
if opt_def.max_value is not None and val > opt_def.max_value:
|
||||
|
||||
82
server/src/wled_controller/core/filters/noise_gate.py
Normal file
82
server/src/wled_controller/core/filters/noise_gate.py
Normal file
@@ -0,0 +1,82 @@
|
||||
"""Noise gate postprocessing filter."""
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import numpy as np
|
||||
|
||||
from wled_controller.core.filters.base import FilterOptionDef, PostprocessingFilter
|
||||
from wled_controller.core.filters.image_pool import ImagePool
|
||||
from wled_controller.core.filters.registry import FilterRegistry
|
||||
|
||||
|
||||
@FilterRegistry.register
|
||||
class NoiseGateFilter(PostprocessingFilter):
|
||||
"""Suppresses small per-pixel color changes below threshold.
|
||||
|
||||
Prevents shimmer / flicker on static or near-static content by keeping
|
||||
the previous pixel value when the max per-channel delta is below the
|
||||
configured threshold.
|
||||
"""
|
||||
|
||||
filter_id = "noise_gate"
|
||||
filter_name = "Noise Gate"
|
||||
|
||||
def __init__(self, options: Dict[str, Any]):
|
||||
super().__init__(options)
|
||||
self._prev_frame: Optional[np.ndarray] = None
|
||||
# Pre-allocated scratch buffers (avoid per-frame allocation)
|
||||
self._i16_cur: Optional[np.ndarray] = None
|
||||
self._i16_prev: Optional[np.ndarray] = None
|
||||
self._buf_shape: Optional[tuple] = None
|
||||
|
||||
@classmethod
|
||||
def get_options_schema(cls) -> List[FilterOptionDef]:
|
||||
return [
|
||||
FilterOptionDef(
|
||||
key="threshold",
|
||||
label="Threshold",
|
||||
option_type="int",
|
||||
default=10,
|
||||
min_value=1,
|
||||
max_value=50,
|
||||
step=1,
|
||||
),
|
||||
]
|
||||
|
||||
def process_image(self, image: np.ndarray, image_pool: ImagePool) -> Optional[np.ndarray]:
|
||||
threshold = self.options["threshold"]
|
||||
h, w, c = image.shape
|
||||
shape = (h, w, c)
|
||||
|
||||
if self._prev_frame is None or self._prev_frame.shape != shape:
|
||||
self._prev_frame = image.copy()
|
||||
self._i16_cur = np.empty(shape, dtype=np.int16)
|
||||
self._i16_prev = np.empty(shape, dtype=np.int16)
|
||||
self._buf_shape = shape
|
||||
return None
|
||||
|
||||
# Reallocate scratch if shape changed
|
||||
if self._buf_shape != shape:
|
||||
self._i16_cur = np.empty(shape, dtype=np.int16)
|
||||
self._i16_prev = np.empty(shape, dtype=np.int16)
|
||||
self._buf_shape = shape
|
||||
|
||||
# Compute per-channel absolute delta using int16 to avoid uint8 underflow
|
||||
np.copyto(self._i16_cur, image, casting="unsafe")
|
||||
np.copyto(self._i16_prev, self._prev_frame, casting="unsafe")
|
||||
self._i16_cur -= self._i16_prev
|
||||
np.abs(self._i16_cur, out=self._i16_cur)
|
||||
|
||||
# Per-pixel: max channel delta
|
||||
max_delta = self._i16_cur.max(axis=2) # shape (h, w)
|
||||
|
||||
# Boolean mask: True where pixel is stable (keep previous)
|
||||
mask = max_delta < threshold
|
||||
mask_3d = mask[:, :, np.newaxis] # broadcast to (h, w, 1)
|
||||
|
||||
# Apply: keep previous where stable, otherwise take new
|
||||
np.where(mask_3d, self._prev_frame, image, out=image)
|
||||
|
||||
# Update stored frame
|
||||
np.copyto(self._prev_frame, image)
|
||||
return None
|
||||
118
server/src/wled_controller/core/filters/palette_quantization.py
Normal file
118
server/src/wled_controller/core/filters/palette_quantization.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""Palette quantization postprocessing filter."""
|
||||
|
||||
import re
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import numpy as np
|
||||
|
||||
from wled_controller.core.filters.base import FilterOptionDef, PostprocessingFilter
|
||||
from wled_controller.core.filters.image_pool import ImagePool
|
||||
from wled_controller.core.filters.registry import FilterRegistry
|
||||
|
||||
_PRESETS = {
|
||||
"warm": "#FF4500,#FF6347,#FF8C00,#FFD700,#FFA500,#DC143C",
|
||||
"cool": "#0000FF,#1E90FF,#00BFFF,#00CED1,#4682B4,#6A5ACD",
|
||||
"neon": "#FF00FF,#00FFFF,#00FF00,#FFFF00,#FF0000,#0000FF",
|
||||
"pastel": "#FFB3BA,#FFDFBA,#FFFFBA,#BAFFC9,#BAE1FF,#E8BAFF",
|
||||
"monochrome": "#000000,#404040,#808080,#C0C0C0,#FFFFFF",
|
||||
"rgb": "#FF0000,#00FF00,#0000FF",
|
||||
"rainbow": "#FF0000,#FF7F00,#FFFF00,#00FF00,#0000FF,#4B0082,#8F00FF",
|
||||
"earth": "#8B4513,#A0522D,#D2691E,#228B22,#006400,#2F4F4F",
|
||||
}
|
||||
|
||||
_HEX_RE = re.compile(r"^#?([0-9a-fA-F]{6})$")
|
||||
|
||||
|
||||
def _parse_hex_colors(color_str: str) -> Optional[np.ndarray]:
|
||||
"""Parse comma-separated hex colors into (N, 3) uint8 array."""
|
||||
colors = []
|
||||
for part in color_str.split(","):
|
||||
part = part.strip()
|
||||
m = _HEX_RE.match(part)
|
||||
if m:
|
||||
h = m.group(1)
|
||||
colors.append((int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)))
|
||||
if not colors:
|
||||
return None
|
||||
return np.array(colors, dtype=np.uint8)
|
||||
|
||||
|
||||
@FilterRegistry.register
|
||||
class PaletteQuantizationFilter(PostprocessingFilter):
|
||||
"""Maps every pixel to the nearest color in a palette."""
|
||||
|
||||
filter_id = "palette_quantization"
|
||||
filter_name = "Palette Quantization"
|
||||
|
||||
def __init__(self, options: Dict[str, Any]):
|
||||
super().__init__(options)
|
||||
self._palette: Optional[np.ndarray] = None
|
||||
self._rebuild_palette()
|
||||
|
||||
def _rebuild_palette(self):
|
||||
mode = self.options.get("mode", "preset")
|
||||
if mode == "custom":
|
||||
color_str = self.options.get("colors", "")
|
||||
self._palette = _parse_hex_colors(color_str)
|
||||
else:
|
||||
preset_name = self.options.get("preset", "rainbow")
|
||||
preset_str = _PRESETS.get(preset_name, _PRESETS["rainbow"])
|
||||
self._palette = _parse_hex_colors(preset_str)
|
||||
|
||||
@classmethod
|
||||
def get_options_schema(cls) -> List[FilterOptionDef]:
|
||||
return [
|
||||
FilterOptionDef(
|
||||
key="mode",
|
||||
label="Mode",
|
||||
option_type="select",
|
||||
default="preset",
|
||||
min_value=None,
|
||||
max_value=None,
|
||||
step=None,
|
||||
choices=[
|
||||
{"value": "preset", "label": "Preset"},
|
||||
{"value": "custom", "label": "Custom"},
|
||||
],
|
||||
),
|
||||
FilterOptionDef(
|
||||
key="preset",
|
||||
label="Preset Palette",
|
||||
option_type="select",
|
||||
default="rainbow",
|
||||
min_value=None,
|
||||
max_value=None,
|
||||
step=None,
|
||||
choices=[{"value": k, "label": k.capitalize()} for k in _PRESETS],
|
||||
),
|
||||
FilterOptionDef(
|
||||
key="colors",
|
||||
label="Custom Colors (hex, comma-separated)",
|
||||
option_type="string",
|
||||
default="#FF0000,#00FF00,#0000FF",
|
||||
min_value=None,
|
||||
max_value=None,
|
||||
step=None,
|
||||
max_length=500,
|
||||
),
|
||||
]
|
||||
|
||||
def process_image(self, image: np.ndarray, image_pool: ImagePool) -> Optional[np.ndarray]:
|
||||
if self._palette is None or len(self._palette) == 0:
|
||||
return None
|
||||
|
||||
palette = self._palette.astype(np.int16)
|
||||
flat = image.reshape(-1, 3)
|
||||
total = flat.shape[0]
|
||||
chunk_size = 32768
|
||||
|
||||
for start in range(0, total, chunk_size):
|
||||
end = min(start + chunk_size, total)
|
||||
chunk = flat[start:end].astype(np.int16)
|
||||
# Squared Euclidean distance: (C, 1, 3) - (1, N, 3) -> (C, N, 3) -> sum -> (C, N)
|
||||
diff = chunk[:, np.newaxis, :] - palette[np.newaxis, :, :]
|
||||
dist_sq = (diff * diff).sum(axis=2)
|
||||
nearest = dist_sq.argmin(axis=1)
|
||||
flat[start:end] = self._palette[nearest]
|
||||
|
||||
return None
|
||||
Reference in New Issue
Block a user