Skip to content

Commit

Permalink
Run black and isort on sources
Browse files Browse the repository at this point in the history
Internal-tag: [#50994]
Signed-off-by: Krzysztof Obłonczek <koblonczek@internships.antmicro.com>
  • Loading branch information
koblonczek authored and rw1nkler committed Nov 8, 2023
1 parent ad83cdc commit 426873c
Show file tree
Hide file tree
Showing 15 changed files with 262 additions and 120 deletions.
4 changes: 2 additions & 2 deletions fpga_topwrap/amaranth_helpers.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down
12 changes: 8 additions & 4 deletions fpga_topwrap/design.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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):
Expand Down
15 changes: 9 additions & 6 deletions fpga_topwrap/design_to_kpm_dataflow_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
]

Expand All @@ -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))
Expand Down
22 changes: 13 additions & 9 deletions fpga_topwrap/ipconnect.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
8 changes: 7 additions & 1 deletion fpga_topwrap/ipwrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
10 changes: 3 additions & 7 deletions fpga_topwrap/kpm_dataflow_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -16,6 +15,7 @@
get_dataflow_ip_connections,
get_dataflow_ip_nodes,
get_metanode_property_value,
is_external_metanode,
)


Expand Down Expand Up @@ -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"],
}
5 changes: 3 additions & 2 deletions fpga_topwrap/kpm_dataflow_validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
11 changes: 4 additions & 7 deletions fpga_topwrap/yamls_to_kpm_spec_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand Down Expand Up @@ -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"
]
)
)
Expand Down
24 changes: 18 additions & 6 deletions tests/tests_build/test_hier_design.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"])
Expand All @@ -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)
Expand All @@ -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
Expand Down
Loading

0 comments on commit 426873c

Please sign in to comment.