I have made a small application that uses a Rich to show a Live view in several Panels.
Is there a way to put standard (or rich) print statements into a particular panel?
Being able to show logging output in its own dedicated panel would also work.
I feel like this would be a very common use case but I haven't found any documentation. I think the answer could be in using the Console.capture() method but I can't figure it out.
I figure out how you can manage to that. First, we need to intercept the stdout process of Rich logger. We start with a class:
from collections import deque
class Logger():
_instance = None
def __init__(self):
self.messages = deque(["sa"])
self.size = 10
def __new__(class_, *args, **kwargs):
if not isinstance(class_._instance, class_):
class_._instance = object.__new__(class_, *args, **kwargs)
return class_._instance
def write(self, message):
self.messages.extend(message.splitlines())
while len(self.messages) > self.size:
self.messages.popleft()
def flush(self):
pass
which is a Singleton class. We need to pass this class into console API as
from rich.console import Console
c = Console(file=Logger(), width=150)
with some width. Then, we create a logging handler
from rich.logging import RichHandler
r = RichHandler(console=c)
This will be our logging handler as
import logging
logging.basicConfig(
level="NOTSET", format=FORMAT, datefmt="[%X]", handlers=[r]
)
logger = logging.getLogger("rich")
Later on, we need to use our Logger class called somewhere you manage your layout. For me, it is inside a Dashboard class.
class Dashboard:
def __init__(self):
self.log_std = Logger()
def update(self, new_parameters):
self.layout["logs"].update(Panel(Text(
"\n".join(self.log_std.messages), justify="right"), padding=(1, 2),
title="[b red]Logs",
))
Each time I call update method, it updates my layout.
My layout is more complex, self.layout["logs"] where I show the logs.
So I took @Mete Yildirim's answer as inspiration and came up with a slight variation that uses an existing logging handler instead of creating a new logger.
The logging.handlers module has a BufferingHandler() that I used for prototyping my solution. By passing the handler into rich, I can snoop its contents and print them into my rich Panel without mods to logging.
LOG_BUFFER_MAX_MSGS = 20
# Set up the main/root logger
main_logger = logging.getLogger()
main_logger.setLevel(logging.DEBUG)
# Instantiate the buffering handler we will watch from within Rich
buffering_handler = BufferingHandler(capacity=LOG_BUFFER_MAX_MSGS)
main_logger.addHandler(buffering_handler)
# Local logger
log = logging.getLogger("rich")
# Create a basic Rich layout
layout = Layout(name="root")
def get_log():
"""
We call this method from within Rich to snoop the messages
within the BufferingHandler and put them in a form Rich
can display.
Check the BufferingHandler() source to see how we can access
its data.
"""
log_messages = []
for li in buffering_handler.buffer:
log_messages.append(li.msg)
return Panel("\n".join(log_messages))
# Run Rich, displaying each log message to the screen
with Live(layout, refresh_per_second=4) as live:
while True:
layout["root"].update(get_log())
time.sleep(0.25)
To test the above, we can generate some log messages in the background:
def create_log_messages():
msgs = (
"Test message 1",
"Test message 2",
"Test message 3",
)
for li in msgs:
log.info(li)
time.sleep(2)
threading.Thread(target=create_log_messages).start()
Shortcomings:
The BufferingHandler() will clear it's buffer when the number of lines exceeds the value passed into its capacity argument.
I would prefer that it delete old messages instead, but that will require either overloading the existing BufferHandler() implementation or writing a new Handler. The BufferHandler() code is short, so simply writing a new one shouldn't be too much effort.
I had a similar problem where I could use print to display and object with the styling from rich but once I put the object in a panel, it wasn't styling it anymore.
Just with print:
from rich import print
print(self.model)
To solve it, I used pretty printing:
from rich import print
from rich.panel import Panel
from rich.pretty import Pretty
prettyModel = Pretty(self.model)
print(Panel(prettyModel))
I hope it can help.
My solution violates some rules, but I got logging to work asynchronously. I subclassed RichHandler to update a layout instead of printing to stdout. I was not able to figure out how to adjust the vertical cropping or the number of messages to fit the available screen space, so instead, I limited the number of messages to five. Hopefully, someone wiser will pick this up and make it better.
class RichHandlerPanel(RichHandler):
"""Send logs to a layout."""
def __init__(
self,
layout: Layout,
level: Union[int, str] = logging.NOTSET,
console: Optional[Console] = None,
panel_title: str = "Log",
max_display: int = 5,
**kwargs
) -> None:
super().__init__(level=level, console=console, **kwargs)
self.layout = layout
self.full_log = None
self.panel_title = panel_title
self.max_display = max_display
def emit(self, record: LogRecord) -> None:
"""Invoke by logging. This is a copy of the original with a change on how emit is done."""
message = self.format(record)
traceback = None
if (
self.rich_tracebacks
and record.exc_info
and record.exc_info != (None, None, None)
):
exc_type, exc_value, exc_traceback = record.exc_info
assert exc_type is not None
assert exc_value is not None
traceback = Traceback.from_exception(
exc_type,
exc_value,
exc_traceback,
width=self.tracebacks_width,
extra_lines=self.tracebacks_extra_lines,
theme=self.tracebacks_theme,
word_wrap=self.tracebacks_word_wrap,
show_locals=self.tracebacks_show_locals,
locals_max_length=self.locals_max_length,
locals_max_string=self.locals_max_string,
suppress=self.tracebacks_suppress,
)
message = record.getMessage()
if self.formatter:
record.message = record.getMessage()
formatter = self.formatter
if hasattr(formatter, "usesTime") and formatter.usesTime():
record.asctime = formatter.formatTime(record, formatter.datefmt)
message = formatter.formatMessage(record)
message_renderable = self.render_message(record, message)
log_renderable = self.render(
record=record, traceback=traceback, message_renderable=message_renderable
)
if not self.full_log:
self.full_log = log_renderable
else:
for r in range(log_renderable.row_count):
self.full_log.add_row(
*[
log_renderable.columns[c]._cells[r]
for c in range(len(log_renderable.columns))
]
)
while len(self.full_log.rows) > self.max_display:
for c in range(len(log_renderable.columns)):
self.full_log.columns[c]._cells.pop(0)
self.full_log.rows.pop(0)
try:
p = Panel(self.full_log, title=self.panel_title, title_align="left")
self.layout.update(p)
except Exception:
self.handleError(record)
When instantiating the handler, I provide the layout cell:
def _init_logging(self) -> None:
# Handler
self._log_handler = RichHandlerPanel(
level=logging.INFO, layout=self.layout["log"]
)
# Logger
self._logger = logging.getLogger()
self._logger.addHandler(self._log_handler)
The simplest solution I found is by utilizing console protocal to make a console itself renderable. Here is an example:
from rich.console import Console
import os
class ConsolePanel(Console):
def __init__(self,*args,**kwargs):
console_file = open(os.devnull,'w')
super().__init__(record=True,file=console_file,*args,**kwargs)
def __rich_console__(self,console,options):
texts = self.export_text(clear=False).split('\n')
for line in texts[-options.height:]:
yield line
by adding a __rich_console__ method to the Console class, it can be directly treated as other renderable objects, and you can put it anywhere in your layout.
Here is a simple example with two consoles placed vertically
if __name__=='__main__':
from rich.layout import Layout
from rich.live import Live
import time
from datetime import datetime
class Interface():
def __init__(self) -> None:
self.console:list[ConsolePanel] = [ConsolePanel() for _ in range(2)]
def get_renderable(self):
layout = Layout()
layout.split_column(
Layout(self.console[0],name='top'),
Layout(self.console[1],name='bottom',size=6)
)
layout.children[0]
return layout
db = Interface()
with Live(get_renderable=db.get_renderable):
while True:
time.sleep(1)
db.console[0].print(datetime.now().ctime()+'='*100)
db.console[1].print(datetime.now().ctime())
