From 426873c22d370fab678fca7d7f411df987cd1645 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Krzysztof=20Ob=C5=82onczek?= Date: Mon, 6 Nov 2023 16:25:50 +0100 Subject: [PATCH] Run black and isort on sources MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Internal-tag: [#50994] Signed-off-by: Krzysztof Obłonczek --- fpga_topwrap/amaranth_helpers.py | 4 +- fpga_topwrap/design.py | 12 +- fpga_topwrap/design_to_kpm_dataflow_parser.py | 15 ++- fpga_topwrap/ipconnect.py | 22 +-- fpga_topwrap/ipwrapper.py | 8 +- fpga_topwrap/kpm_dataflow_parser.py | 10 +- fpga_topwrap/kpm_dataflow_validator.py | 5 +- fpga_topwrap/yamls_to_kpm_spec_parser.py | 11 +- tests/tests_build/test_hier_design.py | 24 +++- tests/tests_build/test_ip_connect.py | 92 +++++++++---- tests/tests_build/test_ip_wrapper.py | 125 ++++++++++++++---- tests/tests_kpm/conftest.py | 1 + tests/tests_kpm/test_kpm_specification.py | 33 +++-- tests/tests_kpm/test_kpm_validation.py | 12 +- tests/tests_parse/test_parse.py | 8 +- 15 files changed, 262 insertions(+), 120 deletions(-) diff --git a/fpga_topwrap/amaranth_helpers.py b/fpga_topwrap/amaranth_helpers.py index c5141793..d94bad95 100644 --- a/fpga_topwrap/amaranth_helpers.py +++ b/fpga_topwrap/amaranth_helpers.py @@ -1,10 +1,10 @@ # Copyright (C) 2021 Antmicro # SPDX-License-Identifier: Apache-2.0 -from amaranth import Signal from enum import Enum +from amaranth import Signal -PortDirection = Enum('PortDirection', ('INOUT', 'OUT', 'IN')) +PortDirection = Enum("PortDirection", ("INOUT", "OUT", "IN")) DIR_INOUT = PortDirection.INOUT DIR_OUT = PortDirection.OUT diff --git a/fpga_topwrap/design.py b/fpga_topwrap/design.py index 4660cbd0..8aab9a16 100644 --- a/fpga_topwrap/design.py +++ b/fpga_topwrap/design.py @@ -16,16 +16,18 @@ def build_design_from_yaml(yamlfile, sources_dir=None, part=None): def get_hierarchies_names(design_descr: dict) -> set: - """ `design_descr` is the "design" section of a design description yaml.""" + """`design_descr` is the "design" section of a design description yaml.""" if "hierarchies" not in design_descr.keys(): return set() return set(design_descr["hierarchies"].keys()) def get_ipcores_names(design_descr: dict) -> set: - """ `design_descr` is the "design" section of a design description yaml.""" + """`design_descr` is the "design" section of a design description yaml.""" design_ports = design_descr["ports"] if "ports" in design_descr.keys() else dict() - design_interfaces = design_descr["interfaces"] if "interfaces" in design_descr.keys() else dict() + design_interfaces = ( + design_descr["interfaces"] if "interfaces" in design_descr.keys() else dict() + ) ports_keys = set(design_ports.keys()) interfaces_keys = set(design_interfaces.keys()) # IP core should be added to the design if its name occurs as a key in "ports" or @@ -42,7 +44,9 @@ def generate_design(ips: dict, design: dict, external: dict) -> IPConnect: # Generate hierarchies and add them to `ipc`. for hier_name in get_hierarchies_names(design): - hier_ipc = generate_design(ips, ipc_hiers[hier_name]["design"], ipc_hiers[hier_name]["external"]) + hier_ipc = generate_design( + ips, ipc_hiers[hier_name]["design"], ipc_hiers[hier_name]["external"] + ) ipc.add_component(hier_name, HierarchyWrapper(hier_name, hier_ipc)) for ip_name in get_ipcores_names(design): diff --git a/fpga_topwrap/design_to_kpm_dataflow_parser.py b/fpga_topwrap/design_to_kpm_dataflow_parser.py index 793a3c4e..10a21486 100644 --- a/fpga_topwrap/design_to_kpm_dataflow_parser.py +++ b/fpga_topwrap/design_to_kpm_dataflow_parser.py @@ -8,13 +8,13 @@ from time import time from typing import List +from .design import get_hierarchies_names, get_ipcores_names from .kpm_common import ( EXT_INOUT_NAME, EXT_INPUT_NAME, EXT_OUTPUT_NAME, get_metanode_property_value, ) -from .design import get_hierarchies_names, get_ipcores_names class IDGenerator(object): @@ -180,9 +180,11 @@ def kpm_nodes_from_design_descr(design_descr: dict, specification: dict) -> List hier_names = get_hierarchies_names(design) if hier_names: - logging.warning(f"Imported design contains hierarchies ({hier_names}) which are not yet " - "supported. The imported design will be incomplete") - + logging.warning( + f"Imported design contains hierarchies ({hier_names}) which are not yet " + "supported. The imported design will be incomplete" + ) + for ip_name in get_ipcores_names(design): ip_type = os.path.splitext(os.path.basename(ips[ip_name]["file"]))[0] spec_node = _get_specification_node_by_type(ip_type, specification) @@ -269,7 +271,8 @@ def _get_external_connections(design_descr: dict) -> list: "port_iface_name": conn["port_iface_name"], "external_name": conn["connection"], } - for conn in ext_connections if conn["ip_name"] not in get_hierarchies_names(design_descr["design"]) + for conn in ext_connections + if conn["ip_name"] not in get_hierarchies_names(design_descr["design"]) # skip external connections from/to hierarchies ] @@ -289,7 +292,7 @@ def _is_ipcore_connection(conn_descr: dict) -> bool: return False if conn_descr["connection"][0] in get_hierarchies_names(design_descr["design"]): return False - return True + return True ipcores_connections = list( filter(_is_ipcore_connection, _get_flattened_connections(design_descr)) diff --git a/fpga_topwrap/ipconnect.py b/fpga_topwrap/ipconnect.py index 39c792a6..2ca6dba7 100644 --- a/fpga_topwrap/ipconnect.py +++ b/fpga_topwrap/ipconnect.py @@ -8,18 +8,18 @@ from amaranth.build import Platform from amaranth.hdl.ast import Const -from .fuse_helper import FuseSocBuilder -from .hierarchy_wrapper import HierarchyWrapper -from .ipwrapper import IPWrapper from .amaranth_helpers import ( - port_direction_to_prefix, - strip_port_prefix, DIR_IN, - DIR_OUT, DIR_INOUT, + DIR_OUT, PortDirection, - WrapperPort + WrapperPort, + port_direction_to_prefix, + strip_port_prefix, ) +from .fuse_helper import FuseSocBuilder +from .hierarchy_wrapper import HierarchyWrapper +from .ipwrapper import IPWrapper class IPConnect(Elaboratable): @@ -139,7 +139,9 @@ def connect_interfaces( f"{comp1_name}:{iface1} - {comp2_name}:{iface2}" ) - def _set_port(self, comp_name: str, port_name: str, external_name: str, external_dir: PortDirection) -> None: + def _set_port( + self, comp_name: str, port_name: str, external_name: str, external_dir: PortDirection + ) -> None: """Set port specified by name as an external port :param comp_name: name of the component - hierarchy or IP core @@ -218,7 +220,9 @@ def get_ports(self) -> list: """Return a list of external ports of this module""" return self._ports - def _set_unconnected_port(self, comp_name: str, port_name: str, iface_name: str, external_dir: PortDirection) -> None: + def _set_unconnected_port( + self, comp_name: str, port_name: str, iface_name: str, external_dir: PortDirection + ) -> None: """Create signal for unconnected port to allow using it as external. This is essential since ports that haven't been used have no signals assigned to them. diff --git a/fpga_topwrap/ipwrapper.py b/fpga_topwrap/ipwrapper.py index c477e227..d7bd995b 100644 --- a/fpga_topwrap/ipwrapper.py +++ b/fpga_topwrap/ipwrapper.py @@ -10,9 +10,15 @@ from amaranth.build import Platform from amaranth.hdl.ast import Cat, Const +from .amaranth_helpers import ( + DIR_IN, + DIR_INOUT, + DIR_OUT, + WrapperPort, + port_direction_to_prefix, +) from .config import config from .interface import get_interface_by_name -from .amaranth_helpers import WrapperPort, port_direction_to_prefix, DIR_IN, DIR_OUT, DIR_INOUT from .parsers import parse_port_map from .util import check_interface_compliance diff --git a/fpga_topwrap/kpm_dataflow_parser.py b/fpga_topwrap/kpm_dataflow_parser.py index 4acbad98..af936b38 100644 --- a/fpga_topwrap/kpm_dataflow_parser.py +++ b/fpga_topwrap/kpm_dataflow_parser.py @@ -7,7 +7,6 @@ EXT_INOUT_NAME, EXT_INPUT_NAME, EXT_OUTPUT_NAME, - is_external_metanode, find_dataflow_interface_by_id, find_dataflow_node_by_interface_id, find_dataflow_node_type_by_name, @@ -16,6 +15,7 @@ get_dataflow_ip_connections, get_dataflow_ip_nodes, get_metanode_property_value, + is_external_metanode, ) @@ -201,10 +201,6 @@ def kpm_dataflow_to_design(dataflow_data, specification) -> dict: return { "ips": ips, - "design": { - "parameters": properties, - "ports": ports, - "interfaces": interfaces - }, - "external": externals["external"] + "design": {"parameters": properties, "ports": ports, "interfaces": interfaces}, + "external": externals["external"], } diff --git a/fpga_topwrap/kpm_dataflow_validator.py b/fpga_topwrap/kpm_dataflow_validator.py index 7bc55700..7a4096a3 100644 --- a/fpga_topwrap/kpm_dataflow_validator.py +++ b/fpga_topwrap/kpm_dataflow_validator.py @@ -246,8 +246,9 @@ def _check_inouts_connections(dataflow_data, specification) -> CheckResult: if connected_inouts: return CheckResult( - MessageType.WARNING, f"Wires connecting inout ports {connected_inouts} are always " - "external in the top module by Amaranth" + MessageType.WARNING, + f"Wires connecting inout ports {connected_inouts} are always " + "external in the top module by Amaranth", ) return CheckResult(MessageType.OK, None) diff --git a/fpga_topwrap/yamls_to_kpm_spec_parser.py b/fpga_topwrap/yamls_to_kpm_spec_parser.py index e5267230..d263e4c7 100644 --- a/fpga_topwrap/yamls_to_kpm_spec_parser.py +++ b/fpga_topwrap/yamls_to_kpm_spec_parser.py @@ -64,10 +64,7 @@ def _ipcore_ports_to_kpm(ports: dict) -> list: } for port in ports["in"] ] - outputs = [ - {"name": port[0], "type": ["port"], "direction": "output"} - for port in ports["out"] - ] + outputs = [{"name": port[0], "type": ["port"], "direction": "output"} for port in ports["out"]] inouts = [ {"name": port[0], "type": ["port"], "direction": "inout", "side": "right"} for port in ports["inout"] @@ -215,15 +212,15 @@ def _generate_ifaces_styling(interfaces_types: list) -> dict: def _get_ifaces_types(specification: dict) -> list: - """Return a list of all interfaces types from specification that are interfaces types. - """ + """Return a list of all interfaces types from specification that are interfaces types.""" return list( set( [ iface_type for node in specification["nodes"] for iface in node["interfaces"] - for iface_type in iface["type"] if iface_type != "port" + for iface_type in iface["type"] + if iface_type != "port" ] ) ) diff --git a/tests/tests_build/test_hier_design.py b/tests/tests_build/test_hier_design.py index 498619f7..d0b49ee9 100644 --- a/tests/tests_build/test_hier_design.py +++ b/tests/tests_build/test_hier_design.py @@ -3,42 +3,49 @@ # Copyright (C) 2023 Antmicro # SPDX-License-Identifier: Apache-2.0 +from pathlib import Path + import pytest +from yaml import Loader, load -from yaml import load, Loader -from pathlib import Path from fpga_topwrap.design import generate_design +from fpga_topwrap.hierarchy_wrapper import HierarchyWrapper from fpga_topwrap.ipconnect import IPConnect from fpga_topwrap.ipwrapper import IPWrapper -from fpga_topwrap.hierarchy_wrapper import HierarchyWrapper @pytest.fixture def hier_design_yaml() -> Path: return Path("tests/data/data_build/hierarchy/design.yml") + @pytest.fixture def hier_design(hier_design_yaml) -> dict: with open(hier_design_yaml, "r") as f: design = load(f, Loader=Loader) return design + @pytest.fixture def counter_hier_name() -> str: return "counter_hier" + @pytest.fixture def pwm_mod_name() -> str: return "pwm" + @pytest.fixture def counter_mod_name() -> str: return "counter" + @pytest.fixture def counter_hier_conns() -> list: return ["in_en_sig_pwm", "top_clk", "top_cnt"] + @pytest.fixture def hier_design_ipconnect(hier_design) -> IPConnect: return generate_design(hier_design["ips"], hier_design["design"], hier_design["external"]) @@ -48,7 +55,10 @@ class TestHierarchyDesign: """Check whether the generated structure consisting of `IPConnect`, `HierarchyWrapper` and `IPWrapper` objects is correct for the test design. """ - def test_hierarchy_structure(self, hier_design_ipconnect, counter_hier_name, pwm_mod_name, counter_mod_name): + + def test_hierarchy_structure( + self, hier_design_ipconnect, counter_hier_name, pwm_mod_name, counter_mod_name + ): # The top IPConnect has two components - `pwm` module and `counter_hier` hierarchy pwm_mod = hier_design_ipconnect._get_component_by_name(pwm_mod_name) counter_hier = hier_design_ipconnect._get_component_by_name(counter_hier_name) @@ -61,8 +71,10 @@ def test_hierarchy_structure(self, hier_design_ipconnect, counter_hier_name, pwm counter_mod = counter_hier.ipc._get_component_by_name(counter_mod_name) assert isinstance(counter_mod, IPWrapper) - def test_connections_of_hierarchy(self, hier_design_ipconnect, counter_hier_name, counter_hier_conns): - """ Check whether the connections from/to the `counter_hier` hierarchy have been + def test_connections_of_hierarchy( + self, hier_design_ipconnect, counter_hier_name, counter_hier_conns + ): + """Check whether the connections from/to the `counter_hier` hierarchy have been created correctly. We should have: * `in_clk` incoming from the top module * `out_cnt` outgoing as external output of the top module diff --git a/tests/tests_build/test_ip_connect.py b/tests/tests_build/test_ip_connect.py index 0e09e72b..5984d989 100644 --- a/tests/tests_build/test_ip_connect.py +++ b/tests/tests_build/test_ip_connect.py @@ -2,9 +2,11 @@ # Copyright (C) 2021-2023 Antmicro # SPDX-License-Identifier: Apache-2.0 -import pytest from pathlib import Path -from fpga_topwrap.amaranth_helpers import DIR_IN, DIR_OUT, DIR_INOUT + +import pytest + +from fpga_topwrap.amaranth_helpers import DIR_IN, DIR_INOUT, DIR_OUT from fpga_topwrap.ipconnect import IPConnect from fpga_topwrap.ipwrapper import IPWrapper @@ -16,18 +18,22 @@ def dmatop_name() -> str: return "DMATop" + @pytest.fixture def axi_dispctrl_name() -> str: return "axi_dispctrl_v1_0" + @pytest.fixture def dmatop_yaml() -> Path: return Path("tests/data/data_build/DMATop.yaml") + @pytest.fixture def axi_dispctrl_yaml() -> Path: return Path("tests/data/data_build/axi_dispctrl_v1_0.yaml") + # ------------------------------------- # IPWrapper and HierarchyWrapper structures # ------------------------------------- @@ -35,24 +41,46 @@ def axi_dispctrl_yaml() -> Path: def dmatop_ipw(dmatop_yaml, dmatop_name) -> IPWrapper: return IPWrapper(dmatop_yaml, dmatop_name, dmatop_name) + @pytest.fixture def axi_dispctrl_ipw(axi_dispctrl_yaml, axi_dispctrl_name) -> IPWrapper: return IPWrapper(axi_dispctrl_yaml, axi_dispctrl_name, axi_dispctrl_name) + # ------------------------------------- # Fixtures for specific tests # ------------------------------------- @pytest.fixture def dmatop_axism0_ports_names() -> list: - return ['i_AXIS_m0_TREADY', 'o_AXIS_m0_TDATA', 'o_AXIS_m0_TLAST', 'o_AXIS_m0_TUSER', 'o_AXIS_m0_TVALID'] + return [ + "i_AXIS_m0_TREADY", + "o_AXIS_m0_TDATA", + "o_AXIS_m0_TLAST", + "o_AXIS_m0_TUSER", + "o_AXIS_m0_TVALID", + ] + @pytest.fixture def axi_dispctrl_axiss0_ports_names() -> list: - return ['i_AXIS_s0_TDATA', 'i_AXIS_s0_TLAST', 'i_AXIS_s0_TUSER', 'i_AXIS_s0_TVALID', 'o_AXIS_s0_TREADY'] + return [ + "i_AXIS_s0_TDATA", + "i_AXIS_s0_TLAST", + "i_AXIS_s0_TUSER", + "i_AXIS_s0_TVALID", + "o_AXIS_s0_TREADY", + ] + @pytest.fixture def ext_axism0_ports_names() -> list: - return ['ext_axis_TDATA', 'ext_axis_TLAST', 'ext_axis_TREADY', 'ext_axis_TUSER', 'ext_axis_TVALID'] + return [ + "ext_axis_TDATA", + "ext_axis_TLAST", + "ext_axis_TREADY", + "ext_axis_TUSER", + "ext_axis_TVALID", + ] class TestIPConnect: @@ -63,10 +91,8 @@ def test_add_component(self, dmatop_ipw, dmatop_name): with pytest.raises(ValueError): ipc._get_component_by_name("non_existing_comp_name") - def test_connect_ports(self, dmatop_ipw, dmatop_name, axi_dispctrl_ipw, axi_dispctrl_name): - """Test connecting two IPs ports. - """ + """Test connecting two IPs ports.""" ipc = IPConnect() ipc.add_component(dmatop_name, dmatop_ipw) ipc.add_component(axi_dispctrl_name, axi_dispctrl_ipw) @@ -87,14 +113,27 @@ def test_connect_ports(self, dmatop_ipw, dmatop_name, axi_dispctrl_ipw, axi_disp # Check function behavior on invalid data with pytest.raises(ValueError): - ipc.connect_ports(dmatop_port_name, dmatop_name, "non_existing_port_name", axi_dispctrl_name) + ipc.connect_ports( + dmatop_port_name, dmatop_name, "non_existing_port_name", axi_dispctrl_name + ) with pytest.raises(ValueError): - ipc.connect_ports(dmatop_port_name, "non_existing_comp_name", axi_dispctrl_port_name, axi_dispctrl_name) - - - def test_connect_interfaces(self, dmatop_ipw, dmatop_name, axi_dispctrl_ipw, axi_dispctrl_name, dmatop_axism0_ports_names, axi_dispctrl_axiss0_ports_names): - """Test connecting two IPs interfaces. - """ + ipc.connect_ports( + dmatop_port_name, + "non_existing_comp_name", + axi_dispctrl_port_name, + axi_dispctrl_name, + ) + + def test_connect_interfaces( + self, + dmatop_ipw, + dmatop_name, + axi_dispctrl_ipw, + axi_dispctrl_name, + dmatop_axism0_ports_names, + axi_dispctrl_axiss0_ports_names, + ): + """Test connecting two IPs interfaces.""" ipc = IPConnect() ipc.add_component(dmatop_name, dmatop_ipw) ipc.add_component(axi_dispctrl_name, axi_dispctrl_ipw) @@ -112,14 +151,16 @@ def test_connect_interfaces(self, dmatop_ipw, dmatop_name, axi_dispctrl_ipw, axi # Check function behavior on invalid data with pytest.raises(ValueError): - ipc.connect_interfaces(dmatop_iface, dmatop_name, "non_existing_iface_name", axi_dispctrl_name) + ipc.connect_interfaces( + dmatop_iface, dmatop_name, "non_existing_iface_name", axi_dispctrl_name + ) with pytest.raises(ValueError): - ipc.connect_interfaces(dmatop_iface, "non_existing_comp_name", axi_dispctrl_iface, axi_dispctrl_name) - + ipc.connect_interfaces( + dmatop_iface, "non_existing_comp_name", axi_dispctrl_iface, axi_dispctrl_name + ) def test_set_port(self, dmatop_ipw, dmatop_name, axi_dispctrl_ipw, axi_dispctrl_name): - """Test setting a port as external in the top module - """ + """Test setting a port as external in the top module""" ipc = IPConnect() ipc.add_component(dmatop_name, dmatop_ipw) ipc.add_component(axi_dispctrl_name, axi_dispctrl_ipw) @@ -135,7 +176,9 @@ def test_set_port(self, dmatop_ipw, dmatop_name, axi_dispctrl_ipw, axi_dispctrl_ sig_clk_external = "ext_clk" ipc._set_port(dmatop_name, sig_dmatop_clk, sig_clk_external, DIR_IN) ipc._set_port(axi_dispctrl_name, sig_axi_dispctrl_clk, sig_clk_external, DIR_IN) - assert len(ipc.get_ports()) == 2 and sig_clk_external in [sig.name for sig in ipc.get_ports()] + assert len(ipc.get_ports()) == 2 and sig_clk_external in [ + sig.name for sig in ipc.get_ports() + ] assert sig_clk_external in [sig.name for sig in getattr(ipc, dmatop_name).values()] assert sig_clk_external in [sig.name for sig in getattr(ipc, axi_dispctrl_name).values()] @@ -150,10 +193,8 @@ def test_set_port(self, dmatop_ipw, dmatop_name, axi_dispctrl_ipw, axi_dispctrl_ with pytest.raises(ValueError): ipc._set_port(dmatop_name, "non_existing_port_name", "sig_ext", DIR_IN) - def test_set_interface(self, dmatop_ipw, dmatop_name, ext_axism0_ports_names): - """Test setting an interface as external in the top module - """ + """Test setting an interface as external in the top module""" ipc = IPConnect() ipc.add_component(dmatop_name, dmatop_ipw) @@ -169,8 +210,7 @@ def test_set_interface(self, dmatop_ipw, dmatop_name, ext_axism0_ports_names): ipc._set_interface("non_existing_comp_name", dmatop_iface, ext_iface_name) def test_set_constant(self, dmatop_ipw, dmatop_name): - """Test setting a constant value on a port of a module - """ + """Test setting a constant value on a port of a module""" ipc = IPConnect() ipc.add_component(dmatop_name, dmatop_ipw) diff --git a/tests/tests_build/test_ip_wrapper.py b/tests/tests_build/test_ip_wrapper.py index 16eb213a..72ee6641 100644 --- a/tests/tests_build/test_ip_wrapper.py +++ b/tests/tests_build/test_ip_wrapper.py @@ -2,10 +2,13 @@ # Copyright (C) 2023 Antmicro # SPDX-License-Identifier: Apache-2.0 -import pytest from pathlib import Path + +import pytest + from fpga_topwrap.ipwrapper import IPWrapper + # ----------------------------------------- # IP core names and description files paths # ----------------------------------------- @@ -13,10 +16,12 @@ def axi_dispctrl_name() -> str: return "axi_dispctrl_v1_0" + @pytest.fixture def axi_dispctrl_yaml() -> Path: return Path("tests/data/data_build/axi_dispctrl_v1_0.yaml") + # ----------------- # IPWrapper objects # ----------------- @@ -24,67 +29,131 @@ def axi_dispctrl_yaml() -> Path: def axi_dispctrl_ipw(axi_dispctrl_name, axi_dispctrl_yaml) -> IPWrapper: return IPWrapper(axi_dispctrl_yaml, axi_dispctrl_name, axi_dispctrl_name, {}) + @pytest.fixture def axi_dispctrl_ipw_overriden(axi_dispctrl_name, axi_dispctrl_yaml) -> IPWrapper: return IPWrapper( axi_dispctrl_yaml, axi_dispctrl_name, axi_dispctrl_name, - {"C_S_AXIS_TDATA_WIDTH": 64, "C_S00_AXI_DATA_WIDTH": 64} + {"C_S_AXIS_TDATA_WIDTH": 64, "C_S00_AXI_DATA_WIDTH": 64}, ) + # ---------------------------------------------------------------------------- # Names of ports that should be created while initializing `IPWrapper` objects # ---------------------------------------------------------------------------- @pytest.fixture def axi_dispctrl_port_names_widths() -> list: - return [('S_AXIS_ACLK', 1), ('LOCKED_I', 1), ('s00_axi_aclk', 1), ('s00_axi_aresetn', 1), - ('FSYNC_O', 1), ('HSYNC_O', 1), ('VSYNC_O', 1), ('DE_O', 1), ('DATA_O', 32), - ('CTL_O', 4), ('VGUARD_O', 1), ('DGUARD_O', 1), ('DIEN_O', 1), ('DIH_O', 1), - ('AXI_s0_AWADDR', 7), ('AXI_s0_AWPROT', 3), ('AXI_s0_AWVALID', 1), ('AXI_s0_WDATA', 32), - ('AXI_s0_WSTRB', 4), ('AXI_s0_WVALID', 1), ('AXI_s0_BREADY', 1), ('AXI_s0_ARADDR', 7), - ('AXI_s0_ARPROT', 3), ('AXI_s0_ARVALID', 1), ('AXI_s0_RREADY', 1), ('AXI_s0_AWREADY', 1), - ('AXI_s0_WREADY', 1), ('AXI_s0_BRESP', 2), ('AXI_s0_BVALID', 1), ('AXI_s0_ARREADY', 1), - ('AXI_s0_RDATA', 32), ('AXI_s0_RRESP', 2), ('AXI_s0_RVALID', 1), ('AXIS_s0_TVALID', 1), - ('AXIS_s0_TDATA', 32), ('AXIS_s0_TLAST', 1), ('AXIS_s0_TUSER', 1), ('AXIS_s0_TREADY', 1) - ] + return [ + ("S_AXIS_ACLK", 1), + ("LOCKED_I", 1), + ("s00_axi_aclk", 1), + ("s00_axi_aresetn", 1), + ("FSYNC_O", 1), + ("HSYNC_O", 1), + ("VSYNC_O", 1), + ("DE_O", 1), + ("DATA_O", 32), + ("CTL_O", 4), + ("VGUARD_O", 1), + ("DGUARD_O", 1), + ("DIEN_O", 1), + ("DIH_O", 1), + ("AXI_s0_AWADDR", 7), + ("AXI_s0_AWPROT", 3), + ("AXI_s0_AWVALID", 1), + ("AXI_s0_WDATA", 32), + ("AXI_s0_WSTRB", 4), + ("AXI_s0_WVALID", 1), + ("AXI_s0_BREADY", 1), + ("AXI_s0_ARADDR", 7), + ("AXI_s0_ARPROT", 3), + ("AXI_s0_ARVALID", 1), + ("AXI_s0_RREADY", 1), + ("AXI_s0_AWREADY", 1), + ("AXI_s0_WREADY", 1), + ("AXI_s0_BRESP", 2), + ("AXI_s0_BVALID", 1), + ("AXI_s0_ARREADY", 1), + ("AXI_s0_RDATA", 32), + ("AXI_s0_RRESP", 2), + ("AXI_s0_RVALID", 1), + ("AXIS_s0_TVALID", 1), + ("AXIS_s0_TDATA", 32), + ("AXIS_s0_TLAST", 1), + ("AXIS_s0_TUSER", 1), + ("AXIS_s0_TREADY", 1), + ] + @pytest.fixture def axi_dispctrl_overriden_port_names_widths() -> list: - return [('S_AXIS_ACLK', 1), ('LOCKED_I', 1), ('s00_axi_aclk', 1), ('s00_axi_aresetn', 1), - ('FSYNC_O', 1), ('HSYNC_O', 1), ('VSYNC_O', 1), ('DE_O', 1), ('DATA_O', 64), - ('CTL_O', 4), ('VGUARD_O', 1), ('DGUARD_O', 1), ('DIEN_O', 1), ('DIH_O', 1), - ('AXI_s0_AWADDR', 7), ('AXI_s0_AWPROT', 3), ('AXI_s0_AWVALID', 1), ('AXI_s0_WDATA', 64), - ('AXI_s0_WSTRB', 8), ('AXI_s0_WVALID', 1), ('AXI_s0_BREADY', 1), ('AXI_s0_ARADDR', 7), - ('AXI_s0_ARPROT', 3), ('AXI_s0_ARVALID', 1), ('AXI_s0_RREADY', 1), ('AXI_s0_AWREADY', 1), - ('AXI_s0_WREADY', 1), ('AXI_s0_BRESP', 2), ('AXI_s0_BVALID', 1), ('AXI_s0_ARREADY', 1), - ('AXI_s0_RDATA', 64), ('AXI_s0_RRESP', 2), ('AXI_s0_RVALID', 1), ('AXIS_s0_TVALID', 1), - ('AXIS_s0_TDATA', 64), ('AXIS_s0_TLAST', 1), ('AXIS_s0_TUSER', 1), ('AXIS_s0_TREADY', 1) - ] + return [ + ("S_AXIS_ACLK", 1), + ("LOCKED_I", 1), + ("s00_axi_aclk", 1), + ("s00_axi_aresetn", 1), + ("FSYNC_O", 1), + ("HSYNC_O", 1), + ("VSYNC_O", 1), + ("DE_O", 1), + ("DATA_O", 64), + ("CTL_O", 4), + ("VGUARD_O", 1), + ("DGUARD_O", 1), + ("DIEN_O", 1), + ("DIH_O", 1), + ("AXI_s0_AWADDR", 7), + ("AXI_s0_AWPROT", 3), + ("AXI_s0_AWVALID", 1), + ("AXI_s0_WDATA", 64), + ("AXI_s0_WSTRB", 8), + ("AXI_s0_WVALID", 1), + ("AXI_s0_BREADY", 1), + ("AXI_s0_ARADDR", 7), + ("AXI_s0_ARPROT", 3), + ("AXI_s0_ARVALID", 1), + ("AXI_s0_RREADY", 1), + ("AXI_s0_AWREADY", 1), + ("AXI_s0_WREADY", 1), + ("AXI_s0_BRESP", 2), + ("AXI_s0_BVALID", 1), + ("AXI_s0_ARREADY", 1), + ("AXI_s0_RDATA", 64), + ("AXI_s0_RRESP", 2), + ("AXI_s0_RVALID", 1), + ("AXIS_s0_TVALID", 1), + ("AXIS_s0_TDATA", 64), + ("AXIS_s0_TLAST", 1), + ("AXIS_s0_TUSER", 1), + ("AXIS_s0_TREADY", 1), + ] + @pytest.fixture def axi_dispctrl_interfaces() -> list: - """ Return a list of interfaces and number of ports belonging to them - """ + """Return a list of interfaces and number of ports belonging to them""" return [("AXI_s0", 19), ("AXIS_s0", 5)] class TestIPWrapper: def test_ports_no_override(self, axi_dispctrl_ipw, axi_dispctrl_port_names_widths): - """Check whether all the ports of `IPWrapper` object have been created - """ + """Check whether all the ports of `IPWrapper` object have been created""" for port_name, port_width in axi_dispctrl_port_names_widths: ipw_port = axi_dispctrl_ipw.get_port_by_name(port_name) assert len(ipw_port) == port_width - def test_ports_override(self, axi_dispctrl_ipw_overriden, axi_dispctrl_overriden_port_names_widths): + def test_ports_override( + self, axi_dispctrl_ipw_overriden, axi_dispctrl_overriden_port_names_widths + ): """Check whether all the ports of `IPWrapper` object have been created, but this time taking into account the overriden parameters values. """ for port_name, port_width in axi_dispctrl_overriden_port_names_widths: ipw_port = axi_dispctrl_ipw_overriden.get_port_by_name(port_name) assert len(ipw_port) == port_width - + def test_get_port_by_name_invalid_name(self, axi_dispctrl_ipw): with pytest.raises(ValueError): axi_dispctrl_ipw.get_port_by_name("non_existing_port_name") diff --git a/tests/tests_kpm/conftest.py b/tests/tests_kpm/conftest.py index d22e7d50..d8a4173e 100644 --- a/tests/tests_kpm/conftest.py +++ b/tests/tests_kpm/conftest.py @@ -6,6 +6,7 @@ import pytest from yaml import Loader, load + from tests.common import read_json_file diff --git a/tests/tests_kpm/test_kpm_specification.py b/tests/tests_kpm/test_kpm_specification.py index 9e02ce61..85bfbf5a 100644 --- a/tests/tests_kpm/test_kpm_specification.py +++ b/tests/tests_kpm/test_kpm_specification.py @@ -1,10 +1,11 @@ # Copyright (C) 2023 Antmicro # SPDX-License-Identifier: Apache-2.0 -import pytest from pathlib import Path -from referencing import Registry + +import pytest from jsonschema import Draft202012Validator +from referencing import Registry from referencing.jsonschema import DRAFT202012 from fpga_topwrap.yamls_to_kpm_spec_parser import ipcore_yamls_to_kpm_spec @@ -32,16 +33,22 @@ def schemas_names() -> list: """Return a list of helper schemas necessary for specification validation. These are "included" by the main `speficication_schema.json`. """ - return ['unresolved_specification_schema', 'metadata_schema'] + return ["unresolved_specification_schema", "metadata_schema"] @pytest.fixture def schemas_registry(schemas_dir, schemas_names) -> Registry: - return Registry().with_resources([ - (schema_name, DRAFT202012.create_resource( - read_json_file(schemas_dir / Path(f"{schema_name}.json")))) - for schema_name in schemas_names - ]) + return Registry().with_resources( + [ + ( + schema_name, + DRAFT202012.create_resource( + read_json_file(schemas_dir / Path(f"{schema_name}.json")) + ), + ) + for schema_name in schemas_names + ] + ) def test_pwm_specification_generation(specification_schema, pwm_ipcores_yamls, schemas_registry): @@ -49,8 +56,9 @@ def test_pwm_specification_generation(specification_schema, pwm_ipcores_yamls, s pwm_specification = ipcore_yamls_to_kpm_spec(pwm_ipcores_yamls) assert len(pwm_specification["nodes"]) == 6 # 3 IP cores + 3 External metanodes - Draft202012Validator(specification_schema, - registry=schemas_registry).validate(pwm_specification) + Draft202012Validator(specification_schema, registry=schemas_registry).validate( + pwm_specification + ) def test_hdmi_specification_generation(specification_schema, hdmi_ipcores_yamls, schemas_registry): @@ -58,5 +66,6 @@ def test_hdmi_specification_generation(specification_schema, hdmi_ipcores_yamls, hdmi_specification = ipcore_yamls_to_kpm_spec(hdmi_ipcores_yamls) assert len(hdmi_specification["nodes"]) == 15 # 12 IP cores + 3 External metanodes - Draft202012Validator(specification_schema, - registry=schemas_registry).validate(hdmi_specification) + Draft202012Validator(specification_schema, registry=schemas_registry).validate( + hdmi_specification + ) diff --git a/tests/tests_kpm/test_kpm_validation.py b/tests/tests_kpm/test_kpm_validation.py index 17c198f9..b88004a3 100644 --- a/tests/tests_kpm/test_kpm_validation.py +++ b/tests/tests_kpm/test_kpm_validation.py @@ -2,9 +2,9 @@ # SPDX-License-Identifier: Apache-2.0 import pytest +from pipeline_manager_backend_communication.misc_structures import MessageType from pytest_lazy_fixtures import lf -from pipeline_manager_backend_communication.misc_structures import MessageType from fpga_topwrap.kpm_dataflow_validator import ( _check_ambigous_ports, _check_duplicate_external_input_interfaces, @@ -12,9 +12,9 @@ _check_duplicate_ip_names, _check_ext_in_to_ext_out_connections, _check_external_inputs_missing_val, + _check_inouts_connections, _check_parameters_values, _check_unconnected_ports_interfaces, - _check_inouts_connections, ) from tests.common import read_json_file @@ -93,7 +93,7 @@ def specification_duplicate_external_input_interfaces(): (_check_duplicate_external_input_interfaces, MessageType.OK), (_check_duplicate_external_out_names, MessageType.OK), (_check_unconnected_ports_interfaces, MessageType.WARNING), - (_check_inouts_connections, MessageType.OK) + (_check_inouts_connections, MessageType.OK), ], ) def test_hdmi_dataflow_validation( @@ -154,11 +154,7 @@ def test_pwm_dataflow_validation(_check_function, expected_result, pwm_dataflow, lf("dataflow_duplicate_external_input_interfaces"), MessageType.ERROR, ), - ( - _check_inouts_connections, - lf("dataflow_inouts_connections"), - MessageType.WARNING - ) + (_check_inouts_connections, lf("dataflow_inouts_connections"), MessageType.WARNING), ], ) def test_invalid_dataflow_validation(dataflow, _check_function, expected_result, pwm_specification): diff --git a/tests/tests_parse/test_parse.py b/tests/tests_parse/test_parse.py index 870c56de..2e17f0c1 100644 --- a/tests/tests_parse/test_parse.py +++ b/tests/tests_parse/test_parse.py @@ -14,7 +14,9 @@ class TestVerilogParse: def axi_axil_adapter_module(self) -> VerilogModule: from fpga_topwrap.verilog_parser import VerilogModuleGenerator - verilog_modules = VerilogModuleGenerator().get_modules("tests/data/data_parse/axi_axil_adapter.v") + verilog_modules = VerilogModuleGenerator().get_modules( + "tests/data/data_parse/axi_axil_adapter.v" + ) assert len(verilog_modules) == 1 return verilog_modules[0] @@ -36,7 +38,9 @@ def axi_axil_adapter_ports_by_dir(self, axi_axil_adapter_ports) -> dict: def seg7_4d_ctrl_modules(self) -> List[VerilogModule]: from fpga_topwrap.verilog_parser import VerilogModuleGenerator - verilog_modules = VerilogModuleGenerator().get_modules("tests/data/data_parse/seg7_4d_ctrl.v") + verilog_modules = VerilogModuleGenerator().get_modules( + "tests/data/data_parse/seg7_4d_ctrl.v" + ) assert len(verilog_modules) > 0 return verilog_modules