Cleanup source code

- enables lint check for unused module imports
- removes unused code
- moves test specific code to the test itself
- fixes typos and spelling mistakes

Signed-off-by: Robert Szczepanski <[email protected]>

.flake8

@@ -1,7 +1,7 @@
 ; Copyright (c) 2023-2024 Antmicro <www.antmicro.com>
 ; SPDX-License-Identifier: Apache-2.0
 [flake8]
-ignore = E203, E266, E501, W503, F403, F401, F405
+ignore = E203, E266, E501, W503, F403, F405
 max-line-length = 100
 max-complexity = 27
 select = B,C,E,F,W,T4,B9

tests/tests_build/test_interconnect.py

@@ -1,7 +1,6 @@
 # Copyright (C) 2023 Antmicro
 # SPDX-License-Identifier: Apache-2.0
 
-import os
 import re
 import shutil
 import subprocess

tests/tests_build/test_interface.py

@@ -1,7 +1,54 @@
 # Copyright (c) 2021-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
+
 from pytest import raises
 
+from topwrap.interface import (
+    get_interface_by_name,
+    get_interface_by_prefix,
+    interface_definitions,
+)
+
+
+def match_interface(ports_matches):
+    """
+    :param ports_matches: a list of pairs (port_name, signal_name), where
+    signal_name is compatible with interface definition.
+    The list must belong to a single instance of the interface
+    :return: interface name if the list matches an interface correctly
+    """
+
+    g_prefix = ports_matches[0][1].split("_")[0]
+    g_instance = ports_matches[0][1].split("_")[1]
+    iface = get_interface_by_prefix(g_prefix)
+
+    for _, signal_name in ports_matches:
+        prefix = signal_name.split("_")[0]
+        instance = signal_name.split("_")[1]
+        signal = signal_name.split("_")[2]
+
+        # every port must belong to the same interface
+        if prefix != iface.prefix:
+            raise ValueError(
+                f"signal prefix: {prefix} does not match"
+                f"the prefix of this interface: {iface.prefix}"
+            )
+        # every port must belong to the same instance
+        if instance != g_instance:
+            raise ValueError(
+                f"interface instance number: {instance} "
+                "does not match with other ports of "
+                f"this interface : {g_instance}"
+            )
+
+        if signal not in iface.signals["required"] + iface.signals["optional"]:
+            raise ValueError(
+                f"signal name: {signal_name} does not match any "
+                f"signal in interface definition: {iface.name}"
+            )
+
+    return {"name": iface.name, "ports": ports_matches}
+
 
 class TestInterfaceDef:
     def test_interfaces_presence(self):
@@ -12,52 +59,41 @@ def test_interfaces_presence(self):
         assert interfaces
 
     def test_predefined(self):
-        from topwrap import interface
-
-        assert interface.interface_definitions, "No predefined interfaces " "could be retrieved"
+        assert interface_definitions, "No predefined interfaces " "could be retrieved"
 
     def test_iface_retrieve_by_name(self):
-        from topwrap import interface
-
         name = "AXI4Stream"
-        assert interface.get_interface_by_name(name)
+        assert get_interface_by_name(name)
 
     def test_iface_retrieve_by_name_negative(self):
-        from topwrap import interface
-
         name = "zxcvbnm"
-        assert interface.get_interface_by_name(name) is None
+        assert get_interface_by_name(name) is None
 
     def test_iface_retrieve_by_prefix(self):
-        from topwrap import interface
-
         prefix = "AXIS"
-        assert interface.get_interface_by_prefix(prefix)
+        assert get_interface_by_prefix(prefix)
 
     def test_iface_retrieve_by_prefix_negative(self):
-        from topwrap import interface
-
         prefix = "zxcvbnm"
-        assert interface.get_interface_by_prefix(prefix) is None
+        assert get_interface_by_prefix(prefix) is None
 
 
 class TestInterfaces:
     def test_iface_match(self):
-        from topwrap import util
-
         ports_correct = (("port1", "AXIS_0_TVALID"), ("port2", "AXIS_0_TREADY"))
         # signal name does not belong to the interface
         ports_incorrect = (("port1", "AXIS_0_TVALID"), ("port2", "AXIS_0_000000"))
         # interface prefixes don't match
         ports_incorrect2 = (("port1", "AXIS_0_TVALID"), ("port2", "AXI_0_TREADY"))
         # interface instances don't match
         ports_incorrect3 = (("port1", "AXIS_0_TVALID"), ("port2", "AXIS_1_TREADY"))
-        correct = util.match_interface(ports_correct)
+        correct = match_interface(ports_correct)
+
         assert correct["name"] == "AXI4Stream"
         assert correct["ports"]
         with raises(ValueError):
-            util.match_interface(ports_incorrect)
+            match_interface(ports_incorrect)
         with raises(ValueError):
-            util.match_interface(ports_incorrect2)
+            match_interface(ports_incorrect2)
         with raises(ValueError):
-            util.match_interface(ports_incorrect3)
+            match_interface(ports_incorrect3)

tests/tests_build/test_ip_connect.py

@@ -8,7 +8,6 @@
 import pytest
 from amaranth.hdl import ast
 
-from topwrap.amaranth_helpers import DIR_IN, DIR_INOUT, DIR_OUT
 from topwrap.ipconnect import IPConnect
 from topwrap.ipwrapper import IPWrapper
 

tests/tests_kpm/conftest.py

@@ -1,9 +1,6 @@
 # Copyright (c) 2023-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
 
-import json
-from pathlib import Path
-
 import pytest
 from common import read_json_file
 from yaml import Loader, load

tests/tests_kpm/test_kpm_export.py

@@ -1,8 +1,6 @@
 # Copyright (c) 2023-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
 
-import os
-
 from topwrap.kpm_dataflow_parser import (
     _kpm_connections_to_external,
     _kpm_connections_to_ports_ifaces,

topwrap/cli.py

@@ -5,12 +5,9 @@
 import logging
 import os
 import subprocess
-import sys
 from pathlib import Path
 
 import click
-from pipeline_manager.scripts.build import script_build
-from pipeline_manager.scripts.run import script_run
 
 from .config import config
 from .design import build_design_from_yaml
@@ -77,7 +74,7 @@ def build_main(sources, design, build_dir, part, iface_compliance, log_level):
     "--use-yosys",
     default=False,
     is_flag=True,
-    help="Use yosys's read_verilog_feature to parse Verilog files",
+    help="Use Yosys command `read_verilog` to parse Verilog files",
 )
 @click.option(
     "--iface-deduce",

topwrap/ipconnect.py

@@ -1,27 +1,16 @@
 # Copyright (c) 2021-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
 from collections import defaultdict
-from logging import error, info, warning
-from os import makedirs, path
+from logging import info, warning
+from os import path
 
-from amaranth import Elaboratable, Fragment, Instance, Module, Signal
+from amaranth import Fragment, Module
 from amaranth.back import verilog
 from amaranth.build import Platform
 from amaranth.hdl.ast import Const
-from soc_generator.gen.wishbone_interconnect import WishboneRRInterconnect
-
-from .amaranth_helpers import (
-    DIR_IN,
-    DIR_INOUT,
-    DIR_OUT,
-    PortDirection,
-    WrapperPort,
-    port_direction_to_prefix,
-    strip_port_prefix,
-)
-from .elaboratable_wrapper import ElaboratableWrapper
+
+from .amaranth_helpers import DIR_IN, DIR_INOUT, DIR_OUT, WrapperPort
 from .fuse_helper import FuseSocBuilder
-from .ipwrapper import IPWrapper
 from .util import removeprefix
 from .wrapper import Wrapper
 

topwrap/ipwrapper.py

@@ -220,7 +220,7 @@ def elaborate(self, platform: Platform) -> Module:
             # fill the empty slices in the list with Signals
             # the list is iterated starting from last position
             # in order not to change indices
-            # or mess up when en alement is inserted
+            # or mess up when en element is inserted
             for i in range(len(ports_sorted) - 2, -1, -1):
                 port1 = ports_sorted[i]
                 port2 = ports_sorted[i + 1]

topwrap/kpm_dataflow_parser.py

@@ -4,7 +4,6 @@
 import re
 
 from .kpm_common import (
-    EXT_INOUT_NAME,
     EXT_INPUT_NAME,
     EXT_OUTPUT_NAME,
     find_dataflow_interface_by_id,

topwrap/kpm_dataflow_validator.py

@@ -1,8 +1,7 @@
 # Copyright (c) 2023-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
-import json
+
 import re
-from enum import Enum
 from typing import NamedTuple, Union
 
 import numexpr as ex
@@ -101,7 +100,7 @@ def _check_ext_in_to_ext_out_connections(dataflow_data, specification) -> CheckR
 
 def _check_ambigous_ports(dataflow_data, specification) -> CheckResult:
     """Check for ports which are connected to another ipcore port
-    and to external meanode at the same time
+    and to external metanode at the same time
     """
     ext_ifaces_ids = get_dataflow_externals_interfaces(dataflow_data).keys()
 

topwrap/kpm_topwrap_client.py

@@ -2,15 +2,14 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging
-import os
 from base64 import b64encode
 from datetime import datetime
 
 import yaml
 from pipeline_manager_backend_communication.communication_backend import (
     CommunicationBackend,
 )
-from pipeline_manager_backend_communication.misc_structures import MessageType, Status
+from pipeline_manager_backend_communication.misc_structures import MessageType
 from pipeline_manager_backend_communication.utils import convert_message_to_string
 
 from .design import build_design
@@ -101,8 +100,6 @@ def dataflow_validate(self, dataflow: dict) -> dict:
             else:
                 return {"type": MessageType.OK.value, "content": "Design is valid"}
 
-            return {"type": MessageType.OK.value}
-
         def dataflow_run(self, dataflow: dict) -> dict:
             logging.info(f"Dataflow run request received from {host}:{port}")
             errors = _kpm_run_handler(dataflow, yamlfiles, build_dir)

topwrap/parsers.py

@@ -42,7 +42,7 @@ def _default_bounds(signal):
 
 
 def parse_interface_definitions(dir_name=DIR):
-    """Parseinterfaces described in YAML files, bundled with the package
+    """Parse interfaces described in YAML files, bundled with the package
 
     :param dir_name: directory that contains YAML files for interfaces
     :raises OSError: when dir_name directory cannot be listed

topwrap/util.py

@@ -1,6 +1,5 @@
 # Copyright (c) 2021-2024 Antmicro <www.antmicro.com>
 # SPDX-License-Identifier: Apache-2.0
-from .interface import get_interface_by_prefix
 
 
 def check_interface_compliance(iface_def, signals):
@@ -20,47 +19,6 @@ def check_interface_compliance(iface_def, signals):
     return True
 
 
-# This might be no longer needed
-def match_interface(ports_matches):
-    """
-    :param ports_matches: a list of pairs (port_name, signal_name), where
-    signal_name is compatible with interface definition.
-    The list must belong to a single instance of the interface
-    :return: interface name if the list matches an interface correctly
-    """
-
-    g_prefix = ports_matches[0][1].split("_")[0]
-    g_instance = ports_matches[0][1].split("_")[1]
-    iface = get_interface_by_prefix(g_prefix)
-
-    for _, signal_name in ports_matches:
-        prefix = signal_name.split("_")[0]
-        instance = signal_name.split("_")[1]
-        signal = signal_name.split("_")[2]
-
-        # every port must belong to the same interface
-        if prefix != iface.prefix:
-            raise ValueError(
-                f"signal prefix: {prefix} does not match"
-                f"the prefix of this interface: {iface.prefix}"
-            )
-        # every port must belong to the same instance
-        if instance != g_instance:
-            raise ValueError(
-                f"interface instance number: {instance} "
-                "does not match with other ports of "
-                f"this interface : {g_instance}"
-            )
-
-        if signal not in iface.signals["required"] + iface.signals["optional"]:
-            raise ValueError(
-                f"signal name: {signal_name} does not match any "
-                f"signal in interface definition: {iface.name}"
-            )
-
-    return {"name": iface.name, "ports": ports_matches}
-
-
 def removeprefix(s: str, prefix: str):
     """Return string with a prefix removed if it contains it
 

topwrap/yamls_to_kpm_spec_parser.py

@@ -3,7 +3,6 @@
 
 import logging
 import os
-from itertools import cycle
 
 from .kpm_common import EXT_INOUT_NAME, EXT_INPUT_NAME, EXT_OUTPUT_NAME
 from .parsers import parse_port_map