diff --git a/tests/test_conan_helper.py b/tests/test_conan_helper.py new file mode 100644 index 0000000..685a2dc --- /dev/null +++ b/tests/test_conan_helper.py @@ -0,0 +1,118 @@ +# Copyright (c) 2023-2024 Contributors to the Eclipse Foundation +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +import os +import sys + +import pytest +from pyfakefs.fake_filesystem import FakeFilesystem +from velocitas_lib import get_workspace_dir + +sys.path.append(os.path.join(os.path.dirname(__file__), "..", "..", "src")) +from shared_utils.conan_helper import add_dependency_to_conanfile # noqa + + +@pytest.fixture +def env(): + os.environ["VELOCITAS_WORKSPACE_DIR"] = "/home/workspace" + + +def test_add_dependency_to_conanfile__only_requires_section(fs: FakeFilesystem, env): + contents = """ +[requires] +""" + + conanfile_path = os.path.join(get_workspace_dir(), "conanfile.txt") + fs.create_file(conanfile_path, contents=contents) + add_dependency_to_conanfile("mydep", "myver") + + expected = """ +[requires] +mydep/myver +""" + + assert expected == open(conanfile_path, encoding="utf-8").read() + + +def test_add_dependency_to_conanfile__multiple_sections(fs: FakeFilesystem, env): + contents = """ +[requires] + +[foo] + +[bar] +""" + + conanfile_path = os.path.join(get_workspace_dir(), "conanfile.txt") + fs.create_file(conanfile_path, contents=contents) + add_dependency_to_conanfile("mydep", "myver") + + expected = """ +[requires] +mydep/myver + +[foo] + +[bar] +""" + + assert expected == open(conanfile_path, encoding="utf-8").read() + + +def test_add_dependency_to_conanfile__no_requires_section(fs: FakeFilesystem, env): + contents = """ +[foo] + +[bar] +""" + + conanfile_path = os.path.join(get_workspace_dir(), "conanfile.txt") + fs.create_file(conanfile_path, contents=contents) + add_dependency_to_conanfile("mydep", "myver") + + expected = """ +[foo] + +[bar] +[requires] +mydep/myver +""" + + assert expected == open(conanfile_path, encoding="utf-8").read() + + +def test_add_dependency_to_conanfile__pre_existing_dep_(fs: FakeFilesystem, env): + contents = """ +[foo] + +[requires] +mydep/myver2 + +[bar] +""" + + conanfile_path = os.path.join(get_workspace_dir(), "conanfile.txt") + fs.create_file(conanfile_path, contents=contents) + add_dependency_to_conanfile("mydep", "myver") + + expected = """ +[foo] + +[requires] +mydep/myver + +[bar] +""" + + assert expected == open(conanfile_path, encoding="utf-8").read() diff --git a/velocitas_lib/__init__.py b/velocitas_lib/__init__.py index 12dae03..c0f45aa 100644 --- a/velocitas_lib/__init__.py +++ b/velocitas_lib/__init__.py @@ -16,11 +16,109 @@ import os import sys from io import TextIOWrapper -from typing import Any, Dict +from typing import Any, Callable, Dict, List, Optional import requests +def to_camel_case(snake_str: str) -> str: + """Return a camel case version of a snake case string. + + Args: + snake_str (str): A snake case string. + + Returns: + str: A camel case version of a snake case string. + """ + return "".join(x.capitalize() for x in snake_str.lower().split("-")) + + +def create_truncated_string(input: str, length: int) -> str: + """Create a truncated version of input if it is longer than length. + Will keep the rightmost characters and cut of the front if it is + longer than allowed. + + Args: + input (str): The input string. + length (int): The allowed overall length. + + Returns: + str: A truncated string which has len() of length. + """ + if len(input) < length: + return input + + return f"...{input[-length+3:]}" # noqa: E226 intended behaviour + + +def replace_in_file(file_path: str, text: str, replacement: str) -> None: + """Replace all occurrences of text in a file with a replacement. + + Args: + file_path (str): The path to the file. + text (str): The text to find. + replacement (str): The replacement for text. + """ + buffer = [] + for line in open(file_path, encoding="utf-8"): + buffer.append(line.replace(text, replacement)) + + with open(file_path, mode="w", encoding="utf-8") as file: + for line in buffer: + file.write(line) + + +def get_valid_arch(arch: str) -> str: + """Return a known architecture for the given `arch`. + + Args: + arch (str): The architecture of the profile. + + Returns: + str: valid architecture. + """ + if "x86_64" in arch or "amd64" in arch: + return "x86_64" + elif "aarch64" in arch or "arm64" in arch: + return "aarch64" + + raise ValueError(f"Unknown architecture: {arch}") + + +def capture_textfile_area( + file: TextIOWrapper, + start_line: str, + end_line: str, + map_fn: Optional[Callable[[str], str]] = None, +) -> List[str]: + """Capture an area of a textfile between a matching start line (exclusive) and the first line matching end_line (exclusive). + + Args: + file (TextIOWrapper): The text file to read from. + start_line (str): The line which triggers the capture (will not be part of the output) + end_line (str): The line which terminates the capture (will not be bart of the output) + map_fn (Optional[Callable[[str], str]], optional): An optional mapping function to transform captured lines. Defaults to None. + + Returns: + List[str]: A list of captured lines. + """ + area_content: List[str] = [] + is_capturing = False + for line in file: + if line.strip() == start_line: + is_capturing = True + elif line.strip() == end_line: + is_capturing = False + elif is_capturing: + line = line.rstrip() + + if map_fn: + line = map_fn(line) + + area_content.append(line) + return area_content + + def require_env(name: str) -> str: """Require and return an environment variable. diff --git a/velocitas_lib/conan_helper.py b/velocitas_lib/conan_helper.py new file mode 100644 index 0000000..a6e6d41 --- /dev/null +++ b/velocitas_lib/conan_helper.py @@ -0,0 +1,166 @@ +# Copyright (c) 2023-2024 Contributors to the Eclipse Foundation +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +import glob +import os +import shutil +import subprocess +from typing import List, Optional, Tuple + +from velocitas_lib import get_workspace_dir + + +def get_required_sdk_version() -> Optional[str]: + """Return the required version of the core SDK. + + Returns: + Optional[str]: The required version or None in case SDK is not a dependency. + """ + sdk_version: Optional[str] = None + with open( + os.path.join(get_workspace_dir(), "conanfile.txt"), encoding="utf-8" + ) as conanfile: + for line in conanfile: + if line.startswith("vehicle-app-sdk/"): + sdk_version = line.split("/", maxsplit=1)[1].split("@")[0].strip() + + return sdk_version + + +def move_generated_sources( + generated_source_dir: str, output_dir: str, include_dir_rel: str, src_dir_rel: str +) -> Tuple[List[str], List[str]]: + """Move generated source code from the generation dir into + headers: / + sources: / + + Args: + generated_source_dir (str): The directory containing the generated sources. + output_dir (str): The root directory to move the generated files to. + include_dir_rel (str): Path relative to output_dir where to move the headers to. + src_dir_rel (str): Path relative to the output_dir where to move the sources to. + + Returns: + Tuple[List[str], List[str]]: A tuple containing + [0] = a list of the paths to all headers + [1] = a list of the paths to all sources + """ + + headers = glob.glob(os.path.join(generated_source_dir, "*.h")) + sources = glob.glob(os.path.join(generated_source_dir, "*.cc")) + + headers_relative = [] + for header in headers: + rel_path = os.path.relpath(header, generated_source_dir) + os.makedirs(os.path.join(output_dir, include_dir_rel), exist_ok=True) + shutil.move(header, os.path.join(output_dir, include_dir_rel, rel_path)) + headers_relative.append(os.path.join(include_dir_rel, rel_path)) + + sources_relative = [] + for source in sources: + rel_path = os.path.relpath(source, generated_source_dir) + os.makedirs(os.path.join(output_dir, src_dir_rel), exist_ok=True) + shutil.move(source, os.path.join(output_dir, src_dir_rel, rel_path)) + sources_relative.append(os.path.join(src_dir_rel, rel_path)) + + return headers_relative, sources_relative + + +def export_conan_project(conan_project_path: str) -> None: + """Export a conan project to the local conan cache. + + Args: + conan_project_path (str): The path to directory containing the project. + """ + env = os.environ.copy() + env["CONAN_REVISIONS_ENABLED"] = "1" + print("Exporting Conan project") + subprocess.check_call( + ["conan", "export", "."], + cwd=conan_project_path, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + env=env, + ) + + +def _find_insertion_index( + lines: List[str], dependency_name: str +) -> Tuple[int, bool, bool]: + """Find an insertion index for the dependency in a conanfile.txt. + + Args: + lines (List[str]): The lines of the original conanfile.txt + dependency_name (str): The name of the dependency (without version) e.g. "grpc" + of the dependency to insert. + + Returns: + Tuple[int, bool, bool]: A tuple consisting of + [0] = Insert index. + [1] = Whether the insert index replaces the original line or not. + [2] = Whether the original file has a requires section or not. + """ + found_index: Optional[int] = None + replace: bool = False + in_requires_section = False + has_requires_section = False + for i in range(0, len(lines)): + stripped_line = lines[i].strip() + if stripped_line == "[requires]": + has_requires_section = True + in_requires_section = True + found_index = i + 1 + elif in_requires_section and stripped_line.startswith("["): + in_requires_section = False + + if in_requires_section: + if len(stripped_line) > 0: + if stripped_line.startswith(dependency_name): + found_index = i + replace = True + + if found_index is None: + found_index = len(lines) + + return (found_index, replace, has_requires_section) + + +def add_dependency_to_conanfile(dependency_name: str, dependency_version: str) -> None: + """Add the dependency name to the project's list of dependencies. + + Args: + dependency_name (str): The dependency to add e.g. grpc + dependency_version (str): The version of the dependency to add e.g. 1.50.1 + """ + conanfile_path = os.path.join(get_workspace_dir(), "conanfile.txt") + + lines = [] + with open(conanfile_path, encoding="utf-8", mode="r") as conanfile: + lines = conanfile.readlines() + + insert_index, replace, has_requires_section = _find_insertion_index( + lines, dependency_name + ) + + dependency_line = f"{dependency_name}/{dependency_version}\n" + if replace: + lines[insert_index] = dependency_line + else: + if not has_requires_section: + lines.insert(insert_index, "[requires]\n") + insert_index = insert_index + 1 + lines.insert(insert_index, dependency_line) + + with open(conanfile_path, encoding="utf-8", mode="w") as conanfile: + conanfile.writelines(lines) diff --git a/velocitas_lib/templates.py b/velocitas_lib/templates.py new file mode 100644 index 0000000..e1cd1ed --- /dev/null +++ b/velocitas_lib/templates.py @@ -0,0 +1,68 @@ +# Copyright (c) 2023-2024 Contributors to the Eclipse Foundation +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +import os +from typing import Dict, List, Optional + + +class CopySpec: + """Copy specification of a single file or directory.""" + + def __init__(self, source_path: str, target_path: Optional[str] = None): + self.source_path = source_path + self.target_path = target_path + + def get_target(self) -> str: + """Get the target path of the copy spec. + + Returns: + str: If a target_path is given explicitly, it will be returned. + Otherwise the source_path will be returned. + """ + if self.target_path is not None: + return self.target_path + return self.source_path + + +def copy_templates( + template_dir: str, + target_dir: str, + template_file_mapping: List[CopySpec], + variables: Dict[str, str], +) -> None: + """Copy templates from the template dir to the target dir. + + Args: + template_dir (str): Path to the directory containing the template files. + target_dir (str): Path to the target directory. + template_file_mapping (Dict[str, str]): A mapping of source path to target path. + variables (Dict[str, str]): Name to value mapping which will be replaced when parsing the template files. + """ + for file_to_copy in template_file_mapping: + with open( + os.path.join( + template_dir, + file_to_copy.source_path, + ), + encoding="utf-8", + ) as file_in: + target_file_path = os.path.join(target_dir, file_to_copy.get_target()) + + os.makedirs(os.path.split(target_file_path)[0], exist_ok=True) + with open(target_file_path, encoding="utf-8", mode="w") as file_out: + lines = file_in.readlines() + for line in lines: + for key, value in variables.items(): + line = line.replace("${{ " + key + " }}", value) + file_out.write(line)