From 151d4a71f07f9cdfc0d2856cb47751c248614962 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Sza=C5=82kowski?= Date: Mon, 14 Oct 2024 13:57:37 +0200 Subject: [PATCH] Use KPM node.name instead of node.layer and make layers more useful --- CHANGELOG.md | 1 + topwrap/design_to_kpm_dataflow_parser.py | 26 ++++++++------- topwrap/kpm_common.py | 2 +- topwrap/kpm_dataflow_parser.py | 2 +- topwrap/yamls_to_kpm_spec_parser.py | 42 ++++++++++++++++-------- 5 files changed, 45 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7546881..3f7abdf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added "IP Cores", "Externals", and "Constants" layers to the GUI which you can hide/show from the settings - Nox session for downloading and packaging FuseSoc libraries ### Fixed diff --git a/topwrap/design_to_kpm_dataflow_parser.py b/topwrap/design_to_kpm_dataflow_parser.py index de7c955..9b604fd 100644 --- a/topwrap/design_to_kpm_dataflow_parser.py +++ b/topwrap/design_to_kpm_dataflow_parser.py @@ -147,7 +147,7 @@ def __init__( self.properties = properties self.interfaces = interfaces - def to_json_format(self) -> dict: + def to_json_format(self) -> JsonType: return { "name": self.type, "id": self.id, @@ -172,7 +172,7 @@ def __init__( self.subgraph = subgraph @override - def to_json_format(self) -> dict: + def to_json_format(self) -> JsonType: node_json = super().to_json_format() node_json["subgraph"] = self.subgraph @@ -270,7 +270,7 @@ def __init__(self, id_from: str, id_to: str) -> None: self.id_from = id_from self.id_to = id_to - def to_json_format(self) -> dict: + def to_json_format(self) -> JsonType: return {"id": self.id, "from": self.id_from, "to": self.id_to} @@ -285,7 +285,7 @@ def __init__( self.nodes = nodes self.id = id if id else IDGenerator().generate_id() - def to_json_format(self) -> dict: + def to_json_format(self) -> JsonType: return { "id": self.id, "nodes": [node.to_json_format() for node in self.nodes], @@ -293,10 +293,10 @@ def to_json_format(self) -> dict: } -def _get_specification_node_by_type(type: str, specification: dict) -> Optional[dict]: +def _get_specification_node_by_type(type: str, specification: JsonType) -> Optional[JsonType]: """Return a node of type `type` from specification""" for node in specification["nodes"]: - if type == node["layer"]: + if type == node["name"]: return node logging.warning(f'Node type "{type}" not found in specification') @@ -357,7 +357,7 @@ def get_kpm_nodes_from_design( def kpm_nodes_from_design_descr( - design_descr: DesignDescription, specification: dict + design_descr: DesignDescription, specification: JsonType ) -> List[KPMDataflowNode]: """Generate KPM dataflow nodes based on Topwrap's design description yaml (e.g. generated from YAML design description) @@ -595,7 +595,7 @@ def kpm_constant_metanodes_from_nodes( def kpm_constant_metanodes_from_design_descr( - design_descr: DesignDescription, specification: dict + design_descr: DesignDescription, specification: JsonType ) -> List[KPMDataflowConstantMetanode]: """Generate a list of constant metanodes based on values assigned to ip core ports of Topwrap's design description @@ -605,7 +605,7 @@ def kpm_constant_metanodes_from_design_descr( def kpm_metanodes_from_design_descr( - design_descr: DesignDescription, specification: dict + design_descr: DesignDescription, specification: JsonType ) -> List[KPMDataflowMetanode]: """Generate a list of all metanodes based on values assigned to ip core ports and an 'external' section of Topwrap's design description @@ -804,7 +804,7 @@ def subgraph_connections_to_metanodes( def create_subgraphs( design_section: DesignSection, - specification: dict, + specification: JsonType, previous_nodes: List[KPMDataflowNode], parent_subgraph_maps: SubgraphMaps, ) -> List[KPMDataflowGraph]: @@ -855,7 +855,7 @@ def create_subgraphs( def create_entry_graph( - design_descr: DesignDescription, specification: dict + design_descr: DesignDescription, specification: JsonType ) -> Tuple[KPMDataflowGraph, SubgraphMaps]: """Creates entry graph for kpm design. Main difference between entry graph and other subgraphs is that the "external" field @@ -879,7 +879,9 @@ def create_entry_graph( ) -def kpm_dataflow_from_design_descr(design_descr: DesignDescription, specification: dict) -> dict: +def kpm_dataflow_from_design_descr( + design_descr: DesignDescription, specification: JsonType +) -> JsonType: """Generate Pipeline Manager dataflow from a design description in Topwrap's yaml format """ diff --git a/topwrap/kpm_common.py b/topwrap/kpm_common.py index 37f0a86..1049a9b 100644 --- a/topwrap/kpm_common.py +++ b/topwrap/kpm_common.py @@ -362,7 +362,7 @@ def find_spec_interface_by_name( ) -> Optional[JsonType]: """Find `name` interface of `ip_type` IP core in `specification`""" for node in specification["nodes"]: - if node["layer"] != node_type: + if node["name"] != node_type: continue for interface in node["interfaces"]: if interface["name"] == iface_name: diff --git a/topwrap/kpm_dataflow_parser.py b/topwrap/kpm_dataflow_parser.py index 46ac3e7..80dcf2c 100644 --- a/topwrap/kpm_dataflow_parser.py +++ b/topwrap/kpm_dataflow_parser.py @@ -166,7 +166,7 @@ def _kpm_nodes_to_ips(dataflow_data: JsonType, specification: JsonType) -> JsonT instance_names = defaultdict(int) for node in get_dataflow_ip_nodes(dataflow_data): for spec_node in specification["nodes"]: - if spec_node["layer"] == node["name"]: + if spec_node["name"] == node["name"]: if "additionalData" not in spec_node: raise KPMExportException( f'IP "{node["name"]}" does not contain the file path inside "additionalData"' diff --git a/topwrap/yamls_to_kpm_spec_parser.py b/topwrap/yamls_to_kpm_spec_parser.py index 02f11fd..643aa63 100644 --- a/topwrap/yamls_to_kpm_spec_parser.py +++ b/topwrap/yamls_to_kpm_spec_parser.py @@ -1,8 +1,8 @@ # Copyright (c) 2023-2024 Antmicro # SPDX-License-Identifier: Apache-2.0 -import logging from dataclasses import dataclass +from enum import Enum from pathlib import Path from typing import Dict, List, Optional @@ -45,11 +45,17 @@ class PropertyType: value: str = "" +class LayerType(Enum): + IP_CORE = "IP Cores" + EXTERNAL = "Externals" + CONSTANT = "Constants" + + @dataclass class NodeType: name: str category: str - layer: str + layer: LayerType properties: List[PropertyType] interfaces: List[InterfaceType] additional_data: Optional[str] = None @@ -87,20 +93,21 @@ def _ipcore_param_to_kpm_value(param: IPCoreParameter) -> str: def _duplicate_ipcore_types_check(specification: JsonType): """Function to check for any duplicate node types in specification.""" - # If the layer is already in types_set then it means that it's a duplicate + # If the name is already in types_set then it means that it's a duplicate types_set = set() duplicates = set() for node in specification["nodes"]: - if node["layer"] in types_set: - duplicates.add(node["layer"]) + if node["name"] in types_set: + duplicates.add(node["name"]) else: - types_set.add(node["layer"]) - for dup in list(duplicates): - logging.warning(f"Multiple IP cores of type '{dup}'") + types_set.add(node["name"]) + + if len(duplicates) > 0: + raise ValueError(f"Duplicate IP cores of types '{', '.join(duplicates)}'") def _generate_ifaces_styling(interfaces_types: List[str]) -> List[InterfaceStyle]: - """Generate interface styling definitinos that style interfaces and their connections. + """Generate interface styling definitions that style interfaces and their connections. :param interfaces_types: a list of interfaces types, e.g. ["iface_AXI4", "iface_AXILite"] @@ -184,7 +191,9 @@ def create_core_node_from_yaml(yamlfile: Path) -> NodeType: ip_ports = _ipcore_ports_to_iface_type(ip_yaml.signals) ip_ifaces = _ipcore_ifaces_to_iface_type(ip_yaml.interfaces) - return NodeType(ip_name, "IPcore", ip_name, ip_props, ip_ports + ip_ifaces, str(yamlfile)) + return NodeType( + ip_name, "IPcore", LayerType.IP_CORE, ip_props, ip_ports + ip_ifaces, str(yamlfile) + ) def create_external_metanode(meta_name: str, interfaces_types: List[str]) -> NodeType: @@ -201,7 +210,9 @@ def create_external_metanode(meta_name: str, interfaces_types: List[str]) -> Nod KPMDataflowExternalMetanode.interface_dir_by_node_name[meta_name], ) - return NodeType(meta_name, METANODE_CATEGORY, meta_name, [metanode_prop], [metanode_iface]) + return NodeType( + meta_name, METANODE_CATEGORY, LayerType.EXTERNAL, [metanode_prop], [metanode_iface] + ) def create_constantant_metanode(interfaces_types: List[str]) -> NodeType: @@ -211,7 +222,9 @@ def create_constantant_metanode(interfaces_types: List[str]) -> NodeType: KPMDataflowMetanodeInterface.CONST_IFACE_NAME, ["port"] + interfaces_types, "output" ) - return NodeType(CONST_NAME, METANODE_CATEGORY, CONST_NAME, [metanode_prop], [metanode_iface]) + return NodeType( + CONST_NAME, METANODE_CATEGORY, LayerType.CONSTANT, [metanode_prop], [metanode_iface] + ) def create_subgraph_metanode() -> NodeType: @@ -220,13 +233,13 @@ def create_subgraph_metanode() -> NodeType: sub_meta_out = InterfaceType(KPMDataflowMetanodeInterface.SUB_IFACE_IN_NAME, ["port"], "input") return NodeType( - SUBGRAPH_METANODE, METANODE_CATEGORY, SUBGRAPH_METANODE, [], [sub_meta_in, sub_meta_out] + SUBGRAPH_METANODE, METANODE_CATEGORY, LayerType.EXTERNAL, [], [sub_meta_in, sub_meta_out] ) def add_node_type_to_specfication(specification_builder: SpecificationBuilder, node: NodeType): """Adds all information from NodeType to specification_builder""" - specification_builder.add_node_type(node.name, node.category, node.layer) + specification_builder.add_node_type(node.name, node.category, node.layer.value) for property in node.properties: specification_builder.add_node_type_property( @@ -252,6 +265,7 @@ def add_metadata_to_specification( "backgroundSize": 15, "layout": "CytoscapeEngine - grid", "twoColumn": True, + "layers": [{"name": lr.value, "nodeLayers": [lr.value]} for lr in LayerType], "navbarItems": [ { "name": "Validate",