feat: typed script parameters with validation and icon-grid selector
Release / create-release (push) Successful in 1s
Lint & Test / test (push) Successful in 10s
Release / build-linux (push) Successful in 36s
Release / build-windows (push) Successful in 1m15s

- Add ScriptParameterConfig model (string, integer, float, boolean, select types)
- Server-side validation at both define-time and execute-time
- Parameters passed as SCRIPT_PARAM_* environment variables
- Web UI parameter editor in script create/edit dialog (add/remove/reorder)
- Icon-grid selector component (ported from wled-screen-controller)
- Replace audio device dropdown with icon-grid selector
- Replace callback event dropdown with icon-grid selector
- Localization for parameter UI (en, ru)
This commit is contained in:
2026-03-25 11:25:03 +03:00
parent 1c0a011342
commit 1410a8d2cb
12 changed files with 1211 additions and 26 deletions
+239 -16
View File
@@ -12,7 +12,7 @@ from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from ..auth import verify_token
from ..config import ScriptConfig, settings
from ..config import ScriptConfig, ScriptParameterConfig, settings
from ..config_manager import config_manager
from ..services.websocket_manager import ws_manager
@@ -24,9 +24,11 @@ logger = logging.getLogger(__name__)
class ScriptExecuteRequest(BaseModel):
"""Request model for script execution with optional arguments."""
"""Request model for script execution with optional parameters."""
args: list[str] = Field(default_factory=list, description="Additional arguments")
params: dict[str, str | int | float | bool] = Field(
default_factory=dict, description="Named parameters (validated against script schema)"
)
class ScriptExecuteResponse(BaseModel):
@@ -41,6 +43,18 @@ class ScriptExecuteResponse(BaseModel):
execution_time: float | None = None
class ScriptParameterInfo(BaseModel):
"""Information about a script parameter."""
type: str
description: str = ""
required: bool = False
default: str | int | float | bool | None = None
min: float | None = None
max: float | None = None
options: list[str] | None = None
class ScriptInfo(BaseModel):
"""Information about an available script."""
@@ -50,6 +64,7 @@ class ScriptInfo(BaseModel):
description: str
icon: str | None = None
timeout: int
parameters: dict[str, ScriptParameterInfo] = Field(default_factory=dict)
@router.get("/list")
@@ -67,11 +82,126 @@ async def list_scripts(_: str = Depends(verify_token)) -> list[ScriptInfo]:
description=config.description,
icon=config.icon,
timeout=config.timeout,
parameters={
pname: ScriptParameterInfo(**pconfig.model_dump())
for pname, pconfig in config.parameters.items()
},
)
for name, config in settings.scripts.items()
]
def _validate_params(
params: dict[str, str | int | float | bool],
param_defs: dict[str, ScriptParameterConfig],
) -> dict[str, str]:
"""Validate parameters against script schema and return env vars.
Args:
params: User-supplied parameter values.
param_defs: Parameter definitions from script config.
Returns:
Dict of environment variables (SCRIPT_PARAM_<NAME> -> str value).
Raises:
HTTPException: On validation failure.
"""
# Reject unknown parameters
unknown = set(params.keys()) - set(param_defs.keys())
if unknown:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Unknown parameters: {', '.join(sorted(unknown))}",
)
env_vars: dict[str, str] = {}
for pname, pdef in param_defs.items():
value = params.get(pname)
# Apply default if missing
if value is None and pdef.default is not None:
value = pdef.default
# Check required
if value is None:
if pdef.required:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Required parameter '{pname}' is missing",
)
continue
# Type validation and coercion
if pdef.type == "integer":
try:
value = int(value)
except (TypeError, ValueError):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be an integer, got: {value!r}",
)
if pdef.min is not None and value < pdef.min:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be >= {pdef.min}, got: {value}",
)
if pdef.max is not None and value > pdef.max:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be <= {pdef.max}, got: {value}",
)
elif pdef.type == "float":
try:
value = float(value)
except (TypeError, ValueError):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be a number, got: {value!r}",
)
if pdef.min is not None and value < pdef.min:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be >= {pdef.min}, got: {value}",
)
if pdef.max is not None and value > pdef.max:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be <= {pdef.max}, got: {value}",
)
elif pdef.type == "boolean":
if isinstance(value, str):
if value.lower() in ("true", "1", "yes"):
value = True
elif value.lower() in ("false", "0", "no"):
value = False
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be a boolean, got: {value!r}",
)
elif not isinstance(value, bool):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be a boolean, got: {value!r}",
)
elif pdef.type == "select":
value = str(value)
if pdef.options and value not in pdef.options:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' must be one of {pdef.options}, got: {value!r}",
)
else:
# string — just convert to str
value = str(value)
env_vars[f"SCRIPT_PARAM_{pname.upper()}"] = str(value)
return env_vars
@router.post("/execute/{script_name}")
async def execute_script(
script_name: str,
@@ -82,7 +212,7 @@ async def execute_script(
Args:
script_name: Name of the script to execute (must be defined in config)
request: Optional arguments to pass to the script
request: Optional parameters to pass to the script
Returns:
Execution result including stdout, stderr, and exit code
@@ -94,26 +224,24 @@ async def execute_script(
detail=f"Script '{script_name}' not found. Use /api/scripts/list to see available scripts.",
)
script_config = settings.scripts[script_name]
args = request.args if request else []
params = request.params if request else {}
# Validate parameters and build env vars
extra_env = _validate_params(params, script_config.parameters)
logger.info(f"Executing script: {script_name}")
try:
# Build command
command = script_config.command
if args:
# Append arguments to command
command = f"{command} {' '.join(args)}"
# Execute in dedicated thread pool to not block the default executor
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
_script_executor,
lambda: _run_script(
command=command,
command=script_config.command,
timeout=script_config.timeout,
shell=script_config.shell,
working_dir=script_config.working_dir,
extra_env=extra_env,
),
)
@@ -140,6 +268,7 @@ def _run_script(
timeout: int,
shell: bool,
working_dir: str | None,
extra_env: dict[str, str] | None = None,
) -> dict[str, Any]:
"""Run a script synchronously.
@@ -148,11 +277,16 @@ def _run_script(
timeout: Timeout in seconds
shell: Whether to run in shell
working_dir: Working directory
extra_env: Additional environment variables (e.g. SCRIPT_PARAM_*)
Returns:
Dict with exit_code, stdout, stderr, execution_time
"""
start_time = time.time()
env = None
if extra_env:
import os
env = {**os.environ, **extra_env}
try:
result = subprocess.run(
command,
@@ -161,6 +295,7 @@ def _run_script(
capture_output=True,
text=True,
timeout=timeout,
env=env,
)
execution_time = time.time() - start_time
return {
@@ -190,6 +325,24 @@ def _run_script(
# Script management endpoints
class ScriptParameterCreateRequest(BaseModel):
"""Request model for a script parameter definition."""
type: str = Field(
..., description="Parameter type: string, integer, float, boolean, select"
)
description: str = Field(default="", description="Parameter description")
required: bool = Field(default=False, description="Whether the parameter is required")
default: str | int | float | bool | None = Field(
default=None, description="Default value if not provided"
)
min: float | None = Field(default=None, description="Minimum value (numeric types only)")
max: float | None = Field(default=None, description="Maximum value (numeric types only)")
options: list[str] | None = Field(
default=None, description="Allowed values (select type only)"
)
class ScriptCreateRequest(BaseModel):
"""Request model for creating or updating a script."""
@@ -200,6 +353,60 @@ class ScriptCreateRequest(BaseModel):
timeout: int = Field(default=30, description="Execution timeout in seconds", ge=1, le=300)
working_dir: str | None = Field(default=None, description="Working directory")
shell: bool = Field(default=True, description="Run command in shell")
parameters: dict[str, ScriptParameterCreateRequest] = Field(
default_factory=dict, description="Named parameters with type and validation rules"
)
def _validate_parameter_definitions(
parameters: dict[str, ScriptParameterCreateRequest],
) -> None:
"""Validate parameter definitions are well-formed.
Args:
parameters: Parameter definitions to validate.
Raises:
HTTPException: If any definition is invalid.
"""
valid_types = {"string", "integer", "float", "boolean", "select"}
for pname, pdef in parameters.items():
if not re.match(r"^[a-zA-Z][a-zA-Z0-9_]*$", pname):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=(
f"Parameter name '{pname}' must start with a letter"
" and contain only alphanumeric characters and underscores"
),
)
if pdef.type not in valid_types:
allowed = ", ".join(sorted(valid_types))
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' has invalid type '{pdef.type}'. Must be one of: {allowed}",
)
if pdef.type == "select":
if not pdef.options or len(pdef.options) == 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}' of type 'select' must have a non-empty 'options' list",
)
if pdef.type not in ("integer", "float"):
if pdef.min is not None or pdef.max is not None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}': 'min'/'max' are only valid for integer/float types",
)
if pdef.min is not None and pdef.max is not None and pdef.min > pdef.max:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Parameter '{pname}': 'min' ({pdef.min}) must be <= 'max' ({pdef.max})",
)
def _validate_script_name(name: str) -> None:
@@ -258,8 +465,16 @@ async def create_script(
detail=f"Script '{script_name}' already exists. Use PUT /api/scripts/update/{script_name} to update it.",
)
# Create script config
script_config = ScriptConfig(**request.model_dump())
# Validate parameter definitions
_validate_parameter_definitions(request.parameters)
# Build ScriptConfig with ScriptParameterConfig instances
data = request.model_dump()
data["parameters"] = {
pname: ScriptParameterConfig(**pdef)
for pname, pdef in data.get("parameters", {}).items()
}
script_config = ScriptConfig(**data)
# Add to config file and in-memory
try:
@@ -306,8 +521,16 @@ async def update_script(
detail=f"Script '{script_name}' not found. Use POST /api/scripts/create/{script_name} to create it.",
)
# Create updated script config
script_config = ScriptConfig(**request.model_dump())
# Validate parameter definitions
_validate_parameter_definitions(request.parameters)
# Build ScriptConfig with ScriptParameterConfig instances
data = request.model_dump()
data["parameters"] = {
pname: ScriptParameterConfig(**pdef)
for pname, pdef in data.get("parameters", {}).items()
}
script_config = ScriptConfig(**data)
# Update config file and in-memory
try: