Fix build directory not existing

Internal-tag: [#65066]
Signed-off-by: bbrzyski <[email protected]>

tests/tests_build/test_design.py

@@ -1,9 +1,11 @@
 # Copyright (c) 2021-2024 Antmicro <www.antmicro.com>
 # 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"))

tests/tests_kpm/test_kpm_client.py

@@ -4,14 +4,15 @@
 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
 
 
 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

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))
     )
 
 

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",

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)
 

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:

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