Add change_interval method to Routine.

Author: EvieePy

docs/changelog.rst

@@ -18,6 +18,10 @@ Master
         - Fix USERSTATE parsing incorrect user
         - Fix errors when event loop is started using `run_until_complete` to call methods prior to :func:`~twitchio.Client.run`
 
+- ext.routines
+    - Additions
+        - Added the :func:`~twitchio.ext.routines.Routine.change_interval` method.
+
 - ext.commands
     - Bug fixes
         - Make sure double-quotes are properly tokenized for bot commands

twitchio/ext/routines/__init__.py

@@ -97,6 +97,9 @@ def __init__(
 
         self._instance = None
 
+        self._args: tuple | None = None
+        self._kwargs: dict | None = None
+
     def __get__(self, instance, owner):
         if instance is None:
             return self
@@ -143,6 +146,8 @@ def start(self, *args, **kwargs) -> asyncio.Task:
         if self._task is not None and not self._task.done() and not self._restarting:
             raise RuntimeError(f"Routine {self._coro.__name__!r} is already running and is not done.")
 
+        self._args, self._kwargs = args, kwargs
+
         self._restarting = False
         self._task = self._loop.create_task(self._routine(*args, **kwargs))
 
@@ -226,6 +231,70 @@ def after_routine(self, coro: Callable) -> None:
 
         self._after = coro
 
+    def change_interval(
+            self,
+            *,
+            seconds: Optional[float] = 0,
+            minutes: Optional[float] = 0,
+            hours: Optional[float] = 0,
+            time: Optional[datetime.datetime] = None,
+            wait_first: Optional[bool] = False,
+    ) -> None:
+        """Method which schedules the running interval of the task to change.
+
+        Parameters
+        ----------
+        seconds: Optional[float]
+            The seconds to wait before the next iteration of the routine.
+        minutes: Optional[float]
+            The minutes to wait before the next iteration of the routine.
+        hours: Optional[float]
+            The hours to wait before the next iteration of the routine.
+        time: Optional[datetime.datetime]
+            A specific time to run this routine at. If a naive datetime is passed, your system local time will be used.
+        wait_first: Optional[bool]
+            Whether to wait the specified time before running the first iteration at the new interval.
+            Defaults to False.
+
+        Raises
+        ------
+        RuntimeError
+            Raised when the time argument and any hours, minutes or seconds is passed together.
+
+
+        .. warning::
+
+            The time argument can not be passed in conjunction with hours, minutes or seconds.
+            This behaviour is intended as it allows the time to be exact every day.
+        """
+        time_ = time
+
+        if any((seconds, minutes, hours)) and time_:
+            raise RuntimeError(
+                "Argument <time> can not be used in conjunction with any <seconds>, <minutes> or <hours> argument(s)."
+            )
+
+        if not time_:
+            delta = compute_timedelta(
+                datetime.datetime.now(datetime.timezone.utc)
+                + datetime.timedelta(seconds=seconds, minutes=minutes, hours=hours)
+            )
+
+            self._delta = delta
+        else:
+            now = datetime.datetime.now(time_.tzinfo)
+            if time_ < now:
+                time_ = datetime.datetime.combine(now.date(), time_.time())
+            if time_ < now:
+                time_ = time_ + datetime.timedelta(days=1)
+
+            self._time = time_
+
+        kwargs: dict = self._kwargs
+        kwargs['force'] = not wait_first
+
+        self.restart(*self._args, **kwargs)
+
     def error(self, coro: Callable):
         """A decorator to assign a coroutine as the error handler for this routine.