ableton-osc/ableton_osc/return_track.py

"""Handles /live/return_track/* OSC addresses (mirrors TrackHandler for return tracks)."""
import logging
from typing import Any, List, Optional, Tuple
from .handler import AbletonOSCHandler

logger = logging.getLogger(__name__)

PROPS_RW = ["name", "mute", "solo", "color", "color_index"]
PROPS_RO = ["output_meter_level", "output_meter_left", "output_meter_right"]


class ReturnTrackHandler(AbletonOSCHandler):
    def init_api(self) -> None:
        self.clear_listeners()

        for prop in PROPS_RW:
            self._add(f"/live/return_track/get/{prop}", self._make_getter(prop))
            self._add(f"/live/return_track/set/{prop}", self._make_setter(prop))
            self._add(f"/live/return_track/start_listen/{prop}",
                      self._make_start_listen(prop))
            self._add(f"/live/return_track/stop_listen/{prop}",
                      self._make_stop_listen(prop))

        for prop in PROPS_RO:
            self._add(f"/live/return_track/get/{prop}", self._make_getter(prop))
            self._add(f"/live/return_track/start_listen/{prop}",
                      self._make_start_listen(prop))
            self._add(f"/live/return_track/stop_listen/{prop}",
                      self._make_stop_listen(prop))

        self._add("/live/return_track/get/volume", self._get_volume)
        self._add("/live/return_track/set/volume", self._set_volume)
        self._add("/live/return_track/get/panning", self._get_panning)
        self._add("/live/return_track/set/panning", self._set_panning)
        self._add("/live/return_track/get/send", self._get_send)
        self._add("/live/return_track/set/send", self._set_send)
        self._add("/live/return_track/get/num_devices", self._get_num_devices)
        self._add("/live/return_track/get/devices/name", self._get_devices_name)

    # ------------------------------------------------------------------

    def _get_return_tracks(self, params) -> List[Tuple[int, Any]]:
        tracks = list(self.song.return_tracks)
        if not params or params[0] == "*":
            return list(enumerate(tracks))
        idx = int(params[0])
        return [(idx, tracks[idx])] if 0 <= idx < len(tracks) else []

    def _get_return_track(self, params) -> Optional[Any]:
        tracks = list(self.song.return_tracks)
        if not params:
            return None
        idx = int(params[0])
        return tracks[idx] if 0 <= idx < len(tracks) else None

    def _make_getter(self, prop: str):
        def handler(params: tuple) -> Optional[tuple]:
            result = []
            for idx, track in self._get_return_tracks(params):
                val = self._get_prop(track, prop)
                if val is not None:
                    result += [idx, val]
            return tuple(result) if result else None
        return handler

    def _make_setter(self, prop: str):
        def handler(params: tuple) -> None:
            if len(params) < 2:
                return None
            track = self._get_return_track(params)
            if track:
                self._set_prop(track, prop, params[1])
            return None
        return handler

    def _make_start_listen(self, prop: str):
        def handler(params: tuple) -> None:
            for idx, track in self._get_return_tracks(params):
                self._register_listener(
                    f"return_track.{idx}.{prop}", track, prop,
                    f"/live/return_track/get/{prop}", (idx,)
                )
        return handler

    def _make_stop_listen(self, prop: str):
        def handler(params: tuple) -> None:
            for idx, _ in self._get_return_tracks(params):
                self._remove_listener(f"return_track.{idx}.{prop}")
        return handler

    def _get_volume(self, params: tuple) -> Optional[tuple]:
        result = []
        for idx, track in self._get_return_tracks(params):
            try:
                result += [idx, track.mixer_device.volume.value]
            except Exception:
                pass
        return tuple(result) if result else None

    def _set_volume(self, params: tuple) -> None:
        if len(params) < 2:
            return None
        track = self._get_return_track(params)
        if track:
            try:
                track.mixer_device.volume.value = float(params[1])
            except Exception as e:
                logger.warning("return_track set volume: %s", e)
        return None

    def _get_panning(self, params: tuple) -> Optional[tuple]:
        result = []
        for idx, track in self._get_return_tracks(params):
            try:
                result += [idx, track.mixer_device.panning.value]
            except Exception:
                pass
        return tuple(result) if result else None

    def _set_panning(self, params: tuple) -> None:
        if len(params) < 2:
            return None
        track = self._get_return_track(params)
        if track:
            try:
                track.mixer_device.panning.value = float(params[1])
            except Exception as e:
                logger.warning("return_track set panning: %s", e)
        return None

    def _get_send(self, params: tuple) -> Optional[tuple]:
        if len(params) < 2:
            return None
        track = self._get_return_track(params)
        send_idx = int(params[1])
        if track:
            try:
                sends = list(track.mixer_device.sends)
                if 0 <= send_idx < len(sends):
                    return (int(params[0]), send_idx, sends[send_idx].value)
            except Exception as e:
                logger.warning("return_track get send: %s", e)
        return None

    def _set_send(self, params: tuple) -> None:
        if len(params) < 3:
            return None
        track = self._get_return_track(params)
        send_idx = int(params[1])
        if track:
            try:
                sends = list(track.mixer_device.sends)
                if 0 <= send_idx < len(sends):
                    sends[send_idx].value = float(params[2])
            except Exception as e:
                logger.warning("return_track set send: %s", e)
        return None

    def _get_num_devices(self, params: tuple) -> Optional[tuple]:
        track = self._get_return_track(params)
        return (len(track.devices),) if track else None

    def _get_devices_name(self, params: tuple) -> Optional[tuple]:
        track = self._get_return_track(params)
        return tuple(d.name for d in track.devices) if track else None