Custom tasks

Implementing tasks with custom widget.

Tasks consist of two parts:

1. a task class derived from yuio.io.TaskBase,

2. a widget derived from yuio.widget.Widget.

To implement one, you will need to supply these parts.

1. You can derive a custom widget from yuio.widget.Task, override some of its protected methods, then derive a custom task from yuio.io.Task, and override its _widget_class.

2. You can build everything from scratch, which we will do in this example.

Custom task example

Let’s build something that looks like status line from Tox.

Implementing a widget

First, we need a widget:

class CustomTaskWidget(yuio.widget.Widget[Never]):  # [1]_
    def __init__(self, envs: list[str]) -> None:
        # All currently running environments.
        self.envs = envs

    def layout(self, rc: yuio.widget.RenderContext) -> tuple[int, int]:
        # Our widget always takes one line.
        return 1, 1  # [2]_

    def draw(self, rc: yuio.widget.RenderContext):  # [3]_
        if spinner_pattern := rc.get_msg_decoration("spinner/pattern"):
            # Draw spinner.
            rc.set_color_path("task/decoration:running")  # [4]_

            # `rc.spinner_state` is a timer synchronized with spinner update rate
            # configured in theme.
            spinner = spinner_pattern[rc.spinner_state % len(spinner_pattern)]
            rc.write(spinner)
            rc.move_pos(1, 0)

        # Draw running environments.
        sep = False
        for env in self.envs:
            if sep:
                rc.set_color_path("task:running")
                rc.write(" | ")
            rc.set_color_path("task/heading:running")
            rc.write(env)
            sep = True

1. We use typing.Never to indicate that our widget doesn’t handle keyboard events.

2. First number is minimum number of lines that this widget can span, second number is maximum number of lines that this widget can span.

3. This method draws task widget using yuio.widget.RenderContext.

4. We reuse color paths from default task (see task/...:{status}), but you can add your own colors using a custom theme.

Implementing a task

Second, we need an actual task:

class CustomTask(yuio.io.TaskBase):
    def __init__(self, envs: list[str] | None = None):
        super().__init__()

        # Our task widget implemented above.
        self._widget = CustomTaskWidget(envs or [])

    def _get_widget(self):
        # This method should return a widget.
        return self._widget

    def _get_priority(self):
        # This method should return an integer priority.
        return 1  # [1]_

    def add_env(self, env: str):
        with self._lock:  # All updates must happen under a lock.
            self._widget.envs.append(env)

            # Check if we're in foreground, and widgets are displayed.
            if self._widgets_are_displayed():
                # If so, notify background thread to update tasks sooner.
                self._request_update()
            else:
                # Otherwise, print the new status.
                yuio.io.info("%s started", env)

    def remove_env(self, env: str):
        with self._lock:
            self._widget.envs.remove(env)

            # Since printing redraws all tasks, we call `_request_update`
            # before `yuio.io.info`. This way, number of redraws is minimized.
            self._request_update()
            yuio.io.info("%s finished", env)

    def __enter__(self):
        # Attach this task to to the top level of the task tree to actually show it.
        # You don't have to do it in `__enter__`, you can do it anywhere you like.
        # For example, `yuio.io.Task` runs `attach` in its constructor.
        self.attach(None)  # [2]_
        return self

1. Priority is used to hide tasks when they don’t fit in one screen. Default priority is 1 for running tasks, and 0 for pending or finished tasks.

2. None attaches to the top of the tree; you can also pass a parent task here.

Putting it all together

Finally, we can use out new task:

Full example

import time

import yuio.io
import yuio.theme
import yuio.widget

from typing import Never


class CustomTaskWidget(yuio.widget.Widget[Never]):  # [1]_
    def __init__(self, envs: list[str]) -> None:
        # All currently running environments.
        self.envs = envs

    def layout(self, rc: yuio.widget.RenderContext) -> tuple[int, int]:
        # Our widget always takes one line.
        return 1, 1  # [2]_

    def draw(self, rc: yuio.widget.RenderContext):  # [3]_
        if spinner_pattern := rc.get_msg_decoration("spinner/pattern"):
            # Draw spinner.
            rc.set_color_path("task/decoration:running")  # [4]_

            # `rc.spinner_state` is a timer synchronized with spinner update rate
            # configured in theme.
            spinner = spinner_pattern[rc.spinner_state % len(spinner_pattern)]
            rc.write(spinner)
            rc.move_pos(1, 0)

        # Draw running environments.
        sep = False
        for env in self.envs:
            if sep:
                rc.set_color_path("task:running")
                rc.write(" | ")
            rc.set_color_path("task/heading:running")
            rc.write(env)
            sep = True


class CustomTask(yuio.io.TaskBase):
    def __init__(self, envs: list[str] | None = None):
        super().__init__()

        # Our task widget implemented above.
        self._widget = CustomTaskWidget(envs or [])

    def _get_widget(self):
        # This method should return a widget.
        return self._widget

    def _get_priority(self):
        # This method should return an integer priority.
        return 1  # [1]_

    def add_env(self, env: str):
        with self._lock:  # All updates must happen under a lock.
            self._widget.envs.append(env)

            # Check if we're in foreground, and widgets are displayed.
            if self._widgets_are_displayed():
                # If so, notify background thread to update tasks sooner.
                self._request_update()
            else:
                # Otherwise, print the new status.
                yuio.io.info("%s started", env)

    def remove_env(self, env: str):
        with self._lock:
            self._widget.envs.remove(env)

            # Since printing redraws all tasks, we call `_request_update`
            # before `yuio.io.info`. This way, number of redraws is minimized.
            self._request_update()
            yuio.io.info("%s finished", env)

    def __enter__(self):
        # Attach this task to to the top level of the task tree to actually show it.
        # You don't have to do it in `__enter__`, you can do it anywhere you like.
        # For example, `yuio.io.Task` runs `attach` in its constructor.
        self.attach(None)  # [2]_
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Detach task from the task tree to hide it.
        self.detach()


class CustomTheme(yuio.theme.DefaultTheme):
    # For further authenticity, change spinner pattern to the one used in Tox.
    spinner_update_rate_ms = 100
    msg_decorations_unicode = {
        "spinner/pattern": "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏",
    }
    msg_decorations_ascii = {
        "spinner/pattern": "|-+x*",
    }


if __name__ == "__main__":
    env_list = ["3.14t", "3.14", "3.13", "3.12", "type", "lint"]

    yuio.io.setup(theme=CustomTheme)

    with CustomTask() as task:
        for env in env_list:
            task.add_env(env)
            time.sleep(0.2)

        time.sleep(5)

        for env in reversed(env_list):
            task.remove_env(env)
            time.sleep(0.2)

    yuio.io.success("Successfully tested %s", yuio.io.And(env_list))