feat: add display notifications for some transport actions (#10)

control_surface/display.py

@@ -59,7 +59,7 @@ def _toggle_text(text: str, toggle_state: bool):
     "redo": "Redo",
     "selected_track_arm": "ArmT",
     "session_record": "SRec",
-    "stop_all_clips": "StCl",
+    "stop_all_clips": "StAl",
     "tap_tempo": "TapT",
     "undo": "Undo",
 }
@@ -133,9 +133,12 @@ class Recording(DefaultNotifications.Recording):
         new = _action_texts["new"]
 
     class Scene(DefaultNotifications.Scene):
-        launch = partial(_scene_notification, prefix="#")
+        launch = partial(_scene_notification, prefix=">")
         select = partial(_scene_notification, prefix="#")
 
+    class Session:
+        stop_all_clips = _action_texts["stop_all_clips"]
+
     class SessionNavigation:
         vertical = partial(_right_align_index, "_")
         horizontal = partial(_right_align_index, "|")
@@ -160,8 +163,8 @@ class EditTrackControl:
             pass
 
     class Transport(DefaultNotifications.Transport):
-        automation_arm = None  # partial(_toggle_text, _action_texts["automation_arm"])
-        # metronome = partial(_toggle_text, _action_texts["metronome"])
+        automation_arm = partial(_toggle_text, _action_texts["automation_arm"])
+        metronome = partial(_toggle_text, _action_texts["metronome"])
         midi_capture = _action_texts["capture_midi"]
         tap_tempo = lambda t: _right_align("T", int(t))  # noqa: E731
 

control_surface/live.py

@@ -11,5 +11,6 @@
     flatten,  # noqa: F401
     lazy_attribute,  # noqa: F401
     listens,  # noqa: F401
+    listens_group,  # noqa: F401
     memoize,  # noqa: F401
 )

control_surface/live.pyi

@@ -1,6 +1,7 @@
 import typing
 
 from ableton.v2.base import Slot as __Slot
+from ableton.v2.base import SlotGroup as __SlotGroup
 
 T = typing.TypeVar("T")
 
@@ -16,4 +17,7 @@ class lazy_attribute(typing.Generic[T]):
 def listens(
     event_path: str, *a, **k
 ) -> typing.Callable[[typing.Callable[..., typing.Any]], __Slot]: ...
+def listens_group(
+    event_path: str, *a, **k
+) -> typing.Callable[[typing.Callable[..., typing.Any]], __SlotGroup]: ...
 def memoize(function: typing.Callable[..., T]) -> typing.Callable[..., T]: ...

control_surface/recording.py

@@ -59,7 +59,7 @@ def update(self):
 
 
 class RecordingComponent(RecordingComponentBase):
-    # Push-style "New" button.
+    # Equivalent to Push's "Duplicate" button when used while clips are playing.
     capture_and_insert_scene_button: ButtonControl.State = ButtonControl(  # type: ignore
         color="Recording.CaptureAndInsertScene",
         pressed_color="Recording.CaptureAndInsertScenePressed",

control_surface/session.py

@@ -1,29 +1,61 @@
-from enum import Enum
+from itertools import count
 from logging import getLogger
+from typing import Any
 
-from ableton.v3.base import task
+from ableton.v3.base import depends, task
 from ableton.v3.control_surface.components import (
     SessionComponent as SessionComponentBase,
 )
+from ableton.v3.control_surface.controls import ButtonControl
 
-from .live import lazy_attribute
+from .live import lazy_attribute, listens, listens_group
 
 logger = getLogger(__name__)
 
 
-# We want to be able to enable/disable the stop all clips button depending on whether
-# clips are playing, and show a triggered indication while clips are stopping. The Live
-# API doesn't expose the necessary properties/events, so we need to compute the status
-# manually.
-class StopAllClipsStatus(Enum):
-    enabled = "enabled"
-    disabled = "disabled"
-    triggered = "triggered"
-
-
 class SessionComponent(SessionComponentBase):
     _tasks: task.TaskGroup  # type: ignore
 
+    # This already exists on the parent, but we need to override its `pressed` listener.
+    #
+    # It would be nice if we could make the color more dynamic (e.g. off when no tracks
+    # playing and blinking while tracks are stopping), but the base component only
+    # invokes update methods on changes to the stop-clips status for tracks within the
+    # session ring, whereas we'd need to update the color on changes to the status for
+    # any track in the set.
+    stop_all_clips_button: Any = ButtonControl(
+        color="Session.StopAllClips", pressed_color="Session.StopAllClipsPressed"
+    )
+
+    @depends(session_ring=None)
+    def __init__(
+        self,
+        *a,
+        name="Session",
+        session_ring=None,
+        scene_component_type=None,
+        clip_slot_component_type=None,
+        clipboard_component_type=None,
+        **k,
+    ):
+        super().__init__(
+            *a,
+            name=name,
+            session_ring=session_ring,
+            scene_component_type=scene_component_type,
+            clip_slot_component_type=clip_slot_component_type,
+            clipboard_component_type=clipboard_component_type,
+            **k,
+        )
+
+        # Updated when the stop-all-clips button is pressed, and again when all clips
+        # have stopped. Tracks whether the button should be blinking.
+        self.__is_stopping_all_clips: bool = False
+
+        assert self.song
+        self.__on_tracks_changed.subject = self.song
+        self._reassign_track_listeners()
+
     def set_launch_selected_scene_button(self, button):
         self.selected_scene().set_launch_button(button)
 
@@ -40,11 +72,47 @@ def set_clip_launch_buttons(self, buttons):
                 )
                 scene.clip_slot(x).set_launch_button(button)
 
-    def _update_stop_clips_led(self, index):
-        super()._update_stop_clips_led(index)
-
-        # We don't want to compute the triggered state for every LED update, since it's
-        # O(n) with the number of tracks. Instead, use a task to throttle updates.
+    @stop_all_clips_button.pressed  # type: ignore
+    def stop_all_clips_button(self, _):
+        assert self.song
+        self.notify(self.notifications.Session.stop_all_clips)
+        self.song.stop_all_clips()
+
+        # We'll get notified via the fired/playing slot index listeners once the stop
+        # has actually been triggered. The listeners get invoked regardless of wether
+        # any clips were actually already playing.
+        self.__is_stopping_all_clips = True
+
+    def update(self):
+        super().update()
+        self._update_stop_all_clips_led()
+
+    @listens_group("fired_slot_index")
+    def __on_any_fired_slot_index_changed(self, _):
+        self.__throttled_update_stop_all_clips_led()
+
+    @listens_group("playing_slot_index")
+    def __on_any_playing_slot_index_changed(self, _):
+        self.__throttled_update_stop_all_clips_led()
+
+    @listens("tracks")
+    def __on_tracks_changed(self):
+        self._reassign_track_listeners()
+
+    # The Stop All Clips button needs to get updates from every track in the set to
+    # check when it can stop appearing as triggered.
+    def _reassign_track_listeners(self):
+        assert self.song
+        assert self.__on_any_fired_slot_index_changed
+
+        tracks = self.song.tracks
+        self.__on_any_fired_slot_index_changed.replace_subjects(tracks, count())
+        self.__on_any_playing_slot_index_changed.replace_subjects(tracks, count())
+
+    def __throttled_update_stop_all_clips_led(self):
+        # We don't want to compute the triggered state for every listener update, since
+        # the computation is O(n) with the number of tracks. Instead, use a task to
+        # throttle updates.
         if self._update_stop_all_clips_led_task.is_killed:
             # Perform the update once at the head of the delay to get immediate LED
             # feedback. In general this should already set the button state correctly.
@@ -62,22 +130,38 @@ def _update_stop_all_clips_led_task(self):
         return update_stop_all_clips_led_task
 
     def _update_stop_all_clips_led(self):
-        status = self._stop_all_clips_status()
         color = "Session.StopAllClips"
-        enabled = True
 
-        if status == StopAllClipsStatus.disabled:
-            enabled = False
-        elif status == StopAllClipsStatus.triggered:
-            color = "Session.StopAllClipsTriggered"
-
-        if self.stop_all_clips_button.enabled != enabled:
-            self.stop_all_clips_button.enabled = enabled
+        # Check for the triggered state, or clear the stopping status if the triggered
+        # state is no longer active.
+        #
+        # Note the Stop All Clips button in the Live UI has slightly different behavior:
+        # if the transport is playing but no clips are playing when it's pressed, it
+        # will blink until playback reaches the next launch quanitization point
+        # (e.g. the next bar). With our logic in this case, the button won't ever reach
+        # a blinking state. There doesn't appear to be a good way to listen for reaching
+        # the next quanitzation point during playback.
+        if self.__is_stopping_all_clips:
+            if self._is_stop_all_clips_maybe_triggered():
+                color = "Session.StopAllClipsTriggered"
+            else:
+                self.__is_stopping_all_clips = False
 
         if self.stop_all_clips_button.color != color:
             self.stop_all_clips_button.color = color
 
-    def _stop_all_clips_status(self) -> StopAllClipsStatus:
+    # Returns true iff:
+    #
+    # - at least one clip has stop triggered
+    # - no clips are playing without a triggered stop
+    #
+    # These conditions can also be met without pressing "Stop All Clips" (e.g. when
+    # playing a single clip and then stopping it), so this logic should generally be
+    # combined with tracking of actual presses of the stop-all-clips button.
+    def _is_stop_all_clips_maybe_triggered(self) -> bool:
+        # This gets the full list of tracks in the set. Unclear what the difference is
+        # between this and `song.tracks`, but this seems to be more canonical for this
+        # context.
         assert self._session_ring
         tracks_to_use = self._session_ring.tracks_to_use()
 
@@ -90,15 +174,12 @@ def _stop_all_clips_status(self) -> StopAllClipsStatus:
             if track.playing_slot_index >= 0 and track.fired_slot_index != -2:
                 # Once we find one playing clip that isn't
                 # stopping, we can return immediately.
-                return StopAllClipsStatus.enabled
+                return False
             elif track.fired_slot_index == -2:
-                # We need to keep iterating to see if there are any
-                # other clips which are playing and not triggered to
-                # stop.
+                # Once we find a stopping clip, we need to keep iterating to see if
+                # there are any other clips which are playing and not triggered to stop.
                 is_stop_triggered = True
 
-        return (
-            StopAllClipsStatus.triggered
-            if is_stop_triggered
-            else StopAllClipsStatus.disabled
-        )
+        # If we get here, no clips are playing without a triggered stop. Return whether
+        # at least one clip is currently triggered.
+        return is_stop_triggered

tests/conftest.py

@@ -1178,8 +1178,10 @@ def should_be_text(
     text: str,
     device_state: DeviceState,
 ):
-    assert device_state.display_text is not None
-    assert device_state.display_text.strip() == text
+    assert device_state.display_text is not None, "Display text not yet set"
+    assert (
+        device_state.display_text.strip() == text
+    ), f'Expected display text to be "{text}", but was "{device_state.display_text.strip()}"'
 
 
 @then(parsers.parse('the display should be scrolling "{text}"'))

tests/test_transport.py

@@ -0,0 +1,56 @@
+from conftest import (
+    Device,
+    DeviceState,
+    cc_action,
+    get_cc_for_key,
+    get_color,
+    stabilize_after_cc_action,
+    sync,
+)
+from pytest_bdd import parsers, scenarios, then
+from typeguard import typechecked
+
+scenarios("transport.feature")
+
+
+# Check that pressing a button twice toggles the light on/off, and toggles the
+# corresponding display between "+{control}" and "-{control}" respectively, but don't
+# care about whether the toggle is initially on or off.
+#
+# This allows checking the behavior of transport buttons whose state is saved globally
+# in Live (i.e. not as part of the Set).
+@then(parsers.parse('key {key_number:d} should toggle the "{control}" status'))
+@sync
+@typechecked
+async def should_toggle_control(
+    key_number: int,
+    control: str,
+    device: Device,
+    device_state: DeviceState,
+):
+    cc = get_cc_for_key(key_number)
+
+    # Check the LED status and the popup display immediately after pressing the button,
+    # and see if it matches the given on/off status for the toggle.
+    def assert_matches_status_after_press(status: bool):
+        # All transport toggles are currently solid yellow. We can parametrize this if
+        # needed in the future.
+        color = get_color(key_number, device_state)
+        expected_color = "solid yellow" if status else "off"
+        assert (
+            color == expected_color
+        ), f"Expected color to be {expected_color}, but was {color}"
+
+        expected_display_text = f'{"+" if status else "-"}{control}'
+        assert (
+            device_state.display_text == expected_display_text
+        ), f'Expected display text to be "{expected_display_text}", but was "{device_state.display_text}"'
+
+    current_status = False if get_color(key_number, device_state) == "off" else True
+
+    for _ in range(2):
+        await cc_action(cc, "press", device)
+        await stabilize_after_cc_action(device)
+
+        current_status = not current_status
+        assert_matches_status_after_press(current_status)

tests/track_controls_modes.feature

@@ -197,7 +197,7 @@ Feature: Track controls modes
   # change.
   Examples:
     | gesture    | k1   | k2   | k3   | k4   | k5   | k6   | k7   | k8   | k9   |
-    | press      | LnSc | Rec  | Play | Met  | SRec | StCl | ArmT | Aut  | TapT |
+    | press      | LnSc | Rec  | Play | Met  | SRec | StAl | ArmT | Aut  | TapT |
     | long-press | AAr  | Undo | CpMD | CpSc | SRec | BaK  | Redo | Quan | NwCl |
 
   Scenario: Navigating through track control edit screens