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
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):
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'
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"),
)
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.layoutBut 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
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
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
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())
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)
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))
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))
TUI in CLI
Now we can see the result in terminal.
We can have other layout as well, such as split vertically.
self.layout.split(
Layout(name="up"),
Layout(name="bottom"),
)
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()
...
Execute: Run
Program Entry Point
We create class instance
dual = DualPanel()
try:
asyncio.run(dual.main())
except KeyboardInterrupt:
print('Goodbye!')
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
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):
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"),
)
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)
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
And the other in detail class named SingleWatch().
4B: Watch Task Class
The many detail class.
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, andself.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
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))
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.layoutWe 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
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
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
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"
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:
You can resize the terminal wider, to shjow the full path.
Or better, you can enjoy the TUI result in video.
How many Process?
You will see in htop, this script has six processes.
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.