Deanimate! (aka, provide a method of stopping application and widget …

…animations) (Textualize#3000)

* Remove duplicated Added section in the CHANGELOG

* Add the ability to stop a running animation

Adds stop_animation to the core animator class, and then exposes it via the
same named methods on App and Widget. Note that a request to stop an
animation that isn't running is treated as a no-op.

* Fix tests so they actually work and test things

This is what happens when you save time using -k to run one test, then add
more but keep just hitting cursor up to rerun the tests. O_o

* Add the ability to stop an animation and jump to the final value

This doesn't address the issue of stopping scheduled animations, that's to
come next, first I just wanted to get the basic approach in place and then
build out from there.

* Add full stopping support to the ScalarAnimation

* Tidy up various bits of documentation in Animator

While I'm in here and moving things around: being various bits of
documentation more in line with how we document these days, and also add
some missing documentation.

* Allow for the full stopping (with end-seeking) of scheduled animations

* Don't spin up a scheduled animation to then not use it

* Be super-careful about getting keys when stopping

* Pop rather than acquire and delete

* Don't implement anything in Animation.stop

See Textualize#3000 (comment)

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 - Added App.begin_capture_print, App.end_capture_print, Widget.begin_capture_print, Widget.end_capture_print https://github.com/Textualize/textual/issues/2952
 - Added the ability to run async methods as thread workers https://github.com/Textualize/textual/pull/2938
+- Added `App.stop_animation` https://github.com/Textualize/textual/issues/2786
+- Added `Widget.stop_animation` https://github.com/Textualize/textual/issues/2786
 
 ### Changed
 

src/textual/_animator.py

@@ -64,6 +64,20 @@ def __call__(self, time: float) -> bool:  # pragma: no cover
         """
         raise NotImplementedError("")
 
+    async def invoke_callback(self) -> None:
+        """Calls the [`on_complete`][Animation.on_complete] callback if one is provided."""
+        if self.on_complete is not None:
+            await invoke(self.on_complete)
+
+    @abstractmethod
+    async def stop(self, complete: bool = True) -> None:
+        """Stop the animation.
+
+        Args:
+            complete: Flag to say if the animation should be taken to completion.
+        """
+        raise NotImplementedError
+
     def __eq__(self, other: object) -> bool:
         return False
 
@@ -117,6 +131,20 @@ def __call__(self, time: float) -> bool:
         setattr(self.obj, self.attribute, value)
         return factor >= 1
 
+    async def stop(self, complete: bool = True) -> None:
+        """Stop the animation.
+
+        Args:
+            complete: Flag to say if the animation should be taken to completion.
+
+        Note:
+            [`on_complete`][Animation.on_complete] will be called regardless
+            of the value provided for `complete`.
+        """
+        if complete:
+            setattr(self.obj, self.attribute, self.end_value)
+        await self.invoke_callback()
+
     def __eq__(self, other: object) -> bool:
         if isinstance(other, SimpleAnimation):
             return (
@@ -176,31 +204,33 @@ def __call__(
 
 
 class Animator:
-    """An object to manage updates to a given attribute over a period of time.
-
-    Attrs:
-        _animations: Dictionary that maps animation keys to the corresponding animation
-            instances.
-        _scheduled: Keys corresponding to animations that have been scheduled but not yet
-            started.
-        app: The app that owns the animator object.
-    """
+    """An object to manage updates to a given attribute over a period of time."""
 
     def __init__(self, app: App, frames_per_second: int = 60) -> None:
+        """Initialise the animator object.
+
+        Args:
+            app: The application that owns the animator.
+            frames_per_second: The number of frames/second to run the animation at.
+        """
         self._animations: dict[AnimationKey, Animation] = {}
-        self._scheduled: set[AnimationKey] = set()
+        """Dictionary that maps animation keys to the corresponding animation instances."""
+        self._scheduled: dict[AnimationKey, Timer] = {}
+        """Dictionary of scheduled animations, comprising of their keys and the timer objects."""
         self.app = app
+        """The app that owns the animator object."""
         self._timer = Timer(
             app,
             1 / frames_per_second,
             name="Animator",
             callback=self,
             pause=True,
         )
-        # Flag if no animations are currently taking place.
+        """The timer that runs the animator."""
         self._idle_event = asyncio.Event()
-        # Flag if no animations are currently taking place and none are scheduled.
+        """Flag if no animations are currently taking place."""
         self._complete_event = asyncio.Event()
+        """Flag if no animations are currently taking place and none are scheduled."""
 
     async def start(self) -> None:
         """Start the animator task."""
@@ -219,11 +249,26 @@ async def stop(self) -> None:
             self._complete_event.set()
 
     def bind(self, obj: object) -> BoundAnimator:
-        """Bind the animator to a given object."""
+        """Bind the animator to a given object.
+
+        Args:
+            obj: The object to bind to.
+
+        Returns:
+            The bound animator.
+        """
         return BoundAnimator(self, obj)
 
     def is_being_animated(self, obj: object, attribute: str) -> bool:
-        """Does the object/attribute pair have an ongoing or scheduled animation?"""
+        """Does the object/attribute pair have an ongoing or scheduled animation?
+
+        Args:
+            obj: An object to check for.
+            attribute: The attribute on the object to test for.
+
+        Returns:
+            `True` if that attribute is being animated for that object, `False` if not.
+        """
         key = (id(obj), attribute)
         return key in self._animations or key in self._scheduled
 
@@ -265,9 +310,10 @@ def animate(
             on_complete=on_complete,
         )
         if delay:
-            self._scheduled.add((id(obj), attribute))
             self._complete_event.clear()
-            self.app.set_timer(delay, animate_callback)
+            self._scheduled[(id(obj), attribute)] = self.app.set_timer(
+                delay, animate_callback
+            )
         else:
             animate_callback()
 
@@ -304,7 +350,10 @@ def _animate(
         ), "An Animation should have a duration OR a speed"
 
         animation_key = (id(obj), attribute)
-        self._scheduled.discard(animation_key)
+        try:
+            del self._scheduled[animation_key]
+        except KeyError:
+            pass
 
         if final_value is ...:
             final_value = value
@@ -374,6 +423,69 @@ def _animate(
         self._idle_event.clear()
         self._complete_event.clear()
 
+    async def _stop_scheduled_animation(
+        self, key: AnimationKey, complete: bool
+    ) -> None:
+        """Stop a scheduled animation.
+
+        Args:
+            key: The key for the animation to stop.
+            complete: Should the animation be moved to its completed state?
+        """
+        # First off, pull the timer out of the schedule and stop it; it
+        # won't be needed.
+        try:
+            schedule = self._scheduled.pop(key)
+        except KeyError:
+            return
+        schedule.stop()
+        # If we've been asked to complete (there's no point in making the
+        # animation only to then do nothing with it), and if there was a
+        # callback (there will be, but this just keeps type checkers happy
+        # really)...
+        if complete and schedule._callback is not None:
+            # ...invoke it to get the animator created and in the running
+            # animations. Yes, this does mean that a stopped scheduled
+            # animation will start running early...
+            await invoke(schedule._callback)
+            # ...but only so we can call on it to run right to the very end
+            # right away.
+            await self._stop_running_animation(key, complete)
+
+    async def _stop_running_animation(self, key: AnimationKey, complete: bool) -> None:
+        """Stop a running animation.
+
+        Args:
+            key: The key for the animation to stop.
+            complete: Should the animation be moved to its completed state?
+        """
+        try:
+            animation = self._animations.pop(key)
+        except KeyError:
+            return
+        await animation.stop(complete)
+
+    async def stop_animation(
+        self, obj: object, attribute: str, complete: bool = True
+    ) -> None:
+        """Stop an animation on an attribute.
+
+        Args:
+            obj: The object containing the attribute.
+            attribute: The name of the attribute.
+            complete: Should the animation be set to its final value?
+
+        Note:
+            If there is no animation running, this is a no-op. If there is
+            an animation running the attribute will be left in the last
+            state it was in before the call to stop.
+        """
+        key = (id(obj), attribute)
+        if key in self._scheduled:
+            await self._stop_scheduled_animation(key, complete)
+        elif key in self._animations:
+            await self._stop_running_animation(key, complete)
+
     async def __call__(self) -> None:
         if not self._animations:
             self._timer.pause()
@@ -387,13 +499,15 @@ async def __call__(self) -> None:
                 animation = self._animations[animation_key]
                 animation_complete = animation(animation_time)
                 if animation_complete:
-                    completion_callback = animation.on_complete
-                    if completion_callback is not None:
-                        await invoke(completion_callback)
                     del self._animations[animation_key]
+                    await animation.invoke_callback()
 
     def _get_time(self) -> float:
-        """Get the current wall clock time, via the internal Timer."""
+        """Get the current wall clock time, via the internal Timer.
+
+        Returns:
+            The wall clock time.
+        """
         # N.B. We could remove this method and always call `self._timer.get_time()` internally,
         # but it's handy to have in mocking situations
         return _time.get_time()

src/textual/app.py

@@ -601,6 +601,18 @@ def animate(
             on_complete=on_complete,
         )
 
+    async def stop_animation(self, attribute: str, complete: bool = True) -> None:
+        """Stop an animation on an attribute.
+
+        Args:
+            attribute: Name of the attribute whose animation should be stopped.
+            complete: Should the animation be set to its final value?
+
+        Note:
+            If there is no animation running, this is a no-op.
+        """
+        await self._animator.stop_animation(self, attribute, complete)
+
     @property
     def debug(self) -> bool:
         """Is debug mode enabled?"""

src/textual/css/scalar_animation.py

@@ -66,6 +66,20 @@ def __call__(self, time: float) -> bool:
 
         return False
 
+    async def stop(self, complete: bool = True) -> None:
+        """Stop the animation.
+
+        Args:
+            complete: Flag to say if the animation should be taken to completion.
+
+        Note:
+            [`on_complete`][Animation.on_complete] will be called regardless
+            of the value provided for `complete`.
+        """
+        if complete:
+            setattr(self.styles, self.attribute, self.final_value)
+        await self.invoke_callback()
+
     def __eq__(self, other: object) -> bool:
         if isinstance(other, ScalarAnimation):
             return (

src/textual/widget.py

@@ -1567,6 +1567,18 @@ def animate(
             on_complete=on_complete,
         )
 
+    async def stop_animation(self, attribute: str, complete: bool = True) -> None:
+        """Stop an animation on an attribute.
+
+        Args:
+            attribute: Name of the attribute whose animation should be stopped.
+            complete: Should the animation be set to its final value?
+
+        Note:
+            If there is no animation running, this is a no-op.
+        """
+        await self.app.animator.stop_animation(self, attribute, complete)
+
     @property
     def _layout(self) -> Layout:
         """Get the layout object if set in styles, or a default layout.

tests/test_animation.py

@@ -1,6 +1,7 @@
 from time import perf_counter
 
 from textual.app import App, ComposeResult
+from textual.reactive import var
 from textual.widgets import Static
 
 
@@ -157,3 +158,56 @@ async def test_schedule_reverse_animations() -> None:
         styles.animate("background", "black", delay=0.05, duration=0.01)
         await pilot.wait_for_scheduled_animations()
         assert styles.background.rgb == (0, 0, 0)
+
+
+class CancelAnimWidget(Static):
+    counter: var[float] = var(23)
+
+
+class CancelAnimApp(App[None]):
+    counter: var[float] = var(23)
+
+    def compose(self) -> ComposeResult:
+        yield CancelAnimWidget()
+
+
+async def test_cancel_app_animation() -> None:
+    """It should be possible to cancel a running app animation."""
+
+    async with CancelAnimApp().run_test() as pilot:
+        pilot.app.animate("counter", value=0, final_value=1000, duration=60)
+        await pilot.pause()
+        assert pilot.app.animator.is_being_animated(pilot.app, "counter")
+        await pilot.app.stop_animation("counter")
+        assert not pilot.app.animator.is_being_animated(pilot.app, "counter")
+
+
+async def test_cancel_app_non_animation() -> None:
+    """It should be possible to attempt to cancel a non-running app animation."""
+
+    async with CancelAnimApp().run_test() as pilot:
+        assert not pilot.app.animator.is_being_animated(pilot.app, "counter")
+        await pilot.app.stop_animation("counter")
+        assert not pilot.app.animator.is_being_animated(pilot.app, "counter")
+
+
+async def test_cancel_widget_animation() -> None:
+    """It should be possible to cancel a running widget animation."""
+
+    async with CancelAnimApp().run_test() as pilot:
+        widget = pilot.app.query_one(CancelAnimWidget)
+        widget.animate("counter", value=0, final_value=1000, duration=60)
+        await pilot.pause()
+        assert pilot.app.animator.is_being_animated(widget, "counter")
+        await widget.stop_animation("counter")
+        assert not pilot.app.animator.is_being_animated(widget, "counter")
+
+
+async def test_cancel_widget_non_animation() -> None:
+    """It should be possible to attempt to cancel a non-running widget animation."""
+
+    async with CancelAnimApp().run_test() as pilot:
+        widget = pilot.app.query_one(CancelAnimWidget)
+        assert not pilot.app.animator.is_being_animated(widget, "counter")
+        await widget.stop_animation("counter")
+        assert not pilot.app.animator.is_being_animated(widget, "counter")