resolve linter errors

Signed-off-by: Wojciech Sipak <[email protected]>

Author: wsipak

xls/modules/zstd/cocotb/channel.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+"""Cocotb interfaces for XLS channels using data, valid, and ready signals."""
+
 from typing import Any, Sequence, Type, Union
 
 import cocotb
@@ -30,6 +32,7 @@
 
 
 class XLSChannel(Bus):
+  """Represents an XLS- channel with ready/valid handshake."""
   _signals = XLS_CHANNEL_SIGNALS
   _optional_signals = XLS_CHANNEL_OPTIONAL_SIGNALS
 
@@ -41,7 +44,7 @@ def __init__(self, entity, name, clk, *, start_now=False, **kwargs: Any):
 
   @cocotb.coroutine
   async def recv_channel(self):
-    """Cocotb coroutine that acts as a proc receiving data from a channel"""
+    """Cocotb coroutine that acts as a proc receiving data from a channel."""
     self.rdy.setimmediatevalue(1)
     while True:
       await RisingEdge(self.clk)
@@ -51,6 +54,7 @@ def start_recv_loop(self):
 
 
 class XLSChannelDriver(BusDriver):
+  """Drives transactions on an XLS channel."""
   _signals = XLS_CHANNEL_SIGNALS
   _optional_signals = XLS_CHANNEL_OPTIONAL_SIGNALS
 
@@ -79,6 +83,7 @@ async def _driver_send(self, transaction: Transaction, sync: bool = True, **kwar
 
 
 class XLSChannelMonitor(BusMonitor):
+   """Monitors and decodes transactions on an XLS channel."""
   _signals = XLS_CHANNEL_SIGNALS
   _optional_signals = XLS_CHANNEL_OPTIONAL_SIGNALS
 

xls/modules/zstd/cocotb/data_generator.py

@@ -12,14 +12,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from pathlib import Path
-from enum import Enum
+"""Module for generating and decompressing test frames."""
+
+import pathlib
+import enum
 
 from xls.common import runfiles
 import subprocess
 import zstandard
 
-class BlockType(Enum):
+class BlockType(enum.Enum):
   RAW = 0
   RLE = 1
   COMPRESSED = 2
@@ -33,10 +35,10 @@ def from_string(s):
     try:
       return BlockType[s]
     except KeyError as e:
-      raise ValueError(str(e))
+      raise ValueError(str(e)) from e
 
 def CallDecodecorpus(args):
-  decodecorpus = Path(runfiles.get_path("decodecorpus", repository = "zstd"))
+  decodecorpus = pathlib.Path(runfiles.get_path("decodecorpus", repository = "zstd"))
   cmd = args
   cmd.insert(0, str(decodecorpus))
   cmd_concat = " ".join(cmd)

xls/modules/zstd/cocotb/memory.py

@@ -12,32 +12,37 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+"""
+Extensions of the cocotb AXI RAM models that initialize
+memory contents from a binary file.
+"""
+
 import os
 
 from cocotbext.axi.axi_ram import AxiRam, AxiRamRead, AxiRamWrite
 from cocotbext.axi.sparse_memory import SparseMemory
 
 
-def init_axi_mem(path: os.PathLike, kwargs):
+def init_axi_mem(path: os.PathLike[str], kwargs):
   with open(path, "rb") as f:
     sparse_mem = SparseMemory(size=kwargs["size"])
     sparse_mem.write(0x0, f.read())
     kwargs["mem"] = sparse_mem
 
 
 class AxiRamReadFromFile(AxiRamRead):
-  def __init__(self, *args, path: os.PathLike, **kwargs):
+  def __init__(self, *args, path: os.PathLike[str], **kwargs):
     init_axi_mem(path, kwargs)
     super().__init__(*args, **kwargs)
 
 
 class AxiRamFromFile(AxiRam):
-  def __init__(self, *args, path: os.PathLike, **kwargs):
+  def __init__(self, *args, path: os.PathLike[str], **kwargs):
     init_axi_mem(path, kwargs)
     super().__init__(*args, **kwargs)
 
 
 class AxiRamWriteFromFile(AxiRamWrite):
-  def __init__(self, *args, path: os.PathLike, **kwargs):
+  def __init__(self, *args, path: os.PathLike[str], **kwargs):
     init_axi_mem(path, kwargs)
     super().__init__(*args, **kwargs)

xls/modules/zstd/cocotb/scoreboard.py

@@ -12,8 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from dataclasses import dataclass
-from queue import Queue
+"""Scoreboard creator for measuring request-response latency.
+
+Tracks transactions from channel monitors, computes cycle-based latency,
+and logs results.
+"""
+
+import dataclasses
+import queue
 
 from cocotb.clock import Clock
 from cocotb.log import SimLog
@@ -23,20 +29,26 @@
 from xls.modules.zstd.cocotb.xlsstruct import XLSStruct
 
 
-@dataclass
+@dataclasses.dataclass
 class LatencyQueueItem:
+  """
+  Data container for items in a latency queue.
+  """
+
   transaction: XLSStruct
   timestamp: int
 
 
 class LatencyScoreboard:
+  """Tracks and reports latency between request and response transactions."""
+
   def __init__(self, dut, clock: Clock, req_monitor: XLSChannelMonitor, resp_monitor: XLSChannelMonitor):
     self.dut = dut
     self.log = SimLog(f"zstd.cocotb.scoreboard.{self.dut._name}")
     self.clock = clock
     self.req_monitor = req_monitor
     self.resp_monitor = resp_monitor
-    self.pending_req = Queue()
+    self.pending_req = queue.Queue()
     self.results = []
 
     self.req_monitor.add_callback(self._req_callback)
@@ -48,19 +60,20 @@ def _current_cycle(self):
   def _req_callback(self, transaction: XLSStruct):
     self.pending_req.put(LatencyQueueItem(transaction, self._current_cycle()))
 
-  def _resp_callback(self, transaction: XLSStruct):
+  def _resp_callback(self, _transaction: XLSStruct):
     latency_item = self.pending_req.get()
     self.results.append(self._current_cycle() - latency_item.timestamp)
 
   def average_latency(self):
     return sum(self.results)/len(self.results)
 
   def report_result(self):
+    """Logs a summary of latency measurements and any unfulfilled requests."""
     if not self.pending_req.empty():
       self.log.warning(f"There are unfulfilled requests from channel {self.req_monitor.name}")
       while not self.pending_req.empty():
         self.log.warning(f"Unfulfilled request: {self.pending_req.get()}")
-    if len(self.results) > 0:
+    if self.results:
       self.log.info(f"Latency report - 1st latency: {self.results[0]}")
     if len(self.results) > 1:
       self.log.info(f"Latency report - 2nd latency: {self.results[1]}")

xls/modules/zstd/cocotb/utils.py

@@ -12,25 +12,29 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+"""Helpers for setting up and running simulations using Icarus Verilog with Cocotb."""
+
 import os
-from pathlib import Path
+import pathlib
 
 import cocotb
-from cocotb.runner import check_results_file, get_runner
+from cocotb.runner import check_results_file
+from cocotb.runner import get_runner
 from cocotb.triggers import ClockCycles
 
 from xls.common import runfiles
 
 
 def setup_com_iverilog():
-  iverilog_path = Path(runfiles.get_path("iverilog", repository = "com_icarus_iverilog"))
-  vvp_path = Path(runfiles.get_path("vvp", repository = "com_icarus_iverilog"))
+  iverilog_path = pathlib.Path(runfiles.get_path("iverilog", repository = "com_icarus_iverilog"))
+  vvp_path = pathlib.Path(runfiles.get_path("vvp", repository = "com_icarus_iverilog"))
   os.environ["PATH"] += os.pathsep + str(iverilog_path.parent)
   os.environ["PATH"] += os.pathsep + str(vvp_path.parent)
-  build_dir = Path(os.environ['BUILD_WORKING_DIRECTORY'], "sim_build")
+  build_dir = pathlib.Path(os.environ['BUILD_WORKING_DIRECTORY'], "sim_build")
   return build_dir
 
 def run_test(toplevel, test_module, verilog_sources):
+  """Builds and runs a Cocotb testbench using Icarus Verilog."""
   build_dir = setup_com_iverilog()
   runner = get_runner("icarus")
   runner.build(
@@ -51,7 +55,7 @@ def run_test(toplevel, test_module, verilog_sources):
 
 @cocotb.coroutine
 async def reset(clk, rst, cycles=1):
-  """Cocotb coroutine that performs the reset"""
+  """Cocotb coroutine that performs the reset."""
   rst.value = 1
   await ClockCycles(clk, cycles)
   rst.value = 0

xls/modules/zstd/cocotb/xlsstruct.py

@@ -12,8 +12,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+"""
+Provides utilities for defining Python representations of XLS structs.
+"""
+
 import random
-from dataclasses import asdict, astuple, dataclass, fields
+from dataclasses import asdict
+from dataclasses import astuple
+from dataclasses import dataclass
+from dataclasses import fields
 
 from cocotb.binary import BinaryValue
 
@@ -22,13 +29,18 @@ class TruncationError(Exception):
   pass
 
 def xls_dataclass(cls):
-  """
-  Class decorator for XLS structs.
+  """Class decorator for XLS structs.
+
   Usage:
 
-  @xls_dataclass
-  class MyStruct(XLSStruct):
-    ...
+    @xls_dataclass
+    class MyStruct(XLSStruct):
+      ...
+  Args:
+    cls (type): The class to decorate.
+
+  Returns:
+    type: The dataclass-decorated class with repr disabled.
   """
   return dataclass(cls, repr=False)
 
@@ -70,8 +82,7 @@ class MyStruct(XLSStruct):
 
   @classmethod
   def _masks(cls):
-    """
-    Returns a list of field-sized bitmasks.
+    """Returns a list of field-sized bitmasks.
 
     For example for fields of widths 2, 3, 4
     returns [2'b11, 3'b111, 4'b1111].
@@ -84,9 +95,7 @@ def _masks(cls):
 
   @classmethod
   def _positions(cls):
-    """
-    Returns a list of start positions in a bit vector for
-    struct's fields.
+    """Returns a list of start positions in a bit vector for struct's fields.
 
     For example for fields of widths 1, 2, 3, 4, 5, 6
     returns [20, 18, 15, 11, 6, 0]
@@ -103,16 +112,12 @@ def _positions(cls):
   @classmethod
   @property
   def total_width(cls):
-    """
-    Returns total bit width of the struct
-    """
+    """Returns total bit width of the struct."""
     return sum(field.type for field in fields(cls))
 
   @property
   def value(self):
-    """
-    Returns struct's value as a Python integer
-    """
+    """Returns struct's value as a Python integer."""
     value = 0
     masks = self._masks()
     positions = self._positions()
@@ -124,31 +129,25 @@ def value(self):
 
   @property
   def binaryvalue(self):
-    """
-    Returns struct's value as a cocotb.binary.BinaryValue
-    """
+    """Returns struct's value as a cocotb.binary.BinaryValue."""
     return BinaryValue(self.binstr)
 
   @property
   def binstr(self):
-    """
-    Returns struct's value as a string with its binary representation
-    """
+    """Returns struct's value as a string with its binary representation."""
     return f"{self.value:>0{self.total_width}b}"
 
   @property
   def hexstr(self):
-    """
-    Returns struct's value as a string with its hex representation
-    (without leading "0x")
+    """Returns struct's value as a string with its hex representation.
+
+    The result does NOT have leading "0x".
     """
     return f"{self.value:>0{self.total_width // 4}x}"
 
   @classmethod
   def from_int(cls, value):
-    """
-    Returns an instance of the struct from Python integer
-    """
+    """Returns an instance of the struct from Python integer."""
     instance = {}
     masks = cls._masks()
     positions = cls._positions()
@@ -158,9 +157,7 @@ def from_int(cls, value):
 
   @classmethod
   def randomize(cls):
-    """
-    Returns an instance of the struct with all fields' values randomized
-    """
+    """Returns an instance of the struct with all fields' values randomized."""
     instance = {}
     for field in fields(cls):
       instance[field.name] = random.randrange(0, 2**field.type)
@@ -171,5 +168,5 @@ def __str__(self):
 
   def __repr__(self):
     classname = self.__class__.__name__
-    fields = [f"{name}={hex(value)}" for name, value in asdict(self).items()]
-    return f"{classname}({', '.join(fields)})"
+    class_fields = [f"{name}={hex(value)}" for name, value in asdict(self).items()]
+    return f"{classname}({', '.join(class_fields)})"