Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 79 additions & 1 deletion src/ableton_cli/commands/_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,23 @@

import json
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any
from typing import Any, TypeVar

from ..errors import AppError, ExitCode

NOTE_KEYS = {"pitch", "start_time", "duration", "velocity", "mute"}
TRACK_INDEX_HINT = "Use a valid track index from 'ableton-cli tracks list'."
DEVICE_INDEX_HINT = "Use a valid device index from 'ableton-cli track info'."
SCENE_INDEX_HINT = "Use a valid scene index from 'scenes list'."
SCENE_SOURCE_HINT = "Use a valid source scene index from 'scenes list'."
SCENE_DESTINATION_HINT = "Use a valid destination scene index from 'scenes list'."
SCENE_NAME_HINT = "Pass a non-empty scene name."
SCENE_INSERT_INDEX_HINT = "Use -1 for append or a non-negative insertion index."
TRACK_NAME_HINT = "Pass a non-empty track name."
VOLUME_VALUE_HINT = "Use a normalized volume value such as 0.75."
PAN_VALUE_HINT = "Use a normalized panning value such as -0.25."

TValue = TypeVar("TValue")


def invalid_argument(message: str, hint: str) -> AppError:
Expand All @@ -34,6 +44,10 @@ def require_device_index(value: int, *, hint: str = DEVICE_INDEX_HINT) -> int:
return require_non_negative("device", value, hint=hint)


def require_scene_index(value: int, *, hint: str = SCENE_INDEX_HINT) -> int:
return require_non_negative("scene", value, hint=hint)


def require_parameter_index(value: int, *, hint: str) -> int:
return require_non_negative("parameter", value, hint=hint)

Expand Down Expand Up @@ -72,6 +86,70 @@ def require_float_in_range(
return value


def require_track_and_value(track: int, value: TValue) -> tuple[int, TValue]:
return require_track_index(track), value


def require_optional_track_index(track: int | None) -> int | None:
if track is None:
return None
return require_track_index(track)


def require_track_and_device(track: int, device: int) -> tuple[int, int]:
return require_track_index(track), require_device_index(device)


def require_scene_and_value(scene: int, value: TValue) -> tuple[int, TValue]:
return require_scene_index(scene), value


def require_track_and_name(track: int, value: str) -> tuple[int, str]:
valid_track = require_track_index(track)
valid_name = require_non_empty_string("name", value, hint=TRACK_NAME_HINT)
return valid_track, valid_name


def require_scene_and_name(scene: int, value: str) -> tuple[int, str]:
valid_scene = require_scene_index(scene)
valid_name = require_non_empty_string("name", value, hint=SCENE_NAME_HINT)
return valid_scene, valid_name


def require_scene_move(from_scene: int, to_scene: int) -> tuple[int, int]:
valid_from_scene = require_non_negative("from", from_scene, hint=SCENE_SOURCE_HINT)
valid_to_scene = require_non_negative("to", to_scene, hint=SCENE_DESTINATION_HINT)
return valid_from_scene, valid_to_scene


def require_scene_insert_index(index: int) -> int:
return require_minus_one_or_non_negative("index", index, hint=SCENE_INSERT_INDEX_HINT)


def require_track_and_volume(track: int, value: float) -> tuple[int, float]:
valid_track = require_track_index(track)
valid_value = require_float_in_range(
"value",
value,
minimum=0.0,
maximum=1.0,
hint=VOLUME_VALUE_HINT,
)
return valid_track, valid_value


def require_track_and_pan(track: int, value: float) -> tuple[int, float]:
valid_track = require_track_index(track)
valid_value = require_float_in_range(
"value",
value,
minimum=-1.0,
maximum=1.0,
hint=PAN_VALUE_HINT,
)
return valid_track, valid_value


def require_non_empty_string(name: str, value: str, *, hint: str) -> str:
stripped = value.strip()
if not stripped:
Expand Down
80 changes: 42 additions & 38 deletions src/ableton_cli/commands/effect.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,17 @@
from __future__ import annotations

from collections.abc import Callable
from collections.abc import Callable, Sequence
from typing import Annotated

import typer

from ..runtime import execute_command, get_client
from ._validation import (
invalid_argument,
require_device_index,
require_non_empty_string,
require_optional_track_index,
require_parameter_index,
require_track_index,
require_track_and_device,
)

_SUPPORTED_EFFECT_TYPES = (
Expand All @@ -31,6 +31,9 @@
parameters_app = typer.Typer(help="Effect parameter listing commands", no_args_is_help=True)
parameter_app = typer.Typer(help="Effect parameter write commands", no_args_is_help=True)

TrackDeviceValidator = Callable[[int, int], tuple[int, int]]
TrackDeviceAction = Callable[[object, int, int], dict[str, object]]


def _normalize_effect_type(value: str) -> str:
parsed = require_non_empty_string("effect_type", value, hint="Pass a non-empty effect type.")
Expand All @@ -43,41 +46,35 @@ def _normalize_effect_type(value: str) -> str:
return normalized


def _require_optional_track_index(track: int | None) -> int | None:
if track is None:
return None
return require_track_index(track)


def _require_track_and_device_index(track: int, device: int) -> tuple[int, int]:
return (
require_track_index(track),
require_device_index(device),
)


def _require_effect_parameter_index(parameter: int) -> int:
return require_parameter_index(
parameter,
hint="Use a valid parameter index from 'ableton-cli effect parameters list'.",
)


def _execute_track_device_command(
def run_track_device_command(
ctx: typer.Context,
*,
command: str,
command_name: str,
track: int,
device: int,
action: Callable[[int, int], dict[str, object]],
fn: TrackDeviceAction,
validators: Sequence[TrackDeviceValidator] | None = None,
) -> None:
active_validators = validators if validators is not None else (require_track_and_device,)

def _run() -> dict[str, object]:
valid_track, valid_device = _require_track_and_device_index(track, device)
return action(valid_track, valid_device)
valid_track = track
valid_device = device
for validator in active_validators:
valid_track, valid_device = validator(valid_track, valid_device)
client = get_client(ctx)
return fn(client, valid_track, valid_device)

execute_command(
ctx,
command=command,
command=command_name,
args={"track": track, "device": device},
action=_run,
)
Expand All @@ -99,9 +96,10 @@ def effect_find(
] = None,
) -> None:
def _run() -> dict[str, object]:
valid_track = _require_optional_track_index(track)
valid_track = require_optional_track_index(track)
valid_type = _normalize_effect_type(effect_type) if effect_type is not None else None
return get_client(ctx).find_effect_devices(track=valid_track, effect_type=valid_type)
client = get_client(ctx)
return client.find_effect_devices(track=valid_track, effect_type=valid_type)

execute_command(
ctx,
Expand All @@ -117,12 +115,12 @@ def effect_parameters_list(
track: TrackArgument,
device: DeviceArgument,
) -> None:
_execute_track_device_command(
run_track_device_command(
ctx,
command="effect parameters list",
command_name="effect parameters list",
track=track,
device=device,
action=lambda valid_track, valid_device: get_client(ctx).list_effect_parameters(
fn=lambda client, valid_track, valid_device: client.list_effect_parameters(
track=valid_track,
device=valid_device,
),
Expand All @@ -138,9 +136,10 @@ def effect_parameter_set(
value: Annotated[float, typer.Argument(help="Target parameter value")],
) -> None:
def _run() -> dict[str, object]:
valid_track, valid_device = _require_track_and_device_index(track, device)
valid_track, valid_device = require_track_and_device(track, device)
valid_parameter = _require_effect_parameter_index(parameter)
return get_client(ctx).set_effect_parameter_safe(
client = get_client(ctx)
return client.set_effect_parameter_safe(
track=valid_track,
device=valid_device,
parameter=valid_parameter,
Expand All @@ -161,12 +160,12 @@ def effect_observe(
track: TrackArgument,
device: DeviceArgument,
) -> None:
_execute_track_device_command(
run_track_device_command(
ctx,
command="effect observe",
command_name="effect observe",
track=track,
device=device,
action=lambda valid_track, valid_device: get_client(ctx).observe_effect_parameters(
fn=lambda client, valid_track, valid_device: client.observe_effect_parameters(
track=valid_track,
device=valid_device,
),
Expand All @@ -181,11 +180,15 @@ def _build_standard_effect_app(effect_type: str, cli_name: str) -> typer.Typer:

@standard_app.command("keys")
def keys(ctx: typer.Context) -> None:
def _run() -> dict[str, object]:
client = get_client(ctx)
return client.list_standard_effect_keys(effect_type)

execute_command(
ctx,
command=f"effect {cli_name} keys",
args={},
action=lambda: get_client(ctx).list_standard_effect_keys(effect_type),
action=_run,
)

@standard_app.command("set")
Expand All @@ -197,13 +200,14 @@ def standard_set(
value: Annotated[float, typer.Argument(help="Target parameter value")],
) -> None:
def _run() -> dict[str, object]:
valid_track, valid_device = _require_track_and_device_index(track, device)
valid_track, valid_device = require_track_and_device(track, device)
valid_key = require_non_empty_string(
"key",
key,
hint="Pass a non-empty stable effect key.",
)
return get_client(ctx).set_standard_effect_parameter_safe(
client = get_client(ctx)
return client.set_standard_effect_parameter_safe(
effect_type=effect_type,
track=valid_track,
device=valid_device,
Expand All @@ -224,12 +228,12 @@ def standard_observe(
track: TrackArgument,
device: DeviceArgument,
) -> None:
_execute_track_device_command(
run_track_device_command(
ctx,
command=f"effect {cli_name} observe",
command_name=f"effect {cli_name} observe",
track=track,
device=device,
action=lambda valid_track, valid_device: get_client(ctx).observe_standard_effect_state(
fn=lambda client, valid_track, valid_device: client.observe_standard_effect_state(
effect_type=effect_type,
track=valid_track,
device=valid_device,
Expand Down
Loading