diff --git a/tests/tests_build/test_design.py b/tests/tests_build/test_design.py index 3a374aa0..aef4aa38 100644 --- a/tests/tests_build/test_design.py +++ b/tests/tests_build/test_design.py @@ -1,9 +1,11 @@ # Copyright (c) 2021-2024 Antmicro # SPDX-License-Identifier: Apache-2.0 +from pathlib import Path + class TestDesign: def test_design(self): from topwrap.design import build_design_from_yaml - build_design_from_yaml("tests/data/data_build/design.yaml", "build") + build_design_from_yaml(Path("tests/data/data_build/design.yaml"), Path("build")) diff --git a/tests/tests_kpm/test_kpm_client.py b/tests/tests_kpm/test_kpm_client.py index ff35e761..7e058a9b 100644 --- a/tests/tests_kpm/test_kpm_client.py +++ b/tests/tests_kpm/test_kpm_client.py @@ -4,6 +4,7 @@ from deepdiff import DeepDiff from pipeline_manager_backend_communication.misc_structures import MessageType +from tests.tests_build.test_design import Path from topwrap.kpm_common import RPCparams from topwrap.kpm_topwrap_client import RPCMethods @@ -11,7 +12,7 @@ class TestClient: @pytest.fixture def default_rpc_params(self): - return RPCparams("127.0.0.1", 9000, [], "build", "") + return RPCparams("127.0.0.1", 9000, [], Path("build"), Path()) def test_specification(self, all_yaml_files, all_specification_files, default_rpc_params): # Testing all cores diff --git a/topwrap/cli.py b/topwrap/cli.py index ab32cffb..ab2a37a4 100644 --- a/topwrap/cli.py +++ b/topwrap/cli.py @@ -6,7 +6,7 @@ import os import subprocess from pathlib import Path -from typing import Optional +from typing import List, Optional import click @@ -83,7 +83,9 @@ def build_main( # following function does make sure that build directory exists # so we don't explicitly create build directory here - build_design_from_yaml(design, build_dir, all_sources, part) + build_design_from_yaml( + Path(design), Path(build_dir), [Path(source) for source in all_sources], part + ) @main.command("parse", help="Parse HDL sources to ip core yamls") @@ -177,7 +179,9 @@ def parse_main(use_yosys, iface_deduce, iface, dest_dir, log_level, files): help="Specify directory name for output files", ) @click.argument("yamlfiles", type=click_r_file, nargs=-1) -def kpm_client_main(host, port, log_level, design, yamlfiles: list[str], build_dir): +def kpm_client_main( + host: str, port: str, log_level: str, design: str, yamlfiles: List[str], build_dir: str +): configure_log_level(log_level) logging.info("Starting kenning pipeline manager client") @@ -188,7 +192,7 @@ def kpm_client_main(host, port, log_level, design, yamlfiles: list[str], build_d loop = asyncio.get_event_loop() loop.run_until_complete( - kpm_run_client(RPCparams(host, port, extended_yamlfiles, build_dir, design)) + kpm_run_client(RPCparams(host, int(port), extended_yamlfiles, Path(build_dir), design)) ) diff --git a/topwrap/design.py b/topwrap/design.py index e18a146d..2652c7e2 100644 --- a/topwrap/design.py +++ b/topwrap/design.py @@ -189,20 +189,17 @@ def save(self, path: Optional[Union[str, Path]]): def build_design_from_yaml( - design_path: Union[str, Path], - build_dir: Union[str, Path], - sources_dir: Collection[Union[str, Path]] = [], + design_path: Path, + build_dir: Path, + sources_dir: Collection[Path] = [], part: Optional[str] = None, ): design_path = Path(design_path) design_dir = design_path.parent - build_dir = Path(build_dir) - - build_dir.mkdir(exist_ok=True) desc = DesignDescription.from_file(design_path) desc.generate_design(design_dir).build( - build_dir=str(build_dir), + build_dir=build_dir, sources_dir=sources_dir, part=part, top_module_name=desc.design.name or "top", diff --git a/topwrap/ipconnect.py b/topwrap/ipconnect.py index db4ad117..4395bcb9 100644 --- a/topwrap/ipconnect.py +++ b/topwrap/ipconnect.py @@ -2,7 +2,8 @@ # SPDX-License-Identifier: Apache-2.0 from logging import info, warning from os import path -from typing import TYPE_CHECKING, Collection, Dict, List, Set, Tuple +from pathlib import Path +from typing import TYPE_CHECKING, Collection, Dict, List, Optional, Set, Tuple from amaranth import Fragment, Module from amaranth.back import verilog @@ -401,12 +402,14 @@ def validate_inout_connections(self, inouts: Collection[Tuple[str, str]]): def build( self, - build_dir="build", + build_dir: Path = Path("build"), template=None, - sources_dir=[], - top_module_name="project_top", - part=None, + sources_dir: Collection[Path] = [], + top_module_name: str = "project_top", + part: Optional[str] = None, ) -> None: + build_dir.mkdir(exist_ok=True) + # This class is used for generating FuseSoC Core file fuse = FuseSocBuilder(part) diff --git a/topwrap/kpm_common.py b/topwrap/kpm_common.py index ccb67367..a48fbf6c 100644 --- a/topwrap/kpm_common.py +++ b/topwrap/kpm_common.py @@ -4,6 +4,7 @@ import logging from collections import defaultdict from dataclasses import dataclass +from pathlib import Path from typing import Dict, List, Optional CONST_NAME = "Constant" @@ -60,9 +61,9 @@ def get_graph_with_id(dataflow_json: dict, graph_id: str) -> Optional[dict]: class RPCparams: host: str port: int - yamlfiles: list - build_dir: str - design: str + yamlfiles: List[str] + build_dir: Path + design: Path def is_external_metanode(node: dict) -> bool: diff --git a/topwrap/kpm_topwrap_client.py b/topwrap/kpm_topwrap_client.py index 8ca5905a..5b5a1eb7 100644 --- a/topwrap/kpm_topwrap_client.py +++ b/topwrap/kpm_topwrap_client.py @@ -4,7 +4,8 @@ import logging from base64 import b64encode from datetime import datetime -from typing import Optional, Tuple +from pathlib import Path +from typing import List, Optional, Tuple import yaml from pipeline_manager_backend_communication.communication_backend import ( @@ -99,25 +100,25 @@ async def frontend_on_connect(self) -> dict: return {} -def _kpm_specification_handler(yamlfiles: list) -> dict: +def _kpm_specification_handler(yamlfiles: List[str]) -> dict: """Return KPM specification containing info about IP cores. The specification is generated from given IP core description YAMLs. """ return ipcore_yamls_to_kpm_spec(yamlfiles) -def _kpm_import_handler(data: str, yamlfiles: list) -> dict: +def _kpm_import_handler(data: str, yamlfiles: List[str]) -> dict: specification = ipcore_yamls_to_kpm_spec(yamlfiles) design_descr = DesignDescription.from_dict(yaml.safe_load(data)) return kpm_dataflow_from_design_descr(design_descr, specification) -def _design_from_kpm_data(data: dict, yamlfiles: list) -> DesignDescription: +def _design_from_kpm_data(data: dict, yamlfiles: List[str]) -> DesignDescription: specification = ipcore_yamls_to_kpm_spec(yamlfiles) return kpm_dataflow_to_design(data, specification) -def _kpm_run_handler(data: dict, yamlfiles: list, build_dir: str) -> list: +def _kpm_run_handler(data: dict, yamlfiles: List[str], build_dir: Path) -> list: """Parse information about design from KPM dataflow format into Topwrap's internal representation and build the design. """ @@ -131,7 +132,7 @@ def _kpm_run_handler(data: dict, yamlfiles: list, build_dir: str) -> list: return messages["errors"] -def _kpm_validate_handler(data: dict, yamlfiles: list) -> dict: +def _kpm_validate_handler(data: dict, yamlfiles: List[str]) -> dict: specification = ipcore_yamls_to_kpm_spec(yamlfiles) return validate_kpm_design(data, specification) @@ -143,7 +144,7 @@ def _generate_design_filename() -> str: return datetime.now().strftime("kpm_design_%Y%m%d_%H%M%S.yaml") -def _kpm_export_handler(dataflow: dict, yamlfiles: list) -> Tuple[str, str]: +def _kpm_export_handler(dataflow: dict, yamlfiles: List[str]) -> Tuple[str, str]: """Convert created dataflow into Topwrap's design description YAML. :param dataflow: dataflow JSON from KPM