Where to Discuss?

Local Group

Preface

Goal: Dual Panel Watchfiles with Rich Live


3: Concurrrent Live: Renderable

Dual panel using async.

As I promise, many changes.

Python Source Code

It would be easier if you read the script first.

Required Package

Now it is complete. We have a table inside panel, inside layout, inside live.

import asyncio

from watchfiles import awatch

from rich.live   import Live
from rich.table  import Table
from rich.panel  import Panel
from rich.layout import Layout

Python Rich: Concurrrent Live: Import

Class: Skeleton

You have alread know most of these bunch of method, in previous article. But I explain it anyway to refresh our memory, before we get down into class refactoring.

# Dual Rich Panel Watchfiles Class Example
class DualWatch:
  filepath_cut   = '/home/epsi/awatch/'
  filepath_right = '/home/epsi/awatch/code-01-base'
  filepath_left  = '/home/epsi/awatch/code-02-enh'

  def make_layout(self):
  def generate_table(self, changes) -> Table:
  def get_panel(self, changes) -> Panel:
  def get_layout(self, layout_name, changes) -> Layout:

  async def do_left(self):
  async def do_right(self):
  async def main(self):

Python Rich: Concurrrent Live: Skeleton

Class: Property

Actuall we only need two property:

  • Filepath to be monitored in left panel
  • Filepath to be monitored in right panel

But the output string seems to be too wide, that I have to replace the string, so it will fit in narrow panel. So here it is:

  • Filepath prefix to cut the string.
class DualWatch:
  filepath_cut   = '/home/epsi/awatch/'
  filepath_right = '/home/epsi/awatch/code-01-base'
  filepath_left  = '/home/epsi/awatch/code-02-enh'

Python Rich: Concurrrent Live: Properties

Constructor

Since we have no variable initialization. There is still no need for class constructor,

Class: Make Layout

This can be called in class constructor, or main.

Here we split layout horizontally, into dual layout.

  def make_layout(self):
    # Define the layout.
    self.layout = Layout(name="root")
    self.layout.split_row(
      Layout(name="left"),
      Layout(name="right"),
    )

Python Rich: Concurrrent Live: Make

We can have other layout as well, such as split vertically.

    self.layout.split(
      Layout(name="up"),
      Layout(name="bottom"),
    )

Renderable: Layout

In live.update(), we call layout.update(). The live.update() itself require renderable layout object, so we have to return the whole layout. But we need to determine which layout, whether it is left or right.

This can be done this way:

  def get_layout(self, layout_name, changes) -> Layout:
      self.layout[layout_name].update(
        self.get_panel(changes))
    return self.layout

But how about empty layout? Don’t we still have to initialize both panel?

We can simply propagate the None to both panel:

    if layout_name == None:
      self.layout['left'].update(
        self.get_panel(None))
      self.layout['right'].update(
        self.get_panel(None))

Now we have complete method as below:

  def get_layout(self, layout_name, changes) -> Layout:
    if layout_name == None:
      self.layout['left'].update(
        self.get_panel(None))
      self.layout['right'].update(
        self.get_panel(None))
    else:
      self.layout[layout_name].update(
        self.get_panel(changes))
    return self.layout

Python Rich: Concurrrent Live: Layout

Renderable: Panel

Now we can propagate the changes, from layout. Either it has values or just None.

  def get_panel(self, changes) -> Panel:
    dataTable = self.generate_table(changes)
    panel = Panel(
      dataTable, title="A Panel",
      subtitle="A [yellow]Thank you")
    return panel

Python Rich: Concurrrent Live: Panel

Beware of this.

We have identical property for each panel.

This means, if we want each panel to have different behaviour, such as title or stuff, we have to refactor class, as base class or descendant.

Renderable: Table

There is no change, but shorter file output.

The flow of changes end here. If it contain None, display header only.

  def generate_table(self, changes) -> Table:
    # Make a new table
    table = Table(expand=True)
    table.add_column("Type")
    table.add_column("File")

    if changes != None:
      for change in changes:
        short = change[1].replace(
          self.filepath_cut, '')
        table.add_row(
          *[str(change[0]), short])

    return table

Python Rich: Concurrrent Live: Table

After preparing renderable, from ‘livetolayouttopaneltotable`, we are going to discuss the asynchronous task.


3: Concurrrent Live: The Tasks

We will discuss the process flow, from program entry point, await for both tasks, and getting renderable layout for each task.

Execute: Run

Class Instance

We create instance of DualWatch class, as program entry point as shown below:

dual = DualWatch()
asyncio.run(dual.main())

Python Rich: Concurrrent Live: Run

Async Main Method: Live

We manage to concurrent two tasks with create_task and await.

  async def main(self):
    self.make_layout()
    self.live = Live(self.get_layout(None, None))

    task_left = asyncio.create_task(
      self.do_left())
    task_right = asyncio.create_task(
      self.do_right())

    await(task_left)
    await(task_right)

Python Rich: Concurrrent Live: Main

How does it works?

We need to get self.live reference, so we create instance:

    self.live = Live(self.get_layout(None, None))

But in order to display empty layout, we have to define our layout first, to get our self.layout.

So we have this line:

    self.make_layout()
    self.live = Live(self.get_layout(None, None))

We can call this self.live reference later with:

        self.live.update(
          self.get_layout('left', changes))

In class refactoring later, we require both self.live, and self.layout.

Asynchronous Task

For each task, we watch the folder, then show the changes in the layout.

For example this left layout.

  async def do_left(self):
    with self.live:
      watchloop = awatch(
        self.filepath_left, recursive=True)

      async for changes in watchloop:
        self.live.update(
          self.get_layout('left', changes))

Python Rich: Concurrrent Live: Left Task

We can do the same with right layout.

  async def do_right(self):
    with self.live:
      watchloop = awatch(
        self.filepath_right, recursive=True)

      async for changes in watchloop:
        self.live.update(
          self.get_layout('right', changes))

Python Rich: Concurrrent Live: Right Task

TUI in CLI

Now we can see the result in terminal.

Python Rich: Concurrrent Live: TUI in CLI: Horizontal

We can have other layout as well, such as split vertically.

    self.layout.split(
      Layout(name="up"),
      Layout(name="bottom"),
    )

Python Rich: Concurrrent Live: TUI in CLI: Vertical

Real life TUI would not be this simple. We would deal with a bunch of library, with their own behaviour. We also have to split task, that force us to split class.


4: Class Refactoring

Let’s get it on!

We finally get into our main course.

This would be long. But this would not be so complex.

But don’t forget this is just a step, to a more complex system. OOP pattern does required to simplify complex stuff.

Python Source Code

Reading a single script, won’t hurt.

For simplicty I store both classes in just one file. We will separate the class later in other article, beyond this article series.

Script: Skeleton

Two clasess in just one script. The relation between these two clasess is one-to-many. One DualPanel class, run SingleWatch concurrently.

class SingleWatch:
  ...

class DualPanel:
  ...

dual = DualPanel()
...

Python Rich: Class Refactoring: Skeleton

Execute: Run

Program Entry Point

We create class instance

dual = DualPanel()
try:
  asyncio.run(dual.main())
except KeyboardInterrupt:
  print('Goodbye!')

Python Rich: Class Refactoring: Run

Required Package

All imported module shown here:

import asyncio, datetime

from watchfiles import awatch

from rich.live   import Live
from rich.table  import Table
from rich.panel  import Panel
from rich.layout import Layout

Python Rich: Class Refactoring: Import

Now we discuss the master-detail or one-to-many, strating from one master, the DualPanel class.


4A: Rich Layout Class

We have the one master class here.

Class: Skeleton

class DualPanel:
  def make_layout(self):
  def get_layout_empty(self) -> Layout:
  async def main(self):

Python Rich: Rich Layout Class: Skeleton

Method: Make Layout

It is still the same as usual.

  def make_layout(self):
    # Define the layout.
    self.layout = Layout(name="root")
    self.layout.split_row(
      Layout(name="left"),
      Layout(name="right"),
    )

Python Rich: Rich Layout Class: Make Layout

Async Main Method: Live

We basically want to instantiate class, to be user by different panel.

    self.watch_left = SingleWatch(...)

    self.watch_right = SingleWatch(...)

And use later, for example:

    self.watch_left = SingleWatch(...)

    task_left = asyncio.create_task(
      self.watch_left.do_task())

    await(task_left)

For both panel, the complete method is here as below:

  async def main(self):
    self.make_layout()
    self.live = Live(None)

    self.watch_left = SingleWatch(
      self.live, self.layout, 'left',
      '/home/epsi/awatch/code-01-base/')

    self.watch_right = SingleWatch(
      self.live, self.layout, 'right',
      '/home/epsi/awatch/code-02-enh/')

    self.live.update(self.get_layout_empty())

    task_left = asyncio.create_task(
      self.watch_left.do_task())
    task_right = asyncio.create_task(
      self.watch_right.do_task())

    await(task_left)
    await(task_right)

Python Rich: Rich Layout Class: Main

We can initialize live first, and diplay later by this code.

Note that live, called twice:

  • The instance
  • The update
    self.live = Live(None)
    self.live.update(self.get_layout_empty())

Method: Empty Layout

Beware of this!

Where do we put all these method?

One in master class as below.

  def get_layout_empty(self) -> Layout:
    self.layout['left'].update(
      self.watch_left.get_panel(None))
    self.layout['right'].update(
      self.watch_right.get_panel(None))
    return self.layout

Python Rich: Rich Layout Class: Empty

And the other in detail class named SingleWatch().


4B: Watch Task Class

The many detail class.

Class: Skeleton

Python Rich: Watch Task Class: Skeleton

class SingleWatch:
  def __init__(self, 
      ref_live, ref_layout, layout_name, filepath):

  def format_table(self, dataTable):
  def generate_table(self, changes) -> Table:
  def get_panel(self, changes) -> Panel:
  def get_layout(self, changes) -> Layout:

  async def do_task(self):

Constructor

Finally, our own constructor

Since we require different variable for each instance. We should create constructor.

The main part related to Rich library here is:

  • self.live, and
  • self.layout.

In a more complex system, these two should exist.

  def __init__(self, 
      ref_live, ref_layout, layout_name, filepath):
    self.live   = ref_live
    self.layout = ref_layout
    self.layout_name = layout_name
    self.filepath    = filepath

Python Rich: Watch Task Class: Constructor

Task

Remember that we call do_task earlier?

    task_left = asyncio.create_task(
      self.watch_left.do_task())
    task_right = asyncio.create_task(
      self.watch_right.do_task())

We can implement the do_task as below. This contain the watchfiles monitoring part.

  async def do_task(self):
    with self.live:
      watchloop = awatch(
        self.filepath, recursive=True)

      async for changes in watchloop:
        self.live.update(
          self.get_layout(changes))

Python Rich: Watch Task Class: Do Task

Renderable: Layout

Again, remember when we live.update earlier?

  def get_layout_empty(self) -> Layout:
    self.layout['left'].update(
      self.watch_left.get_panel(None))
    self.layout['right'].update(
      self.watch_right.get_panel(None))
    return self.layout

We can implement the get_layout() as below:

  def get_layout(self, changes) -> Layout:
    self.layout[self.layout_name].update(
      self.get_panel(changes))
    return self.layout

Python Rich: Watch Task Class: Layout

Renderable: Panel

A few decoration

Now we can move the panel in detail class.

We can also add some text above the table.

  def get_panel(self, changes) -> Panel:
    dataTable = self.generate_table(changes)

    message = Table.grid(padding=1, expand=True)
    message.add_column(no_wrap=True)
    cts = datetime.datetime.now()
    message.add_row(
      "Time Received: [yellow]" + str(cts))
    message.add_row(dataTable)

    panel = Panel(message, title=self.filepath)
    return panel

Python Rich: Watch Task Class: Panel

Renderable: Table

And finally the table itself.

With additional line containing format_table().

  def generate_table(self, changes) -> Table:
    # Make a new table
    table = Table(expand=True)
    table.add_column("Type")
    table.add_column("File")

    self.format_table(table)

    if changes != None:
      for change in changes:
        short = change[1].replace(
          self.filepath, '')
        table.add_row(
          *[str(change[0]), short])

    return table

Python Rich: Watch Task Class: Table

Renderable: Table Formatting

Pretty Color

We require a few aestethic enhancement.

  def format_table(self, dataTable):
    c = dataTable.columns

    c[0].header_style = "bold blue"
    c[1].header_style = "bold green"

    c[0].style = "blue"
    c[1].style = "green"

Python Rich: Watch Task Class: Format

We are done, with class refactoring. It is time to enjoy the display view.

TUI in CLI

The result in terminal could be as below:

Python Rich: Class Refactoring: TUI in CLI: Narrow

You can resize the terminal wider, to shjow the full path.

Python Rich: Class Refactoring: TUI in CLI: Wide

Or better, you can enjoy the TUI result in video.

How many Process?

You will see in htop, this script has six processes.

Python Rich: How many Process?

But why is it? Shouldn’t it be just two, or maybe four tasks?

In order to answer this, you have to test htop, from the very beginning of this article.


Conclusion

We are actually done here. But if you wish you can continue to real use of rich, in the next article.

Rich works well with asyncio, watchfiles, and websocket.

Farewell. We shall meet again.