Excel - Monitor - Multiple Task

   
   
 python  
 monitor  
Python Monitoring Article Series Test

Python - Monitor

Table of Content
Where to Discuss?

Local Group

Preface

Goal: Gather All The Stuff Together

This is a multiparts article.

We need to integrate all panel into just one script. Separate task into its own class. Combine all the different class. Run them in concurrent fashioned, and display each fancy output in its own rich panel.

Terminal User Interface

Yay

I finally made with my first TUI Application.

Python: Multitask: Dark Triple Panels

Good job!

Preparation

$ pip install openpyxl websockets tomli watchfiles rich aiohttp jinja2 aiohttp_jinja2

21: Separate Class

Clean Up

I just realize that python logging library, display all the logs. Not just the web server, but also web socket, and watchfiles. I think I need to separate, the logging class, and the web server class.

Web Server

The web server class along with Jinja2, is turn out to be very short:

Nothing special. I just copy paste from official documentation. But for use with asynchronous.

from aiohttp import web
import jinja2, aiohttp_jinja2

@aiohttp_jinja2.template("index.jinja")
class HomeHandler(web.View):

  async def get(self):
    return {}

class AIOWeb:
  async def task_web(self):
    app = web.Application()

    ...

Python: Class Clean Up: Web Server

This works for python 3.10. For python 3.11, be sure to upgrade aiohttp_jinja2 to 1.5.1.

Logging

The rest is this hacky logging routine. You can obtain the python class here:

I also clean up the method name, to differ the private method access, as shown in skeleton below.

class Log2Panel:
  def __init__(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 do_log(self):

Python: Class Clean Up: Logging Skeleton

Universal logging, can be set in constructor. Do not forget to also initialize, layout stuff.

  def __init__(self,
      ref_live, ref_layout, layout_name):
    # rich: required in descendent class
    self.live   = ref_live
    self.layout = ref_layout
    self.layout_name = layout_name

    ...

    logging.basicConfig(
      stream=self.log_stream,
      level=logging.INFO)

Python: Class Clean Up: Logging Constructor

So we can display all result as below:

Python: Class Clean Up: Logging Output Panel

Panel above display application wide logging. Not just the web server information.

22: Gather Task

Concurrency

Manage multiple task is not scary, but however I still need to carefully manage the code.

Python Source Code

You can obtain the python class here:

Custom Package

We start with iporting our own custom module library.

import asyncio
import xl2web_25c, xl2web_25d
import log2panel_29, aioweb_29

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

Python: Concurrent Task: Custom Package

Skeleton

This script is basically a rich application. Here is the skeleton:

class MultiPanels:
  def __make_layout(self):
  def __get_layout_empty(self) -> Layout:
  def __prepare_watcher(self):
  async def main(self):

Python: Concurrent Task: Skeleton

Empty Panel

No code intact. The same as previous code.

Python: Concurrent Task: Empty Panel

Triple Panel

How you manage your layout is up to yo. However, this is mine.

  def __make_layout(self):
    # Define the layout.
    self.layout = Layout(name="root")

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

    self.layout['top'].split_row(
      Layout(name="left"),
      Layout(name="right"),
    )

Python: Concurrent Task: Triple Panel

Manage Task

Since we have many task, it is better to be well prepared.

Create each class instance, one by one.

  def __prepare_watcher(self):
    self.live = Live()

    self.watch_left = xl2web_25c.Xl2Web_c(
      'sheetGlobal', self.live, self.layout, 'left')

    self.watch_right = xl2web_25d.Xl2Web_d(
      'sheetDaily', self.live, self.layout, 'right')

    self.watch_log = log2panel_29.Log2Panel(
      self.live, self.layout, 'bottom')

Python: Concurrent Task: Manage Task

Await All

Run baby, run…

  async def main(self):
    self.__make_layout()
    self.__prepare_watcher()

    with self.live:
      self.live.update(self.__get_layout_empty())

      task_web = aioweb_29.AIOWeb().task_web
      task_left = asyncio.create_task(
        self.watch_left.main())
      task_right = asyncio.create_task(
        self.watch_right.main())
      task_log = asyncio.create_task(
        self.watch_log.do_log())

Now you can run them all concurrently

      await(task_web())
      await(task_left)
      await(task_right)
      await(task_log)

Python: Concurrent Task: Await All

Note that, the simple web server is not in made in create_task, so we should run it before other tasks. You can manage the simple web server, to be created in create_task as well, if necessary.

Asynchronous Entry Point

Actually, it is just program entry point, in asynchronous fashioned.

class MultiPanels:
   ...

multi = MultiPanels()
try:
  asyncio.run(multi.main())
except KeyboardInterrupt:
  print('Goodbye!')

Python: Concurrent Task: Asynchronous Entry Point

Terminal User Interface

Python: Concurrent Task: Terminal User Interface

Conclusion

Our journey is complete here. There is still also chart, and testing. But I’m still pretty busy these days. Maybe I’ll skip this time.

What do you think?