You've already forked MicroPythonOS
mirror of
https://github.com/m5stack/MicroPythonOS.git
synced 2026-05-20 11:51:27 -07:00
Replace binary logging with source copy
This commit is contained in:
Thomas Farstrike
@@ -1,9 +1,13 @@
|
||||
import sys
|
||||
import logging
|
||||
|
||||
from mpos import Activity, DisplayMetrics, BuildInfo, DeviceInfo
|
||||
|
||||
class About(Activity):
|
||||
|
||||
logger = logging.getLogger(__file__)
|
||||
logger.setLevel(logging.INFO)
|
||||
|
||||
def _add_label(self, parent, text, is_header=False, margin_top=DisplayMetrics.pct_of_height(5)):
|
||||
"""Helper to create and add a label with text."""
|
||||
label = lv.label(parent)
|
||||
@@ -31,7 +35,7 @@ class About(Activity):
|
||||
self._add_label(screen, f"Free space {path}: {free_space} bytes")
|
||||
self._add_label(screen, f"Used space {path}: {used_space} bytes")
|
||||
except Exception as e:
|
||||
print(f"About app could not get info on {path} filesystem: {e}")
|
||||
self.logger.warning(f"About app could not get info on {path} filesystem: {e}")
|
||||
|
||||
def onCreate(self):
|
||||
screen = lv.obj()
|
||||
@@ -98,7 +102,7 @@ class About(Activity):
|
||||
import esp32
|
||||
self._add_label(screen, f"Temperature: {esp32.mcu_temperature()} °C")
|
||||
except Exception as e:
|
||||
print(f"Could not get ESP32 hardware info: {e}")
|
||||
self.logger.warning(f"Could not get ESP32 hardware info: {e}")
|
||||
|
||||
# Partition info (ESP32 only)
|
||||
try:
|
||||
@@ -110,12 +114,12 @@ class About(Activity):
|
||||
self._add_label(screen, f"Next update partition: {next_partition}")
|
||||
except Exception as e:
|
||||
error = f"Could not find partition info because: {e}\nIt's normal to get this error on desktop."
|
||||
print(error)
|
||||
self.logger.warning(error)
|
||||
self._add_label(screen, error)
|
||||
|
||||
# Machine info
|
||||
try:
|
||||
print("Trying to find out additional board info, not available on every platform...")
|
||||
self.logger.info("Trying to find out additional board info, not available on every platform...")
|
||||
self._add_label(screen, f"{lv.SYMBOL.POWER} Machine Info", is_header=True)
|
||||
import machine
|
||||
self._add_label(screen, f"machine.freq: {machine.freq()}")
|
||||
@@ -127,12 +131,12 @@ class About(Activity):
|
||||
self._add_label(screen, f"machine.reset_cause(): {machine.reset_cause()}")
|
||||
except Exception as e:
|
||||
error = f"Could not find machine info because: {e}\nIt's normal to get this error on desktop."
|
||||
print(error)
|
||||
self.logger.warning(error)
|
||||
self._add_label(screen, error)
|
||||
|
||||
# Freezefs info (production builds only)
|
||||
try:
|
||||
print("Trying to find out freezefs info")
|
||||
self.logger.info("Trying to find out freezefs info")
|
||||
self._add_label(screen, f"{lv.SYMBOL.DOWNLOAD} Frozen Filesystem", is_header=True)
|
||||
import freezefs_mount_builtin
|
||||
self._add_label(screen, f"freezefs_mount_builtin.date_frozen: {freezefs_mount_builtin.date_frozen}")
|
||||
@@ -147,7 +151,7 @@ class About(Activity):
|
||||
# BUT which will still have the frozen-inside /lib folder. So the user will be able to install apps into /builtin
|
||||
# but they will not be able to install libraries into /lib.
|
||||
error = f"Could not get freezefs_mount_builtin info because: {e}\nIt's normal to get an exception if the internal storage partition contains an overriding /builtin folder."
|
||||
print(error)
|
||||
self.logger.warning(error)
|
||||
self._add_label(screen, error)
|
||||
|
||||
# Display info
|
||||
@@ -159,7 +163,7 @@ class About(Activity):
|
||||
dpi = DisplayMetrics.dpi()
|
||||
self._add_label(screen, f"Dots Per Inch (dpi): {dpi}")
|
||||
except Exception as e:
|
||||
print(f"Could not get display info: {e}")
|
||||
self.logger.warning(f"Could not get display info: {e}")
|
||||
|
||||
# Disk usage info
|
||||
self._add_label(screen, f"{lv.SYMBOL.DRIVE} Storage", is_header=True)
|
||||
|
||||
Binary file not shown.
@@ -0,0 +1,254 @@
|
||||
from micropython import const
|
||||
import io
|
||||
import sys
|
||||
import time
|
||||
|
||||
CRITICAL = const(50)
|
||||
ERROR = const(40)
|
||||
WARNING = const(30)
|
||||
INFO = const(20)
|
||||
DEBUG = const(10)
|
||||
NOTSET = const(0)
|
||||
|
||||
_DEFAULT_LEVEL = const(WARNING)
|
||||
|
||||
_level_dict = {
|
||||
CRITICAL: "CRITICAL",
|
||||
ERROR: "ERROR",
|
||||
WARNING: "WARNING",
|
||||
INFO: "INFO",
|
||||
DEBUG: "DEBUG",
|
||||
NOTSET: "NOTSET",
|
||||
}
|
||||
|
||||
_loggers = {}
|
||||
_stream = sys.stderr
|
||||
_default_fmt = "%(levelname)s:%(name)s:%(message)s"
|
||||
_default_datefmt = "%Y-%m-%d %H:%M:%S"
|
||||
|
||||
|
||||
class LogRecord:
|
||||
def set(self, name, level, message):
|
||||
self.name = name
|
||||
self.levelno = level
|
||||
self.levelname = _level_dict[level]
|
||||
self.message = message
|
||||
self.ct = time.time()
|
||||
self.msecs = int((self.ct - int(self.ct)) * 1000)
|
||||
self.asctime = None
|
||||
|
||||
|
||||
class Handler:
|
||||
def __init__(self, level=NOTSET):
|
||||
self.level = level
|
||||
self.formatter = None
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
def setLevel(self, level):
|
||||
self.level = level
|
||||
|
||||
def setFormatter(self, formatter):
|
||||
self.formatter = formatter
|
||||
|
||||
def format(self, record):
|
||||
return self.formatter.format(record)
|
||||
|
||||
|
||||
class StreamHandler(Handler):
|
||||
def __init__(self, stream=None):
|
||||
super().__init__()
|
||||
self.stream = _stream if stream is None else stream
|
||||
self.terminator = "\n"
|
||||
|
||||
def close(self):
|
||||
if hasattr(self.stream, "flush"):
|
||||
self.stream.flush()
|
||||
|
||||
def emit(self, record):
|
||||
if record.levelno >= self.level:
|
||||
self.stream.write(self.format(record) + self.terminator)
|
||||
|
||||
|
||||
class FileHandler(StreamHandler):
|
||||
def __init__(self, filename, mode="a", encoding="UTF-8"):
|
||||
super().__init__(stream=open(filename, mode=mode, encoding=encoding))
|
||||
|
||||
def close(self):
|
||||
super().close()
|
||||
self.stream.close()
|
||||
|
||||
|
||||
class Formatter:
|
||||
def __init__(self, fmt=None, datefmt=None):
|
||||
self.fmt = _default_fmt if fmt is None else fmt
|
||||
self.datefmt = _default_datefmt if datefmt is None else datefmt
|
||||
|
||||
def usesTime(self):
|
||||
return "asctime" in self.fmt
|
||||
|
||||
def formatTime(self, datefmt, record):
|
||||
if hasattr(time, "strftime"):
|
||||
return time.strftime(datefmt, time.localtime(record.ct))
|
||||
return None
|
||||
|
||||
def format(self, record):
|
||||
if self.usesTime():
|
||||
record.asctime = self.formatTime(self.datefmt, record)
|
||||
return self.fmt % {
|
||||
"name": record.name,
|
||||
"message": record.message,
|
||||
"msecs": record.msecs,
|
||||
"asctime": record.asctime,
|
||||
"levelname": record.levelname,
|
||||
}
|
||||
|
||||
|
||||
class Logger:
|
||||
def __init__(self, name, level=NOTSET):
|
||||
self.name = name
|
||||
self.level = level
|
||||
self.handlers = []
|
||||
self.record = LogRecord()
|
||||
|
||||
def setLevel(self, level):
|
||||
self.level = level
|
||||
|
||||
def isEnabledFor(self, level):
|
||||
return level >= self.getEffectiveLevel()
|
||||
|
||||
def getEffectiveLevel(self):
|
||||
return self.level or getLogger().level or _DEFAULT_LEVEL
|
||||
|
||||
def log(self, level, msg, *args):
|
||||
if self.isEnabledFor(level):
|
||||
if args:
|
||||
if isinstance(args[0], dict):
|
||||
args = args[0]
|
||||
msg = msg % args
|
||||
self.record.set(self.name, level, msg)
|
||||
handlers = self.handlers
|
||||
if not handlers:
|
||||
handlers = getLogger().handlers
|
||||
for h in handlers:
|
||||
h.emit(self.record)
|
||||
|
||||
def debug(self, msg, *args):
|
||||
self.log(DEBUG, msg, *args)
|
||||
|
||||
def info(self, msg, *args):
|
||||
self.log(INFO, msg, *args)
|
||||
|
||||
def warning(self, msg, *args):
|
||||
self.log(WARNING, msg, *args)
|
||||
|
||||
def error(self, msg, *args):
|
||||
self.log(ERROR, msg, *args)
|
||||
|
||||
def critical(self, msg, *args):
|
||||
self.log(CRITICAL, msg, *args)
|
||||
|
||||
def exception(self, msg, *args, exc_info=True):
|
||||
self.log(ERROR, msg, *args)
|
||||
tb = None
|
||||
if isinstance(exc_info, BaseException):
|
||||
tb = exc_info
|
||||
elif hasattr(sys, "exc_info"):
|
||||
tb = sys.exc_info()[1]
|
||||
if tb:
|
||||
buf = io.StringIO()
|
||||
sys.print_exception(tb, buf)
|
||||
self.log(ERROR, buf.getvalue())
|
||||
|
||||
def addHandler(self, handler):
|
||||
self.handlers.append(handler)
|
||||
|
||||
def hasHandlers(self):
|
||||
return len(self.handlers) > 0
|
||||
|
||||
|
||||
def getLogger(name=None):
|
||||
if name is None:
|
||||
name = "root"
|
||||
if name not in _loggers:
|
||||
_loggers[name] = Logger(name)
|
||||
if name == "root":
|
||||
basicConfig()
|
||||
return _loggers[name]
|
||||
|
||||
|
||||
def log(level, msg, *args):
|
||||
getLogger().log(level, msg, *args)
|
||||
|
||||
|
||||
def debug(msg, *args):
|
||||
getLogger().debug(msg, *args)
|
||||
|
||||
|
||||
def info(msg, *args):
|
||||
getLogger().info(msg, *args)
|
||||
|
||||
|
||||
def warning(msg, *args):
|
||||
getLogger().warning(msg, *args)
|
||||
|
||||
|
||||
def error(msg, *args):
|
||||
getLogger().error(msg, *args)
|
||||
|
||||
|
||||
def critical(msg, *args):
|
||||
getLogger().critical(msg, *args)
|
||||
|
||||
|
||||
def exception(msg, *args, exc_info=True):
|
||||
getLogger().exception(msg, *args, exc_info=exc_info)
|
||||
|
||||
|
||||
def shutdown():
|
||||
for k, logger in _loggers.items():
|
||||
for h in logger.handlers:
|
||||
h.close()
|
||||
_loggers.pop(logger, None)
|
||||
|
||||
|
||||
def addLevelName(level, name):
|
||||
_level_dict[level] = name
|
||||
|
||||
|
||||
def basicConfig(
|
||||
filename=None,
|
||||
filemode="a",
|
||||
format=None,
|
||||
datefmt=None,
|
||||
level=WARNING,
|
||||
stream=None,
|
||||
encoding="UTF-8",
|
||||
force=False,
|
||||
):
|
||||
if "root" not in _loggers:
|
||||
_loggers["root"] = Logger("root")
|
||||
|
||||
logger = _loggers["root"]
|
||||
|
||||
if force or not logger.handlers:
|
||||
for h in logger.handlers:
|
||||
h.close()
|
||||
logger.handlers = []
|
||||
|
||||
if filename is None:
|
||||
handler = StreamHandler(stream)
|
||||
else:
|
||||
handler = FileHandler(filename, filemode, encoding)
|
||||
|
||||
# Fix from https://github.com/micropython/micropython-lib/pull/1077 is on the line below:
|
||||
handler.setLevel(NOTSET)
|
||||
handler.setFormatter(Formatter(format, datefmt))
|
||||
|
||||
logger.setLevel(level)
|
||||
logger.addHandler(handler)
|
||||
|
||||
|
||||
if hasattr(sys, "atexit"):
|
||||
sys.atexit(shutdown)
|
||||
@@ -252,7 +252,8 @@ class AppManager:
|
||||
print(f"execute_script: reading script_source took {read_time}ms")
|
||||
script_globals = {
|
||||
'lv': lv,
|
||||
'__name__': "__main__"
|
||||
'__name__': "__main__", # in case the script wants this
|
||||
'__file__': compile_name
|
||||
}
|
||||
print(f"Thread {thread_id}: starting script")
|
||||
import sys
|
||||
|
||||
@@ -1,4 +0,0 @@
|
||||
freeze('/tmp/', 'boot.py') # Hardware initialization - this file is copied from boot_fri3d-2024.py to /tmp by the build script to have it named boot.py
|
||||
freeze('../internal_filesystem/', 'main.py') # User Interface initialization
|
||||
freeze('../internal_filesystem/lib', '') # Additional libraries
|
||||
freeze('../freezeFS/', 'freezefs_mount_builtin.py') # Built-in apps
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Tests for the logging module to ensure logger and handler level filtering works correctly."""
|
||||
|
||||
import unittest
|
||||
import sys
|
||||
import io
|
||||
import logging
|
||||
|
||||
# Add lib to path so we can import logging
|
||||
sys.path.insert(0, 'MicroPythonOS/internal_filesystem/lib')
|
||||
|
||||
|
||||
class TestLoggingLevels(unittest.TestCase):
|
||||
"""Test that logger levels work correctly with handlers."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
# Clear any existing loggers
|
||||
logging._loggers.clear()
|
||||
# Reset basicConfig
|
||||
logging.basicConfig(force=True)
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after tests."""
|
||||
logging.shutdown()
|
||||
logging._loggers.clear()
|
||||
|
||||
def test_child_logger_info_level_with_root_handlers(self):
|
||||
"""Test that a child logger can set INFO level and log INFO messages using root handlers."""
|
||||
# Capture output
|
||||
stream = io.StringIO()
|
||||
logging.basicConfig(stream=stream, level=logging.WARNING, force=True)
|
||||
|
||||
# Create child logger and set to INFO
|
||||
logger = logging.getLogger("test_child")
|
||||
logger.setLevel(logging.INFO)
|
||||
|
||||
# Log at different levels
|
||||
logger.debug("debug message")
|
||||
logger.info("info message")
|
||||
logger.warning("warning message")
|
||||
|
||||
output = stream.getvalue()
|
||||
|
||||
# Should NOT have debug (below INFO)
|
||||
self.assertTrue("debug message" not in output)
|
||||
# Should have info (at INFO level)
|
||||
self.assertTrue("info message" in output)
|
||||
# Should have warning (above INFO)
|
||||
self.assertTrue("warning message" in output)
|
||||
|
||||
def test_root_logger_warning_level(self):
|
||||
"""Test that root logger at WARNING level filters correctly."""
|
||||
stream = io.StringIO()
|
||||
logging.basicConfig(stream=stream, level=logging.WARNING, force=True)
|
||||
|
||||
logger = logging.getLogger()
|
||||
|
||||
logger.debug("debug message")
|
||||
logger.info("info message")
|
||||
logger.warning("warning message")
|
||||
|
||||
output = stream.getvalue()
|
||||
|
||||
# Should NOT have debug or info
|
||||
self.assertTrue("debug message" not in output)
|
||||
self.assertTrue("info message" not in output)
|
||||
# Should have warning
|
||||
self.assertTrue("warning message" in output)
|
||||
|
||||
def test_child_logger_debug_level(self):
|
||||
"""Test that a child logger can set DEBUG level."""
|
||||
stream = io.StringIO()
|
||||
logging.basicConfig(stream=stream, level=logging.WARNING, force=True)
|
||||
|
||||
logger = logging.getLogger("test_debug")
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
logger.debug("debug message")
|
||||
logger.info("info message")
|
||||
logger.warning("warning message")
|
||||
|
||||
output = stream.getvalue()
|
||||
|
||||
# Should have all messages
|
||||
self.assertIn("debug message", output)
|
||||
self.assertIn("info message", output)
|
||||
self.assertIn("warning message", output)
|
||||
|
||||
def test_multiple_child_loggers_different_levels(self):
|
||||
"""Test that multiple child loggers can have different levels."""
|
||||
stream = io.StringIO()
|
||||
logging.basicConfig(stream=stream, level=logging.WARNING, force=True)
|
||||
|
||||
logger1 = logging.getLogger("app1")
|
||||
logger1.setLevel(logging.DEBUG)
|
||||
|
||||
logger2 = logging.getLogger("app2")
|
||||
logger2.setLevel(logging.ERROR)
|
||||
|
||||
logger1.debug("app1 debug")
|
||||
logger1.info("app1 info")
|
||||
logger2.debug("app2 debug")
|
||||
logger2.info("app2 info")
|
||||
logger2.error("app2 error")
|
||||
|
||||
output = stream.getvalue()
|
||||
|
||||
# app1 should log debug and info
|
||||
self.assertTrue("app1 debug" in output)
|
||||
self.assertTrue("app1 info" in output)
|
||||
# app2 should NOT log debug or info
|
||||
self.assertTrue("app2 debug" not in output)
|
||||
self.assertTrue("app2 info" not in output)
|
||||
# app2 should log error
|
||||
self.assertTrue("app2 error" in output)
|
||||
|
||||
def test_handler_level_does_not_filter(self):
|
||||
"""Test that handler level is NOTSET and doesn't filter messages."""
|
||||
stream = io.StringIO()
|
||||
logging.basicConfig(stream=stream, level=logging.INFO, force=True)
|
||||
|
||||
# Get the root logger and check handler level
|
||||
root_logger = logging.getLogger()
|
||||
self.assertEqual(len(root_logger.handlers), 1)
|
||||
handler = root_logger.handlers[0]
|
||||
|
||||
# Handler level should be NOTSET (0) so it doesn't filter
|
||||
self.assertEqual(handler.level, logging.NOTSET)
|
||||
|
||||
def test_child_logger_notset_level_uses_root_level(self):
|
||||
"""Test that a child logger with NOTSET level uses root logger's level."""
|
||||
stream = io.StringIO()
|
||||
logging.basicConfig(stream=stream, level=logging.WARNING, force=True)
|
||||
|
||||
logger = logging.getLogger("test_notset")
|
||||
# Don't set logger level, it should default to NOTSET
|
||||
|
||||
logger.debug("debug message")
|
||||
logger.info("info message")
|
||||
logger.warning("warning message")
|
||||
|
||||
output = stream.getvalue()
|
||||
|
||||
# Should use root logger's WARNING level
|
||||
self.assertTrue("debug message" not in output)
|
||||
self.assertTrue("info message" not in output)
|
||||
self.assertTrue("warning message" in output)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user