Preface
Goal: Add common feature such as configuration, terminal interface, and web server
This is a multiparts article.
Wouldn’t it be nice if we can have colorful logging in a panel?
The main idea here is using StringIO instead of easy top use Rich Handler.
We also need to put everything in rich panel, so that we don’t need to load each script manually.
18: Handling Log With Rich
Python Source Code
For convenience, I share the script.
Preparation
As usual, we need to prepare AIOHTTP, aong with Jinja2.
import logging, sys
import jinja2, aiohttp_jinja2
from aiohttp import web
from rich.logging import RichHandler
@aiohttp_jinja2.template("index.jinja")
class HomeHandler(web.View):
async def get(self):
return {}
Note that we have import RichHandler in above statement.
Rich Handler
The main idea is to use RichHandler().
As in official documentation we can do this:
FORMAT = "%(message)s"
logging.basicConfig(
level=logging.DEBUG, format=FORMAT,
datefmt="[%X]", handlers=[RichHandler()])
Main Program
And the rest is the same as previous.
def main():
FORMAT = "%(message)s"
logging.basicConfig(
level=logging.DEBUG, format=FORMAT,
datefmt="[%X]", handlers=[RichHandler()])
app = web.Application()
# setup jinja2
aiohttp_jinja2.setup(app,
loader=jinja2.FileSystemLoader('../templates'))
app.router.add_get('/', HomeHandler, name="index")
app.router.add_static('/', '../site')
web.run_app(app)
main()Running The Script
Consider running the script:
❯ python 27-log2rich.py[08:18:01] DEBUG Using selector: selector_events.py:54
EpollSelector
======== Running on http://0.0.0.0:8080 ========
(Press CTRL+C to quit)
[08:18:40] INFO 127.0.0.1 [24/Jan/2023:01:18:40 web_log.py:206
+0000] "GET / HTTP/1.1" 200 7238
"-" "Mozilla/5.0 (X11; Linux
x86_64; rv:109.0) Gecko/20100101
Firefox/109.0"
INFO 127.0.0.1 [24/Jan/2023:01:18:40 web_log.py:206
+0000] "GET /bulma.min.css
HTTP/1.1" 200 0
"http://localhost:8080/"
"Mozilla/5.0 (X11; Linux x86_64;
rv:109.0) Gecko/20100101
Firefox/109.0" 
The issue with using RichHandler is,
it is not flexible, that we cannot tweak things.
The code become too long and complex.
And it is also not responsive.
So we have to find other way, to make things simple.
19: Capture Logging with StringIO
Plain Without Rich using Delay
Nothing scary here, just a short script.
Python Source Code
Our script, ready to serve.
Package Requirement
As usual, we import stuff.
Adding StringIO module.
import logging, asyncio
import jinja2, aiohttp_jinja2
from aiohttp import web
from io import StringIO
@aiohttp_jinja2.template("index.jinja")
class HomeHandler(web.View):
async def get(self):
return {}
Skeleton
Very light, all we have to do is, manage two task concurrently.
class HomeHandler(web.View):
async def get(self):
class LayoutExample:
def __init__(self):
async def task_web(self):
async def do_log(self):
async def main(self):Logging Stream
Do not forget to initialize.
class LayoutExample:
def __init__(self):
self.log_stream = StringIO()
We are going to add this self.log_stream ,
as logging stream later.
async def task_web(self):
logging.basicConfig(
stream=self.log_stream,
level=logging.DEBUG)Task Web
The web.run_app(app) is blocking.
Since we use asynchronous we have to go lower lever a bit:
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', 8080)
await site.start()The complete method is as below:
async def task_web(self):
logging.basicConfig(
stream=self.log_stream,
level=logging.DEBUG)
app = web.Application()
# setup jinja2
aiohttp_jinja2.setup(app,
loader=jinja2.FileSystemLoader(
'../templates'))
app.router.add_get(
'/', HomeHandler, name="index")
app.router.add_static('/', '../site')
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', 8080)
await site.start()
Task Log
Since we have already get our logging stream. Now we can just get the value anytime.
async def do_log(self):
while True:
print(self.log_stream.getvalue())
await asyncio.sleep(1)
I felt like using delay every one second is a bad idea. I’d rather use event driven signal instead. but unfortunately this is something that I cannot afford. We are going to deal with this later.
Main Method
I forget why, but we do need to wrap task web in a task.
I’ll figure it out later for reader later.
I’m pretty busy right now.
async def main(self):
task_log = asyncio.create_task(
self.do_log())
await(self.task_web())
await(task_log)And finally we can run the script in main program.
example = LayoutExample()
asyncio.run(example.main())
Running The Script
Consider running the script. This will show empty lines, until you open something in browser.
❯ python 27-log2io.pyINFO:aiohttp.access:127.0.0.1 [24/Jan/2023:01:20:33 +0000] "GET / HTTP/1.1" 200 7238 "-" "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/109.0"
INFO:aiohttp.access:127.0.0.1 [24/Jan/2023:01:20:33 +0000] "GET /favicon.ico HTTP/1.1" 404 0 "http://localhost:8080/" "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/109.0"
I know this is not an elegant way to log about something. That is why we have to continue into next section. This is going to be a little bit long, because we have to integrate with rich panel.
20: Manage Log in Rich Panel
Although the code looks long, half part of has already been discussed.
Python Source Code
Most of the time, having ready script is, better than not having one.
Skeleton
The skeleton is going to be long
class LayoutExample:
MAXLINE = 40
def __init__(self):
def make_layout(self):
def get_lines(self):
def capture(self, message):
def split(self, line):
def get_panel(self) -> Panel:
def get_layout(self) -> Layout:
async def task_web(self):
async def do_log(self):
async def main(self):
Nothing is complicated. Really.
Recquired Package
Import Rich
Just a bunch of rich package.
import logging, asyncio
import jinja2, aiohttp_jinja2
from aiohttp import web
from io import StringIO
from rich.console import Console
from rich.text import Text
from rich.table import Table
from rich.panel import Panel
from rich.layout import Layout
from rich.live import Live
Brief Summary
Consider this overview of the script
Constructor
We need to create maximum buffer. So our memory could be limited
class LayoutExample:
MAXLINE = 40And then we initialize some stuff.
class LayoutExample:
MAXLINE = 40
def __init__(self):
self.console = Console()
self.log_stream = StringIO()
self.last_msg = None
self.lines = None
self.make_layout()
Logging Task
We meet again with the do_log method.
But this time it has a bunch of code.
async def do_log(self):
with self.live:
while True:
self.lines = self.get_lines()
last_msg = self.lines[-1:]
if last_msg != self.last_msg:
self.live.update(
self.get_layout())
self.last_msg = last_msg
await asyncio.sleep(1)
I felt like using delay every one second is a bad idea. I’d rather use event driven signal instead. but unfortunately this is something that I cannot afford.
So all I do is comparing the last message for every delay. If it changed, then we can update layout.
We can find the last lines of an array, by using this pythonic notation.
last_msg = self.lines[-1:]Getting The Lines for StringIO
Getting the lines is simply:
def get_lines(self):
log_str = self.log_stream.getvalue()
lines = log_str.splitlines()Since want to limit the buffer, we have to calculate the buffer for each delay.
def get_lines(self):
log_str = self.log_stream.getvalue()
lines = log_str.splitlines()
if len(lines) > self.MAXLINE:
lines = lines[- self.MAXLINE:]
self.log_stream.truncate(0)
self.log_stream.seek(0)
self.log_stream.write("\n".join(lines)+"\n")
return linesWe can just create new log stream, instead of erasing the content:
self.log_stream = StringIO() 
Splitting Lines
Suppose that we have a log message like this one:
INFO:aiohttp.access:192.168.0.67 [23/Jan/2023:17:26:52 +0000]We can separate the message into parts such as:
INFO
aiohttp.access
192.168.0.67 [23/Jan/2023:17:26:52 +0000]And discarded the second part
INFO
192.168.0.67 [23/Jan/2023:17:26:52 +0000]This can be done by using this method:
def split(self, line):
pos1 = line.find(':', 0)
pos2 = line.find(':', pos1+1)
col1 = line[:pos1]
col2 = line[pos1+1:pos2]
col3 = self.capture(line[pos2+1:])
return [col1, col3]
We use capture to make the text colorful.
Capture Rich Color
I found this ide from stackoverflow:
All we need to do is to print the text,
and capture the output into string.
And get it back as rich Text object.
# https://stackoverflow.com/questions/73512663/
def capture(self, message):
with self.console.capture() as capture:
self.console.print(message)
str_text = capture.get()
str_text = Text.from_ansi(str_text)
return str_text
We are done with StringIO,
now we can setup our panel.
Logging Task
Consider go back to our logging task.
async def do_log(self):
with self.live:
while True:
self.lines = self.get_lines()
last_msg = self.lines[-1:]
if last_msg != self.last_msg:
self.live.update(
self.get_layout())
self.last_msg = last_msg
await asyncio.sleep(1)The Live.update required us a layout.
In that function, we can update the layout .
Make Layout
Preparation
Just one layout:
def make_layout(self):
# Define the layout
self.layout = Layout(name="root")
Getting Layout
This method just updating the panel.
Propagating the update signal,
from Live to Layout to Panel.
def get_layout(self) -> Layout:
self.layout.update(
self.get_panel())
return self.layout
Getting Panel
It is just translating lines in buffer into table in panel.
def get_panel(self) -> Panel:
table = Table.grid(padding=0, expand=True)
table.add_column(no_wrap=True,
min_width=8, overflow="fold", style="blue")
table.add_column()
for i, line in enumerate(self.lines[::-1]):
table.add_row(*self.split(line))
panel = Panel(table,
title="[b]Web [yellow]Logs")
return panel
Actually this is the core idea.
Panel Limitation
Reverse
Rich TUI Panel is not scrollable, so we can’t see the last messages, because it buried, oveflowed under the table.
The quick and dirty solution is to show in LIFO. Last log message will be shown at top most. We can do this by reverse the array with below code
self.lines[::-1]Then use the array in the for loop.
for i, line in enumerate(self.lines[::-1]):
table.add_row(*self.split(line))Task Web
The Web Task is identical to previous example. It is verbatim copy. No changes intact.
async def task_web(self):
logging.basicConfig(
stream=self.log_stream,
level=logging.DEBUG)
app = web.Application()
# setup jinja2
aiohttp_jinja2.setup(app,
loader=jinja2.FileSystemLoader(
'../templates'))
app.router.add_get(
'/', HomeHandler, name="index")
app.router.add_static('/', '../site')
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', 8080)
await site.start()
Main Method
We can finally run this two coroutine task in concurrent fashioned.
async def main(self):
self.live = Live(None)
task_log = asyncio.create_task(
self.do_log())
await(self.task_web())
await(task_log)And in main program
Main Program
Execute: Program Entry Point
example = LayoutExample()
asyncio.run(example.main())
Running The Script
It is responsive, as you can see below:
What is Next 🤔?
We need to integrate all panel into just one script.
Consider continue reading [ Excel - Monitor - Multiple Task ].