Merge branch 'master' into lekcyjna/transactron-multi-method-request

constants/ecp5_platforms.py

@@ -55,7 +55,7 @@ def __init__(self, pins: Iterable[str]):
     def p(self, count: int = 1):
         return " ".join([self.pin_bag.pop() for _ in range(count)])
 
-    def named_pin(self, names: list[str]):
+    def named_pin(self, names: Iterable[str]):
         for name in names:
             if name in self.pin_bag:
                 self.pin_bag.remove(name)

requirements-dev.txt

@@ -9,7 +9,7 @@ myst-parser==0.18.0
 numpydoc==1.5.0
 parameterized==0.8.1
 pre-commit==2.16.0
-pyright==1.1.308
+pyright==1.1.332
 Sphinx==5.1.1
 sphinx-rtd-theme==1.0.0
 sphinxcontrib-mermaid==0.8.1

test/common.py

@@ -12,7 +12,7 @@
 
 from transactron.core import SignalBundle, Method, TransactionModule
 from transactron.lib import AdapterBase, AdapterTrans
-from transactron._utils import method_def_helper
+from transactron._utils import mock_def_helper
 from coreblocks.utils import ValueLike, HasElaborate, HasDebugSignals, auto_debug_signals, LayoutLike, ModuleConnector
 from .gtkw_extension import write_vcd_ext
 
@@ -363,7 +363,7 @@ def method_handle(
             for _ in range(extra_settle_count + 1):
                 yield Settle()
 
-        ret_out = method_def_helper(self, function, **arg)
+        ret_out = mock_def_helper(self, function, arg)
         yield from self.method_return(ret_out or {})
         yield
 

test/regression/cocotb.py

@@ -73,7 +73,7 @@ async def start(self):
 
         while True:
             while not (self.bus.stb.value and self.bus.cyc.value):
-                await clock_edge_event
+                await clock_edge_event  # type: ignore
 
             sig_m = WishboneMasterSignals()
             self.bus.sample(sig_m)
@@ -124,10 +124,10 @@ async def start(self):
             )
 
             for _ in range(self.delay):
-                await clock_edge_event
+                await clock_edge_event  # type: ignore
 
             self.bus.drive(sig_s)
-            await clock_edge_event
+            await clock_edge_event  # type: ignore
             self.bus.drive(WishboneSlaveSignals())
 
 

test/test_core.py

@@ -263,7 +263,7 @@ def test_asm_source(self):
         self.base_dir = "test/asm/"
         self.bin_src = []
 
-        with tempfile.NamedTemporaryFile() as asm_tmp:
+        with tempfile.NamedTemporaryFile() as asm_tmp, tempfile.NamedTemporaryFile() as bin_tmp:
             subprocess.check_call(
                 [
                     "riscv64-unknown-elf-as",
@@ -276,9 +276,10 @@ def test_asm_source(self):
                     self.base_dir + self.source_file,
                 ]
             )
-            code = subprocess.check_output(
-                ["riscv64-unknown-elf-objcopy", "-O", "binary", "-j", ".text", asm_tmp.name, "/dev/stdout"]
+            subprocess.check_call(
+                ["riscv64-unknown-elf-objcopy", "-O", "binary", "-j", ".text", asm_tmp.name, bin_tmp.name]
             )
+            code = bin_tmp.read()
             for word_idx in range(0, len(code), 4):
                 word = code[word_idx : word_idx + 4]
                 bin_instr = int.from_bytes(word, "little")

transactron/_utils.py

@@ -1,7 +1,7 @@
 import itertools
 import sys
 from inspect import Parameter, signature
-from typing import Optional, TypeAlias, TypeVar
+from typing import Any, Concatenate, Optional, TypeAlias, TypeGuard, TypeVar
 from collections.abc import Callable, Iterable, Mapping
 from amaranth import *
 from coreblocks.utils._typing import LayoutLike
@@ -15,11 +15,13 @@
     "Graph",
     "GraphCC",
     "get_caller_class_name",
+    "def_helper",
     "method_def_helper",
 ]
 
 
 T = TypeVar("T")
+U = TypeVar("U")
 
 
 class Scheduler(Elaboratable):
@@ -122,24 +124,35 @@ def _graph_ccs(gr: ROGraph[T]) -> list[GraphCC[T]]:
 MethodLayout: TypeAlias = LayoutLike
 
 
-def method_def_helper(method, func: Callable[..., T], arg=None, /, **kwargs) -> T:
+def has_first_param(func: Callable[..., T], name: str, tp: type[U]) -> TypeGuard[Callable[Concatenate[U, ...], T]]:
+    parameters = signature(func).parameters
+    return (
+        len(parameters) >= 1
+        and next(iter(parameters)) == name
+        and parameters[name].kind in {Parameter.POSITIONAL_OR_KEYWORD, Parameter.POSITIONAL_ONLY}
+        and parameters[name].annotation in {Parameter.empty, tp}
+    )
+
+
+def def_helper(description, func: Callable[..., T], tp: type[U], arg: U, /, **kwargs) -> T:
     parameters = signature(func).parameters
     kw_parameters = set(
         n for n, p in parameters.items() if p.kind in {Parameter.POSITIONAL_OR_KEYWORD, Parameter.KEYWORD_ONLY}
     )
-    if (
-        len(parameters) == 1
-        and "arg" in parameters
-        and parameters["arg"].kind in {Parameter.POSITIONAL_OR_KEYWORD, Parameter.POSITIONAL_ONLY}
-        and parameters["arg"].annotation in {Parameter.empty, Record}
-    ):
-        if arg is None:
-            arg = kwargs
+    if len(parameters) == 1 and has_first_param(func, "arg", tp):
         return func(arg)
     elif kw_parameters <= kwargs.keys():
         return func(**kwargs)
     else:
-        raise TypeError(f"Invalid method definition/mock for {method}: {func}")
+        raise TypeError(f"Invalid {description}: {func}")
+
+
+def mock_def_helper(tb, func: Callable[..., T], arg: Mapping[str, Any]) -> T:
+    return def_helper(f"mock definition for {tb}", func, Mapping[str, Any], arg, **arg)
+
+
+def method_def_helper(method, func: Callable[..., T], arg: Record) -> T:
+    return def_helper(f"method definition for {method}", func, Record, arg, **arg.fields)
 
 
 def get_caller_class_name(default: Optional[str] = None) -> tuple[Optional[Elaboratable], str]:

transactron/core.py

@@ -1113,7 +1113,7 @@ def body(
 
     def _validate_arguments(self, arg_rec: Record) -> ValueLike:
         if self.validate_arguments is not None:
-            return self.ready & method_def_helper(self, self.validate_arguments, arg_rec, **arg_rec.fields)
+            return self.ready & method_def_helper(self, self.validate_arguments, arg_rec)
         return self.ready
 
     def __call__(
@@ -1259,7 +1259,7 @@ def decorator(func: Callable[..., Optional[RecordDict]]):
         ret_out = None
 
         with method.body(m, ready=ready, out=out, validate_arguments=validate_arguments) as arg:
-            ret_out = method_def_helper(method, func, arg, **arg.fields)
+            ret_out = method_def_helper(method, func, arg)
 
         if ret_out is not None:
             m.d.top_comb += assign(out, ret_out, fields=AssignType.ALL)