diff --git a/.copier-answers.yaml b/.copier-answers.yaml index 1256427..ce7a3b5 100644 --- a/.copier-answers.yaml +++ b/.copier-answers.yaml @@ -1,5 +1,5 @@ # Changes here will be overwritten by Copier -_commit: 5ff720b +_commit: '4372075' _src_path: https://github.com/python-project-templates/base.git add_docs: true add_extension: python diff --git a/Makefile b/Makefile index f6b0a6a..0d476e1 100644 --- a/Makefile +++ b/Makefile @@ -25,16 +25,16 @@ lint-py: ## lint python with ruff python -m ruff format --check dau_sim lint-docs: ## lint docs with mdformat and codespell - python -m mdformat --check README.md docs/src - python -m codespell_lib README.md docs/src + python -m mdformat --check README.md $(wildcard docs/src) + python -m codespell_lib README.md $(wildcard docs/src) fix-py: ## autoformat python code with ruff python -m ruff check --fix dau_sim python -m ruff format dau_sim fix-docs: ## autoformat docs with mdformat and codespell - python -m mdformat README.md docs/src - python -m codespell_lib --write README.md docs/src + python -m mdformat README.md $(wildcard docs/src) + python -m codespell_lib --write README.md $(wildcard docs/src) lint: lint-py lint-docs ## run all linters lints: lint @@ -50,9 +50,9 @@ check-dist: ## check python sdist and wheel with check-dist check-dist -v check-types: ## check python types with ty - ty check --python $$(which python) + python -m ty check -checks: check-dist +checks: check-dist check-types # Alias check: checks diff --git a/dau_sim/api.py b/dau_sim/api.py index 438888a..33d6523 100644 --- a/dau_sim/api.py +++ b/dau_sim/api.py @@ -1,4 +1,4 @@ -from datetime import timedelta +from datetime import datetime, timedelta from ccflow import BaseModel from pydantic import ConfigDict @@ -14,7 +14,7 @@ class SimulationResult(BaseModel): model_config = ConfigDict(frozen=True) module_name: str - traces: tuple[tuple[str, tuple[tuple[object, int], ...]], ...] + traces: tuple[tuple[str, tuple[tuple[datetime, int], ...]], ...] def latest(self, signal: str) -> int | None: """Return the most recent value for a signal, if available.""" @@ -94,4 +94,4 @@ def write_vcd( signals: list[str] | None = None, exclude: list[str] | None = None, ) -> None: - self._compiled.write_vcd(path, dict(result.traces), timescale=timescale, signals=signals, exclude=exclude) + self._compiled.write_vcd(path, {name: list(values) for name, values in result.traces}, timescale=timescale, signals=signals, exclude=exclude) diff --git a/dau_sim/backends/cocotb_backend.py b/dau_sim/backends/cocotb_backend.py index 27541e9..0d556a4 100644 --- a/dau_sim/backends/cocotb_backend.py +++ b/dau_sim/backends/cocotb_backend.py @@ -25,7 +25,7 @@ import time import types from collections.abc import Callable -from typing import Any +from typing import Any, cast from dau_sim.compiler.compile import ( _build_domain_info, @@ -700,9 +700,9 @@ def stop(self) -> None: # Simulator module — drop-in replacement for cocotb.simulator -def _create_simulator_module(engine: SimulationEngine) -> types.ModuleType: +def _create_simulator_module(engine: SimulationEngine) -> Any: """Create a module object that mimics ``cocotb.simulator``.""" - mod = types.ModuleType("cocotb.simulator") + mod: Any = types.ModuleType("cocotb.simulator") mod.__package__ = "cocotb" # Constants @@ -752,7 +752,7 @@ def register_nextstep_callback(func: Callable, *args: Any) -> DauSimCallback: def register_value_change_callback(signal: DauSimHandle, func: Callable, edge: int, *args: Any) -> DauSimCallback: return engine.register_value_change_callback(signal, func, edge, *args) - def get_root_handle(name: str | None = None) -> DauSimHandle: # noqa: ARG001 (cocotb simulator API) + def get_root_handle(name: str | None = None) -> DauSimHandle | None: # noqa: ARG001 (cocotb simulator API) return engine._root_handle def get_sim_time() -> tuple[int, int]: @@ -837,9 +837,9 @@ def run_cocotb( else: # Try Amaranth try: - from dau_sim.frontends.amaranth_frontend import lower_amaranth_design + from dau_sim.frontends.amaranth_frontend import from_amaranth - ir_module = lower_amaranth_design(design) + ir_module = from_amaranth(design) except (ImportError, TypeError): raise TypeError(f"Cannot lower design of type {type(design).__name__}. Pass an Amaranth Elaboratable or a dau-sim IR Module.") @@ -866,6 +866,8 @@ def _run_with_patched_simulator( import cocotb._event_loop import cocotb._gpi_triggers import cocotb.simtime + + cocotb = cast(Any, cocotb) # isort: on # Save originals diff --git a/dau_sim/compiler/codegen.py b/dau_sim/compiler/codegen.py index ebde16f..fbdf82d 100644 --- a/dau_sim/compiler/codegen.py +++ b/dau_sim/compiler/codegen.py @@ -9,6 +9,8 @@ from __future__ import annotations +from collections.abc import Callable + from dau_sim.ir.expr import ( Binary, BinaryOp, @@ -67,7 +69,7 @@ def compile_block( stmts: tuple[Stmt, ...], *, name: str = "_compiled_block", - ) -> callable: + ) -> Callable[..., None]: """Compile a statement block into a callable ``fn(S, changed)``.""" self._counter = 0 lines: list[str] = [] @@ -93,7 +95,7 @@ def compile_expr_block( write_signals: set[str], *, name: str = "_compiled_block", - ) -> callable: + ) -> Callable[..., None]: """Compile a block with local-variable optimization. Generates code that loads all read/written signals into locals @@ -173,12 +175,11 @@ def _compile_stmts(self, stmts: tuple[Stmt, ...], indent: int) -> list[str]: def _compile_stmt(self, stmt: Stmt, indent: int) -> list[str]: pad = " " * indent - st = type(stmt) - if st is Assign: + if type(stmt) is Assign: return self._compile_assign(stmt, indent) - if st is IfElse: + if type(stmt) is IfElse: lines: list[str] = [] cond_expr = self._emit_expr(stmt.cond) lines.append(f"{pad}if {cond_expr}:") @@ -190,28 +191,28 @@ def _compile_stmt(self, stmt: Stmt, indent: int) -> list[str]: lines.extend(else_ if else_ else [f"{pad} pass"]) return lines - if st is Switch: + if type(stmt) is Switch: return self._compile_switch(stmt, indent) - if st is Assert: + if type(stmt) is Assert: cond_expr = self._emit_expr(stmt.cond) msg = repr(stmt.message or "assertion failed") return [f"{pad}if not ({cond_expr}): raise AssertionError({msg})"] - if st is Print: + if type(stmt) is Print: args = ", ".join(self._emit_expr(a) for a in stmt.args) return [f"{pad}print({stmt.format_str!r}.format({args}))"] - if st is Finish: + if type(stmt) is Finish: return [f"{pad}raise _SimulationFinish({stmt.exit_code})"] - if st is Delay: + if type(stmt) is Delay: return [] # no-op in synchronous execution - if st is ReadMem: + if type(stmt) is ReadMem: return [] # handled at init-block level - raise TypeError(f"Unknown statement type: {st.__name__}") + raise TypeError(f"Unknown statement type: {type(stmt).__name__}") def _compile_assign(self, stmt: Assign, indent: int) -> list[str]: pad = " " * indent @@ -260,36 +261,35 @@ def _compile_switch(self, stmt: Switch, indent: int) -> list[str]: def _emit_expr(self, expr: Expr) -> str: """Emit a Python expression string for an IR expression.""" - et = type(expr) - if et is Const: + if type(expr) is Const: return self._emit_const(expr) - if et is SignalRef: + if type(expr) is SignalRef: return self._emit_signal_ref(expr) - if et is Unary: + if type(expr) is Unary: return self._emit_unary(expr) - if et is Binary: + if type(expr) is Binary: return self._emit_binary(expr) - if et is Mux: + if type(expr) is Mux: sel = self._emit_expr(expr.sel) t = self._emit_expr(expr.if_true) f = self._emit_expr(expr.if_false) return self._emit_mask(f"({t} if {sel} else {f})", expr.shape) - if et is Concat: + if type(expr) is Concat: return self._emit_concat(expr) - if et is Slice: + if type(expr) is Slice: return self._emit_slice(expr) - if et is SysRandom: + if type(expr) is SysRandom: return self._emit_sysrandom(expr) - raise TypeError(f"Unknown expression type: {et.__name__}") + raise TypeError(f"Unknown expression type: {type(expr).__name__}") def _emit_const(self, expr: Const) -> str: width = expr.shape.width @@ -465,12 +465,11 @@ def _compile_stmt_local( local_reads: set[str], ) -> list[str]: pad = " " * indent - st = type(stmt) - if st is Assign: + if type(stmt) is Assign: return self._compile_assign_local(stmt, indent, local_reads) - if st is IfElse: + if type(stmt) is IfElse: lines: list[str] = [] cond_expr = self._emit_expr_local(stmt.cond, local_reads) lines.append(f"{pad}if {cond_expr}:") @@ -482,28 +481,28 @@ def _compile_stmt_local( lines.extend(else_ if else_ else [f"{pad} pass"]) return lines - if st is Switch: + if type(stmt) is Switch: return self._compile_switch_local(stmt, indent, local_reads) - if st is Assert: + if type(stmt) is Assert: cond_expr = self._emit_expr_local(stmt.cond, local_reads) msg = repr(stmt.message or "assertion failed") return [f"{pad}if not ({cond_expr}): raise AssertionError({msg})"] - if st is Print: + if type(stmt) is Print: args = ", ".join(self._emit_expr_local(a, local_reads) for a in stmt.args) return [f"{pad}print({stmt.format_str!r}.format({args}))"] - if st is Finish: + if type(stmt) is Finish: return [f"{pad}raise _SimulationFinish({stmt.exit_code})"] - if st is Delay: + if type(stmt) is Delay: return [] - if st is ReadMem: + if type(stmt) is ReadMem: return [] - raise TypeError(f"Unknown statement type: {st.__name__}") + raise TypeError(f"Unknown statement type: {type(stmt).__name__}") def _compile_assign_local( self, @@ -565,12 +564,11 @@ def _compile_switch_local( def _emit_expr_local(self, expr: Expr, local_reads: set[str]) -> str: """Emit expression using local variables for signal reads where possible.""" - et = type(expr) - if et is Const: + if type(expr) is Const: return self._emit_const(expr) - if et is SignalRef: + if type(expr) is SignalRef: if expr.name in local_reads: local = self._local_name(expr.name) width = expr.shape.width @@ -582,31 +580,31 @@ def _emit_expr_local(self, expr: Expr, local_reads: set[str]) -> str: return f"({local} & {mask})" return self._emit_signal_ref(expr) - if et is Unary: + if type(expr) is Unary: a = self._emit_expr_local(expr.operand, local_reads) return self._emit_unary_raw(expr.op, a, expr.operand.shape, expr.shape) - if et is Binary: + if type(expr) is Binary: left = self._emit_expr_local(expr.left, local_reads) right = self._emit_expr_local(expr.right, local_reads) return self._emit_binary_raw(expr.op, left, right, expr.left.shape, expr.right.shape, expr.shape) - if et is Mux: + if type(expr) is Mux: sel = self._emit_expr_local(expr.sel, local_reads) t = self._emit_expr_local(expr.if_true, local_reads) f = self._emit_expr_local(expr.if_false, local_reads) return self._emit_mask(f"({t} if {sel} else {f})", expr.shape) - if et is Concat: + if type(expr) is Concat: return self._emit_concat_local(expr, local_reads) - if et is Slice: + if type(expr) is Slice: return self._emit_slice_local(expr, local_reads) - if et is SysRandom: + if type(expr) is SysRandom: return self._emit_sysrandom(expr) - raise TypeError(f"Unknown expression type: {et.__name__}") + raise TypeError(f"Unknown expression type: {type(expr).__name__}") def _emit_unary_raw(self, op: UnaryOp, a: str, a_shape: Shape, out_shape: Shape) -> str: if op is UnaryOp.NOT: @@ -710,7 +708,7 @@ def build_run_loop( comb_components: tuple, init_values: dict[str, int], traced: tuple[str, ...] = (), - ) -> callable: + ) -> Callable[..., None]: """Generate one function that runs the whole simulation. ``fn(S, clock_arr, tc, n_ticks, ticks, vals)`` advances ``n_ticks`` @@ -871,45 +869,43 @@ def collect_reads_writes(stmts: tuple[Stmt, ...]) -> tuple[set[str], set[str]]: def _collect_rw_stmt(stmt: Stmt, reads: set[str], writes: set[str]) -> None: - st = type(stmt) - if st is Assign: + if type(stmt) is Assign: writes.add(stmt.target) _collect_rw_expr(stmt.value, reads) - elif st is IfElse: + elif type(stmt) is IfElse: _collect_rw_expr(stmt.cond, reads) for s in stmt.then_body: _collect_rw_stmt(s, reads, writes) for s in stmt.else_body: _collect_rw_stmt(s, reads, writes) - elif st is Switch: + elif type(stmt) is Switch: _collect_rw_expr(stmt.test, reads) for _, body in stmt.cases: for s in body: _collect_rw_stmt(s, reads, writes) - elif st is Assert: + elif type(stmt) is Assert: _collect_rw_expr(stmt.cond, reads) - elif st is Print: + elif type(stmt) is Print: for a in stmt.args: _collect_rw_expr(a, reads) def _collect_rw_expr(expr: Expr, reads: set[str]) -> None: - et = type(expr) - if et is SignalRef: + if type(expr) is SignalRef: reads.add(expr.name) - elif et is Unary: + elif type(expr) is Unary: _collect_rw_expr(expr.operand, reads) - elif et is Binary: + elif type(expr) is Binary: _collect_rw_expr(expr.left, reads) _collect_rw_expr(expr.right, reads) - elif et is Mux: + elif type(expr) is Mux: _collect_rw_expr(expr.sel, reads) _collect_rw_expr(expr.if_true, reads) _collect_rw_expr(expr.if_false, reads) - elif et is Concat: + elif type(expr) is Concat: for p in expr.parts: _collect_rw_expr(p, reads) - elif et is Slice: + elif type(expr) is Slice: _collect_rw_expr(expr.value, reads) - elif et is SysRandom and expr.seed is not None: + elif type(expr) is SysRandom and expr.seed is not None: _collect_rw_expr(expr.seed, reads) diff --git a/dau_sim/compiler/compile.py b/dau_sim/compiler/compile.py index b69f700..18f07c6 100644 --- a/dau_sim/compiler/compile.py +++ b/dau_sim/compiler/compile.py @@ -1,10 +1,19 @@ from __future__ import annotations import math +from collections.abc import Callable from datetime import datetime, timedelta +from typing import TYPE_CHECKING, Any import csp -from csp import ts + +if TYPE_CHECKING: + from typing import Annotated, TypeAlias, TypeVar + + T = TypeVar("T") + ts: TypeAlias = Annotated[T, "CSP stream"] # noqa: PYI042 +else: + from csp import ts from dau_sim.compiler.codegen import CodeGen, collect_reads_writes from dau_sim.compiler.depanalysis import ( @@ -53,9 +62,8 @@ def _exec_stmt( changed_signals: set[str] | None = None, ) -> None: """Execute a single statement, mutating current signal values.""" - stmt_type = type(stmt) - if stmt_type is Assign: + if type(stmt) is Assign: target = stmt.target val = mask_value(eval_expr(stmt.value, current), shapes[target]) old_val = current.get(target) @@ -64,13 +72,13 @@ def _exec_stmt( current[target] = val return - if stmt_type is IfElse: + if type(stmt) is IfElse: cond = eval_expr(stmt.cond, current) body = stmt.then_body if cond else stmt.else_body _exec_stmts(body, current, shapes, mem_state, changed_signals) return - if stmt_type is Switch: + if type(stmt) is Switch: test = eval_expr(stmt.test, current) default_stmts: tuple[Stmt, ...] = () for pattern, stmts in stmt.cases: @@ -83,25 +91,25 @@ def _exec_stmt( _exec_stmts(default_stmts, current, shapes, mem_state, changed_signals) return - if stmt_type is Assert: + if type(stmt) is Assert: cond = eval_expr(stmt.cond, current) if not cond: msg = stmt.message or "assertion failed" raise AssertionError(msg) return - if stmt_type is Print: + if type(stmt) is Print: args = [eval_expr(a, current) for a in stmt.args] print(stmt.format_str.format(*args)) return - if stmt_type is Finish: + if type(stmt) is Finish: raise SimulationFinish(stmt.exit_code) - if stmt_type is Delay: + if type(stmt) is Delay: return # Delay is a no-op in synchronous execution; handled by InitBlock runner - if stmt_type is ReadMem: + if type(stmt) is ReadMem: if mem_state is None: raise RuntimeError("$readmemh/$readmemb requires mem_state (use in initial block)") _exec_readmem(stmt, mem_state) @@ -430,10 +438,10 @@ def _edge_fires(edge: EdgePolarity, old_clk: int, new_clk: int) -> bool: def _sim_tick( tick: ts[bool], init_signals: ts[dict], - comb_order: ts[object], - shapes: ts[object], - memories: ts[object], - mem_init: ts[object], + comb_order: ts[list], + shapes: ts[dict], + memories: ts[tuple], + mem_init: ts[dict], ) -> ts[dict]: """Combinational-only simulation node (no clock edge semantics). @@ -479,10 +487,10 @@ def _sim_tick( def _sim_tick_4( tick: ts[bool], init_signals: ts[dict], - comb_order: ts[object], - shapes: ts[object], - memories: ts[object], - mem_init: ts[object], + comb_order: ts[list], + shapes: ts[dict], + memories: ts[tuple], + mem_init: ts[dict], ) -> ts[dict]: """Four-state combinational-only simulation node.""" with csp.state(): @@ -523,13 +531,13 @@ def _sim_tick_4( def _sim_engine_seq( tick: ts[bool], init_signals: ts[dict], - comb_components: ts[object], - comb_component_index: ts[object], - domain_info: ts[object], - shapes: ts[object], - init_values: ts[object], - memories: ts[object], - mem_init: ts[object], + comb_components: ts[tuple], + comb_component_index: ts[dict], + domain_info: ts[dict], + shapes: ts[dict], + init_values: ts[dict], + memories: ts[tuple], + mem_init: ts[dict], ) -> ts[dict]: """Clock-aware simulation engine (two-state). @@ -662,13 +670,13 @@ def _sim_engine_seq( def _sim_engine_seq_4( tick: ts[bool], init_signals: ts[dict], - comb_components: ts[object], - comb_component_index: ts[object], - domain_info: ts[object], - shapes: ts[object], - init_values: ts[object], - memories: ts[object], - mem_init: ts[object], + comb_components: ts[tuple], + comb_component_index: ts[dict], + domain_info: ts[dict], + shapes: ts[dict], + init_values: ts[dict], + memories: ts[tuple], + mem_init: ts[dict], ) -> ts[dict]: """Clock-aware simulation engine (four-state).""" with csp.state(): @@ -813,12 +821,12 @@ def _affected_component_ids_arr( def _sim_engine_compiled( tick: ts[bool], init_signals: ts[dict], - domain_info: ts[object], - shapes: ts[object], - init_values: ts[object], - memories: ts[object], - mem_init: ts[object], - compiled_meta: ts[object], + domain_info: ts[dict], + shapes: ts[dict], + init_values: ts[dict], + memories: ts[tuple], + mem_init: ts[dict], + compiled_meta: ts[dict], return_traces: bool = True, ) -> ts[dict]: """Clock-aware compiled simulation engine (two-state). @@ -1173,7 +1181,7 @@ def _build_compiled_meta(self) -> dict: sig_index = cg.signal_index # Compile per-domain seq block functions - seq_fns: dict[str, callable] = {} + seq_fns: dict[str, Callable[..., None]] = {} for dom_i, (dname, dinfo) in enumerate(self._domain_info.items()): sbs = dinfo["seq_blocks"] if not sbs: @@ -1192,7 +1200,7 @@ def _build_compiled_meta(self) -> dict: ) # Compile per-component comb functions - comb_fns: list[callable] = [] + comb_fns: list[Callable[..., None]] = [] for comp_i, comp in enumerate(self._comb_components): all_stmts = [] for assignment in comp.assignments: @@ -1353,7 +1361,7 @@ def run( # $finish in init block → return single-point traces if not return_traces: return {} - starttime = datetime(2000, 1, 1) # noqa: DTZ001 # naive simulation epoch anchor, not a real timestamp + starttime = datetime(2000, 1, 1) # naive simulation epoch anchor, not a real timestamp return {name: [(starttime, init.get(name, 0))] for name in out_names} # Persist any $readmemh/$readmemb changes back to mem_init @@ -1427,7 +1435,7 @@ def sim_graph(): else: csp.add_graph_output("__sink__", _trace_sink(all_signals)) - starttime = datetime(2000, 1, 1) # noqa: DTZ001 # naive simulation epoch anchor, not a real timestamp + starttime = datetime(2000, 1, 1) # naive simulation epoch anchor, not a real timestamp endtime = starttime + clock_period * cycles raw = csp.run(sim_graph, starttime=starttime, endtime=endtime, output_numpy=output_numpy) @@ -1533,7 +1541,7 @@ def sim_graph(): else: csp.add_graph_output("__sink__", _trace_sink(all_signals)) - starttime = datetime(2000, 1, 1) # noqa: DTZ001 # naive simulation epoch anchor, not a real timestamp + starttime = datetime(2000, 1, 1) # naive simulation epoch anchor, not a real timestamp endtime = starttime + tick_period * total_ticks raw = csp.run(sim_graph, starttime=starttime, endtime=endtime, output_numpy=output_numpy) @@ -1567,7 +1575,7 @@ def _run_generated_loop( run(S, clock_arr, 0, total_ticks, ticks, vals) if not return_traces: return {} - starttime = datetime(2000, 1, 1) # noqa: DTZ001 # naive simulation epoch anchor, not a real timestamp + starttime = datetime(2000, 1, 1) # naive simulation epoch anchor, not a real timestamp if output_numpy: import numpy as np @@ -1582,15 +1590,15 @@ def _collect_traces( all_names: list[str], four_state: bool, output_numpy: bool, - ) -> dict[str, list[tuple[datetime, int]]]: - result: dict[str, list[tuple[datetime, int]]] = {} + ) -> dict[str, Any]: + result: dict[str, Any] = {} for name in all_names: if name in raw: if output_numpy: times, values = raw[name] if four_state: values = [v.to_int for v in values] - result[name] = (times, values) # type: ignore[assignment] + result[name] = (times, values) continue if four_state: result[name] = [(t, v.to_int) for t, v in raw[name]] diff --git a/dau_sim/compiler/depanalysis.py b/dau_sim/compiler/depanalysis.py index b5cb99e..186f1a5 100644 --- a/dau_sim/compiler/depanalysis.py +++ b/dau_sim/compiler/depanalysis.py @@ -9,6 +9,7 @@ from __future__ import annotations from collections import deque +from collections.abc import Sequence from dataclasses import dataclass from dau_sim.ir.expr import Binary, Concat, Const, Expr, Mux, SignalRef, Slice, Unary @@ -123,7 +124,7 @@ def __str__(self) -> str: return f"Combinational loop detected: {path}" -def build_assignments(stmts_list: list[tuple[int, tuple[Stmt, ...]]]) -> list[Assignment]: +def build_assignments(stmts_list: Sequence[tuple[int, tuple[Stmt, ...]]]) -> list[Assignment]: """Build Assignment objects from indexed statement groups. Each entry is (index, stmts) where stmts is a tuple of statements diff --git a/dau_sim/compiler/eval.py b/dau_sim/compiler/eval.py index bfa1d51..4443246 100644 --- a/dau_sim/compiler/eval.py +++ b/dau_sim/compiler/eval.py @@ -56,9 +56,7 @@ def eval_expr(expr: Expr, signals: dict[str, int]) -> int: Returns an int value truncated to the expression's shape. """ - expr_type = type(expr) - - if expr_type is Const: + if type(expr) is Const: width = expr.shape.width if width == 0: return 0 @@ -68,7 +66,7 @@ def eval_expr(expr: Expr, signals: dict[str, int]) -> int: raw -= 1 << width return raw - if expr_type is SignalRef: + if type(expr) is SignalRef: width = expr.shape.width if width == 0: return 0 @@ -78,34 +76,34 @@ def eval_expr(expr: Expr, signals: dict[str, int]) -> int: raw -= 1 << width return raw - if expr_type is Unary: + if type(expr) is Unary: a = eval_expr(expr.operand, signals) return _eval_unary(expr.op, a, expr.operand.shape, expr.shape) - if expr_type is Binary: + if type(expr) is Binary: left = eval_expr(expr.left, signals) right = eval_expr(expr.right, signals) return _eval_binary(expr.op, left, right, expr.left.shape, expr.right.shape, expr.shape) - if expr_type is Mux: + if type(expr) is Mux: sel = eval_expr(expr.sel, signals) if sel: return mask_value(eval_expr(expr.if_true, signals), expr.shape) return mask_value(eval_expr(expr.if_false, signals), expr.shape) - if expr_type is Concat: + if type(expr) is Concat: result = 0 for part in expr.parts: result = (result << part.shape.width) | (eval_expr(part, signals) & _width_mask(part.shape.width)) return mask_value(result, expr.shape) - if expr_type is Slice: + if type(expr) is Slice: val = eval_expr(expr.value, signals) # Extract bits [low:high) extracted = (val >> expr.low) & _width_mask(expr.high - expr.low) return mask_value(extracted, expr.shape) - if expr_type is SysRandom: + if type(expr) is SysRandom: if expr.seed is not None: seed_val = eval_expr(expr.seed, signals) _sys_random_rng.seed(seed_val) diff --git a/dau_sim/config/__init__.py b/dau_sim/config/__init__.py index f77ef19..a292185 100644 --- a/dau_sim/config/__init__.py +++ b/dau_sim/config/__init__.py @@ -88,7 +88,7 @@ def compose_config( def profile_names(*, config_dir: str | None = None, version_base: str | None = None) -> tuple[str, ...]: result = _load_base_config(config_dir=config_dir, version_base=version_base, debug=True) - return tuple(sorted(result.group_options.get("profile/profiles", ()))) + return tuple(sorted((result.group_options or {}).get("profile/profiles", ()))) def _load_base_config( @@ -102,7 +102,7 @@ def _load_base_config( return base_load_config( root_config_dir=parent_dir, root_config_name="base", - config_dir=config_dir, + config_dir=config_dir or "", overrides=list(overrides or ()), version_base=version_base, basepath=parent_dir, diff --git a/dau_sim/frontends/pyslang_frontend.py b/dau_sim/frontends/pyslang_frontend.py index 7b7b1cf..8aa8f48 100644 --- a/dau_sim/frontends/pyslang_frontend.py +++ b/dau_sim/frontends/pyslang_frontend.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import pyslang as ps from dau_sim.ir.expr import ( @@ -25,8 +27,11 @@ from dau_sim.ir.stmt import Assign, IfElse, Stmt from dau_sim.ir.types import EdgePolarity, PortDirection, ResetStyle, Shape -_ps_ast = getattr(ps, "ast", ps) -_ps_syntax = getattr(ps, "syntax", ps) +if TYPE_CHECKING: + from pyslang import ast as _ps_ast, syntax as _ps_syntax +else: + _ps_ast = getattr(ps, "ast", ps) + _ps_syntax = getattr(ps, "syntax", ps) __all__ = ( "parse_sv", @@ -58,7 +63,7 @@ _ps_ast.BinaryOperator.LogicalOr: BinaryOp.LOGIC_OR, } -_UNOP_MAP: dict[_ps_ast.UnaryOperator, UnaryOp] = { +_UNOP_MAP: dict[_ps_ast.UnaryOperator, UnaryOp | None] = { _ps_ast.UnaryOperator.BitwiseNot: UnaryOp.NOT, _ps_ast.UnaryOperator.Minus: UnaryOp.NEG, _ps_ast.UnaryOperator.Plus: None, # unary + is identity diff --git a/dau_sim/integrations/cocotb.py b/dau_sim/integrations/cocotb.py index 2f84552..f693e8e 100644 --- a/dau_sim/integrations/cocotb.py +++ b/dau_sim/integrations/cocotb.py @@ -11,6 +11,7 @@ from collections.abc import Mapping, Sequence from pathlib import Path +from typing import Any DEFAULT_BUILD_ARGS = ("--timing", "-Wno-fatal") @@ -59,7 +60,7 @@ def run_cocotb_testbench( include_dirs = sorted({source_path.parent for source_path in source_paths}) runner = get_runner("verilator") - build_kwargs = {"parameters": dict(parameters)} if parameters else {} + build_kwargs: dict[str, Any] = {"parameters": dict(parameters)} if parameters else {} runner.build( sources=source_paths, hdl_toplevel=hdl_toplevel, diff --git a/dau_sim/tests/test_cli.py b/dau_sim/tests/test_cli.py index ed618dd..1c276e1 100644 --- a/dau_sim/tests/test_cli.py +++ b/dau_sim/tests/test_cli.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import Any import typer from typer.testing import CliRunner @@ -58,12 +59,12 @@ def test_run_sv_command_invokes_composed_task(tmp_path: Path, monkeypatch) -> No src = tmp_path / "adder.sv" src.write_text("module adder(input logic [7:0] a, output logic [7:0] y); assign y = a; endmodule") - captured: dict[str, object] = {} + captured: dict[str, Any] = {} def fake_run_request_config(request_kind, request_name, *, overrides=None, **_kwargs): captured["request_kind"] = request_kind captured["request_name"] = request_name - captured["overrides"] = list(overrides) + captured["overrides"] = list(overrides or ()) return RunSvResult(module_name="adder", cycles=3, latest={"y": 7}, vcd_path=tmp_path / "out.vcd") monkeypatch.setattr("dau_sim.config.run_request_config", fake_run_request_config) @@ -98,7 +99,7 @@ def test_perf_sv_command_invokes_composed_task(tmp_path: Path, monkeypatch) -> N captured = {} def fake_run_request_config(request_kind, request_name, *, overrides, **kwargs): - captured.update(request_kind=request_kind, request_name=request_name, overrides=list(overrides), kwargs=kwargs) + captured.update(request_kind=request_kind, request_name=request_name, overrides=list(overrides or ()), kwargs=kwargs) return PerfSvResult( benchmark=BenchmarkResult(compile_seconds_median=0.1, run_seconds_median=0.2, cycles_per_second=50.0), node_separation=NodeSeparationStats( @@ -138,7 +139,7 @@ def test_a_relative_config_dir_resolves_against_the_callers_directory(tmp_path: src.write_text("module adder(input logic [7:0] a, output logic [7:0] y); assign y = a; endmodule") overlay = tmp_path / "overlay" overlay.mkdir() - captured: dict[str, object] = {} + captured: dict[str, Any] = {} def fake_run_request_config(request_kind, request_name, *, overrides=None, config_dir=None, **_kwargs): captured["config_dir"] = config_dir diff --git a/dau_sim/tests/test_cocotb_backend.py b/dau_sim/tests/test_cocotb_backend.py index 2ad01ed..2a6d0ea 100644 --- a/dau_sim/tests/test_cocotb_backend.py +++ b/dau_sim/tests/test_cocotb_backend.py @@ -1,3 +1,5 @@ +from typing import Any, cast + """Tests for the cocotb backend (pure-Python GPI replacement). Tests cover: @@ -645,6 +647,7 @@ def test_register_value_change(self): engine = SimulationEngine(_make_comb_module()) engine.build_handle_hierarchy() mod = _create_simulator_module(engine) + assert engine._root_handle is not None h = engine._root_handle.get_handle_by_name("a") cb = mod.register_value_change_callback(h, lambda *a: None, RISING) assert isinstance(cb, DauSimCallback) @@ -671,12 +674,16 @@ def _setup_cocotb(self, module): import cocotb + cocotb = cast(Any, cocotb) + # isort: off import cocotb.handle # must precede cocotb._gpi_triggers (circular import) import cocotb._event_loop import cocotb._gpi_triggers import cocotb._test_manager import cocotb.simtime + + cocotb = cast(Any, cocotb) # isort: on engine = SimulationEngine(module) @@ -719,7 +726,7 @@ def _setup_cocotb(self, module): from cocotb._base_triggers import Event from cocotb._test_manager import TestManager - test = TestManager.__new__(TestManager) + test: Any = TestManager.__new__(TestManager) test._tasks = {} test._excs = [] test._finishing = False @@ -737,11 +744,15 @@ def _teardown_cocotb(self): import cocotb + cocotb = cast(Any, cocotb) + # isort: off import cocotb.handle # must precede cocotb._gpi_triggers (circular import) import cocotb._gpi_triggers import cocotb._test_manager import cocotb.simtime + + cocotb = cast(Any, cocotb) # isort: on cocotb.handle._handle2obj.clear() @@ -761,6 +772,8 @@ def test_timer_trigger(self): """A Timer(100) trigger should fire after 100 sim time steps.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup_cocotb(_make_comb_module()) try: from cocotb._gpi_triggers import Timer @@ -792,6 +805,8 @@ def test_rising_edge_trigger(self): """RisingEdge should fire when signal goes 0→1.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup_cocotb(_make_counter_module()) try: from cocotb._gpi_triggers import RisingEdge, Timer @@ -831,6 +846,8 @@ def test_counter_via_cocotb_clock(self): """Use cocotb.clock.Clock to drive the counter and verify count increments.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup_cocotb(_make_counter_module()) try: from cocotb._gpi_triggers import RisingEdge @@ -867,6 +884,8 @@ def test_combinational_read(self): """Verify that combinational logic settles after signal writes.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup_cocotb(_make_comb_module()) try: from cocotb._gpi_triggers import Timer @@ -901,6 +920,8 @@ def test_timer_and_edge_same_timestep(self): """Timer+edge combination should complete when both happen together.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup_cocotb(_make_comb_module()) try: from cocotb._gpi_triggers import RisingEdge, Timer @@ -935,3 +956,20 @@ async def waiter(): assert result[0] == 100 finally: self._teardown_cocotb() + + +def test_run_cocotb_lowers_amaranth_design(monkeypatch): + from dau_sim.backends import cocotb_backend + + design = object() + module = _make_comb_module() + captured = [] + + def lower(value): + assert value is design + return module + + monkeypatch.setattr("dau_sim.frontends.amaranth_frontend.from_amaranth", lower) + monkeypatch.setattr(cocotb_backend, "_run_with_patched_simulator", lambda engine, _simulator, _test_module: captured.append(engine._module)) + cocotb_backend.run_cocotb(design, test_module="test_design") + assert captured == [module] diff --git a/dau_sim/tests/test_cocotb_examples.py b/dau_sim/tests/test_cocotb_examples.py index b684c37..708388f 100644 --- a/dau_sim/tests/test_cocotb_examples.py +++ b/dau_sim/tests/test_cocotb_examples.py @@ -1,3 +1,5 @@ +from typing import Any, cast + """End-to-end tests based on cocotb's official examples. Each test translates a cocotb example HDL design into dau-sim IR, sets up @@ -170,11 +172,15 @@ def _setup(self, module): """Build engine, patch cocotb.simulator, return engine.""" # isort: off import cocotb + + cocotb = cast(Any, cocotb) import cocotb.handle # must precede cocotb._gpi_triggers (circular import) import cocotb._event_loop import cocotb._gpi_triggers import cocotb._test_manager import cocotb.simtime + + cocotb = cast(Any, cocotb) # isort: on engine = SimulationEngine(module) @@ -217,7 +223,7 @@ def _setup(self, module): from cocotb._base_triggers import Event from cocotb._test_manager import TestManager - test = TestManager.__new__(TestManager) + test: Any = TestManager.__new__(TestManager) test._tasks = {} test._excs = [] test._finishing = False @@ -234,10 +240,14 @@ def _setup(self, module): def _teardown(self): # isort: off import cocotb + + cocotb = cast(Any, cocotb) import cocotb.handle # must precede cocotb._gpi_triggers (circular import) import cocotb._gpi_triggers import cocotb._test_manager import cocotb.simtime + + cocotb = cast(Any, cocotb) # isort: on cocotb.handle._handle2obj.clear() @@ -256,6 +266,8 @@ def _teardown(self): def _run_coroutine(self, coro, max_steps=10_000_000): """Schedule *coro* and run the engine.""" import cocotb + + cocotb = cast(Any, cocotb) from cocotb.task import Task task = Task(coro) @@ -276,6 +288,8 @@ def test_dff_d_propagates_to_q(self): """Test that d propagates to q on each rising clock edge.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_dff()) try: from cocotb._gpi_triggers import RisingEdge @@ -324,6 +338,8 @@ def test_dff_holds_value_without_clock(self): """Verify q doesn't change when no clock edge occurs.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_dff()) try: from cocotb._gpi_triggers import RisingEdge, Timer @@ -369,6 +385,8 @@ def test_adder_basic(self): """Test for 5 + 10 = 15.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_adder()) try: from cocotb._gpi_triggers import Timer @@ -389,6 +407,8 @@ def test_adder_randomised(self): """Test for adding 2 random numbers multiple times.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_adder()) try: from cocotb._gpi_triggers import Timer @@ -415,6 +435,8 @@ def test_adder_overflow(self): """Verify the 5-bit output captures the carry for max inputs.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_adder()) try: from cocotb._gpi_triggers import Timer @@ -448,6 +470,8 @@ def test_quickstart_1(self): """ import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_simple_counter()) try: from cocotb._gpi_triggers import RisingEdge, Timer @@ -502,6 +526,8 @@ def test_quickstart_2_enable_counter(self): """ import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_simple_counter()) try: from cocotb._gpi_triggers import FallingEdge, RisingEdge @@ -550,6 +576,8 @@ def test_counter_set_din(self): """Verify the 'set' input loads din into counter.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_simple_counter()) try: from cocotb._gpi_triggers import RisingEdge @@ -597,6 +625,8 @@ def test_counter_reset_during_count(self): """Verify rst resets counter even while ena is active.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_simple_counter()) try: from cocotb._gpi_triggers import RisingEdge @@ -652,6 +682,8 @@ def test_counter_start_soon(self): """Verify start_soon() works for concurrent clock + test coroutines.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_simple_counter()) try: from cocotb._gpi_triggers import FallingEdge, RisingEdge @@ -704,6 +736,8 @@ def test_counter_holds_when_disabled(self): """Counter should hold its value when neither ena nor set is active.""" import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_make_simple_counter()) try: from cocotb._gpi_triggers import RisingEdge diff --git a/dau_sim/tests/test_compiler.py b/dau_sim/tests/test_compiler.py index b217da1..8cd142a 100644 --- a/dau_sim/tests/test_compiler.py +++ b/dau_sim/tests/test_compiler.py @@ -264,6 +264,7 @@ class TestRewriter: def test_rewrite_signalref(self): expr = SignalRef(shape=Shape(8), name="count") result = prefix_expr(expr, "child") + assert isinstance(result, SignalRef) assert result.name == "child.count" def test_rewrite_const_unchanged(self): @@ -274,12 +275,17 @@ def test_rewrite_const_unchanged(self): def test_rewrite_binary_expr(self): expr = Binary(shape=Shape(8), op=BinaryOp.ADD, left=SignalRef(shape=Shape(8), name="a"), right=SignalRef(shape=Shape(8), name="b")) result = prefix_expr(expr, "sub") + assert isinstance(result, Binary) + assert isinstance(result.left, SignalRef) assert result.left.name == "sub.a" + assert isinstance(result.right, SignalRef) assert result.right.name == "sub.b" def test_rewrite_unary_expr(self): expr = Unary(shape=Shape(8), op=UnaryOp.NOT, operand=SignalRef(shape=Shape(8), name="x")) result = prefix_expr(expr, "p") + assert isinstance(result, Unary) + assert isinstance(result.operand, SignalRef) assert result.operand.name == "p.x" def test_rewrite_mux_expr(self): @@ -290,25 +296,36 @@ def test_rewrite_mux_expr(self): if_false=SignalRef(shape=Shape(8), name="b"), ) result = prefix_expr(expr, "m") + assert isinstance(result, Mux) + assert isinstance(result.sel, SignalRef) assert result.sel.name == "m.sel" + assert isinstance(result.if_true, SignalRef) assert result.if_true.name == "m.a" + assert isinstance(result.if_false, SignalRef) assert result.if_false.name == "m.b" def test_rewrite_concat_expr(self): expr = Concat(shape=Shape(16), parts=(SignalRef(shape=Shape(8), name="hi"), SignalRef(shape=Shape(8), name="lo"))) result = prefix_expr(expr, "c") + assert isinstance(result, Concat) + assert isinstance(result.parts[0], SignalRef) assert result.parts[0].name == "c.hi" + assert isinstance(result.parts[1], SignalRef) assert result.parts[1].name == "c.lo" def test_rewrite_slice_expr(self): expr = Slice(shape=Shape(4), value=SignalRef(shape=Shape(8), name="val"), low=0, high=4) result = prefix_expr(expr, "s") + assert isinstance(result, Slice) + assert isinstance(result.value, SignalRef) assert result.value.name == "s.val" def test_rewrite_assign_stmt(self): stmt = Assign(target="out", value=SignalRef(shape=Shape(1), name="in_sig")) result = prefix_stmt(stmt, "child") + assert isinstance(result, Assign) assert result.target == "child.out" + assert isinstance(result.value, SignalRef) assert result.value.name == "child.in_sig" def test_rewrite_ifelse_stmt(self): @@ -318,7 +335,10 @@ def test_rewrite_ifelse_stmt(self): else_body=(Assign(target="o", value=Const(shape=Shape(1), value=0)),), ) result = prefix_stmt(stmt, "x") + assert isinstance(result, IfElse) + assert isinstance(result.cond, SignalRef) assert result.cond.name == "x.en" + assert isinstance(result.then_body[0], Assign) assert result.then_body[0].target == "x.o" def test_rewrite_switch_stmt(self): @@ -327,7 +347,10 @@ def test_rewrite_switch_stmt(self): cases=((0, (Assign(target="y", value=Const(shape=Shape(1), value=0)),)),), ) result = prefix_stmt(stmt, "p") + assert isinstance(result, IrSwitch) + assert isinstance(result.test, SignalRef) assert result.test.name == "p.sel" + assert isinstance(result.cases[0][1][0], Assign) assert result.cases[0][1][0].target == "p.y" def test_prefix_stmts_batch(self): @@ -336,6 +359,9 @@ def test_prefix_stmts_batch(self): Assign(target="b", value=SignalRef(shape=Shape(1), name="c")), ) result = prefix_stmts(stmts, "mod") + assert isinstance(result[0], Assign) + assert isinstance(result[1], Assign) + assert isinstance(result[1].value, SignalRef) assert result[0].target == "mod.a" assert result[1].target == "mod.b" assert result[1].value.name == "mod.c" diff --git a/dau_sim/tests/test_ir.py b/dau_sim/tests/test_ir.py index bbc79ac..2ba9069 100644 --- a/dau_sim/tests/test_ir.py +++ b/dau_sim/tests/test_ir.py @@ -1,3 +1,5 @@ +from typing import Any, cast + """Tests for IR data structures: types, expressions, module construction, validation, and printing.""" import unittest @@ -46,7 +48,7 @@ def test_shape_basics(): def test_shape_frozen(): s = Shape(4) try: - s.width = 5 + cast(Any, s).width = 5 raise AssertionError("Should have raised") except AttributeError: pass @@ -96,12 +98,12 @@ class TestIRNodes(unittest.TestCase): def test_delay_frozen(self): d = Delay(ticks=5) with self.assertRaises(ValidationError): - d.ticks = 10 + cast(Any, d).ticks = 10 def test_finish_frozen(self): f = Finish(exit_code=1) with self.assertRaises(ValidationError): - f.exit_code = 2 + cast(Any, f).exit_code = 2 def test_finish_default_exit_code(self): f = Finish() @@ -110,7 +112,7 @@ def test_finish_default_exit_code(self): def test_init_block_frozen(self): ib = InitBlock(stmts=()) with self.assertRaises(ValidationError): - ib.stmts = () + cast(Any, ib).stmts = () def _make_counter_module() -> Module: @@ -166,7 +168,9 @@ def test_module_construction(): assert m.name == "counter" assert len(m.ports) == 3 assert len(m.seq_blocks) == 1 - assert m.signal_by_name("count").shape == Shape(4) + count = m.signal_by_name("count") + assert count is not None + assert count.shape == Shape(4) def test_module_all_signal_names(): diff --git a/dau_sim/tests/test_protocol_conformance.py b/dau_sim/tests/test_protocol_conformance.py index 946074c..ff2a9e4 100644 --- a/dau_sim/tests/test_protocol_conformance.py +++ b/dau_sim/tests/test_protocol_conformance.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Any, cast + import pytest from dau_sim.compiler import compile_module @@ -120,6 +122,8 @@ class TestCocotbStreamContractMonitor(CocotbExampleTestBase): def test_context_manager_attaches_to_prefixed_interface(self) -> None: import cocotb + cocotb = cast(Any, cocotb) + engine = self._setup(_broken_stream_module("mutates_payload")) violations: list[str] = [] try: diff --git a/dau_sim/tests/test_selectors.py b/dau_sim/tests/test_selectors.py index fb90ee5..a2f61ed 100644 --- a/dau_sim/tests/test_selectors.py +++ b/dau_sim/tests/test_selectors.py @@ -6,7 +6,7 @@ from dau_sim.adapters.selectors import match_signals, select_signals SAMPLE_NAMES = ["clk", "rst", "en", "count", "data_0", "data_1", "data_2", "internal_reg"] -SAMPLE_TRACES = {name: [(datetime(2000, 1, 1), i)] for i, name in enumerate(SAMPLE_NAMES)} # noqa: DTZ001 # naive simulation-epoch trace timestamp +SAMPLE_TRACES = {name: [(datetime(2000, 1, 1), i)] for i, name in enumerate(SAMPLE_NAMES)} # naive simulation-epoch trace timestamp class TestMatchSignals(unittest.TestCase): diff --git a/dau_sim/tests/test_vcd.py b/dau_sim/tests/test_vcd.py index e59c062..7276c93 100644 --- a/dau_sim/tests/test_vcd.py +++ b/dau_sim/tests/test_vcd.py @@ -152,7 +152,7 @@ class TestSubNanosecondTicks: def test_a_picosecond_timescale_counts_picoseconds(self): """The old parser rounded "1ps" up to a nanosecond, so a 1ps VCD was a thousand times too short.""" - base = datetime(2024, 1, 1) # noqa: DTZ001 # naive simulation-epoch trace timestamp + base = datetime(2024, 1, 1) # naive simulation-epoch trace timestamp stamps = [base, base + timedelta(microseconds=1), base + timedelta(microseconds=2)] assert _datetime_to_timescale_ticks(stamps, timescale_ps=parse_timescale_ps("1ps")) == [0, 1_000_000, 2_000_000] assert _datetime_to_timescale_ticks(stamps, timescale_ps=parse_timescale_ps("1ns")) == [0, 1_000, 2_000] @@ -379,7 +379,7 @@ def test_timestamp_spacing(self): def test_no_module_fallback(self): """Without module metadata, signals default to 1-bit wire.""" - t0 = datetime(2000, 1, 1) # noqa: DTZ001 # naive simulation-epoch trace timestamp + t0 = datetime(2000, 1, 1) # naive simulation-epoch trace timestamp traces = { "x": [(t0, 0), (t0 + timedelta(microseconds=1), 1)], } @@ -390,7 +390,7 @@ def test_no_module_fallback(self): assert "$scope module top $end" in vcd def test_custom_scope(self): - t0 = datetime(2000, 1, 1) # noqa: DTZ001 # naive simulation-epoch trace timestamp + t0 = datetime(2000, 1, 1) # naive simulation-epoch trace timestamp traces = {"x": [(t0, 0)]} vcd = traces_to_vcd(traces, scope="my_dut") assert "$scope module my_dut $end" in vcd @@ -428,7 +428,7 @@ def test_write_vcd_file(self): def test_standalone_write_vcd(self): """write_vcd standalone function should work.""" - t0 = datetime(2000, 1, 1) # noqa: DTZ001 # naive simulation-epoch trace timestamp + t0 = datetime(2000, 1, 1) # naive simulation-epoch trace timestamp traces = {"sig": [(t0, 1), (t0 + timedelta(microseconds=1), 0)]} with tempfile.NamedTemporaryFile(suffix=".vcd", delete=False) as f: diff --git a/pyproject.toml b/pyproject.toml index 7be6e88..f162dd4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,6 +48,7 @@ develop = [ "codespell", "hatchling", "mdformat", + "mdformat-frontmatter", "mdformat-tables>=1", "pytest", "pytest-benchmark", @@ -93,7 +94,6 @@ ignore-words-list = ["inout"] [tool.coverage.run] branch = true omit = [ - # a bare directory path matches no file; coverage omit takes globs "dau_sim/tests/integration/*", ] @@ -184,3 +184,19 @@ pages = [ "docs/src/benchmarks.md", ] use-autoapi = false + +# These annotations describe HDL ports rather than Python types. +[[tool.ty.overrides]] +include = [ + "dau_sim/tests/test_frontend_amaranth.py", + "dau_sim/tests/test_testbench.py", + "dau_sim/tests/test_vcd.py", +] +[tool.ty.overrides.rules] +invalid-type-form = "ignore" + +# CSP nodes may skip emitting a value on a tick. +[[tool.ty.overrides]] +include = ["dau_sim/compiler/compile.py"] +[tool.ty.overrides.rules] +invalid-return-type = "ignore"