ableton-osc/ableton_osc/handler.py

"""Base handler with get/set/listen/call infrastructure."""
import logging
from functools import partial
from typing import Any, Callable, Dict, Optional, Tuple

logger = logging.getLogger(__name__)


class AbletonOSCHandler:
    def __init__(self, manager):
        self.manager = manager
        self.osc_server = manager.osc_server
        self._listener_store: Dict[str, Callable] = {}  # key -> (target, remove_fn)

    @property
    def song(self):
        return self.manager.song

    # ------------------------------------------------------------------
    # Registration helpers
    # ------------------------------------------------------------------

    def _add(self, address: str, callback: Callable) -> None:
        self.osc_server.add_handler(address, callback)

    def _send(self, address: str, args: tuple) -> None:
        self.osc_server.send_all_clients(address, args)

    # ------------------------------------------------------------------
    # Generic property get
    # ------------------------------------------------------------------

    def _get_prop(self, target: Any, prop: str) -> Any:
        try:
            return getattr(target, prop)
        except Exception as e:
            logger.warning("get %s: %s", prop, e)
            return None

    def _set_prop(self, target: Any, prop: str, value: Any) -> None:
        try:
            setattr(target, prop, value)
        except Exception as e:
            logger.warning("set %s=%s: %s", prop, value, e)

    # ------------------------------------------------------------------
    # Listener management
    # ------------------------------------------------------------------

    def _register_listener(self, key: str, target: Any, prop: str,
                           notify_address: str, notify_args_prefix: tuple = ()) -> None:
        """Attach a Live listener and remember it so we can remove it later."""
        self._remove_listener(key)

        add_fn_name = f"add_{prop}_listener"
        remove_fn_name = f"remove_{prop}_listener"

        add_fn = getattr(target, add_fn_name, None)
        remove_fn = getattr(target, remove_fn_name, None)
        if add_fn is None:
            logger.warning("No listener support for %s", prop)
            return

        def callback():
            val = self._get_prop(target, prop)
            if val is None:
                return
            if isinstance(val, (list, tuple)):
                self._send(notify_address, notify_args_prefix + tuple(val))
            else:
                self._send(notify_address, notify_args_prefix + (val,))

        add_fn(callback)
        self._listener_store[key] = (remove_fn, callback)
        # Send current value immediately
        callback()

    def _remove_listener(self, key: str) -> None:
        entry = self._listener_store.pop(key, None)
        if entry is None:
            return
        remove_fn, callback = entry
        try:
            if remove_fn and remove_fn(callback):
                pass
            elif remove_fn:
                remove_fn(callback)
        except Exception as e:
            logger.warning("remove listener %s: %s", key, e)

    def clear_listeners(self) -> None:
        for key in list(self._listener_store.keys()):
            self._remove_listener(key)

    # ------------------------------------------------------------------
    # Handler factories for the common get/set/listen pattern
    # ------------------------------------------------------------------

    def _make_getter(self, get_target: Callable, prop: str,
                     prefix_from_params: Callable = None) -> Callable:
        """Returns a handler fn that reads target.prop and returns the value."""
        def handler(params: tuple) -> Optional[tuple]:
            target = get_target(params)
            if target is None:
                return None
            val = self._get_prop(target, prop)
            if val is None:
                return None
            prefix = prefix_from_params(params) if prefix_from_params else ()
            if isinstance(val, (list, tuple)):
                return prefix + tuple(val)
            return prefix + (val,)
        return handler

    def _make_setter(self, get_target: Callable, prop: str,
                     value_index: int = 0) -> Callable:
        """Returns a handler fn that sets target.prop from params."""
        def handler(params: tuple) -> None:
            target = get_target(params)
            if target is None:
                return None
            value = params[value_index]
            self._set_prop(target, prop, value)
            return None
        return handler

    def init_api(self) -> None:
        """Override in subclasses to register all OSC handlers."""
        pass