pw_console: Add test-mode fake logger task
Adds an asyncio background task to generate fake log messages. This
can be enabled via the --test-mode command line flag.
Test: pw-console --test-mode
No-Docs-Update-Reason: Test mode is temporary for log viewer dev.
Change-Id: I4bbb7221c4340d46899d2e65b6855ea533f036ef
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/49025
Pigweed-Auto-Submit: Anthony DiGirolamo <tonymd@google.com>
Commit-Queue: Auto-Submit <auto-submit@pigweed.google.com.iam.gserviceaccount.com>
Reviewed-by: Joe Ethier <jethier@google.com>
diff --git a/pw_console/py/pw_console/__main__.py b/pw_console/py/pw_console/__main__.py
index c516897..05171f6 100644
--- a/pw_console/py/pw_console/__main__.py
+++ b/pw_console/py/pw_console/__main__.py
@@ -23,7 +23,7 @@
import pw_cli.log
import pw_cli.argument_types
-from pw_console.console_app import embed
+from pw_console.console_app import embed, FAKE_DEVICE_LOGGER_NAME
_LOG = logging.getLogger(__package__)
@@ -80,16 +80,20 @@
pw_cli.log.install(args.loglevel, True, False, args.logfile)
+ global_vars = None
default_loggers: List = []
- # TODO: Add test-mode loggers here.
- # if args.test_mode:
- # default_loggers = [
- # # Don't include pw_console package logs (_LOG) in the log pane UI.
- # # Add the fake logger for test_mode.
- # logging.getLogger(FAKE_DEVICE_LOGGER_NAME)
- # ]
+ if args.test_mode:
+ default_loggers = [
+ # Don't include pw_console package logs (_LOG) in the log pane UI.
+ # Add the fake logger for test_mode.
+ logging.getLogger(FAKE_DEVICE_LOGGER_NAME)
+ ]
+ # Give access to adding log messages from the repl via: `LOG.warning()`
+ global_vars = dict(LOG=default_loggers[0])
- embed(loggers=default_loggers)
+ embed(global_vars=global_vars,
+ loggers=default_loggers,
+ test_mode=args.test_mode)
if args.logfile:
print(f'Logs saved to: {args.logfile}')
diff --git a/pw_console/py/pw_console/console_app.py b/pw_console/py/pw_console/console_app.py
index fe2e0da..d2447df 100644
--- a/pw_console/py/pw_console/console_app.py
+++ b/pw_console/py/pw_console/console_app.py
@@ -49,6 +49,10 @@
_LOG = logging.getLogger(__package__)
+# Fake logger for --test-mode
+FAKE_DEVICE_LOGGER_NAME = 'fake_device.1'
+_FAKE_DEVICE_LOG = logging.getLogger(FAKE_DEVICE_LOGGER_NAME)
+
class FloatingMessageBar(ConditionalContainer):
"""Floating message bar for showing status messages."""
@@ -267,14 +271,54 @@
"""Redraw the prompt_toolkit UI."""
self.application.invalidate()
- async def run(
- self,
- # TODO: remove pylint disable line.
- test_mode=False # pylint: disable=unused-argument
- ):
+ async def run(self, test_mode=False):
"""Start the prompt_toolkit UI."""
- unused_result = await self.application.run_async(
- set_exception_handler=True)
+ if test_mode:
+ background_log_task = asyncio.create_task(self.log_forever())
+
+ try:
+ unused_result = await self.application.run_async(
+ set_exception_handler=True)
+ finally:
+ if test_mode:
+ background_log_task.cancel()
+
+ async def log_forever(self):
+ """Test mode async log generator coroutine that runs forever."""
+ message_count = 0
+ # Sample log lines:
+ # Log message [= ] # 291
+ # Log message [ = ] # 292
+ # Log message [ = ] # 293
+ # Log message [ = ] # 294
+ # Log message [ = ] # 295
+ # Log message [ = ] # 296
+ # Log message [ = ] # 297
+ # Log message [ = ] # 298
+ # Log message [ = ] # 299
+ # Log message [ =] # 300
+ while True:
+ await asyncio.sleep(2)
+ bar_size = 10
+ position = message_count % bar_size
+ bar_content = " " * (bar_size - position - 1) + "="
+ if position > 0:
+ bar_content = "=".rjust(position) + " " * (bar_size - position)
+ new_log_line = 'Log message [{}] # {}'.format(
+ bar_content, message_count)
+ if message_count % 10 == 0:
+ new_log_line += (" Lorem ipsum dolor sit amet, consectetur "
+ "adipiscing elit.") * 8
+ # TODO: Test log lines that include linebreaks.
+ # if message_count % 11 == 0:
+ # new_log_line += inspect.cleandoc(""" [PYTHON] START
+ # In []: import time;
+ # def t(s):
+ # time.sleep(s)
+ # return 't({}) seconds done'.format(s)""")
+
+ message_count += 1
+ _FAKE_DEVICE_LOG.info(new_log_line)
def embed(
