threading.py

"""
Threading utilities.
"""
from threading import Thread, Event


class StoppableThread(Thread):
    """
    A :cls:`Thread` with a :meth:`.stop` function and :attr:`.stopped`
    attribute.
    """

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.stopped = Event()

    def stop(self) -> None:
        """
        Tell the thread it should stop.

        This method merely signals to the thread that it should stop.  Calling
        code must subsequently call the :meth:`.join` method to wait for the
        thread to actually stop.  It is an error to call :meth:`.stop` on a
        thread which isn't alive (running).

        The thread's running code must periodically check if :attr:`.stopped`
        has been set, otherwise this method will have no effect.  It can do
        this in a loop, for example, with a short wait time::

            while not self.stopped.wait(0.2):
                # periodic work here
                ...       
        """
        assert self.is_alive(), "Thread not alive"
        self.stopped.set()