Text-based user interface
Repype comes with a built-in text-based user interface TUI that is based on Textual:
python -m repype.textual examples/segmentation
The argument examples/segmentation is the path of the root task directory.
Built-in capabilities
For demonstration purposes, we load the TUI programmatically:
[3]:
from repype.textual.app import Repype
from repype.textual.demo import run
demo = await run(
Repype(path = root_task_path),
size = (113, 24),
)
demo.screenshot()
repype — Manage tasks ▼ Loaded tasks ┗━━ ▼ [ ] /tmp/tmp564tqd0a/examples/segmentation (pending) ┗━━ ▼ [ ] sigma=2 (pending) Tasks: 0 queued / 2 pending a Add sub-task e Edit task d Delete task r Run tasks R Reset task x Toggle task q Exit
It can be seen from the screen above, that 2 tasks are found and loaded.
The TUI can be navigated by using the keyboard and the hotkeys displayed at the bottom line of the screen. It is also possible to use the mouse, if this is supported by the terminal emulator.
One or more tasks can be queued by pressing the the x key:
[4]:
await demo.press('down', 'x')
demo.screenshot()
repype — Manage tasks ▼ Loaded tasks └── ▼ [x] /tmp/tmp564tqd0a/examples/segmentation (pending) ┗━━ ▼ [ ] sigma=2 (pending) Tasks: 1 queued / 2 pending a Add sub-task e Edit task d Delete task r Run tasks R Reset task x Toggle task q Exit
The queued tasks can then be run by pressing the r key. If no task is queued and the r key is pressed, the currently selected task will be run. Only pending tasks can be run (i.e. those which are runnable and haven’t been completed yet).
Now lets run the queued task:
[5]:
await demo.press('r')
await demo.wait(1)
demo.screenshot()
repype — Run tasks ▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔ ▼ /tmp/tmp564tqd0a/examples/segmentation Starting from scratch (1/1) Processing: B2--W00026--P00001--Z00000--T00000--dapi.tif Running stage: download ━━━━━━━━━━━━━━━━━━━━━━━╸━━━━━━━━ --% --:--:-- ^c Cancel ESC Close
[6]:
await demo.wait_for_condition(lambda: demo.app.batch.task_process is None)
demo.screenshot()
repype — Run tasks ▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔ ▼ /tmp/tmp564tqd0a/examples/segmentation (done) Starting from scratch (1/1) Processing: B2--W00026--P00001--Z00000--T00000--dapi.tif Results have been stored ^c Cancel ESC Close
[7]:
await demo.press('escape')
demo.screenshot()
repype — Manage tasks ▼ Loaded tasks └── ▼ [ ] /tmp/tmp564tqd0a/examples/segmentation ┗━━ ▼ [ ] sigma=2 (pending) Tasks: 0 queued / 1 pending a Add sub-task e Edit task d Delete task r Run tasks R Reset task x Toggle task q Exit
Custom status updates
First, lets update the Segmentation stage so it produces a custom status update:
[8]:
import repype.status
import tests.test_repype
class Segmentation(tests.test_repype.Segmentation):
def process(self, image, pipeline, config, status = None):
repype.status.update(status, info = 'custom', text = 'Text', number = 42)
return super().process(image, pipeline, config, status)
To employ the new implementation programmatically, we override the create_pipeline method of the Task class:
[9]:
import repype.task
class Task(repype.task.Task):
def create_pipeline(self, *args, **kwargs):
pipeline = super().create_pipeline(*args, **kwargs)
pipeline.stages[pipeline.find('segmentation')] = Segmentation()
return pipeline
We then extend the RunScreen implementation to handle the custom status update:
[10]:
import repype.textual.run
class RunScreen(repype.textual.run.RunScreen):
def custom_format(self, positions, status, intermediate):
if isinstance(status, dict) and status.get('info') == 'custom':
return f'*** {status["text"]} ({status["number"]})'
demo.app.screen.run_screen_cls = RunScreen
Then we employ the new Task implementation in the TUI and observe that the previously finished task is pending again, due to the changed implementation:
[11]:
import repype.batch
demo.app.batch = repype.batch.Batch(Task)
demo.app.screen.update_task_tree()
demo.screenshot()
repype — Manage tasks ▼ Loaded tasks └── ▼ [ ] /tmp/tmp564tqd0a/examples/segmentation (pending) ┗━━ ▼ [ ] sigma=2 (pending) Tasks: 0 queued / 2 pending a Add sub-task e Edit task d Delete task r Run tasks R Reset task x Toggle task q Exit
We run the updated task py pressing r:
[12]:
await demo.press('r')
demo.screenshot()
repype — Manage tasks ▼ Loaded tasks └── ▼ [ ] /tmp/tmp564tqd0a/examples/segmentation (pending) ┗━━ ▼ [ ] sigma=2 (pending) █▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀█ █ █ █ No tasks queued. Run the selected task? █ █ /tmp/tmp564tqd0a/examples/segmentation █ █ █ █ █ █ ▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔ ▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔ █ █ Yes No █ █ ▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁ ▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁▁ █ █ █ █ █ █▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄█ Tasks: 0 queued / 2 pending a Add sub-task e Edit task d Delete task r Run tasks R Reset task x Toggle task q Exit
[13]:
await demo.press('enter')
await demo.wait_for_condition(lambda: demo.app.batch.task_process is None)
demo.screenshot()
repype — Run tasks ▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔▔ ▼ /tmp/tmp564tqd0a/examples/segmentation (done) Picking up from: /tmp/tmp564tqd0a/examples/segmentation (segmentation) (1/1) Processing: B2--W00026--P00001--Z00000--T00000--dapi.tif *** Text (42) Results have been stored ^c Cancel ESC Close