gtui

Task scheduler & executor with text-based user interface, able to:

• Declare task & dependency as a graph
• Execute task graph
• Show task status, stdout & log
• Provide a nice user interface

Why this

I always have a lot of scripts to run. I want to execute them parallelly in a certain order and be able to check their status, output & log in an easy way. So I built this package.

Installation

$ pip install gtui

Note: it's not supported on windows, you need to use cygwin or wsl.

Quickstart

Let's say we have some helloworld tasks & their dependencies like this:

We need to create Task and add them to TaskGraph:

import time
import logging
from gtui import Task, TaskGraph

def foo(n):
    logging.info('foo(%s) is called', n)
    print('Start to sleep %s seconds.' % n)
    time.sleep(n)
    print('Hello World!')

t1 = Task('t1', func=foo, args=[0.1])
t2 = Task('t2', func=foo, args=[1])
t3 = Task('t3', func=foo, args=[1])
t4 = Task('t4', func=foo, args=[0.5])

g = TaskGraph()
g.add_task(t1)
g.add_task(t2, waiting_for=t1)
g.add_task(t3, waiting_for=t1)
g.add_task(t4, waiting_for=[t2, t3])

g.run()

TaskGraph.run starts the text user interface, and you can navigate through tasks, see their status, output and logs:

Keybindings:

• t : toggle tail -f mode, will follow text when enabled
• tab : switch between output & log
• j/k : select previous/next task
• h/l : page up/down
• ↑/↓ : scroll up/down one line
• y : copy text
• q : exit

Task & TaskGraph

Task defines what to do and has an unique name:

# foo(*args, **kwargs) will be called
t = Task(name='foo', func=foo, args=[1, 2], kwargs={'foo': 'bar'})

TaskGraph defines execution order of a set of tasks, it provides method to declare task & dependency:

g = TaskGraph()
g.add_task(t1)                              # added task t1
g.add_task(t2, waiting_for=t1)              # added task t2, t2 runs after t1 finishes
g.add_tasks([t3, t4])                       # added task t3, t4
g.add_dependency(t3, waiting_for=[t1, t2])  # declare t3 to run after t1 & t2 finish

When TaskGraph contains a cycle denpendency, run method will throw a ValueError. You can also use has_cycle to check:

> g = TaskGraph()
> g.add_tasks([t1, t2])
> g.add_dependency(t1, waiting_for=t2)
> g.add_dependency(t2, waiting_for=t1)
> g.has_cycle()
[t1, t2, t1]

Run Options

TaskGraph.run provides some options:

g.run(
  title='Demo',           # Text shown at the left bottom corner
  callback=None,          # A function called when execution fail or succeed
  log_formatter=None,     # An instance of logging.Formatter, to specify the log format
  exit_on_success=False   # whether exit tui when execution succeed
)

callback can be used to notify the execution result, it will be called with an boolean indicating whether execution succeed. gtui.callback has some common callbacks:

# emit a desktop notification, use osascript on mac and notify-send on linux
from gtui import callback
g.run(callback.desktop_nofity(title='Plz See Here!', success_msg='Success', fail_msg='Fail'))

Possible Problem with Stdout

Writing to stdout will break the TUI display. gtui runs each task in a new thread with sys.stdout replaced so functions like print will just work fine. When creating a new thread inside a task, gtui.IORedirectedThread can be used to achieve the same result:

from gtui import IORedirectedThread

t = IORedirectedThread(target=print, args=['hello world'])
t.start()
t.join()

content = t.get_stdout_content()

However, gtui doesn't try to deal with other cases so you should take care of it by yourself.