Fix code style issues identified by newer versions of ruff. Also:

* Consistently name `net_connections`'s return values `wire_src_dict` and
  `wire_dst_dict`.
* Remove some unnecessary assignments.
* Update `prngs.py` to use `IntEnum`s.

Author: fdxmw

pyrtl/analysis.py

@@ -222,7 +222,7 @@ def _multiplier_stdcell_estimate(width):
     def _memory_read_estimate(mem):
         # http://www.cs.ucsb.edu/~sherwood/pubs/ICCD-srammodel.pdf
         # ROM is assumed to be same delay as SRAM (perhaps optimistic?)
-        bits, ports, is_rom = _bits_ports_and_isrom_from_memory(mem)
+        bits, ports, _is_rom = _bits_ports_and_isrom_from_memory(mem)
         tech_in_um = 0.130
         return 270 * tech_in_um**1.38 * bits**0.25 * ports**1.30 + 1.05
 
@@ -279,7 +279,7 @@ def critical_path(
                  the critical paths (which themselves are lists of nets) as the second
         """
         critical_paths = []  # storage of all completed critical paths
-        wire_src_map, dst_map = self.block.net_connections()
+        wire_src_dict, _wire_dst_dict = self.block.net_connections()
 
         def critical_path_pass(old_critical_path, first_wire):
             if isinstance(first_wire, (Input, Const, Register)):
@@ -289,7 +289,7 @@ def critical_path_pass(old_critical_path, first_wire):
             if len(critical_paths) >= cp_limit:
                 raise self._TooManyCPsError()
 
-            source = wire_src_map[first_wire]
+            source = wire_src_dict[first_wire]
             critical_path = [source]
             critical_path.extend(old_critical_path)
             arg_max_time = max(self.timing_map[arg_wire] for arg_wire in source.args)
@@ -500,7 +500,7 @@ def paths(
         # present as the destination "net" of Output wires in the dst_nets map. That
         # would overly complicate this algorithm: we will assume all values() in the
         # dst_nets map are logic nets only. We set this to False for explicitness...
-        _, dst_nets = block.net_connections(include_virtual_nodes=False)
+        _wire_src_dict, dst_nets = block.net_connections(include_virtual_nodes=False)
     else:
         # ... or make sure it's not present otherwise.
         for output in block.wirevector_subset(cls=Output):
@@ -594,9 +594,9 @@ def fanout(w: WireVector) -> int:
 
     :return: Integer fanout count.
     """
-    _, dst_nets = w._block.net_connections()
-    if w not in dst_nets:
+    _wire_src_dict, wire_dst_dict = w._block.net_connections()
+    if w not in wire_dst_dict:
         return 0
 
-    all_args = [arg for net in dst_nets[w] for arg in net.args]
+    all_args = [arg for net in wire_dst_dict[w] for arg in net.args]
     return len(list(filter(lambda arg: arg is w, all_args)))

pyrtl/core.py

@@ -565,8 +565,8 @@ def net_connections(
         A :class:`WireVector`'s `source` is the :class:`LogicNet` that sets the
         :class:`WireVector`'s value.
 
-        A :class:`WireVector`'s `sinks` are the :class:`LogicNets<LogicNet>` that use
-        the :class:`WireVector`'s value.
+        A :class:`WireVector`'s `destinations` are the :class:`LogicNets<LogicNet>` that
+        use the :class:`WireVector`'s value.
 
         This information helps when building a graph representation for the ``Block``.
         See :func:`net_graph` for an example.
@@ -577,15 +577,15 @@ def net_connections(
 
         :param include_virtual_nodes: If ``True``, external `sources` (such as an
             :class:`Inputs<Input>` and :class:`Consts<Const>`) will be represented as
-            wires that set themselves, and external `sinks` (such as
+            wires that set themselves, and external `destinations` (such as
             :class:`Outputs<Output>`) will be represented as wires that use themselves.
             If ``False``, these nodes will be excluded from the results.
 
         :return: Two dictionaries. The first maps :class:`WireVectors<WireVector>` to
                  the :class:`LogicNet` that creates their signal (``wire_src_dict``).
                  The second maps :class:`WireVectors<WireVector>` to a list of
                  :class:`LogicNets<LogicNet>` that use their signal
-                 (``wire_sink_dict``).
+                 (``wire_dst_dict``).
         """
         src_list = {}
         dst_list = {}
@@ -645,16 +645,16 @@ def __iter__(self):
         """
         from pyrtl.wire import Const, Input, Register
 
-        src_dict, dest_dict = self.net_connections()
+        _wire_src_dict, wire_dst_dict = self.net_connections()
         to_clear = self.wirevector_subset((Input, Const, Register))
         cleared = set()
         remaining = self.logic.copy()
         try:
             while len(to_clear):
                 wire_to_check = to_clear.pop()
                 cleared.add(wire_to_check)
-                if wire_to_check in dest_dict:
-                    for gate in dest_dict[
+                if wire_to_check in wire_dst_dict:
+                    for gate in wire_dst_dict[
                         wire_to_check
                     ]:  # loop over logicnets not yet returned
                         if all(
@@ -804,7 +804,7 @@ def sanity_check_memory_sync(self, wire_src_dict=None):
             return  # nothing to check here
 
         if wire_src_dict is None:
-            wire_src_dict, wdd = self.net_connections()
+            wire_src_dict, _wire_dst_dict = self.net_connections()
 
         from pyrtl.wire import Const, Input
 

pyrtl/helperfuncs.py

@@ -880,9 +880,9 @@ def val_to_formatted_str(val: int, format: str, enum_set=None) -> str:
     if type == "s":
         rval = str(val_to_signed_integer(val, bitwidth))
     elif type == "x":
-        rval = hex(val)[2:]  # cuts off '0x' at the start
+        rval = f"{val:x}"
     elif type == "b":
-        rval = bin(val)[2:]  # cuts off '0b' at the start
+        rval = f"{val:b}"
     elif type == "u":
         rval = str(int(val))  # nothing fancy
     elif type == "e":

pyrtl/passes.py

@@ -79,7 +79,7 @@ def optimize(
     return block
 
 
-def _get_inverter_chains(wire_creator, wire_users):
+def _get_inverter_chains(wire_src_dict, wire_dst_dict):
     """Returns all inverter chains in the block.
 
     The function returns a list of inverter chains in the block. Each inverter chain is
@@ -98,7 +98,7 @@ def _get_inverter_chains(wire_creator, wire_users):
     # Build a list of inverter chains. Each inverter chain is a list of WireVectors,
     # from source to destination.
     inverter_chains = []
-    for current_dest, current_creator in wire_creator.items():
+    for current_dest, current_creator in wire_src_dict.items():
         if current_creator.op != "~":
             # Skip non-inverters.
             continue
@@ -107,7 +107,7 @@ def _get_inverter_chains(wire_creator, wire_users):
         # a WireVector).
         current_arg = current_creator.args[0]
         # current_users is the number of LogicNets that use current_dest.
-        current_users = len(wire_users[current_dest])
+        current_users = len(wire_dst_dict[current_dest])
 
         # Add the current inverter to the end of this inverter chain.
         append_to = None
@@ -117,7 +117,7 @@ def _get_inverter_chains(wire_creator, wire_users):
         for inverter_chain in inverter_chains:
             chain_arg = inverter_chain[0]
             chain_dest = inverter_chain[-1]
-            chain_users = len(wire_users[chain_dest])
+            chain_users = len(wire_dst_dict[chain_dest])
 
             if chain_dest is current_arg and chain_users == 1:
                 # This chain's only destination is the current inverter. Append the
@@ -168,9 +168,9 @@ def _optimize_inverter_chains(block, skip_sanity_check=False):
         B -w-> Y
     """
 
-    # wire_creator maps from WireVector to the LogicNet that defines its value.
-    # wire_users maps from WireVector to a list of LogicNets that use its value.
-    wire_creator, wire_users = block.net_connections()
+    # wire_src_dict maps from WireVector to the LogicNet that defines its value.
+    # wire_dst_dict maps from WireVector to a list of LogicNets that use its value.
+    wire_src_dict, wire_dst_dict = block.net_connections()
 
     new_logic = set()
     net_removal_set = set()
@@ -199,9 +199,9 @@ def _optimize_inverter_chains(block, skip_sanity_check=False):
     # will be mapped to A and E will be mapped to C. Hence, when finding the replacement
     # of E, we have to first query the dict to get C, and then query the dict again on C
     # to get A.
-    wire_src_dict = _ProducerList()
+    wire_producer = _ProducerList()
 
-    for inverter_chain in _get_inverter_chains(wire_creator, wire_users):
+    for inverter_chain in _get_inverter_chains(wire_src_dict, wire_dst_dict):
         # If len(inverter_chain) = n, there are n-1 inverters in the chain. We only
         # remove inverters if there are at least two inverters in a chain.
         if len(inverter_chain) > 2:
@@ -215,10 +215,10 @@ def _optimize_inverter_chains(block, skip_sanity_check=False):
             wires_to_remove = inverter_chain[start_idx:]
             wire_removal_set.update(wires_to_remove)
             # Remove inverters used in the chain.
-            inverters_to_remove = {wire_creator[wire] for wire in wires_to_remove}
+            inverters_to_remove = {wire_src_dict[wire] for wire in wires_to_remove}
             net_removal_set.update(inverters_to_remove)
             # Map the end wire of the inverter chain to the beginning wire.
-            wire_src_dict[inverter_chain[-1]] = inverter_chain[start_idx - 1]
+            wire_producer[inverter_chain[-1]] = inverter_chain[start_idx - 1]
 
     # This loop recreates the block with inverter chains removed. It adds each LogicNet
     # in the original block to the new block if it is not marked for removal, and
@@ -230,7 +230,7 @@ def _optimize_inverter_chains(block, skip_sanity_check=False):
                 LogicNet(
                     net.op,
                     net.op_param,
-                    args=tuple(wire_src_dict.find_producer(x) for x in net.args),
+                    args=tuple(wire_producer.find_producer(x) for x in net.args),
                     dests=net.dests,
                 )
             )
@@ -1042,7 +1042,7 @@ def direct_connect_outputs(block=None):
     # NOTE: would use transform.all_nets(), but it becomes tricky when we want to remove
     # more than just the current net on a single pass
     block = working_block(block)
-    _, dst_nets = block.net_connections()
+    _wire_src_dict, wire_dst_dict = block.net_connections()
 
     nets_to_remove = set()
     nets_to_add = set()
@@ -1053,10 +1053,10 @@ def direct_connect_outputs(block=None):
             continue
 
         dest_wire = net.dests[0]
-        if dest_wire not in dst_nets or len(dst_nets[dest_wire]) > 1:
+        if dest_wire not in wire_dst_dict or len(wire_dst_dict[dest_wire]) > 1:
             continue
 
-        dst_net = dst_nets[dest_wire][0]
+        dst_net = wire_dst_dict[dest_wire][0]
         if dst_net.op != "w" or not isinstance(dst_net.dests[0], Output):
             continue
 
@@ -1105,7 +1105,7 @@ def two_way_fanout(block=None):
 
     block = working_block(block)
 
-    _, dst_map = block.net_connections()
+    _wire_src_dict, wire_dst_dict = block.net_connections()
     # Two-pass approach: Remember which nets will need to change, in case there are
     # multiple arguments which will be changing along the way.
     nets_to_update = collections.defaultdict(list)
@@ -1114,7 +1114,7 @@ def two_way_fanout(block=None):
         if curr_fanout > 1:
             s = _make_tree(wire, block, curr_fanout)
             curr_ix = 0
-            for dst_net in dst_map[wire]:
+            for dst_net in wire_dst_dict[wire]:
                 for i, arg in enumerate(dst_net.args):
                     if arg is wire:
                         nets_to_update[dst_net].append((wire, i, s[curr_ix]))

pyrtl/rtllib/matrix.py

@@ -1570,7 +1570,7 @@ def matrix_wv_to_list(
 
     :return: A Python list of lists.
     """
-    value = bin(matrix_wv)[2:].zfill(rows * columns * bits)
+    value = f"{matrix_wv:b}".zfill(rows * columns * bits)
 
     result = [[0 for _ in range(columns)] for _ in range(rows)]
 

pyrtl/rtllib/muxes.py

@@ -235,7 +235,9 @@ def _check_finalized(self):
 
     def option(self, select_val, *data_signals):
         self._check_finalized()
-        instr, ib = pyrtl.infer_val_and_bitwidth(select_val, self.signal_wire.bitwidth)
+        instr = pyrtl.infer_val_and_bitwidth(
+            select_val, self.signal_wire.bitwidth
+        ).value
         if instr in self.instructions:
             msg = f"instruction {select_val} already exists"
             raise pyrtl.PyrtlError(msg)

pyrtl/rtllib/prngs.py

@@ -1,3 +1,4 @@
+import enum
 import random
 from math import ceil, log2
 
@@ -145,26 +146,30 @@ def prng_xoroshiro128(
     counter = pyrtl.Register(counter_bitwidth, "counter")
     gen_done = counter == gen_cycles - 1
     state = pyrtl.Register(1)
-    WAIT, GEN = (pyrtl.Const(x) for x in range(2))
+
+    class State(enum.IntEnum):
+        WAIT = 0
+        GEN = 1
+
     with pyrtl.conditional_assignment:
         with load:
             s0.next |= seed[:64]
             s1.next |= seed[64:]
-            state.next |= WAIT
+            state.next |= State.WAIT
         with req:
             counter.next |= 0
             s0.next |= s0_next
             s1.next |= s1_next
             rand.next |= pyrtl.concat(rand, output)
-            state.next |= GEN
-        with state == GEN:
+            state.next |= State.GEN
+        with state == State.GEN:
             with ~gen_done:
                 counter.next |= counter + 1
                 s0.next |= s0_next
                 s1.next |= s1_next
                 rand.next |= pyrtl.concat(rand, output)
 
-    ready = ~load & ~req & (state == GEN) & gen_done
+    ready = ~load & ~req & (state == State.GEN) & gen_done
     return ready, rand[-bitwidth:]  # return MSBs because LSBs are less random
 
 
@@ -263,34 +268,43 @@ def csprng_trivium(
     init_done = counter == init_cycles
     gen_done = counter == gen_cycles - 1
     state = pyrtl.Register(2)
-    WAIT, INIT, GEN = (pyrtl.Const(x) for x in range(3))
+
+    class State(enum.IntEnum):
+        WAIT = 0
+        INIT = 1
+        GEN = 2
+
     with pyrtl.conditional_assignment:
         with load:
             counter.next |= 0
             a.next |= key
             b.next |= iv
             c.next |= pyrtl.concat(pyrtl.Const("3'b111"), pyrtl.Const(0, 108))
-            state.next |= INIT
+            state.next |= State.INIT
         with req:
             counter.next |= 0
             a.next |= a_next
             b.next |= b_next
             c.next |= c_next
             rand.next |= pyrtl.concat(rand, *output)
-            state.next |= GEN
-        with state == INIT:
+            state.next |= State.GEN
+        with state == State.INIT:
             with ~init_done:
                 counter.next |= counter + 1
                 a.next |= a_next
                 b.next |= b_next
                 c.next |= c_next
-        with state == GEN:
+        with state == State.GEN:
             with ~gen_done:
                 counter.next |= counter + 1
                 a.next |= a_next
                 b.next |= b_next
                 c.next |= c_next
                 rand.next |= pyrtl.concat(rand, *output)
 
-    ready = ~load & ~req & ((state == INIT) & init_done | (state == GEN) & gen_done)
+    ready = (
+        ~load
+        & ~req
+        & ((state == State.INIT) & init_done | (state == State.GEN) & gen_done)
+    )
     return ready, rand

pyrtl/simulation.py

@@ -1594,7 +1594,7 @@ def __len__(self):
             raise PyrtlError(msg)
         # return the length of the list of some element in the dictionary (all should be
         # the same)
-        wire, value_list = next(x for x in self.trace.items())
+        _wire, value_list = next(x for x in self.trace.items())
         return len(value_list)
 
     def add_step(self, value_map):

pyrtl/transform.py

@@ -73,10 +73,12 @@ def wire_transform(
     :param block: The Block to replace wires on. Defaults to the :ref:`working_block`.
     """
     block = working_block(block)
-    src_nets, dst_nets = block.net_connections(include_virtual_nodes=False)
+    wire_src_dict, wire_dst_dict = block.net_connections(include_virtual_nodes=False)
     for orig_wire in block.wirevector_subset(select_types, exclude_types):
         new_src, new_dst = transform_func(orig_wire)
-        replace_wire_fast(orig_wire, new_src, new_dst, src_nets, dst_nets, block)
+        replace_wire_fast(
+            orig_wire, new_src, new_dst, wire_src_dict, wire_dst_dict, block
+        )
 
 
 def all_wires(transform_func):
@@ -96,9 +98,9 @@ def replace_wires(wire_map, block=None):
     :param block: block to operate over (defaults to :ref:`working_block`)
     """
     block = working_block(block)
-    src_nets, dst_nets = block.net_connections(include_virtual_nodes=False)
+    wire_src_dict, wire_dst_dict = block.net_connections(include_virtual_nodes=False)
     for old_w, new_w in wire_map.items():
-        replace_wire_fast(old_w, new_w, new_w, src_nets, dst_nets, block)
+        replace_wire_fast(old_w, new_w, new_w, wire_src_dict, wire_dst_dict, block)
 
 
 def replace_wire_fast(orig_wire, new_src, new_dst, src_nets, dst_nets, block=None):

tests/rtllib/test_libutils.py

@@ -16,7 +16,7 @@ def test_successful_partition(self):
     def test_failing_partition(self):
         w = pyrtl.WireVector(14)
         with self.assertRaises(pyrtl.PyrtlError):
-            _ = libutils.partition_wire(w, 4)
+            libutils.partition_wire(w, 4)
 
     def test_partition_sim(self):
         pyrtl.reset_working_block()