diff --git a/src/yosys_mau/task_loop/logging.py b/src/yosys_mau/task_loop/logging.py index fbbb6ee..1877e3e 100644 --- a/src/yosys_mau/task_loop/logging.py +++ b/src/yosys_mau/task_loop/logging.py @@ -17,7 +17,10 @@ current_task_or_none, root_task, ) -from .context import task_context +from .context import ( + TaskContextDict, + task_context, +) Level = Literal["debug", "info", "warning", "error"] @@ -120,14 +123,12 @@ class LogContext: level: Level = "info" """The minimum log level to display/log. + + Can be overridden for named destinations with `destination_levels`. This does not stop `LogEvent` of smaller levels to be emitted. It is only used to filter which messages to actually print/log. Hence, it does not affect any user installed `LogEvent` - handlers. - - When logging to multiple destinations, currently there is no way to specify this per - destination. - """ + handlers.""" log_format: Callable[[LogEvent], str] = default_formatter """The formatter used to format log messages. @@ -145,6 +146,14 @@ class LogContext: Like `log_format` this is looked up by the log writing task, not the emitting task. """ + destination_levels: TaskContextDict[str, Level] = TaskContextDict() + """The minimum log level to display/log for named destinations. + + Like `log_format` this is looked up by the log writing task, not the emitting task. If the + current destination has no key:value pair in this dictionary, the `level` will be looked up by + the task which emit the log. + """ + def log(*args: Any, level: Level = "info", cls: type[LogEvent] = LogEvent) -> LogEvent: """Produce log output. @@ -297,7 +306,10 @@ def log_exception(exception: BaseException, raise_error: bool = True) -> LoggedE def start_logging( - file: IO[Any] | None = None, err: bool = False, color: bool | None = None + file: IO[Any] | None = None, + err: bool = False, + color: bool | None = None, + destination_label: str | None = None, ) -> None: """Start logging all log events reaching the current task. @@ -310,6 +322,8 @@ def start_logging( :param color: Whether to use colors. Defaults to ``True`` for terminals and ``False`` otherwise. When the ``NO_COLOR`` environment variable is set, this will be ignored and no colors will be used. + :param destination_label: Used to look up destination specific log level filtering. + Used with `LogContext.destination_levels`. """ if _no_color: color = False @@ -318,7 +332,14 @@ def log_handler(event: LogEvent): if file and file.closed: remove_log_handler() return - source_level = _level_order[event.source[LogContext].level] + emitter_default = event.source[LogContext].level + if destination_label: + destination_level = LogContext.destination_levels.get( + destination_label, emitter_default + ) + else: + destination_level = emitter_default + source_level = _level_order[destination_level] event_level = _level_order[event.level] if event_level < source_level: return diff --git a/tests/task_loop/test_logging.py b/tests/task_loop/test_logging.py index 5e7186e..2ee19f6 100644 --- a/tests/task_loop/test_logging.py +++ b/tests/task_loop/test_logging.py @@ -140,6 +140,158 @@ def main(): ] +@pytest.mark.parametrize( + "label,expected", + [ + ("default", [2, 3, 4, 6]), + ("debug", [1, 2, 3, 4, 5, 6]), + ("info", [2, 4, 5, 6]), + ("warning", [5, 6]), + ("error", [6]), + ("varied", [1, 2, 4, 6]), + ], +) +def test_log_destinations(label: str, expected: list[str]): + log_output = io.StringIO() + + def main(): + tl.LogContext.time_format = fixed_time + tl.logging.start_logging(file=log_output, destination_label=label) + + # tl.LogContext.level = "info" # implied + tl.LogContext.destination_levels["info"] = "info" + tl.LogContext.destination_levels["debug"] = "debug" + tl.LogContext.destination_levels["warning"] = "warning" + tl.LogContext.destination_levels["error"] = "error" + + tl.LogContext.destination_levels["varied"] = "debug" + tl.log_debug("line 1") + tl.log("line 2") + + tl.LogContext.level = "debug" + tl.LogContext.destination_levels["varied"] = "warning" + tl.log_debug("line 3") + + del tl.LogContext.destination_levels["varied"] + tl.LogContext.destination_levels[""] = "warning" + tl.log("line 4") + + tl.LogContext.level = "error" + tl.log_warning("line 5") + tl.log_error("line 6", raise_error=False) + + tl.run_task_loop(main) + + trimmed_output = [int(x[-1]) for x in log_output.getvalue().splitlines()] + assert trimmed_output == expected + + +@pytest.mark.parametrize("task", ["root", "task1", "task2"]) +@pytest.mark.parametrize("label", ["debug", "info", "warning", "mixed1", "mixed2"]) +def test_nested_destinations(task: str, label: str): + log_output = io.StringIO() + + async def main(): + tl.LogContext.time_format = fixed_time + tl.LogContext.scope = "?root?" + if task == "root": + tl.logging.start_logging(file=log_output, destination_label=label) + tl.LogContext.destination_levels["mixed1"] = "warning" + + tl.LogContext.destination_levels["debug"] = "debug" + tl.LogContext.destination_levels["info"] = "info" + tl.LogContext.destination_levels["warning"] = "warning" + tl.LogContext.destination_levels["error"] = "error" + tl.LogContext.destination_levels["source"] = "warning" + + tl.log("line 0") + sync_event = asyncio.Event() + + async def run_task1(): + tl.LogContext.scope = "?root?task1?" + if task == "task1": + tl.logging.start_logging(file=log_output, destination_label=label) + tl.LogContext.destination_levels["mixed1"] = "info" + + tl.LogContext.destination_levels["mixed2"] = "debug" if task == "root" else "info" + + task2 = tl.Task(on_run=run_task2) + tl.log("line 2") + + await task2.started + + tl.log_debug("line 4") + + sync_event.set() + + await task2.finished + + tl.log("line 6") + + async def run_task2(): + tl.LogContext.scope = "?root?task1?task2?" + if task == "task2": + tl.logging.start_logging(file=log_output, destination_label=label) + tl.LogContext.destination_levels["mixed1"] = "debug" + + tl.LogContext.destination_levels["mixed2"] = "debug" if task == "task1" else "error" + + tl.log_debug("line 3") + + await sync_event.wait() + tl.log_warning("line 5") + + task1 = tl.Task(on_run=run_task1) + + tl.log("line 1") + + await task1.finished + + tl.log("line 7") + + tl.run_task_loop(main) + + reference_list = [ + "12:34:56 ?root?: line 0", + "12:34:56 ?root?: line 1", + "12:34:56 ?root?task1?: line 2", + "12:34:56 ?root?task1?task2?: DEBUG: line 3", + "12:34:56 ?root?task1?: DEBUG: line 4", + "12:34:56 ?root?task1?task2?: WARNING: line 5", + "12:34:56 ?root?task1?: line 6", + "12:34:56 ?root?: line 7", + ] + + label_map: dict[str, list[int]] = { + "debug": [0, 1, 2, 3, 4, 5, 6, 7], + "info": [0, 1, 2, 5, 6, 7], + "warning": [5], + } + + if label in label_map: + filtered_list = [x for i, x in enumerate(reference_list) if i in label_map[label]] + expected = [x for x in filtered_list if task in x.split("?")] + else: + if label == "mixed1": + task_map: dict[str, list[int]] = { + "root": [5], + "task1": [2, 5, 6], + "task2": [3, 5], + } + elif label == "mixed2": + task_map: dict[str, list[int]] = { + "root": [0, 1, 2, 5, 6, 7], + "task1": [2, 5, 6], + "task2": [], + } + else: + assert False, f"unknown label {label}" + expected = [x for i, x in enumerate(reference_list) if i in task_map[task]] + + print(log_output.getvalue()) + assert log_output.getvalue().splitlines() == expected + + def test_exception_logging(): log_output = io.StringIO()