Есть ли способ указать выходную ячейку, в которой функция должна печатать свои выходные данные?
В моем конкретном случае у меня запущено несколько потоков, каждый из которых имеет регистратор. Выходные данные регистратора печатаются в любой работающей ячейке, мешая предполагаемому выводу этой ячейки. Есть ли способ заставить регистратор печатать, например, только на cell #1
?
🤔 А знаете ли вы, что...
С Python можно создавать кросс-платформенные приложения для Windows, macOS и Linux.
Вы можете использовать следующий подход:
getLogger()
) в QueueHandler, чтобы накопить сообщения журнала в queue.Queue
.QueueListener
, как следует из названия, будет прослушивать новые элементы в очереди регистрации. Он передаст новые элементы в StreamHandler
, который их фактически распечатает.Предполагая, что мы хотим напечатать ниже ячейки 1, это может выглядеть следующим образом:
# Cell 1
import logging, queue, threading, time
from logging.handlers import QueueHandler, QueueListener
log_queue = queue.Queue(-1)
logging.getLogger().addHandler(QueueHandler(log_queue))
listener = QueueListener(log_queue, logging.StreamHandler())
listener.start()
В ячейке 2 мы смоделируем некоторую деятельность:
# Cell 2
def log_activity_1():
while True:
logging.getLogger().warning("Activity 1")
time.sleep(1)
threading.Thread(target=log_activity_1, daemon=True).start()
И аналогично в ячейке 3:
# Cell 3
def log_activity_2():
while True:
logging.getLogger().warning("Activity 2")
time.sleep(2)
threading.Thread(target=log_activity_2, daemon=True).start()
Вывод будет происходить, по сути, в режиме реального времени, под ячейкой, содержащей вызов listener.start()
, то есть в ячейке 1 (и только там) в нашем случае. Это будет выглядеть так, как и ожидалось: для каждого зарегистрированного «Действия 2» мы будем видеть зарегистрированное «Действие 1» попеременно и примерно в два раза чаще, поскольку мы спим 2 секунды в первом случае и 1 секунду во втором:
После завершения обработки мы можем остановить QueueListener
(программно или вручную) с помощью listener.stop()
— точнее, мы должны остановить прослушиватель таким образом, следуя его документации: если вы не вызовете [stop()
] до выхода приложения , в очереди могут остаться записи, которые не будут обработаны.
Вы можете создать оболочку, которая будет перенаправлять все напечатанное в другую ячейку.
# Cell 1
import logging
from io import StringIO
from IPython.display import display, HTML
import sys
# Set up a StringIO object to capture output
output_capture = StringIO()
ch = logging.StreamHandler(output_capture)
ch.setLevel(logging.INFO)
# Configure the logger
logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(ch)
# Create a display area
display_id = display(HTML(''), display_id=True)
# Function to update the display
def update_display():
output_contents = output_capture.getvalue()
display_id.update(HTML(f"<pre>{output_contents}</pre>"))
def capture_output(func):
def wrapper(*args, **kwargs):
# Redirect stdout to our StringIO object
old_stdout = sys.stdout
sys.stdout = output_capture
try:
# Call the original function
result = func(*args, **kwargs)
finally:
# Restore stdout
sys.stdout = old_stdout
# Update the display
update_display()
return result
return wrapper
В следующей ячейке:
# Cell 2
@capture_output
def my_function():
print('Hello world!')
# Now when you call my_function(), anything printed within the function will be displayed in the first cell
my_function()