ready to review

test/transactron/testing/test_log.py

@@ -17,9 +17,9 @@ def elaborate(self, platform):
         m = TModule()
 
         with m.If(self.input == 42):
-            log.warn(m, True, "Log triggered under Amaranth If value+3=0x{:x}", self.input + 3)
+            log.warning(m, True, "Log triggered under Amaranth If value+3=0x{:x}", self.input + 3)
 
-        log.warn(m, self.input[0] == 0, "Input is even! input={}, counter={}", self.input, self.counter)
+        log.warning(m, self.input[0] == 0, "Input is even! input={}, counter={}", self.input, self.counter)
 
         m.d.sync += self.counter.eq(self.counter + 1)
 
@@ -47,6 +47,21 @@ def elaborate(self, platform):
         return m
 
 
+class AssertionTest(Elaboratable):
+    def __init__(self):
+        self.input = Signal()
+        self.output = Signal()
+
+    def elaborate(self, platform):
+        m = TModule()
+
+        m.d.comb += self.output.eq(self.input & ~self.input)
+
+        log.assertion(m, self.input == self.output, "Output differs")
+
+        return m
+
+
 class TestLog(TestCaseWithSimulator):
     @patch("sys.stdout", new_callable=StringIO)
     def test_log(self, stdout):
@@ -78,3 +93,18 @@ def proc():
 
         extected_out = "Input is different than output! input=0x1 output=0x0"
         self.assertIn(extected_out, stdout.getvalue())
+
+    @patch("sys.stdout", new_callable=StringIO)
+    def test_assertion(self, stdout):
+        m = AssertionTest()
+
+        def proc():
+            yield
+            yield m.input.eq(1)
+
+        with self.assertRaises(AssertionError):
+            with self.run_simulation(m) as sim:
+                sim.add_sync_process(proc)
+
+        extected_out = "Output differs"
+        self.assertIn(extected_out, stdout.getvalue())

transactron/lib/log.py

@@ -1,8 +1,9 @@
+import os
 import operator
 from functools import reduce
 from dataclasses import dataclass, field
 from dataclasses_json import dataclass_json
-from typing import TypeAlias, Optional
+from typing import TypeAlias
 
 from amaranth import *
 from amaranth.tracer import get_src_loc
@@ -27,6 +28,13 @@
 
 
 def parse_level(str: str) -> LogLevel:
+    """Parse the log level from a string.
+
+    The level can be either a non-negative integer or a string representation
+    of one of the predefined levels.
+
+    Raises an exception if the level cannot be parsed.
+    """
     for level, name in _level_to_name.items():
         if str.upper() == name:
             return level
@@ -40,33 +48,55 @@ def parse_level(str: str) -> LogLevel:
     raise ValueError("Log level must be either {error, warn, info, debug} or a non-negative integer.")
 
 
-def get_level_by_name(level_str: str) -> Optional[LogLevel]:
-    for level, name in _level_to_name.items():
-        if level_str == name:
-            return level
-
-    return None
-
-
 def get_level_name(level: LogLevel) -> str:
+    """Return the textual representation of logging level 'level'.
+
+    If the level is not one of the predefined levels, string 'Level {level}'
+    is returned.
+    """
     if level in _level_to_name:
         return _level_to_name[level]
-    return "Level {}".format(level)
+    return f"Level {level}"
 
 
 @dataclass_json
 @dataclass(frozen=True)
 class LogRecordInfo:
+    """Simulator-backend-agnostic information about a log record that can
+    be serialized and used outside the Amaranth context.
+
+    Attributes
+    ----------
+    level : LogLevel
+        The severity level of the log.
+    format_str : str
+        The template of the message. Should follow PEP 3101 standard.
+    location : SrcLoc
+        Source location of the log.
+    """
+
     level: LogLevel
     format_str: str
     location: SrcLoc
 
     def format(self, *args) -> str:
+        """Format the log message with a set of concrete arguments."""
+
         return self.format_str.format(*args)
 
 
 @dataclass(frozen=True)
 class LogRecord(LogRecordInfo):
+    """A LogRecord instance represents an event being logged.
+
+    Attributes
+    ----------
+    trigger : Signal
+        Amaranth signal triggering the log.
+    fields : Signal
+        Amaranth signals that will be used to format the message.
+    """
+
     trigger: Signal
     fields: list[Signal] = field(default_factory=list)
 
@@ -77,17 +107,32 @@ class LogKey(ListKey[LogRecord]):
 
 
 def log(m: ModuleLike, level: LogLevel, format: str, trigger: ValueLike, *args, src_loc_at: int = 0):
-    """
+    """Registers a hardware log record with the given severity.
+
+    Hardware logs are evaluated and printed during simulation, so both
+    the trigger and the format fields are Amaranth values, i.e.
+    signals or arbitrary Amaranth expressions.
 
     Parameters
     ----------
+    m : ModuleLike
+        The module for which the log record is added.
+    trigger : ValueLike
+        If the value of this Amaranth expression is true, the log will reported.
+    format : str
+        The format of the message as defined in PEP 3101.
+    *args
+        Amaranth values that will be read during simulation and used to format
+        the message.
     src_loc_at : int, optional
         How many stack frames below to look for the source location, used to
         identify the failing assertion.
     """
 
-    # TODO: make the location relative to the root of the project
-    src_loc = get_src_loc(src_loc_at + 1)
+    def local_src_loc(src_loc: SrcLoc):
+        return (os.path.relpath(src_loc[0]), src_loc[1])
+
+    src_loc = local_src_loc(get_src_loc(src_loc_at + 1))
 
     trigger_signal = Signal()
     m.d.comb += trigger_signal.eq(trigger)
@@ -104,26 +149,82 @@ def log(m: ModuleLike, level: LogLevel, format: str, trigger: ValueLike, *args,
 
 
 def debug(m: ModuleLike, trigger: ValueLike, format: str, *args, **kwargs):
+    """Log a message with severity 'DEBUG'.
+
+    See `log.log` function for more details.
+    """
     log(m, DEBUG, format, trigger, *args, **kwargs)
 
 
 def info(m: ModuleLike, trigger: ValueLike, format: str, *args, **kwargs):
+    """Log a message with severity 'INFO'.
+
+    See `log.log` function for more details.
+    """
     log(m, INFO, format, trigger, *args, **kwargs)
 
 
-def warn(m: ModuleLike, trigger: ValueLike, format: str, *args, **kwargs):
+def warning(m: ModuleLike, trigger: ValueLike, format: str, *args, **kwargs):
+    """Log a message with severity 'WARNING'.
+
+    See `log.log` function for more details.
+    """
     log(m, WARNING, format, trigger, *args, **kwargs)
 
 
 def error(m: ModuleLike, trigger: ValueLike, format: str, *args, **kwargs):
+    """Log a message with severity 'ERROR'.
+
+    This severity level has special semantics. If a log with this serverity
+    level is triggered, the simulation will be terminated.
+
+    See `log.log` function for more details.
+    """
     log(m, ERROR, format, trigger, *args, **kwargs)
 
 
-def get_logs(level: LogLevel) -> list[LogRecord]:
+def assertion(m: ModuleLike, value: Value, format: str = "", *args, **kwargs):
+    """Define an assertion.
+
+    This function might help find some hardware bugs which might otherwise be
+    hard to detect. If `value` is false, it will terminate the simulation or
+    it can also be used to turn on a warning LED on a board.
+
+    Internally, this is a convenience wrapper over log.error.
+
+    See `log.log` function for more details.
+    """
+    error(m, ~value, format, *args, **kwargs)
+
+
+def get_log_records(level: LogLevel) -> list[LogRecord]:
+    """Get log records for the given severity level.
+
+    This function returns all log records with the severity bigger or equal
+    to the specified level.
+
+    Parameters
+    ----------
+    level : LogLevel
+        The minimum severity level.
+    """
+
     dependencies = DependencyContext.get()
     all_logs = dependencies.get_dependency(LogKey())
     return [rec for rec in all_logs if rec.level >= level]
 
 
 def get_trigger_bit(level: LogLevel) -> Value:
-    return reduce(operator.or_, [rec.trigger for rec in get_logs(level)], C(0))
+    """Get a trigger bit for the given severity level.
+
+    The signal returned by this function is high whenever the trigger signal
+    of any of the records with the severity bigger or equal to the specified
+    level is high.
+
+    Parameters
+    ----------
+    level : LogLevel
+        The minimum severity level.
+    """
+
+    return reduce(operator.or_, [rec.trigger for rec in get_log_records(level)], C(0))

transactron/testing/log.py

@@ -16,7 +16,7 @@ def handle_logs():
         if not (yield combined_trigger):
             return
 
-        for record in log.get_logs(level):
+        for record in log.get_log_records(level):
             if not (yield record.trigger):
                 continue