#!/usr/bin/env python3 # coding=utf-8 """ CMake Build System Helper A comprehensive build automation script for CMake projects using CMakePresets.json. Provides a unified interface for configuring, building, testing, and packaging with preset management, toolchain handling, and interactive menu selection. Features: - Interactive curses-based preset selection menu - Support for CMakePresets.json and CMakeUserPresets.json - Automated toolchain configuration per platform - Build configuration, execution, testing, and packaging workflows - Package installation for development dependencies (Linux/Windows/Cygwin) - Colored logging with customizable verbosity - Environment variable management and macro expansion Supported Platforms: - Linux (native, cross-compile for Windows/ARM) - Windows Author: Arjan van Olphen """ from __future__ import annotations import sys # sys.dont_write_bytecode = True import argparse import configparser import copy import hashlib import logging import stat import platform import re import shutil import time import socket import subprocess import os import json import threading import ctypes import zipfile import tempfile import fnmatch import glob import io import base64 import http.client import ssl import uuid from datetime import datetime from dataclasses import dataclass from enum import Enum, auto from string import Template from typing import List, Any, Dict, Optional, Tuple from pathlib import Path from abc import ABC, abstractmethod from urllib.request import urlopen from urllib.parse import quote, urlsplit # Auto-install for Windows the not standard 'curses' required module. try: import curses except ImportError as import_ex: if sys.platform == "win32": subprocess.check_call([sys.executable, "-m", "pip", "install", "windows-curses"]) import curses else: raise import_ex # Convenient type. ParsedArguments = argparse.Namespace # Template for 'build.ini' toolchain configuration file. # noinspection SpellCheckingInspection INI_TEMPLATE = r""" ; File for adding environment during the nested calls of the script. [config] ; The default Qt version and also the initial 'RUN_QT_VER' environment variable and passed as ; env var 'SF_QT_VERSION' to the docker container. qt-ver=6.12.0 ; Optional download location passed to the Docker container as 'SF_QT_DOWNLOAD_BASE' env var. ;qt-download-base=https://nexus.scanframe.com/repository/shared/library/qt ; Override the default docker image using a smaller one without the QT libraries. docker-image=nexus.scanframe.com/amd64/gnu-cpp:24.04-6.12.0 ; Custom container name for this project to prevent conflicts with other projects. ;docker-container-name=my_cpp_builder ; Base for downloading toolchains and qt libraries. library-base-url=https://nexus.scanframe.com/repository/shared/library ; Sematic versioning type impact map. [config-ver-type-map] fix=patch,Bugfix feat=minor,Feature build=patch,Build Tool/Process chore=patch,Chore ci=patch,CI Configuration docs=patch,Documentation style=patch,Code Formatting/Styling refactor=patch,Code Refactoring perf=patch,Performance test=none,Test Addition/Modification revert=patch,Revert of Commit ; Section for optional include file which is merged. [__include__] user=user.ini nexus=.nexus-credentials.ini ; Pulse audio server config. Need volume mapping from host. [pulse-audio] ; Location for the pulse audio server to find the users socket (check with: pactl info). PULSE_SERVER=unix:/run/user/${UID}/pulse/native ; To get the Qt theme as on the host. QT_QPA_PLATFORMTHEME=xdgdesktopportal DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/${UID}/bus [qt-12-fix] ; Prevent displaying Wine warnings. WINEDEBUG=fixme-all ;WINEDEBUG=fixme-all,+seh,+mscms,+loaddll ; Required to since Wine 11.0 is missing function 'WcsCreateIccProfile' in 'mscms.dll' ; and is overridden by our own DLL placed in the system32 directory. But still this below is needed. WINEDLLOVERRIDES=mscms=n ; Environment added when running in Wine natively. ; It needs Python 3.12, CMake 4.2, Ninja 1.13 and NSIS 3.11 which can be installed but also shared through the toolchain directory. ; Windows Git is not completely working since it is using Cygwin libraries and a Git server/client is devised to overcome it. [env.wine@] __inherit__=qt-12-fix ; Convenient variable for common base directory. TOOL_ROOT=Z:${RUN_DIR}\lib\toolchain\win-${MACHINE}-combi ; Configuration for tools as cmake, git and ninja for within the Wine environment. WINEPATH=${TOOL_ROOT}\cmake\bin;${TOOL_ROOT}\bin;${TOOL_ROOT}\python;${TOOL_ROOT}\nsis ; Start of port numbers to look for an unused one for the git server. ;GIT_SERVER_PORT=8000 ; Alternate Wine prefix directory. ;WINEPREFIX= ; Environment added before running Docker. [env.docker@] ; Nothing yet. ; Environment added before running with the compiler msvc natively. [env.msvc@] SF_EXEC_DIR_SUFFIX=-msvc ; Location of the root of the MSVC toolchain in the Wine environment. ; The rest is environment as below is configured in the file CMakePresets.json to allow multiple compilers to be configured in the same project. MSVC_ROOT=${RUN_DIR}\lib\toolchain\w64-${MACHINE}-msvc-2022 ; Overrides QT_VER_DIR for subcommand 'run'. RUN_QT_VER_DIR=${RUN_DIR}\lib\qt\w64-${MACHINE}\${RUN_QT_VER} ; Even though this variable is not used by Windows it is used to locate the DLL's for running the application. LD_LIBRARY_PATH=${RUN_QT_VER_DIR}\msvc_64\bin PATH=${LD_LIBRARY_PATH};${PATH} ; Environment added before running with the compiler msvc in Wine. [env.msvc.wine@] __inherit__=env.msvc@ ; Environment added before running Wine in the Docker container. [env.wine.docker@] __inherit__=pulse-audio,qt-12-fix ; Overrides QT_VER_DIR for subcommand 'run'. RUN_QT_VER_DIR=Z:\home\${USER}\lib\qt\w64-${MACHINE}\${RUN_QT_VER} ; Environment added before running with the compiler msvc in Wine in the Docker container. [env.msvc.wine.docker@] ; The Docker container is build with the MSVC toolchain. (fuse-zip mounted in the home directory). MSVC_ROOT=Z:\home\${USER}\toolchain\w64-${MACHINE}-msvc-2022 SF_EXEC_DIR_SUFFIX=-msvc RUN_QT_VER_DIR=Z:\home\${USER}\lib\qt\w64-${MACHINE}\${RUN_QT_VER} PATH=${RUN_QT_VER_DIR}\msvc_64\bin;${PATH} ; Environment added before running the 'gnu' compiler natively. [env.gnu@] SF_EXEC_DIR_SUFFIX=-gnu ; Normally the RUN_PATH is dealing with this but when compiled differently it must be set. LD_LIBRARY_PATH=${RUN_DIR}/lib/qt/lnx-${MACHINE}/${RUN_QT_VER}/gcc_64/lib ; Make the 'appimagetool' and designer available from the PATH. PATH=${RUN_DIR}/lib/qt/lnx-${MACHINE}/${RUN_QT_VER}/gcc_64/bin:${PATH}:${RUN_DIR}/lib/toolchain/lnx-${MACHINE}-combi/bin ; Environment added before running the 'gnu' compiler in the Docker container. [env.gnu.docker@] __inherit__=pulse-audio SF_EXEC_DIR_SUFFIX=-gnu ; Normally the RUN_PATH is dealing with this but when compiled differently it must be set. LD_LIBRARY_PATH=/home/${USER}/lib/qt/lnx-${MACHINE}/${RUN_QT_VER}/gcc_64/lib ; Make the 'appimagetool' available from the PATH. PATH=${HOME}/lib/qt/lnx-${MACHINE}/${RUN_QT_VER}/gcc_64/bin:${PATH}:${HOME}/toolchain/lnx-${MACHINE}-combi/bin ; Environment added when running wine natively to execute the Windows cross-compiled targets. [env.gw@] __inherit__=qt-12-fix SF_EXEC_DIR_SUFFIX=-gw WINEPATH=Z:\usr\x86_64-w64-mingw32\lib;Z:\usr\lib\gcc\x86_64-w64-mingw32\13-posix;${RUN_DIR}\lib\qt\win-${MACHINE}\${RUN_QT_VER}\mingw_64\bin ; Overrides QT_VER_DIR for subcommand 'run'. RUN_QT_VER_DIR=${RUN_DIR}/lib/qt/win-${MACHINE}/${RUN_QT_VER} ; Environment added when running wine natively to execute the Aarch64 cross-compiled targets. [env.ga@] SF_EXEC_DIR_SUFFIX=-ga # Normally the RUN_PATH is dealing with this but when compiled differently it must be set. LD_LIBRARY_PATH=${RUN_DIR}/lib/qt/lnx-aarch64/${RUN_QT_VER}/gcc_64/lib ; Make the 'appimagetool' available from the PATH. PATH=${RUN_DIR}/lib/qt/lnx-aarch64/${RUN_QT_VER}/gcc_64/bin:${PATH}:${RUN_DIR}/lib/toolchain/lnx-aarch-combi/bin ; Environment added before running the 'mingw' compiler natively. [env.mingw@] SF_EXEC_DIR_SUFFIX=-mingw ; Even though this variable is not used by Windows it is used to locate the DLL's for running the application. LD_LIBRARY_PATH=${RUN_DIR}\lib\toolchain\w64-${MACHINE}-mingw-1320-posix\bin;${RUN_DIR}\lib\qt\w64-${MACHINE}\${RUN_QT_VER}\mingw_64\bin;lib PATH=${LD_LIBRARY_PATH};${PATH} [env.mingw.wine@] __inherit__=env.mingw@,qt-12-fix ; Environment added before running the 'mingw' compiler in Docker and Wine. [env.mingw.wine.docker@] SF_EXEC_DIR_SUFFIX=-mingw PATH=Z:\home\${USER}\toolchain\w64-${MACHINE}-mingw-1320-posix\bin;Z:\home\${USER}\lib\qt\w64-${MACHINE}\${RUN_QT_VER}\mingw_64\bin;lib;${PATH} ; Environment added before running the 'ga' compiler in Docker. [env.ga.docker@] __inherit__=pulse-audio ; Puts the binary in 'bin/lnx64-ga'. SF_EXEC_DIR_SUFFIX=-ga ; Make the 'appimagetool' available from the PATH. PATH=${HOME}/lib/qt/lnx-aarch64/${RUN_QT_VER}/gcc_64/bin:${PATH}:${HOME}/toolchain/lnx-${MACHINE}-combi/bin ; Environment added before running the 'gw' compiler in the Docker container. [env.gw.docker@] __inherit__=pulse-audio,qt-12-fix SF_EXEC_DIR_SUFFIX=-gw ; Provides compiler std libraries to be found. WINEPATH=Z:\usr\x86_64-w64-mingw32\lib;Z:\usr\lib\gcc\x86_64-w64-mingw32\13-posix; # Optional for allowing the .exe files to be executed from Linux. Required compiler std libraries are also part of the Qt library. ;WINEPATH=Z:\home\${USER}\lib\qt\win-${MACHINE}\${RUN_QT_VER}\mingw_64\bin;lib ; Overrides QT_VER_DIR since Wine does not pass any 'QT_' prefixed variables. RUN_QT_VER_DIR=/home/${USER}/lib/qt/win-${MACHINE}/${RUN_QT_VER} """.replace('\r', '') def is_wine() -> bool: """ Tells if this script is run from within Linux Wine. :return: True when in Wine, False otherwise. """ def _is_wine() -> bool: try: # Load ntdll and check for the Wine-specific version function if hasattr(ctypes, "windll"): ntdll = ctypes.windll.ntdll return hasattr(ntdll, "wine_get_version") except (AttributeError, OSError): pass return False if not hasattr(is_wine, "flag"): is_wine.flag = _is_wine() return is_wine.__getattribute__("flag") def is_docker() -> bool: """ Checks if running in Docker. :return: True when Docker is active, False otherwise. """ def _is_docker() -> bool: # When running in Wine, check a different filepath. # noinspection SpellCheckingInspection fn = f"{os.sep}.dockerenv" if not is_wine() else f"Z:{os.sep}.dockerenv" return os.path.isfile(fn) if not hasattr(is_docker, "flag"): is_docker.flag = _is_docker() return is_docker.__getattribute__("flag") def is_debugger_active() -> bool: """ Checks if a debugger is active. :return: True when a debugger is active, False otherwise. """ # Check PEP 669 monitoring API (Python 3.14+) mon = getattr(sys, "monitoring", None) if mon is not None: try: # Actually trying to access 'sys.monitoring.DEBUGGER_ID'. if mon.get_tool(mon.DEBUGGER_ID) is not None: return True except Exception: pass return False def start_git_server(port: int) -> int: """ Starts the listener in a background thread. :param port: Start port to try to listen on. :return: The actual port listened on and zero on failure. """ def translate_path(wine_path: str): """ Converts a path from Wine to the host. :param wine_path: :return: """ # Standardize separators wine_path = wine_path.replace('\\', '/') # Extract drive letter (e.g., 'C:') if len(wine_path) < 2 or wine_path[1] != ':': return wine_path # Already a relative or Unix-style path drive_letter = wine_path[0:2].lower() # 'c:' path_suffix = wine_path[2:].lstrip('/') # Locate the 'dosdevices' directory. wine_prefix = RUN_ENV.get('WINEPREFIX', os.path.expanduser('~/.wine')) dos_devices_path = os.path.join(wine_prefix, 'dosdevices') # The symlink for a drive is exactly the drive letter (e.g., ~/.wine/dosdevices/c:) drive_link = os.path.join(dos_devices_path, drive_letter) if os.path.islink(drive_link): # Resolve the symlink to the actual Linux path target_root = os.path.realpath(drive_link) return os.path.join(target_root, path_suffix) # Fallback if the link doesn't exist (unlikely for active drives) return wine_path def server_loop(): """ Thread function. :return: None """ while True: conn, addr = server.accept() try: raw_data = conn.recv(4096).decode('utf-8') if not raw_data: continue payload = json.loads(raw_data) # Assemble the command. cmd = ['git'] for arg in payload['args']: # Convert the argument when matching a Windows full path with forward slashes. if bool(re.match(r'^[a-zA-Z]:/(?:[^\\/:*?"<>|\r\n]+/)*[^\\/:*?"<>|\r\n]*$', arg)): cmd.append(translate_path(arg)) else: cmd.append(arg) # Execute Git and capture everything logger.info(f"= Git server: {' '.join(cmd)}") proc = subprocess.run(cmd, cwd=translate_path(payload['cwd']), # Captures both stdout and stderr capture_output=True, # Keep as bytes for raw data transfer text=False ) # Prepare the response using hex encoding prevents JSON breakages. # noinspection PyUnresolvedReferences response = {"exit_code": proc.returncode, "stdout": proc.stdout.hex(), "stderr": proc.stderr.hex()} conn.sendall(json.dumps(response).encode('utf-8')) finally: conn.close() # Try to bind to the port, incrementing until successful server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) counter: int = 0 server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) while True: try: server.bind(('127.0.0.1', port + counter)) break except OSError: counter += 1 # When tried too many ports, bailout with an error. if counter > 20: return 0 server.listen(10) logger.info(f"= Git Proxy Listener started on port '{port + counter}'.") # Launch the main server loop in its own thread threading.Thread(target=server_loop, daemon=True).start() return port + counter def _widen_console_buffer() -> None: """Widen the Windows console screen buffer so long lines are not wrapped. Wine's console emulation defaults to a narrow buffer width (often 80 columns) which causes the output to wrap before reaching the actual terminal edge. """ if sys.platform != "win32" or not is_wine(): return try: import ctypes from ctypes import wintypes class WtCoord(ctypes.Structure): _fields_ = [("X", wintypes.SHORT), ("Y", wintypes.SHORT)] class WtSmallRect(ctypes.Structure): _fields_ = [ ("Left", wintypes.SHORT), ("Top", wintypes.SHORT), ("Right", wintypes.SHORT), ("Bottom", wintypes.SHORT), ] class WtConsoleScreenBufferInfo(ctypes.Structure): _fields_ = [ ("dwSize", WtCoord), ("dwCursorPosition", WtCoord), ("wAttributes", wintypes.WORD), ("srWindow", WtSmallRect), ("dwMaximumWindowSize", WtCoord), ] std_output_handle = ctypes.c_ulong(-11 & 0xFFFFFFFF) kernel32 = ctypes.windll.kernel32 # noinspection unresolved-references handle = kernel32.GetStdHandle(std_output_handle) info = WtConsoleScreenBufferInfo() # noinspection unresolved-references if not kernel32.GetConsoleScreenBufferInfo(handle, ctypes.byref(info)): return # Use the COLUMNS environment variable when available, otherwise a large default. target_width = 1024 # target_width = int(os.environ.get("COLUMNS", 1024)) if info.dwSize.X < target_width: new_size = WtCoord(target_width, info.dwSize.Y) # noinspection unresolved-references kernel32.SetConsoleScreenBufferSize(handle, new_size) except (OSError, AttributeError, ValueError): pass class DebugMode(Enum): """ Defines the logging and reporting of commands being executed. """ REPORT = auto() """Reports the executed command and executes it as well.""" REPORT_ONLY = auto() """Reports the executed command only, so no execution takes place.""" SILENT = auto() """Does not report in any case.""" # Flag determining the terminal is dumb or needs to be dumb is the case of a CI pipeline. TERM_DUMB: bool = bool(not sys.stderr.isatty() or os.environ.get("CI") or os.environ.get("TERM") in ["dumb", "unknown"]) # Enumeration of types. class PresetTypes(Enum): """ Enumerate of CMake preset types. """ CONFIGURE = "configure" BUILD = "build" TEST = "test" PACKAGE = "package" WORKFLOW = "workflow" class ColoredFormatter(logging.Formatter): """ A logging formatter that applies color coding based on message content and log level. This class allows customization of log messages by applying ANSI color codes to the messages. The color applied depends on the prefix of the log message and the logging level of the record. """ class ColorCodes(Enum): """ Defines the colors for logging. """ BLACK = "\033[30m" if not TERM_DUMB else "" RED = "\033[31m" if not TERM_DUMB else "" GREEN = "\033[32m" if not TERM_DUMB else "" YELLOW = "\033[1;33m" if not TERM_DUMB else "" BLUE = "\033[1;34m" if not TERM_DUMB else "" MAGENTA = "\033[35m" if not TERM_DUMB else "" CYAN = "\033[36m" if not TERM_DUMB else "" WHITE = "\033[1;37m" if not TERM_DUMB else "" RESET = "\033[0m" if not TERM_DUMB else "" def format(self, record: logging.LogRecord): """ Formats a log message with optional color-coded prefixes based on the content. """ # Strip white space before checking. msg = record.getMessage().strip() color = self.ColorCodes.RESET if msg.startswith("-"): color = self.ColorCodes.CYAN elif msg.startswith("~"): color = self.ColorCodes.BLUE elif msg.startswith("#"): color = self.ColorCodes.YELLOW elif msg.startswith("="): color = self.ColorCodes.GREEN elif msg.startswith(":"): color = self.ColorCodes.MAGENTA elif msg.startswith("!"): color = self.ColorCodes.RED elif record.levelno >= logging.ERROR: color = self.ColorCodes.RED color_code = color return f"{color_code.value}{record.getMessage().expandtabs(2)}{self.ColorCodes.RESET.value}" # Setup logging. logger = logging.getLogger(__name__) handler = logging.StreamHandler(sys.stderr) handler.setFormatter(ColoredFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO) # To calculate the time elapsed. start_time = time.time() def create_config_parser(ini_path: str, cfg: configparser.ConfigParser | None = None) -> configparser.ConfigParser: """Creates the config parser.""" create_flag = True if cfg is None else False if not cfg: logger.info(f"# Configuration file: {ini_path}") cfg = configparser.ConfigParser() cfg.optionxform = str # Create the file if it does not exist yet. if create_flag and not os.path.exists(ini_path) and not is_wine() and not is_docker(): logger.info(f"# Creating non-existing configuration file: {ini_path}") with open(ini_path, "w", encoding="utf-8") as file: file.write(INI_TEMPLATE) if len(cfg.read(ini_path)) == 0: logger.info(f": Unable to read file: {ini_path}") if cfg.has_section(section := "__include__"): items = cfg.items(section) cfg.remove_section(section) for key, file in items: logger.info(f"# Configuration include file: {file}") create_config_parser(ini_path=str(os.path.join(RUN_DIR, file)), cfg=cfg) return cfg # Show the Python used. logger.info(f"# Python {sys.version} on {sys.platform}.") # The run directory using on Linux of the current file/symlink location and on Windows the current working directory. # It is impossible on Windows to detect the location of an executed symlink when executed without the 'python' executable. RUN_DIR = os.getcwd() if sys.platform == "win32" else os.path.dirname(os.path.abspath(__file__)) # Environment variables from this process. PARENT_ENV = os.environ.copy() # Linux has a current working directory environment variable and Windows does not. if 'PWD' not in PARENT_ENV: PARENT_ENV['PWD'] = os.getcwd() # List of optional environment variable names when missing no exception is raised. ENV_OPTIONAL = ["SF_BIN_DIR_SUFFIX", "WINEPATH", "LD_LIBRARY_PATH", "SF_EXEC_DIR_SUFFIX"] # List of ignored environment variables set when a CI pipeline is active. ENV_IGNORED = ["SF_EXEC_DIR_SUFFIX"] if PARENT_ENV.get("CI") else [] # In Linux and Docker it could be the TEMP environment variable is not set. if 'TEMP' not in PARENT_ENV: PARENT_ENV['TEMP'] = tempfile.gettempdir() # Architecture of the host system (uname -m). PARENT_ENV['MACHINE'] = platform.machine() # In Wine 'AMD64' is returned. if PARENT_ENV['MACHINE'] == "AMD64": PARENT_ENV['MACHINE'] = "x86_64" # Add the RUN_DIR so it is available in the '.toolchain-*' files. PARENT_ENV['RUN_DIR'] = RUN_DIR # The prefix 'RUN_' is needed since Wine does not pass environment variables prefixed with 'QT_'. PARENT_ENV['RUN_QT_VER'] = "0.0.0" # Add the user id to the parent environment so it can be used. if sys.platform != "win32": PARENT_ENV['UID'] = f"{os.getuid()}" # Environment variables for running a command with. RUN_ENV = PARENT_ENV.copy() # Global debug flag for system commands. DEBUG_FLAG = False # Holds the current configuration preset name. CONFIG_PRESET = None # Default SSH port for ssh daemon. SSHD_PORT = 8022 # Name of the project subdirectory. PROJECT_SUBDIR = os.path.basename(RUN_DIR) # Get the configuration of the script. CONFIG = create_config_parser(os.path.join(RUN_DIR, str(os.path.splitext(os.path.basename(__file__))[0] + ".ini"))) # Directory to store the CMake library files. CMAKE_LIB_SUBDIR = ["cmake", "lib"] def get_github_release(owner: str, repo: str, assets_wildcard: Optional[str] = None, release_tag: Optional[str] = None ) -> Optional[Dict[str, Any]]: """ Fetch GitHub release information and assets. :param owner: Repository owner :param repo: Repository name :param assets_wildcard: Filter assets by wildcard. :param release_tag: Specific release version (e.g., "1.2.3"). If None, fetches the latest release. :return: Dictionary with 'release', 'assets', and 'url' fields, or None if not found. - release: Release version string - assets: List of asset download info (name, url, size) - url: Base URL for the release """ import requests try: if release_tag is None: # Fetch the latest release. api_url = f"https://api.github.com/repos/{owner}/{repo}/releases/latest" else: # Fetch the specific release by tag api_url = f"https://api.github.com/repos/{owner}/{repo}/releases/tags/{release_tag}" resp = requests.get(api_url, headers={"Accept": "application/vnd.github.v3+json"}, timeout=15) if resp.status_code == 404: return None resp.raise_for_status() data = resp.json() # Extract release info release_tag = data.get("tag_name", "").lstrip('v') assets_list = [] for asset in data.get("assets", []): asset_name = asset.get("name", "") if assets_wildcard and not fnmatch.fnmatch(asset_name, assets_wildcard): continue assets_list.append({ "name": asset_name, "url": asset.get("browser_download_url", ""), "size": asset.get("size", 0), "digest": asset.get("digest", None) }) return { "release": release_tag, "assets": assets_list, "url": data.get("html_url", "") } except requests.RequestException: return None def extract_by_url(url: str, dest_dir: str, new_dir_name: Optional[str] = None, digest: Optional[str] = None) -> bool: """ Extracts the url of a given compressed file to the given destination directory. After extraction the directory is renamed to new_dir_name. """ import tarfile import requests import hashlib try: # Create the destination directory if it doesn't exist. dest_path = Path(dest_dir) dest_path.mkdir(parents=True, exist_ok=True) # Download the file resp = requests.get(url, timeout=60, stream=True) resp.raise_for_status() # Determine file extension from URL filename = url.split('/')[-1] # Download to the temporary file. sha256_hash = hashlib.sha256() if digest else None with tempfile.NamedTemporaryFile(delete=False, suffix=f"_{filename}") as tmp_file: for chunk in resp.iter_content(chunk_size=1024 * 4): tmp_file.write(chunk) if sha256_hash: sha256_hash.update(chunk) tmp_path = tmp_file.name # Verify digest if provided. if digest: computed_digest = sha256_hash.hexdigest() expected_digest = digest.split(':', 1)[1] if ':' in digest else digest if computed_digest != expected_digest: Path(tmp_path).unlink(missing_ok=True) logger.error(f"! Digest mismatch (expected/got): {expected_digest} / {computed_digest}") return False else: logger.info(f"= Digest matches: {computed_digest}") else: logger.warning(f": No digest for this download: {filename}") try: # Extract based on the file type. if filename.endswith(('.tar.gz', '.tgz')): with tarfile.open(tmp_path, 'r:gz') as tar: # filter="data" blocks dangerous features like absolute paths or files escaping the destination folder. tar.extractall(dest_path, filter="data") # Get the root directory name from the archive members = tar.getmembers() if members: root_dir = members[0].name.split('/')[0] else: return False elif filename.endswith('.tar.bz2'): with tarfile.open(tmp_path, 'r:bz2') as tar: tar.extractall(dest_path) members = tar.getmembers() if members: root_dir = members[0].name.split('/')[0] else: return False elif filename.endswith('.zip'): with zipfile.ZipFile(tmp_path, 'r') as zip_ref: zip_ref.extractall(dest_path) # Get the root directory name from the archive. names = zip_ref.namelist() if names: root_dir = names[0].split('/')[0] else: return False else: return False # Rename the extracted directory if new_dir_name is provided if new_dir_name and root_dir != new_dir_name: old_path = dest_path / root_dir new_path = dest_path / new_dir_name if old_path.exists(): # Remove destination if it exists if new_path.exists(): shutil.rmtree(new_path) old_path.rename(new_path) # Signal success. return True finally: # Clean up temporary file Path(tmp_path).unlink(missing_ok=True) except Exception as e: logger.error(f"Error extracting {url}: {e}") return False def get_config_section(section: str, fail: bool = True) -> Dict[str, str]: """ Gets a configuration section as Dict of key-value pairs. """ if not CONFIG.has_section(section): logger.error(f"! Configuration section '{section}' does not exist.") if fail: raise RuntimeError(f"Missing configuration section '{section}' !") return {} return dict(CONFIG.items(section)) def get_merged_config_section(section: str, fail: bool = True) -> Dict[str, str]: """ Gets an assembled dictionary of key-value pairs using inheritance. """ # Final merged result (Child values override parents) merged_data: Dict[str, str] = {} if not CONFIG.has_section(section): if fail: logger.error(f"! Configuration section '{section}' does not exist.") raise RuntimeError(f"Missing configuration section '{section}' !") else: return merged_data # Track visited sections to detect redundancy and prevent infinite loops visited = set() # Process queue (Breadth-First traversal of inheritance) queue = [section] while queue: current_section = queue.pop(0) # Report if a section is encountered more than once in the tree if current_section in visited: logger.warning(f": Notice: Section [{current_section}] was inherited more than once.") continue visited.add(current_section) # parser.items() provides the keys/values for the section current_items = dict(CONFIG.items(current_section) if current_section else {}) # Extract inheritance instructions if inheritance_val := current_items.pop("__inherit__", None): # noinspection PyUnresolvedReferences parents = [p.strip() for p in inheritance_val.split(",")] for parent in parents: if parent and not CONFIG.has_section(parent): logger.warning(f"Warning: Parent section [{parent}] (inherited by [{current_section}]) does not exist.") else: queue.append(parent) # Merge values: only add if the key doesn't already exist. for key, value in current_items.items(): if key not in merged_data: merged_data[key] = value return merged_data def remove_files_from_tree(dir_name: Path, wild_cards: list[str]) -> None: """ Removes any of the matching files listed in the pattern. :param dir_name: Directory to search in recursively. :param wild_cards: Wild card file patterns to look for. """ if not os.path.isdir(dir_name): logger.warning(f": Directory '{dir_name}' does not exist or is not a directory.") return # Single timestamp for all. timestamp = time.strftime("%Y%m%dT%H%M%S", time.localtime()) # Walk the tree. for root, dirs, files in os.walk(dir_name): for filename in files: for pattern in wild_cards: if fnmatch.fnmatch(filename, pattern): file_path = os.path.join(root, filename) try: new_file_path = f"{file_path}-{timestamp}" # Use a shorted path name for logging when in the run/project directory. disp_name = new_file_path if RUN_DIR != new_file_path[:len(RUN_DIR)] else new_file_path[len(RUN_DIR) + 1:] if not DEBUG_FLAG: logger.info(f"~ Renaming file to: {disp_name}") os.rename(file_path, new_file_path) else: logger.info(f"~ Skipped renaming file to: {disp_name}") except OSError as ex: logger.warning(f": Failed to rename file '{file_path}': {ex}") break def post_request_http(url: str, username: str | None = None, password: str | None = None, verbose: bool = False) -> tuple[int, str]: """Makes a POST request to an HTTP endpoint with optional basic authentication.""" parsed_url = urlsplit(url) if parsed_url.scheme not in ["http", "https"] or not parsed_url.hostname: raise ValueError(f"Unsupported or invalid URL: {url}") # request_target = parsed_url.path or "/" if parsed_url.query: request_target += f"?{parsed_url.query}" headers = { "Accept": "application/json", } # Check if username and password are provided and use them for basic authentication. if username and password: headers["Authorization"] = "Basic " + base64.b64encode( f"{username}:{password}".encode("utf-8") ).decode("ascii") # When debugging, report only. if DEBUG_FLAG: logger.info(f"~ Not posting: POST {url}") return 200, "" if verbose: logger.info(f"~ Posting: POST {url}") connection_class = http.client.HTTPSConnection if parsed_url.scheme == "https" else http.client.HTTPConnection connection_args: Dict[str, Any] = {} if parsed_url.scheme == "https": connection_args["context"] = ssl.create_default_context() # noinspection bad-argument-type connection = connection_class(parsed_url.hostname, parsed_url.port, **connection_args) try: connection.request("POST", request_target, headers=headers) response = connection.getresponse() response_body = response.read().decode("utf-8", errors="replace") if verbose: logger.info(f"~ Nexus response: HTTP {response.status} {response.reason}") return response.status, response_body finally: connection.close() def upload_file_http(url: str, upload_file: str, username: str, password: str, method: str = "PUT", multipart_field: str | None = None, content_type: str = "application/octet-stream", verbose: bool = False) -> tuple[int, str]: """Streams a file to an HTTP endpoint using basic authentication.""" parsed_url = urlsplit(url) if parsed_url.scheme not in ["http", "https"] or not parsed_url.hostname: raise ValueError(f"Unsupported or invalid upload URL: {url}") request_target = parsed_url.path or "/" if parsed_url.query: request_target += f"?{parsed_url.query}" file_size = os.path.getsize(upload_file) headers = { "Authorization": "Basic " + base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii"), "Accept": "application/json", } prefix = b"" suffix = b"" if multipart_field: boundary = f"----PythonNexusBoundary{uuid.uuid4().hex}" filename = Path(upload_file).name fallback_filename = filename.encode("ascii", errors="replace").decode("ascii").replace('\\', '\\\\').replace('"', '\\"') encoded_filename = quote(filename, safe="") prefix = ( f"--{boundary}\r\n" f"Content-Disposition: form-data; name=\"{multipart_field}\"; filename=\"{fallback_filename}\"; " f"filename*=UTF-8''{encoded_filename}\r\n" f"Content-Type: {content_type}\r\n\r\n" ).encode("ascii") suffix = f"\r\n--{boundary}--\r\n".encode("ascii") headers["Content-Type"] = f"multipart/form-data; boundary={boundary}" else: headers["Content-Type"] = content_type headers["Content-Length"] = str(len(prefix) + file_size + len(suffix)) if DEBUG_FLAG: logger.info(f"~ Not uploading: {method} {url} <- {upload_file}") return 200, "" if verbose: logger.info(f"~ Uploading: {method} {url} <- {upload_file} ({file_size} bytes)") connection_class = http.client.HTTPSConnection if parsed_url.scheme == "https" else http.client.HTTPConnection connection_args: Dict[str, Any] = {} if parsed_url.scheme == "https": connection_args["context"] = ssl.create_default_context() # noinspection bad-argument-type connection = connection_class(parsed_url.hostname, parsed_url.port, **connection_args) try: connection.putrequest(method, request_target) for header_name, header_value in headers.items(): connection.putheader(header_name, header_value) connection.endheaders() if prefix: connection.send(prefix) with open(upload_file, "rb") as file: while chunk := file.read(1024 * 1024): connection.send(chunk) if suffix: connection.send(suffix) response = connection.getresponse() response_body = response.read().decode("utf-8", errors="replace") if verbose: logger.info(f"~ Nexus response: HTTP {response.status} {response.reason}") return response.status, response_body finally: connection.close() def download_file_http(url: str, dest_file: str, username: str, password: str, verbose: bool = False) -> tuple[int, str]: """Downloads a file from an HTTP endpoint using basic authentication.""" parsed_url = urlsplit(url) if parsed_url.scheme not in ["http", "https"] or not parsed_url.hostname: raise ValueError(f"Unsupported or invalid download URL: {url}") request_target = parsed_url.path or "/" if parsed_url.query: request_target += f"?{parsed_url.query}" headers = { "Authorization": "Basic " + base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii"), } if DEBUG_FLAG: logger.info(f"~ Not downloading: GET {url} -> {dest_file}") return 200, "" if verbose: logger.info(f"~ Downloading: GET {url} -> {dest_file}") connection_class = http.client.HTTPSConnection if parsed_url.scheme == "https" else http.client.HTTPConnection connection_args: Dict[str, Any] = {} if parsed_url.scheme == "https": connection_args["context"] = ssl.create_default_context() # noinspection bad-argument-type connection = connection_class(parsed_url.hostname, parsed_url.port, **connection_args) try: connection.request("GET", request_target, headers=headers) response = connection.getresponse() if response.status < 200 or response.status >= 300: response_body = response.read().decode("utf-8", errors="replace") if verbose: logger.info(f"~ Nexus response: HTTP {response.status} {response.reason}") return response.status, response_body os.makedirs(os.path.dirname(dest_file) or ".", exist_ok=True) with open(dest_file, "wb") as file: while chunk := response.read(1024 * 1024): file.write(chunk) if verbose: logger.info(f"~ Nexus response: HTTP {response.status} {response.reason}") return response.status, "" finally: connection.close() def query_nexus_search(url: str, username: str, password: str, verbose: bool = False) -> tuple[int, Any]: """Performs an HTTP GET request to Nexus search API and returns parsed JSON data.""" parsed_url = urlsplit(url) if parsed_url.scheme not in ["http", "https"] or not parsed_url.hostname: raise ValueError(f"Unsupported or invalid search URL: {url}") request_target = parsed_url.path or "/" if parsed_url.query: request_target += f"?{parsed_url.query}" headers = { "Authorization": "Basic " + base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii"), "Accept": "application/json", } if DEBUG_FLAG: logger.info(f"~ Not querying: GET {url}") return 200, {"items": []} if verbose: logger.info(f"~ Querying: GET {url}") connection_class = http.client.HTTPSConnection if parsed_url.scheme == "https" else http.client.HTTPConnection connection_args: Dict[str, Any] = {} if parsed_url.scheme == "https": connection_args["context"] = ssl.create_default_context() # noinspection bad-argument-type connection = connection_class(parsed_url.hostname, parsed_url.port, **connection_args) try: connection.request("GET", request_target, headers=headers) response = connection.getresponse() response_body = response.read().decode("utf-8", errors="replace") if verbose: logger.info(f"~ Nexus response: HTTP {response.status} {response.reason}") if response.status < 200 or response.status >= 300: return response.status, response_body try: data = json.loads(response_body) return response.status, data except json.JSONDecodeError as ex: logger.error(f"! Failed to parse JSON response: {ex}") return response.status, response_body finally: connection.close() def remove_tree(dir_name: Path) -> None: """ Remove the complete passed directory tree, even when some files are read-only. Needed to remove Git read-only files or directories. :param dir_name: Directory name as a Path object or string. :return: None """ # noinspection PyUnusedLocal def remove_readonly(func, path, exc_info): """ Change the file mode before deleting to allow deletion. """ # Check if the file is writable before changing it. if os.path.exists(path) and not os.access(path, os.W_OK): # Clear the read-only bit. os.chmod(path, stat.S_IWRITE) # Retry the deletion. func(path) shutil.rmtree(dir_name, onerror=remove_readonly) def menu_selection(options: dict[Any, str], title: str | None = "Make a Selection", caption: str = "Select an option?" ) -> Any | None: """ Curses menu styled after the Linux 'dialog' utility. Returns: Associated key (Any) or 'None' if canceled. """ # Convert dictionary values to a list for positional rendering and keys to a list for retrieval by index. option_keys = list(options.keys()) option_values = list(options.values()) # Key conversion for when using in Wine. wine_conversion = { 450: curses.KEY_UP, 456: curses.KEY_DOWN, 452: curses.KEY_LEFT, 454: curses.KEY_RIGHT, 449: curses.KEY_PPAGE, 455: curses.KEY_NPAGE, 447: curses.KEY_HOME, 448: curses.KEY_END, } if is_wine() else {} def _get_key(win): """ Determines and returns a meaningful key representation or raw key code based on input from the provided window object. :param win: A window object that provides the `getch` method to capture input. :type win: Any :return: Returns a string representing directional keys ("UP", "DOWN", "LEFT", "RIGHT") if the input matches predefined key codes. Otherwise, returns the raw key code. :rtype: Union[str, int] """ key = win.getch() if key in wine_conversion: key = wine_conversion[key] return key def _menu(std_scr): # Initialize Colors curses.start_color() # Some terminals require this to enable bright colors with A_BOLD # noinspection PyBroadException try: curses.use_default_colors() except Exception: pass # Pair 1: Cyan on a blue background for the screen. curses.init_pair(1, curses.COLOR_CYAN, curses.COLOR_BLUE) # Pair 2: Black text on a gray background for a dialog box. curses.init_pair(2, curses.COLOR_BLACK, curses.COLOR_WHITE) # Pair 3: Bright White text on a blue background for selection. curses.init_pair(3, curses.COLOR_WHITE, curses.COLOR_BLUE) # Pair 4: Shadow (Black on Black). curses.init_pair(4, curses.COLOR_BLACK, curses.COLOR_BLACK) # curses.curs_set(0) std_scr.keypad(True) # -Startup Resize Trigger. curses.ungetch(curses.KEY_RESIZE) current_row = 0 # Track the first visible item in the list. top_offset = 0 while True: # Always get current screen dimensions at the start of the loop sh, sw = std_scr.getmaxyx() # Setup Background std_scr.bkgd(' ', curses.color_pair(1)) std_scr.erase() # Main Screen Title - Trimmed if needed safe_title = title[:max(1, sw - 4)] if sw > 2: std_scr.addstr(0, 1, f"{safe_title}", curses.A_BOLD | curses.color_pair(1)) std_scr.addstr(1, 1, "\u2500" * (sw - 2), curses.A_BOLD | curses.color_pair(1)) # Calculate Dimensions dynamically based on current sh, sw max_box_height = max(1, sh - 6) needed_height = len(option_values) + 4 height = min(needed_height, max_box_height) # Calculate width based on caption and the longest display string width = max(len(opt) for opt in [caption] + option_values) + 10 width = min(width, sw - 4) # Ensure minimum window size to prevent newwin errors height = max(3, height) width = max(10, width) start_y = max(0, (sh - height) // 2) start_x = max(0, (sw - width) // 2) # Calculate how many items can be displayed. visible_count = max(0, height - 4) # Adjust the scrolling of the window. if current_row < top_offset: top_offset = current_row elif current_row >= top_offset + visible_count: top_offset = max(0, current_row - visible_count + 1) # Draw Shadow if start_y + height < sh and start_x + width < sw: # noinspection PyBroadException try: shadow = curses.newwin(height, width, start_y + 1, start_x + 1) shadow.bkgd(' ', curses.color_pair(4)) std_scr.noutrefresh() shadow.noutrefresh() except Exception: std_scr.noutrefresh() else: std_scr.noutrefresh() # Setup Dialog Window # noinspection PyBroadException try: menu_win = curses.newwin(height, width, start_y, start_x) menu_win.bkgd(' ', curses.color_pair(2)) menu_win.box() menu_win.keypad(True) # Draw the Caption ON TOP of the border if width > 4: trimmed_caption = caption[:width - 4] menu_win.addstr(0, 2, f" {trimmed_caption} ", curses.color_pair(2)) # Draw the options with scrolling logic for i in range(visible_count): idx = i + top_offset if idx >= len(option_values): break row_text = option_values[idx] y, x = i + 2, 4 item_width = width - 8 if item_width > 0: display_text = row_text[:item_width].ljust(item_width) if idx == current_row: menu_win.attron(curses.color_pair(3) | curses.A_BOLD) menu_win.addstr(y, x, display_text) menu_win.attroff(curses.color_pair(3) | curses.A_BOLD) else: menu_win.addstr(y, x, display_text, curses.color_pair(2)) menu_win.refresh() except Exception: std_scr.refresh() menu_win = std_scr # Input Handling key = _get_key(menu_win) if key == curses.KEY_UP and current_row > 0: current_row -= 1 elif key == curses.KEY_DOWN and current_row < len(option_values) - 1: current_row += 1 elif key == curses.KEY_PPAGE: current_row = max(0, current_row - max(1, visible_count)) elif key == curses.KEY_NPAGE: current_row = min(max(0, len(option_values) - 1), current_row + max(1, visible_count)) elif key == curses.KEY_HOME: current_row = 0 elif key == curses.KEY_END: current_row = max(0, len(option_values) - 1) elif key in [curses.KEY_ENTER, 10, 13]: # Return the key associated with the selected value return option_keys[current_row] # Keys to quit on. elif key in [27, ord('q'), ord('Q')]: return None elif key == curses.KEY_RESIZE: try: curses.update_lines_cols() except AttributeError: pass std_scr.clear() rv = curses.wrapper(_menu) # Write something to asure that lines are written after existing text in the console. logger.info(f"# Menu selection: {rv}") return rv def ask_selection(options: dict[Any, str], title: str | None = "Make a Selection", caption: str = "Select an option?" ) -> Any | None: """ Displays a numeric selection menu from a dictionary and returns the corresponding key. - options: Dict where value is displayed and key is returned. - Q-key : returns None. - Invalid input: prompts again. """ # Copy the arguments to pass on to the menu function. func_args = locals().copy() # Check if the 'fancy' attribute flag is set. fancy_menu = ask_selection.__getattribute__("fancy") if hasattr(ask_selection, "fancy") else True # When a dumb terminal is detected, use the curses menu. if not TERM_DUMB and fancy_menu: return menu_selection(**func_args) # Continue as a normal console menu. print(title) # When all keys are boolean act differently. if all(isinstance(k, bool) for k in options): while True: reply = input(f"{caption} (y/n/q): ").strip().lower() if reply == 'y': return True if reply == 'n': return False if reply == 'q': break else: # Create a stable list of keys to map numbers to the original dict keys. keys_list = list(options.keys()) while True: print("[0] Quit") for i, key in enumerate(keys_list, start=1): print(f"[{i}] {options[key]}") reply = input(f"\n{caption} ").strip().lower() if reply == 'q' or reply == '0': break if reply.isdigit(): selection_idx = int(reply) - 1 if 0 <= selection_idx < len(keys_list): return keys_list[selection_idx] else: print(f"Error: Please enter a number between 1 and {len(options)}.") else: print("Error: Invalid input. Enter a number or 'q' to quit.") return None class CallbackEnvironment(dict): """ Callback for getting environment variables and when missing raises an Exception. """ def __init__(self, environment: Dict[str, str], context: Optional[Dict[str, str]] = None, note_str: str | None = None ): super().__init__() self.note = note_str self.context = context self.environment = environment def __getitem__(self, key): value = None if key in self.environment: value = self.environment.get(key, None) elif self.context is not None and key in self.context: value = self.context[key] # Check if still not found. if value is None: if key in ENV_OPTIONAL: logger.debug(f": Missing environment variable '{key}' returning empty string instead.") return "" exception = RuntimeError(f"Missing environment variable '{key}' !") exception.add_note(str(self.note)) raise exception return value def set_environment(compiler_type: str | None = None) -> None: """Sets the run environment according to the configuration in the ini-file.""" # Key name used for inheritance. inherit_key = "__inherit__" def fix_wine_path(wine_path: str) -> str: """ Fixes the given Wine path by resolving nested symlinks into a usable path. This is useful when sharing toolchain directories between projects using symlinks. """ result: List[str] = [] for d in wine_path.split(";"): if d.startswith("Z:"): d = os.path.realpath(d[2:].replace("\\", "/")) result += ["Z:" + d.replace("/", "\\")] else: if d.startswith("/"): d = "Z:" + d.replace("/", "\\") result += [d] return ";".join(result) def get_config_inheritance(_section: str) -> List[str]: """ Gets the inheritance of the passed section. """ visited = [] queue = [_section] while queue: sec = queue.pop(0) # Report if a section is encountered more than once in the tree if sec in visited: logger.warning(f"Notice: Section [{sec}] was inherited more than once.") continue visited.append(sec) # Add the inherited sections to the queue. secs = [p.strip() for p in CONFIG.get(sec, inherit_key, fallback="").split(",") if p.strip()] secs.reverse() queue += secs visited.reverse() return visited # Reset the RUN_ENV dictionary to the parent environment. global RUN_ENV # Start with a fresh copy of the parent environment. RUN_ENV = PARENT_ENV.copy() # Assemble the name of the section. parts: List[str] = ["env"] parts += [str(compiler_type if compiler_type is not None else sys.platform)] if is_wine(): parts += ["wine"] if is_docker(): parts += ["docker"] section = '.'.join(parts) + '@' system_section = section + platform.node() # Check if the system-named entry exists in the configuration. if CONFIG.has_section(system_section): logger.info(f"# Using environment configuration '{system_section}'.") section = system_section else: logger.info(f"# Using environment configuration '{section}' instead of '{system_section}'.") for cur_section in get_config_inheritance(section): for key, value in get_config_section(cur_section).items(): if key == inherit_key: continue if key not in ENV_IGNORED: RUN_ENV[key] = Template(value).safe_substitute( CallbackEnvironment(environment=RUN_ENV, note_str=f"Configuration section: {section}")) # Fix the path since it must be with Windows backslashes only. if key == "WINEPATH": RUN_ENV[key] = fix_wine_path(RUN_ENV.get(key, "")) logger.info(f"~ Environment Set: {key}={RUN_ENV[key]}") else: logger.info(f"~ Environment Ignored: {key}") def get_compiler_type(preset_name: str, preset_type: PresetTypes = PresetTypes.CONFIGURE) -> str | None: """ Gets the compiler type from the field 'vendor/compiler' of the 'configure' preset. Alternatively, the 'cacheVariables/SF_COMPILER/value' field is tried. :param preset_name: Name of preset. :param preset_type: Type of preset defaulting to CONFIGURE. :return: None when not found and a string otherwise. """ if (cpn := get_configure_preset_name(preset_type, preset_name)) is not None: if (pn := get_preset_by_name(PresetTypes.CONFIGURE, cpn)) is not None: # Use the 'vendor' from the preset configuration first. compiler_type = pn.get("vendor", {}).get("compiler", None) # Try the cache variable when not found. if compiler_type is not None: logger.debug(f"# Compiler type from field 'vendor/compiler': {compiler_type}") else: compiler_type = pn.get("cacheVariables", {}).get("SF_COMPILER", {}).get("value", None) if compiler_type is not None: logger.debug(f"# Compiler type from field 'cacheVariables/SF_COMPILER': {compiler_type}") # Return the type. return compiler_type else: logger.warning(f"! No {PresetTypes.CONFIGURE.value} preset found for '{preset_type.value}/{preset_name}'!") return None def set_environment_by_preset(preset_name: str, preset_type: PresetTypes = PresetTypes.CONFIGURE) -> bool: """ Sets the toolchain environment using the preset field 'vendor/compiler' of the 'configure' preset. Alternatively, the 'cacheVariables/SF_COMPILER/value' field is tried. :param preset_name: Name of preset. :param preset_type: Type of preset defaulting to CONFIGURE. :return: True if the toolchain environment was set, False otherwise. """ compiler_type = get_compiler_type(preset_name, preset_type) if compiler_type is not None: # Modify the environment. set_environment(compiler_type) return True return False def expand_macros(preset: Optional[dict], value: Any, is_path: bool = False, context: Dict[str, str] | None = None) -> Any: """ Recursively expands macros, substituting environment variables in strings from CMakePresets.json. """ if value is None: return None if isinstance(value, dict): return {k: expand_macros(preset, value=v) for k, v in value.items()} if isinstance(value, list): return [expand_macros(preset, v) for v in value] if not isinstance(value, str): return value preset_name = preset.get("name", "unknown") def env_replacer(match): """Callback function for regular expression substitution.""" env_src = match.group(1) var_name = match.group(2) return CallbackEnvironment(environment=RUN_ENV, context=context if env_src == "env" else {}, note_str=f"Replacing in preset '{preset_name}' variable '${env_src}{{{var_name}}}'.")[var_name] pat = r'\$(env|penv)\{([^}]+)\}' while re.search(pat, value): value = re.sub(pat, env_replacer, value) if is_path and sys.platform == 'win32': value = value.replace('/', os.sep) # Expand CMakePreset macros at the end. value = value.replace("${presetName}", preset_name) value = value.replace("${sourceDir}", RUN_DIR) value = value.replace("${sourceParentDir}", os.path.dirname(RUN_DIR)) value = value.replace("${fileDir}", RUN_DIR) value = value.replace("${pathListSep}", os.pathsep) value = value.replace("${hostSystemName}", "Windows" if sys.platform == 'win32' else "Linux") value = value.replace("${dollar}", "$") return value def run_command(cmd_list: List[str], input_data: bytes | None = None, shell: bool = False, capture_output: bool = False, check: bool = True, cwd: str | None = None, dbg_mode: DebugMode = DebugMode.REPORT) -> subprocess.CompletedProcess: """ Utility to run shell commands. Raises 'subprocess.CalledProcessError' if the command fails. """ cwd = os.getcwd() if cwd is None else cwd # When debugging, and the command is to report only. if DEBUG_FLAG and dbg_mode == DebugMode.REPORT_ONLY: cmd_str = " ".join(cmd_list) logger.info(f"~ Not executing from ({cwd}): {cmd_str}") # Simulate a completion. return subprocess.CompletedProcess(args=cmd_list, returncode=0) # Report when not mode is not silent. if dbg_mode != DebugMode.SILENT or DEBUG_FLAG: cmd_str = " ".join(cmd_list) logger.info(f"~ Executing from({cwd}): {cmd_str}") # Raises a 'CalledProcessError' exception on error. try: result = subprocess.run(cmd_list, shell=shell, cwd=cwd, check=check, env=RUN_ENV, capture_output=capture_output, input=input_data) return result except Exception as ex: ex.add_note(f"Subprocess: {' '.join(cmd_list)}") raise ex def get_merged_presets() -> dict: """ Get the merged presets from CMakePresets.json and CMakeUserPresets.json. :return: A dictionary containing the merged presets. """ # Use a cached version of the merged presets if available. if hasattr(get_merged_presets, "merged_data"): return get_merged_presets.merged_data else: get_merged_presets.merged_data = {} # Make 'merged' reference the attribute. merged = get_merged_presets.merged_data def deep_merge(_source, _destination): """ Recursively merges the contents of the source dictionary into the destination dictionary. If a key exists in both dictionaries and its value is also a dictionary, this function will recursively merge those nested dictionaries. For non-dictionary values, the value from the source dictionary will overwrite or add to the destination. """ # Make a deep copy since at some point '_source' get corrupted somehow. for _key, _value in _source.items(): if isinstance(_value, dict) and _key in _destination and isinstance(_destination[_key], dict): # Recursively merge nested dictionaries. deep_merge(_value, _destination[_key]) else: # Overwrite or add the value. if _key not in _destination: _destination[_key] = copy.deepcopy(_value) return _destination base_path = os.path.join(RUN_DIR, "CMakePresets.json") user_path = os.path.join(RUN_DIR, "CMakeUserPresets.json") # Array Fields: Concatenate preset arrays (configure, build, test, etc.) preset_type_fields = [f"{item.value}Presets" for item in list(PresetTypes)] # Load files (handle missing UserPresets gracefully) with open(base_path, 'r') as f: merged.update(json.load(f)) if os.path.exists(user_path): with open(user_path, 'r') as f: data = json.load(f) # Versioning: UserPresets version typically takes precedence for the union. merged['version'] = max(merged.get('version', 1), data.get('version', 1)) if int(merged['version']) < 6: raise RuntimeError(f"{base_path} file is required to be version 6 or higher!") for field in preset_type_fields: if field in data: # Initialize the field in merged if it doesn't exist. if field not in merged: merged[field] = [] # Overwrite Rule: If a user preset has the same name as a base preset, the user preset overrides it. user_presets_dict = {p['name']: p for p in data[field]} # Remove existing presets in a base that usernames override. merged[field] = [p for p in merged[field] if p['name'] not in user_presets_dict] # Append all user presets (including the overrides) merged[field].extend(data[field]) # Vendor Maps: Shallow merge vendor-specific data if "vendor" in data: merged.setdefault("vendor", {}).update(data["vendor"]) # Merge all in inherited presets. for field in preset_type_fields: # Assemble a dictionary of presets by name in tuple containing a hash for a sanity check. presets: dict[str, tuple[str, dict]] = {} for preset in merged[field]: presets.setdefault(preset['name'], (hashlib.md5(f"{preset}".encode()).hexdigest(), preset)) # Merge inherited presets. for preset in merged[field]: inherits = preset.get("inherits", None) if inherits: cur_preset_name = preset['name'] cur_index = list(presets).index(cur_preset_name) # Iterate through the inherited presets and merge them in. for preset_name in inherits: # Can only merge a preset when defined earlier. if cur_index > list(presets).index(preset_name): deep_merge(presets[preset_name][1], preset) # Update the current hash only. presets[cur_preset_name] = (hashlib.md5(f"{presets[cur_preset_name][1]}".encode()).hexdigest(), presets[cur_preset_name][1]) # Sanity check on the changed after the deep-merge. for key in list(presets): if presets[key][0] != hashlib.md5(f"{presets[key][1]}".encode()).hexdigest(): logger.debug(f"! Preset '{key}' has changed due to merge of '{preset_name}' into '{cur_preset_name}'!") else: raise RuntimeError(f"Preset '{cur_preset_name}' inherits from '{preset_name}' before it was defined!") # Sanity check after merge. if cur_preset_name != preset['name']: logger.debug(f"! Failed to merge properly.") # Return the cached value. return get_merged_presets.merged_data def get_valid_presets(preset_type: PresetTypes) -> list[str]: """ Gets a list of valid presets of a specified type from the command output. :param preset_type: The type of preset to query. :return: A list of strings representing the names of the valid presets. """ # Use a cached version of the merged presets if available. if hasattr(get_valid_presets, "result"): return get_valid_presets.result lines = run_command(["cmake", "--list-presets", preset_type.value], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode("utf-8") get_valid_presets.result = re.findall(r'^\s+\"([a-zA-Z_\-]+)\"\s+-', lines, re.MULTILINE) return get_valid_presets.result def select_preset(preset_type: PresetTypes | str | None = None) -> str | None: """ Selects a preset or displays information on all available presets from CMakePresets.json. :param preset_type: """ # When showing information and a dialog is eminent, check if this is a dumb terminal. if TERM_DUMB and preset_type is not None: raise RuntimeError("Cannot select preset using a dialog when the terminal is dumb!") if preset_type is None: logger.info("# Information on all presets.") elif type(preset_type) is str: logger.info(f"# Information on presets named: {preset_type}") # Only used when a preset_type was passed. options: dict[str, str] = {} # Get valid configure presets to dismiss other presets by. valid_configure_presets = get_valid_presets(preset_type=PresetTypes.CONFIGURE) # Retrieve the preset data from CMakePresets.json data = get_merged_presets() for pt in list(PresetTypes) if type(preset_type) is not PresetTypes else [preset_type]: key = f"{pt.value}Presets" # Retrieve the by CMake considered valid presets. valid_presets = get_valid_presets(preset_type=pt) presets = data.get(key, []) # Filter out hidden presets. visible_presets = [] for p in presets: if p.get("name") in valid_presets: # Special check for workflow presets. The 'configure' preset [0] must be valid. if pt == PresetTypes.WORKFLOW: if len(steps := p.get("steps", [])): # The first step is always the 'configure' step, add the workflow entry when it is valid. if steps[0].get("type") == PresetTypes.CONFIGURE.value and steps[0].get("name") in valid_configure_presets: visible_presets += [p] else: visible_presets += [p] if visible_presets: for p in visible_presets: name = p.get("name", "") if type(preset_type) is str and preset_type != name: logger.debug(f"# Skipping {pt.value} preset: {name}") continue display_name = p.get("displayName", "") description = p.get("description", "") info_line = name if display_name: info_line += f" ({display_name})" if description: info_line += f": {description}" if type(preset_type) is PresetTypes: options[name] = info_line continue else: # noinspection unresolved-references logger.info(f"\t- {pt.value.title()}: {info_line}") # When showing 'test' presets also show the tests if available. if key == "configurePresets": cvs = p.get("cacheVariables", {}) for cvn in cvs: cve = cvs.get(cvn, {}) logger.info(f"\t\t~ {cvn}:{cve.get("type", "")}={cve.get("value", "")}") # When showing 'test' presets also show the tests if available. if key == "testPresets": lines = run_command(["ctest", "--preset", name, "--show-only"], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode("utf-8") lines = re.findall(r'^\s+(Test #.*)$', lines, re.MULTILINE) if len(lines): for line in lines: logger.info(f"\t\t~ {line}") else: logger.info(f"\t\t: Need cmake configuration step for this information.") # When showing 'workflow' presets also show the steps. if key == "workflowPresets": steps = p.get("steps", []) if steps: index = 0 for s in steps: index += 1 logger.info(f"\t\t~ Step #{index}: {s.get("type", "")}({s.get("name", "")})") # When no preset type is passed, just log the found entries. if type(preset_type) is not PresetTypes: return None # Return the selected preset string. # noinspection unresolved-references return ask_selection(options, title=f"{preset_type.value.title()} Selection", caption="Select a preset:") def select_target(preset_type: PresetTypes, preset_name: str) -> str | None: """ Selects a target for a specific preset type and name. :param preset_type: Preset type :param preset_name: Name of the preset. :return: The target name when selected, 'None' when not. """ if preset_type == PresetTypes.BUILD: lines = run_command(["cmake", "--build", "--preset", preset_name, "--target", "help"], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode("utf-8") options: dict[str, str] = {} for trg in re.findall(r'^([^/\s]+):', lines, re.MULTILINE): # Remove noise by skipping targets with slashes or _autogen in the name. if "/" in trg or "_autogen" in trg: continue options[trg] = trg return ask_selection(options, title=f"{preset_type.value.title()} Selection", caption="Select a target:") if preset_type == PresetTypes.TEST: cmd = ["ctest", "--preset", preset_name, "--show-only=json-v1"] data = json.loads(run_command(cmd, capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode("utf-8")) if type(data) is dict: tests = data.get("tests", []) options: dict[str, str] = {} for test in tests: name = test.get("name", "") label_props = [v for v in test.get("properties", []) if v['name'] == 'LABELS'] labels = "" if not len(label_props) else ", ".join(label_props[0]['value']) options[name] = f"{name} ({labels})" # Return the selected preset string. return ask_selection(options, title=f"{preset_type.value.title()} Selection", caption="Select a target:") return None def get_preset_by_name(preset_type: PresetTypes, preset_name: str) -> dict | None: """ Gets a typed preset dictionary by name. :param preset_type: Type of the preset, e.g. 'configure', 'build', 'test', 'package', 'workflow'. :param preset_name: Name of the preset defined in CMakePresets.json or CMakeUserPresets.json. :return: The preset dictionary or None when not found. """ for p in get_merged_presets().get(f"{preset_type.value}Presets", []): if p.get("name") == preset_name: return p return None def get_configure_preset_name(preset_type: PresetTypes, preset_name: str) -> str | None: """ Gets the 'configure' preset name from a given other type of preset. :param preset_type: Type of the preset, e.g. 'configure', 'build', 'test', 'package', 'workflow'. :param preset_name: Name of the preset defined in CMakePresets.json or CMakeUserPresets.json. :return: The preset dictionary or None when not found. """ # Nothing special here. if preset_type == PresetTypes.CONFIGURE: return preset_name if preset_type in [PresetTypes.BUILD, PresetTypes.TEST, PresetTypes.PACKAGE]: p = get_preset_by_name(preset_type, preset_name) if p is not None: return p.get("configurePreset", None) if preset_type == PresetTypes.WORKFLOW: p = get_preset_by_name(preset_type, preset_name) if p is not None: steps = p.get("steps", "") # Sanity check on the first step. if len(steps) and all(k in steps[0] for k in ["name", "type"]) and steps[0]["type"] == PresetTypes.CONFIGURE: return steps[0]["name"] return None class HelpAction(argparse._HelpAction): """Action handler for printing the help the intended way, which is otherwise not possible.""" class HelpException(Exception): """Exception class as a hack to avoid parser.exit() being called and print help the intended way.""" def __init__(self): super().__init__("Help requested.") def __call__(self, parser, namespace, values, option_string=None): parser.print_help() raise HelpAction.HelpException class SubCommand(ABC): """Subcommand handler class.""" # Static member to registry all subcommands. registry: Dict[str, SubCommand] = {} # Get the script's name. script = os.path.basename(__file__) def __init__(self, command: str, aliases: list[str] | None = None): """Initializes the SubCommand with a specific parser instance.""" # Command name. self.command: str = command # Command aliases. self.aliases: List[str] = aliases if aliases is not None else [] # Holds the sub-parser instance for this subcommand. self.parser: argparse.ArgumentParser | None = None # Get the configuration section. self._config = get_config_section("config", fail=True) def register(self) -> SubCommand: """Registers this class in to the static member of this same class.""" # Automatically register the instance upon creation. SubCommand.registry[self.command] = self return self @abstractmethod def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: """Create a sub parser for the command.""" self.parser = subparsers.add_parser(name=self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help=f"Run with command '{self.command}'.", description="") if self.parser is None: raise ValueError("Parser cannot be None") return self.parser @abstractmethod def options(self, parser: argparse.ArgumentParser): """Add options to the sub parser.""" parser.add_argument("-h", "--help", action=HelpAction, default=False, help="Shows the command's help.") # Configure the command line options. parser.add_argument("-d", "--dry-run", action="store_true", help="Show executed commands without executing them.") def parse_args(self, args: List[str], show_help: bool = True) -> argparse.Namespace | None: """ Parses the command line arguments for testing. :param args: Argument list. :param show_help: Show help on failure to parse. :return: None on failure and parsed arguments on success. """ parser = self.options(argparse.ArgumentParser(add_help=False, formatter_class=argparse.RawTextHelpFormatter)) parser.help = None parser.epilog = None # noinspection PyBroadException try: if show_help: return parser.parse_args(args, show_help=True) else: return parser.parse_known_args(args)[0] except Exception: if show_help: parser.print_help() return None @abstractmethod def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """Virtual function to be implemented by subclasses.""" # Check if debugging is enabled. global DEBUG_FLAG if args.dry_run: # Set the global debug flag. DEBUG_FLAG = True # Report also debugging. logger.setLevel(logging.DEBUG) logger.debug("# Logger set to level DEBUG.") if is_debugger_active(): logger.debug("# Debugger is active.") return 0 def print_help(self): """Prints the command help when the parser exists.""" if self.parser: self.parser.print_help() pass class SubCommandNative(SubCommand): """Subcommand handler for 'native' command.""" def __init__(self): super().__init__("native", ["n", "_"]) def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: self.parser = subparsers.add_parser(self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help="Run native on this Linux or Windows host.") self.parser.epilog = f""" examples: List files in the configure preset's binary directory: Linux: ./{self.script} {self.command} -p gnu-debug Windows: {self.script} {self.command} -p mingw-debug Run executable in with the working directory as the binary: Linux: ./{self.script} {self.command} -p gnu-debug SF_EXEC_DIR_SUFFIX=-gnu ./{self.script} {self.command} -p gnu-debug -- ./hello-world.bin Windows: SF_EXEC_DIR_SUFFIX=-msvc {self.script} {self.command} -p msvc-debug -- hello-world.exe """ if self.parser is None: raise ValueError("Parser cannot be None") return self.parser def options(self, parser: argparse.ArgumentParser): """Adds standard options to the given parser. """ # Adds the standard help option. super().options(parser) parser.add_argument("-i", "--info", action="store_true", help="Return information on presets.") parser.add_argument("-c", "--clean", action="store_true", help="Remove the built artifacts first (cmake option '--clean-first').") parser.add_argument("-f", "--fresh", action="store_true", help="Optional flag for adding (cmake option --fresh).") parser.add_argument("-D", "--debug", action="store_true", help="Optional flag for adding (cmake option --debug-output).") parser.add_argument("-V", "--verbose", action="store_true", help="Optional flag for adding (cmake/cpack option --verbose).") parser.add_argument("-F", "--fresh-all", action="store_true", help="Clears the build tree of all 'CMakeCache.txt' files.") parser.add_argument("-C", "--wipe", action="store_true", help="Wipe build directory contents.") parser.add_argument("-l", "--list-only", action="store_true", help="Lists tests only.") parser.add_argument("-m", "--make", action="store_true", help="Create build directory/makefiles only.") parser.add_argument("-b", "--build", action="store_true", help="Build target(s) and make config when it does not exist.") parser.add_argument("-B", "--build-only", action="store_true", help="Build target(s) only and fail when the configuration does not exist.") parser.add_argument("-t", "--test", action="store_true", help="Runs ctest using a test-preset.") parser.add_argument("-T", "--test-select", action="store_true", help="Runs ctest using a dialog selecting a test.") parser.add_argument("-R", "--test-regex", type=str, metavar="", help="Regular expression on which test names are to be executed.") parser.add_argument("-L", "--label-regex", type=str, metavar="", help="Regular expression selecting labels of tests to be executed.") parser.add_argument("-p", "--package", action="store_true", help="Create packages.") parser.add_argument("--profile", action="store_true", help="Profile execution of cmake.") parser.add_argument("-w", "--workflow", action="store_true", help="Runs workflow presets.") parser.add_argument("-n", "--target", type=str, metavar="", help="Overrides the build targets set in the preset by a single target.") parser.add_argument("-N", "--target-select", action="store_true", help="Selects a single target single target to build.") parser.add_argument("--no-fancy", action="store_true", help="Disables the fancy menu/dialog for selections.") parser.add_argument("preset", nargs="?", help="Single preset to process and when omitted a dialog is shown to select one.") # Create additional help text. parser.epilog = f""" examples: Get all project presets info: {self.script} --info Get single project presets by name: {self.script} --info gnu-debug Make/Build a preset: {self.script} -mb gnu-debug {self.script} --make -build gnu-debug Run all tests on a preset: {self.script} --test gnu-debug Run specific tests using a regex: {self.script} -t gnu-debug -R '^t_my-test' {self.script} -t gnu-debug -L 'Catch2' Run 'Catch2' labelled tests passing arguments to the test executable: {self.script} -t gnu-debug -L '^Catch2$' -- sf::Hello -section World Package a preset including an additional argument to cpack: {self.script} --package gnu-debug -- -DMY_VAR=4 Workflow (Make/Build/Test/Pack) a preset: {self.script} --workflow gnu-debug When configuring fails due to 'CMakeCache.txt' issues: {self.script} -fm gnu-debug (Applies only to the main project) {self.script} -Fm gnu-debug (Applies to all projects and modules) {self.script} -Cm gnu-debug (Nuclear option which wipes the build directory clean) """ return parser def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """ Handles the default/standard execution of the script. :return: Exit code. """ # Call parent to handle the common dry-run option. super().handle(args, args_left, args_right) # Set the menu to appear not fancy since in Wine it demolishes the terminal. ask_selection.fancy = not args.no_fancy and not is_wine() if args.info: # Passing None will show all available presets. select_preset(args.preset) return 0 # The 'configure' preset name also functioning as a flag as well. config_preset_name: str | None = None # The 'build' preset name also functioning as a flag as well. build_preset_name: str | None = None # Flag to indicate that the build step triggers the make/configure step. make_by_build: bool = False # Binary directory. bin_dir: str = "" if args.build or args.build_only: # Check if a preset was passed and if not, select one. build_preset_name = args.preset if args.preset else select_preset(PresetTypes.BUILD) if build_preset_name is None: return 1 # Get the referenced 'configure' preset name from the 'build' preset. if (config_preset_name := get_configure_preset_name(PresetTypes.BUILD, build_preset_name)) is None: logger.error(f"! Build preset '{build_preset_name}' does not reference a configure preset.") return 1 logger.info(f"# Using configure preset '{config_preset_name}' from build preset '{build_preset_name}'.") # Set the flag triggering a make by the build step. make_by_build = True # When configure step is to be made. if args.make or make_by_build: # Check if a preset was passed and if not, select one. if config_preset_name is None: config_preset_name = args.preset if args.preset else select_preset(PresetTypes.CONFIGURE) if config_preset_name is None: return 1 # Set the environment variables according to the preset's 'configure' preset. set_environment_by_preset(config_preset_name, PresetTypes.CONFIGURE) # Check if the preset was found. config_preset = get_preset_by_name(PresetTypes.CONFIGURE, config_preset_name) # When the preset was not found, exit. if config_preset is None: logger.error(f"! Configure preset with name '{config_preset_name}' not found.") return 1 # Get the binary directory from preset expanding then macros. bin_dir: str | None = expand_macros(config_preset, config_preset.get("binaryDir", None), True) if bin_dir is None: logger.error(f"! Field 'binaryDir' not found for configure preset '{config_preset_name}'.") return 1 else: logger.info(f"# Binary Directory: {bin_dir}") # Get the full path to the binary directory if not already absolute. bin_dir = os.path.abspath(bin_dir) # Check if the build directory needs to have cleaned all 'CMakeLists.txt' files recursively. if args.fresh_all: remove_files_from_tree(Path(str(bin_dir)), ["CMakeCache.txt"]) # When build only, do not configure. if not args.build_only: # Check if the directory should be wiped. if args.wipe and bin_dir: if os.path.exists(bin_dir): logger.info(f"# Wiping directory: {bin_dir}") if not args.dry_run: remove_tree(Path(bin_dir)) # When the build triggered a make and the binary directory exists, make is not needed. if args.make or (make_by_build and not os.path.exists(os.path.join(str(bin_dir), "CMakeCache.txt"))): # os.makedirs(bin_dir, exist_ok=True) # Logic for configuration cmd = ["cmake", "-Wno-author", "--preset", config_preset_name] # cmd.append("--trace") # Add the command option to delete the CMakeCache.txt file. if args.fresh: cmd.append("--fresh") if args.debug: cmd.append("--debug-output") if args.profile: cmd += ["--profiling-format=google-trace", "--profiling-output=profile-make.json"] # Execute the configure command for creating makefiles. run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY) if args.build or args.build_only: # Check if a preset was passed and if not, select one. if not os.path.exists(os.path.join(str(bin_dir), "CMakeCache.txt")): logger.error(f"! Missing build directory: {bin_dir}") return 1 cmd: List[str] = ["cmake", "--build", "--preset", str(build_preset_name)] cmd += ["--parallel", str(os.cpu_count())] if args.profile: cmd += ["--profiling-format=google-trace", "--profiling-output=profile-build.json"] if args.clean: cmd.append("--clean-first") if args.target_select: args.target = select_target(PresetTypes.BUILD, str(build_preset_name)) if args.target: cmd.extend(["--target", args.target]) logger.debug(f"# Select build target: {args.target}") if args.verbose: cmd.append("--verbose") if not args.target_select or args.target_select and not args.target is None: run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY) if args.test or args.list_only or args.test_select or args.test_regex: preset_name = args.preset if args.preset else select_preset(PresetTypes.TEST) if preset_name is None: return 0 # Set the environment variables according to the preset's 'configure' preset. set_environment_by_preset(preset_name, PresetTypes.TEST) if args.test_select and not args.list_only: target = select_target(PresetTypes.TEST, preset_name) if target is None: return 0 cmd = ["ctest", "--preset", preset_name] cmd.extend(['--tests-regex', f"^{target}$"]) if args.verbose: cmd.append("--verbose") run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY) else: cmd = ["ctest", "--preset", preset_name] if args.list_only: cmd.append("--show-only") if args.test_regex: cmd.extend(["--tests-regex", args.test_regex]) if args.label_regex: cmd.extend(["--label-regex", args.label_regex]) if args.verbose: cmd.append("--verbose") # Add the additional arguments from the command line to the executable run by ctest. if args_right: # Add separator as ctest requires it to pass the arguments to the test executable. cmd.append("--") cmd += args_right run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY) if args.package: preset_name = args.preset if args.preset else select_preset(PresetTypes.PACKAGE) if preset_name is None: return 0 # Set the environment variables according to the 'configure' preset. set_environment_by_preset(preset_name, PresetTypes.PACKAGE) cmd = ["cpack", "--preset", preset_name] if args.verbose: cmd.append("--verbose") # Add the additional arguments from the command line to cpack. if args_right: cmd += args_right run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY) if args.workflow: preset_name = args.preset if args.preset else select_preset(PresetTypes.WORKFLOW) if preset_name is None: return 0 # Set the environment variables according to the 'configure' preset. set_environment_by_preset(preset_name, PresetTypes.PACKAGE) cmd = ["cmake", "--workflow", "--preset", preset_name] run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY) if args.verbose: cmd.append("--verbose") logger.info("# Build script completed successfully.") return 0 class SubCommandWine(SubCommand): """Subcommand handler for the 'wine' command.""" def __init__(self): super().__init__("wine", ["w"]) def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: self.parser = subparsers.add_parser(self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help="Run in Wine on Linux. (uses a Git client/server solution)") if self.parser is None: raise ValueError("Parser cannot be None") return self.parser def options(self, parser: argparse.ArgumentParser): """Adds options to the given parser for the wine command.""" # Adds the standard help option. super().options(parser) # parser.add_argument("command", type=str, nargs="?", choices=["run"], help=f""" run -- - Runs a command in Wine . """) parser.add_argument("-g", "--git-server", action="store_true", help="Force start of git-server in the background.") # Define Epilog for the help message. parser.epilog = f""" examples: Compile using Microsoft Visual C++ on Linux ./{self.script} {self.command} -- -b msvc-debug """ def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """ Handles the 'wine' command execution of the script. :return: Exit code. """ # Call parent to handle the common dry-run option. super().handle(args, args_left, args_right) # noinspection PyDeprecation if shutil.which("wine") is None: logger.error(f"! The 'wine' command not found. Please install required packages.") return 1 # Set the run environment for wine since start_git_server() uses it. set_environment("wine") # Add this env var since Windows/Wine cannot handle symlinks. RUN_ENV["PY_SCRIPT_DIR"] = os.path.dirname(os.path.realpath(__file__)) # When there are arguments on the right, and the first argument does not start with a hyphen. if not args.git_server and args_right and args_right[0][:1] == '-': # Split the arguments again on the second '--' and parse the right command arguments and show help when invalid. if cmd_args := SubCommandNative().parse_args(split_arguments(args_right)[0], False): # Some commands demand the Git server to be present. args.git_server = cmd_args.make or cmd_args.build or cmd_args.workflow # Start running the git server only when needed or forced. if args.git_server: # Get the port or path for the Git server. port: int = start_git_server(int(RUN_ENV.get('GIT_SERVER_PORT', '8000'))) # Check for an error when the actual port is zero. if port == 0: return 1 # set the environment variables for the Git client script. RUN_ENV["GIT_SERVER_PORT"] = str(port) # Assemble the arguments to execute and start with wine itself. arguments: List[str] = ["wine"] # When the command is run. if args.command == "run": # When no the right arguments exist, execute "cmd". if not args_right: arguments += ["cmd"] else: arguments += ["python", str(self.script)] # Execute those arguments. if args_right: arguments += args_right # Suppress Wine fix-me messages when 'WINEDEBUG' is not set. if "WINEDEBUG" not in RUN_ENV: RUN_ENV["WINEDEBUG"] = 'fixme-all' logger.debug(f"# Running: WINEPATH='{RUN_ENV.get("WINEPATH", "")}' {' '.join(arguments)}") return run_command(arguments, dbg_mode=DebugMode.REPORT_ONLY).returncode class SubCommandDocker(SubCommand): """Subcommand handler for the 'docker' command.""" def __init__(self): super().__init__("docker", ["d"]) # The default container name. self._container: str = self._config.get("docker-container-name", "cpp_builder") def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: self.parser = subparsers.add_parser(self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help="Run in a docker environment(Linux only).") if self.parser is None: raise ValueError("Parser cannot be None") return self.parser def options(self, parser: argparse.ArgumentParser): """Adds options to the given parser for the docker command.""" # Adds the standard help option. super().options(parser) # Platform detection logic based on machine architecture. machine = platform.machine() default_platform = "arm64" if machine == 'aarch64' else "amd64" # Define Epilog for the help message. parser.epilog = f""" examples: Show the targets using the {default_platform} platform docker image: {self.script} --platform {default_platform} -- --info Show the uname information of the arm64 container without QT libraries: {self.script} --platform arm64 run -- uname -a """ parser.add_argument("command", type=str, nargs="?", choices=["pull", "run", "start", "wstart", "stop", "kill", "prune", "attach", "status", "sshd", "versions"], help=f""" pull - Pulls the docker image from the Docker registry. run -- - Runs a command as user 'user' in the container using Docker command. start - Starts a container named '{self._container}' in the background. wstart - Starts a container named '{self._container}' in the background and a wineserver for speed. attach - Attaches to the running container named '{self._container}'. status - Returns info of the running container '{self._container}'. prune - Remove unused data and anonymous volumes. stop/kill - Stops/Kills the container named '{self._container}'. versions - Shows versions of most installed applications within the container. sshd - Starts sshd service on port {SSHD_PORT} to allow remote control. """) # Check if a fixed docker image is set. if "docker-image" in self._config.keys(): parser.description = f""" Docker image from configuration: {self._config["docker-image"]} This ignores the options: --platform' """ parser.add_argument("-D", "--debug", action="store_true", dest="flag_debug", default=False, help=f"Sets the DEBUG environment flag for the entrypoint script.") parser.add_argument("-p", "--platform", default=default_platform, choices=['amd64', 'arm64'], help=f"Platform part forming the Docker image (default: '{default_platform}').") parser.add_argument("-n", "--no-build-dir", action="store_false", dest="flag_build_dir", default=True, help="Docker project builds in a regular cmake-build directory as a native build would.") def get_container_id(self) -> str: """Returns the ID of the running container if it exists.""" try: result = subprocess.run(["docker", "ps", "--filter", f"name={self._container}", "--quiet"], capture_output=True, text=True, check=True) return result.stdout.strip() except subprocess.CalledProcessError: return "" def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """ Handles the 'docker' command execution of the script. :return: Exit code. """ # Call parent to handle the common dry-run option. super().handle(args, args_left, args_right) def docker_command(options: List[str], image: str, cmd_args: List[str]) -> List[str]: """Determines whether to use 'docker exec' on a running container or 'docker run' for a new one.""" if self.get_container_id(): # If the container is already running (detached), use exec. full_cmd = ["docker", "exec"] if sys.stdin.isatty(): full_cmd += ["--interactive", "--tty"] full_cmd += [self._container, "sudo", "--login", "--user=user", "--"] full_cmd += cmd_args else: # Otherwise start a fresh container. full_cmd = ["docker", "run"] + options + [image] + cmd_args return full_cmd # Construct the Docker image name. img_name = f"nexus.scanframe.com/{args.platform}/gnu-cpp:24.04" # Check if the image name is overridden. img_name = get_config_section("config", fail=True).get("docker-image", img_name) logger.info(f"# Docker image used: {img_name}") # Prepare standard Docker options for running the container. docker_opts = ["--platform", f"linux/{args.platform}", "--rm", "--device", "/dev/fuse", "--cap-add", "SYS_ADMIN", "--cap-add", "SYS_PTRACE", "--security-opt", "apparmor:unconfined", "--security-opt", "seccomp=unconfined", "--hostname", platform.node(), "--user", "0:0", "--env", f"LOCAL_USER={os.getuid()}:{os.getgid()}", "--network", "host"] # Pass the Qt version to the container so the entry point downloads them from. if qt_ver := get_config_section("config", fail=True).get("qt-ver"): docker_opts += ["--env", f"SF_QT_VERSION={qt_ver}"] if args.flag_debug: docker_opts += ["--env", "DEBUG=1"] if qt_download_base := get_config_section("config", fail=True).get("qt-download-base"): docker_opts += ["--env", f"QT_DOWNLOAD_BASE={qt_download_base}"] # Only add TTY and interactive flags when stdin is a terminal. if sys.stdin.isatty(): docker_opts += ["--tty", "--interactive"] # Handle X11 Display forwarding if available on the host. if os.environ.get("DISPLAY"): # noinspection SpellCheckingInspection xauth = os.environ.get("XAUTHORITY") if not xauth: # noinspection SpellCheckingInspection xauth = Path.home() / ".Xauthority" # noinspection SpellCheckingInspection docker_opts += ["--env", "DISPLAY", "--volume", f"{xauth}:/home/user/.Xauthority:ro"] if "XDG_RUNTIME_DIR" in RUN_ENV: audio_socket = f"{RUN_ENV["XDG_RUNTIME_DIR"]}/pulse/native" # And the pulse socket is valid. if os.path.exists(audio_socket): logger.info(f"# Mapping pulse audio socket: {audio_socket}") docker_opts += ["--volume", f"{RUN_ENV["XDG_RUNTIME_DIR"]}:/run/user/{os.getuid()}"] # Add the graphics card for acceleration. The Piper library uses this. if os.path.exists("/dev/dri"): logger.info(f"# Granting access to the GPU") docker_opts += ["--device", "/dev/dri"] # When 'DBUS_SESSION_BUS_ADDRESS' environment variable is present. if "DBUS_SESSION_BUS_ADDRESS" in RUN_ENV: # This is needed when a Qt application wants to match the host's theme. docker_opts += ["--volume", f"/home/{RUN_ENV["USER"]}/.config:/home/user/.config"] # Map the project root directory. docker_opts += ["--volume", f"{RUN_DIR}:/mnt/project/{PROJECT_SUBDIR}:rw"] # Configure a specific build directory volume if requested. if args.flag_build_dir: build_dir = Path(RUN_DIR) / "cmake-build" / f"docker-{args.platform}-{qt_ver}".strip("-") build_dir.mkdir(parents=True, exist_ok=True) docker_opts += ["--volume", f"{build_dir}:/mnt/project/{PROJECT_SUBDIR}/cmake-build:rw"] # Set the working directory within the container. docker_opts += ["--workdir", f"/mnt/project/{PROJECT_SUBDIR}/"] # Determine the specific command and its arguments. command = args.command # Set the environment before calling a docker command. set_environment("docker") # Command Execution logic based on the provided command. if command == "pull": run_command(["docker", "pull", img_name], dbg_mode=DebugMode.REPORT_ONLY) elif command == "status": if not self.get_container_id(): logger.warning(f": Container '{self._container}' is not running.") return 1 if (exit_code := run_command(["docker", "ps", "--filter", f"name={self._container}"], dbg_mode=DebugMode.REPORT_ONLY).returncode) != 0: return exit_code return run_command(docker_command(docker_opts, img_name, ["ps", "ax"]), dbg_mode=DebugMode.REPORT_ONLY).returncode # elif command in ["stop", "kill"]: cntr_id = self.get_container_id() if cntr_id: logger.info(f"# Container ID '{cntr_id}' found. Performing {command}...") return run_command(["docker", command, cntr_id], dbg_mode=DebugMode.REPORT_ONLY).returncode else: logger.info(f": Container '{self._container}' is not running.") # elif command in ["start", "wstart"]: if self.get_container_id(): logger.warning(f": Container '{self._container}' is already running.") return 1 cmd = ["docker", "run"] + docker_opts + ["--name", self._container, "--detach", img_name] if command == "wstart": cmd += ["/bin/bash", "-c", "sleep infinity && wineserver -p"] else: cmd += ["sleep", "infinity"] return run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY).returncode # elif command == "attach": if not self.get_container_id(): logger.warning(f": Container '{self._container}' is not running.") return 1 cmd = ["docker", "exec"] if sys.stdin.isatty(): cmd += ["--interactive", "--tty"] cmd += [self._container, "sudo", "--login", "--user=user", "--"] return run_command(cmd + (args_right or []), dbg_mode=DebugMode.REPORT_ONLY).returncode # elif command == "sshd": cache_dir = Path.home() / "tmp" / f"{self._container}-cache" cache_dir.mkdir(parents=True, exist_ok=True) sshd_cmd = ["docker", "run"] + docker_opts + ["--name", self._container, "--volume", f"{cache_dir}:/home/user/.cache:rw", "--detach", img_name, "sudo", "--", "/usr/sbin/sshd", "-e", "-D", "-p", str(SSHD_PORT)] if run_command(sshd_cmd, dbg_mode=DebugMode.REPORT_ONLY).returncode == 0: logger.info(f"# SSHD service started on port {SSHD_PORT}. Connect with 'ssh -p {SSHD_PORT} user@localhost'.") else: return 1 # elif command == "run": return run_command(docker_command(docker_opts, img_name, (args_right or [])), dbg_mode=DebugMode.REPORT_ONLY).returncode # elif command == "versions": script: list[str] = CMAKE_LIB_SUBDIR + ["bin", "versions.sh"] target_script = os.path.join(RUN_DIR, *script) if not os.path.exists(target_script): logger.warning(f": Script '{os.path.join(*script)}' not found.") return 1 # Form the path for execution inside the container. target_script = f"/mnt/project/{PROJECT_SUBDIR}/{'/'.join(script)}" return run_command(docker_command(docker_opts, img_name, ["bash", target_script]), dbg_mode=DebugMode.REPORT_ONLY).returncode else: # The default behavior is to execute the discovered build script. target_script = f"/mnt/project/{PROJECT_SUBDIR}/{self.script}" return run_command(docker_command(docker_opts, img_name, [target_script] + (args_right or [])), dbg_mode=DebugMode.REPORT_ONLY).returncode return 0 class SubCommandInstall(SubCommand): """Subcommand handler for the 'install' command.""" def __init__(self): super().__init__("install", ["i"]) def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: self.parser = subparsers.add_parser(self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help="Install required build tools or a quick start boilerplate project.") if self.parser is None: raise ValueError("Parser cannot be None") return self.parser def options(self, parser: argparse.ArgumentParser): """Adds options to the given parser for the create command.""" # Adds the standard help option. super().options(parser) # Configure the command line options. parser.add_argument("-p", "--project", action="store_true", help="Install the cmake project directories and files from the template repository.") choices: List[str] = ["win-tools", "lnx-tools", "mingw", "msvc", "msvc-alt"] if sys.platform != "win32": choices.append("doxygen") parser.add_argument("-t", "--toolchain", type=str, choices=choices, help="""Install a portable toolchains for in Windows or Wine with which the Qt library is build. Choices are: win-tools - Multiple tools as CMake, Ninja, NSIS and Git client for Wine(Linux Only). lnx-tools - Multiple tools as appimagetool, runtime-aarch64 and runtime-x86_64. mingw - MinGW x86_64 v13.2.0 posix + msvcrt compiler compatible with the Qt library. msvc - MSVC 2022 x86_64 compatible with the Qt library preassembled from a Nexus repository. msvc-alt - MSVC 2022 x86_64 compatible with the Qt library from Microsoft itself. (Windows only) doxygen - Doxygen latest released version. (Linux only) """) choices: List[str] = [] if sys.platform == "win32": choices.append("win") else: choices = ["dce", "dio", "lnx", "win"] if platform.processor() == 'x86_64': choices += ["arm", "qemu"] parser.add_argument("-r", "--required", type=str, choices=choices, help="""Install required packages using the Debian 'apt' package manager on Linux or 'WinGet' for Windows. Choices are depended on the host platform: Linux: dce - Install 'docker-ce' latest version using an external source. dio - Install 'docker.io' package bundled with the distro. qemu - Packages for running another architecture in a docker container. lnx - Packages for architecture x86_64 or aarch64. arm - Packages x86_64 for aarch64 GCC x86_64 cross-compile. win - Packages x86_64 for Windows MinGW x86_64 cross-compile. Windows: win - Windows WinGet packages for build tools except a compiler(s). """) parser.add_argument("-e", "--env-file", type=str, metavar="", help="Create an environment file for the toolchain configured by the given configure preset.") def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """ Handles the 'create' command execution of the script. :return: Exit code. """ # Call parent to handle the common dry-run option. super().handle(args, args_left, args_right) if args.required: self.install_packages(args.required) # Check if to create project items when requested. if args.project: if not self.create_project(): return 1 if args.env_file: if not self.create_env_file(args.env_file): return 1 if args.toolchain: if not self.install_toolchain(args.toolchain): return 1 return 0 # noinspection PyMethodMayBeStatic def create_env_file(self, preset_name: str) -> bool: """Creates an environment file for the toolchain of the given 'configure' preset.""" # Get the compiler type and the preset itself. toolchain = get_compiler_type(preset_name) preset = get_preset_by_name(PresetTypes.CONFIGURE, preset_name) if toolchain is None or preset is None: logger.error(f"! Could not determine toolchain or find preset '{preset_name}'.") return False # Get the section name for the toolchain. parts = ["env", toolchain] if is_wine(): parts += ["wine"] if is_docker(): parts += ["docker"] section = '.'.join(parts) + '@' # Assemble the filename based on the toolchain and environment. ext = ".bat" if sys.platform == "win32" or is_wine() else ".sh" env_filename = f".tc-env.{'.'.join(parts[1:])}{ext}" env_file = os.path.join(RUN_DIR, env_filename) # Check if the section exists. if not CONFIG.has_section(section): logger.error(f"! No environment configuration section [{section}] found in the ini-file.") return False # Call set_environment to populate RUN_ENV with expanded macros. set_environment(toolchain) # Collect the variables that are defined in the section or its ancestors. # We want to exclude variables that are just inherited from the process environment and NOT modified. def get_config_inheritance(_section: str) -> List[str]: """ Gets the configuration inheritance for the given section. :param _section: :return: """ inherit_key = "__inherit__" visited = [] queue = [_section] while queue: sec = queue.pop(0) if sec in visited: continue visited.append(sec) secs = [p.strip() for p in CONFIG.get(sec, inherit_key, fallback="").split(",") if p.strip()] secs.reverse() queue += secs return visited config_vars = set() for cur_section in get_config_inheritance(section): if CONFIG.has_section(cur_section): for key in CONFIG[cur_section]: if key not in ["__inherit__"]: # configparser keys are case-insensitive and converted to lowercase. # We need to find the actual key name if possible, or just use it. # But RUN_ENV has the correct case. config_vars.add(key.upper()) # Add environment variables from the preset itself. preset_env = preset.get("environment", {}) for key, value in preset_env.items(): expanded_value = expand_macros(preset, value, context=preset_env) # Only update RUN_ENV if it's not already set by the toolchain section (which takes precedence for tools). # Actually, we want both, but if they overlap, which one should win? # Usually, presets might override some things. Let's update RUN_ENV with preset variables. RUN_ENV[key] = expanded_value config_vars.add(key.upper()) export_vars = ["SF_EXEC_DIR_SUFFIX", "LD_LIBRARY_PATH", "WINEPATH", "MSVC_ROOT"] lines = [] if sys.platform == "win32" or is_wine(): lines.append("@echo off") for key in sorted(config_vars): if key in RUN_ENV and key in export_vars: val = RUN_ENV[key] lines.append(f"set {key}={val}") # Special handling for LD_LIBRARY_PATH to prefix the PATH. if key == "LD_LIBRARY_PATH": lines.append(f"set PATH=%LD_LIBRARY_PATH%;%PATH%") else: lines.append("#!/usr/bin/env bash") for key in sorted(config_vars): if key in RUN_ENV and key in export_vars: val = RUN_ENV[key] lines.append(f"export {key}=\"{val}\"") # Special handling for LD_LIBRARY_PATH to prefix the PATH. if key == "LD_LIBRARY_PATH": lines.append(f"# set PATH=%LD_LIBRARY_PATH%;%PATH%") try: with open(env_file, "w") as f: f.write("\n".join(lines) + "\n") logger.info(f"# Environment file for preset '{preset_name}' with toolchain '{toolchain}': {env_file}") return True except OSError as e: logger.error(f"! Failed to create environment file '{env_file}': {e}") return False def install_toolchain(self, toolchain: str) -> bool: """Installs toolchains.""" def get_file_from_url(url: str, suffix: str | None = None) -> str: """Copies the file from the url to a temporary file.""" import requests with (tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp): try: req = requests.get(url, stream=True) if not req.ok: ex = FileNotFoundError(f"Failed to get file: {req.url}") ex.add_note(f"Reason: {req.status_code} {req.reason}") raise ex for chunk in req.iter_content(chunk_size=8192 * 16): tmp.write(chunk) return tmp.name finally: req.close() tmp.close() # Assemble the installation directory. install_dir = os.path.join(RUN_DIR, "lib", "toolchain") # When the installation directory is a symlink bailout. if os.path.islink(install_dir): logger.error(f": Installation directory '{install_dir}' is symlinked so it cannot be overwritten.") return False # Make sure it is fully created. os.makedirs(install_dir, exist_ok=True) # Get the base URL for downloading libraries. base_url = self._config.get(field := "library-base-url") if not base_url: logger.error(f"! Missing field '{field}' in configuration section.") return False match toolchain: case "win-tools": logger.info("# Installing Multiple tools as CMake, Ninja, NSIS and Git client for Wine.") if not extract_by_url(url := f"{base_url}/toolchain/win-x86_64-combi.tar.gz", os.path.join(install_dir, "win-x86_64-combi")): logger.error(f"! Failed to uncompress: {url}") return False case "lnx-tools": logger.info("# Installing Multiple tools as appimagetool, runtime-aarch64 and runtime-x86_64.") if not extract_by_url(url := f"{base_url}/toolchain/lnx-{platform.machine()}-combi.tar.gz", install_dir): logger.error(f"! Failed to uncompress: {url}") return False case "mingw": logger.info("# Installing MinGW x86_64 v13.2.0 posix + msvcrt compiler compatible with the Qt library.") if not extract_by_url(url := f"{base_url}/toolchain/w64-x86_64-mingw-1320-posix.tar.gz", install_dir): logger.error(f"! Failed to uncompress: {url}") return False case "msvc": logger.info("# Installing MSVC 2022 x86_64 compatible with the Qt library. (preassembled)") if not extract_by_url(url := f"{base_url}/toolchain/w64-x86_64-msvc-2022.tar.gz", install_dir): logger.error(f"! Failed to uncompress: {url}") return False case "msvc-alt": logger.info("# Installing MSVC 2022 x86_64 compatible with the Qt library.") # Download a Python file for downloading the MSVC. py_file = get_file_from_url( url="https://raw.githubusercontent.com/Scanframe/sf-cygwin-bin/refs/heads/master/portable-msvc.py", suffix=".py") if run_command([sys.executable, py_file, "--vs", "2022", "--target", "x64", "--accept-license"], cwd=install_dir, dbg_mode=DebugMode.REPORT_ONLY).returncode != 0: logger.error("! Failed to execute installer script.") return False case "doxygen": logger.info("# Installing Doxygen latest released version.") doxygen_dir = os.path.join(install_dir, "doxygen") if os.path.exists(doxygen_dir): logger.warning(f": Doxygen directory already exists: {doxygen_dir}") return False release = get_github_release("doxygen", "doxygen", "*.linux.bin.tar.gz") if release and release["assets"]: extract_by_url(release["assets"][0]["url"], install_dir, "doxygen", release["assets"][0]["digest"]) return False # Signal success. return True @staticmethod def create_project() -> bool: """ Creates and initializes a project by configuring Git, setting up basic files and directories, and incorporating a CMake submodule template. This function ensures that the required project structure is set up and optionally adds files or directories to a Git repository. :return: True indicates success and False indicates failure. """ # noinspection PyDeprecation if shutil.which("git") is None: logger.info("! Git is required and not installed run with option '--required' first!") return False # Assemble the git file or directory path. git_path = os.path.join(RUN_DIR, ".git") # Assemble the cmake/lib submodule directory path. dir_cmake_lib = str(os.path.join(*([RUN_DIR] + CMAKE_LIB_SUBDIR))) dir_tpl = os.path.join(RUN_DIR, dir_cmake_lib, "tpl") py_script = os.path.splitext(os.path.basename(__file__)) # Template files and their destinations. tpl_files = [("default.clang-format", [".clang-format"]), ("default.gitignore", [".gitignore"]), ("git-pre-commit-hook.sh", [".git", "hooks", "pre-commit"]), ("CMakePresets.json", ["CMakePresets.json"]), ("CMakeLists.cmake", ["CMakeLists.txt"]), ("README.md", ["README.md"]), ("AGENTS.md", ["AGENTS.md"]), ("", [str(py_script[0] + py_script[1])]), ("", [str(py_script[0] + ".ini")])] # Template files not to be added to git. tpl_files_ignored = [("user.cmake", ["user.cmake"]), ("CMakeUserPresets.json", ["CMakeUserPresets.json"])] # Template directories to copy from and to using lists. tpl_dirs = [(["cpack"], ["cmake", "cpack"])] # Check if Git is part of the project. if not os.path.exists(git_path): if ask_selection(options={True: "Yes", False: "No"}, title="Project is not a git repository!", caption="Initialize git repository?"): logger.info("# Initializing git repository...") if run_command(["git", "init", "--initial-branch=main"], dbg_mode=DebugMode.REPORT_ONLY).returncode != 0: logger.error("! Failed to initialize git repository with main branch!") return False else: logger.info("# Breaking by skipping git repository initialization.") return False # Check if the repository was installed 'cmake/lib' submodule. if not os.path.isdir(dir_cmake_lib): # Suggest installing the cmake project template. if ask_selection(options={True: "Yes", False: "No"}, title="Project Helper Repository", caption=f"Add submodule project helper in '{'/'.join(CMAKE_LIB_SUBDIR)}'?"): clone_options = { "main@https://github.com/Scanframe/sf-cmake.git": "GitHub Scanframe 'sf-cmake.git'", "main@https://git.scanframe.com/library/cmake-lib.git": "GitLab Scanframe 'cmake-lib.git'", "dev-hotfix@https://github.com/Scanframe/sf-cmake.git": "GitHub Scanframe 'sf-cmake.git' (hotfix)", "dev-hotfix@https://git.scanframe.com/library/cmake-lib.git": "GitLab Scanframe 'cmake-lib.git' (hotfix)" } # Only add these options when '__DEV' is set. if RUN_ENV.get("__DEV"): clone_options["zipfile@https://www.scanframe.com/export/cmake-lib.zip"] = "Zipped (dev only)" if selected := ask_selection( options=clone_options, title="Project template repository?"): branch, repo = selected.split("@") if repo: if repo[-4:] == ".git": if ask_selection(options={True: "Git Submodule", False: "Standalone Repository"}, title="Helper Repository Type", caption="Add repository as?"): cmd = ["git", "submodule", "add", "--branch", branch, "--", repo, '/'.join(CMAKE_LIB_SUBDIR)] else: cmd = ["git", "clone", "--branch", branch, "--", repo, '/'.join(CMAKE_LIB_SUBDIR)] if run_command(cmd, dbg_mode=DebugMode.REPORT_ONLY).returncode != 0: logger.error(f"! Failed to add submodule in '{'/'.join(CMAKE_LIB_SUBDIR)}'!") return False elif repo[-4:] == ".zip": with urlopen(repo) as response: # Use BytesIO to treat the downloaded bytes as a file-like object with zipfile.ZipFile(io.BytesIO(response.read()), 'r') as zf: for info in zf.infolist(): extracted_path = zf.extract(info, RUN_DIR) # Only attempt to restore Unix permissions if on a Unix-like system if sys.platform != "win32" and info.create_system == 3: unix_attributes = info.external_attr >> 16 if unix_attributes > 0: try: os.chmod(extracted_path, unix_attributes) except OSError: # Handle cases where chmod might fail (e.g., read-only filesystem) pass else: logger.info("# Breaking by skipping git submodule '{'/'.join(CMAKE_LIB_SUBDIR)}' installation.") return False # Check if the git submodule was installed and the template root is available. if os.path.isdir(dir_cmake_lib): # Add all directories from the tpl directory. for d in [d for d in os.listdir(os.path.join(dir_tpl, "root")) if os.path.isdir(os.path.join(dir_tpl, "root", d))]: tpl_dirs.append((["root", d], [d])) if ask_selection(options={True: "Yes", False: "No"}, title="Clone/copy CMake Submodule as Helper & Template Project", caption="Copy into the project root when not existing?"): # Iterate through all subdirectories in the cmake/lib submodule template root directory. for entry in tpl_dirs: # Assemble the full path of the destination subdirectory. dir_dest = os.path.join(RUN_DIR, *entry[1]) # Check if the destination subdirectory already exists and skip it if it does. if os.path.exists(dir_dest): logger.info(f"# Skipping subdirectory '{dir_dest}' since it exists already.") else: dir_src = os.path.join(dir_tpl, *entry[0]) logger.info(f"# Copying subdirectory '{dir_dest}' into the project root directory.") if not DEBUG_FLAG: if os.path.isdir(dir_src): shutil.copytree(dir_src, dir_dest, dirs_exist_ok=False) else: logger.warning(f": Source directory copytree '{dir_src}' missing.") else: logger.debug(f"~ Not copying dir tree from '{dir_src}'") # Iterate through all the files with their final destinations. for entry in tpl_files + tpl_files_ignored: # From the source filepath. src_file = os.path.join(dir_tpl, "root", entry[0]) # Form the actual destination filepath. dst_file = os.path.join(RUN_DIR, *entry[1]) # Check if the destination file already exists and skip it if it does. if os.path.exists(dst_file): logger.info(f"# Skipping file '{os.path.join(*entry[1])}' since it exists already.") else: logger.info(f"# Copying file '{entry[0]}' into '{os.path.join(*entry[1])}'.") if not DEBUG_FLAG: # Ignore if the source file does not exist. if len(src_file) > 0: if os.path.isfile(src_file): shutil.copy(src_file, dst_file) else: logger.warning(f": Source file copy '{src_file}' missing.") else: logger.debug(f"~ Not copying file '{src_file}'") if ask_selection(options={True: "Yes", False: "No"}, title="Add directories and files and to the Git repository", caption="Add to Git repository?"): for entry in tpl_dirs: # Assemble the full path of the destination subdirectory. dest = str(os.path.join(*entry[1])) if not os.path.isdir(os.path.join(RUN_DIR, dest)): logger.warning(f": Directory '{dest}' does not exist.") else: logger.info(f"# Adding directory '{dest}' to git repository.") if run_command(["git", "add", dest], dbg_mode=DebugMode.REPORT_ONLY).returncode != 0: logger.error(f"! Git failed to add directory '{dest}'.") for entry in tpl_files: # Assemble the full path of the destination file. dest = str(os.path.join(*entry[1])) if not os.path.isfile(os.path.join(RUN_DIR, dest)): logger.warning(f": File '{dest}' does not exist.") else: logger.info(f"# Adding file '{dest}' to git repository.") if run_command(["git", "add", dest], dbg_mode=DebugMode.REPORT_ONLY).returncode != 0: logger.error(f"! Git failed to add file '{dest}'.") return True @staticmethod def setup_cross_repo() -> None: """ On an x86_64 host, add the arm64 ports sources file (if missing) and register arm64 as a foreign dpkg architecture. """ arch_foreign: str = "arm64" sources_host_path: str = f"/etc/apt/sources.list.d/ubuntu.sources" sources_foreign_path: str = f"/etc/apt/sources.list.d/ubuntu-{arch_foreign}.sources" # logger.info(f"~ Adding 'Architectures' to Ubuntu sources file: {sources_host_path}") run_command(["sudo", "sed", "--in-place", "/^Types: deb$/{$!N; /^Types: deb\\nArchitectures:/!s/^Types: deb\\n/Types: deb\\nArchitectures: amd64 i386\\n/}", sources_host_path], dbg_mode=DebugMode.REPORT_ONLY) # if os.path.exists(sources_foreign_path): logger.info(f"~ APT Sources file already exists: {sources_foreign_path}") return sources_content: str = f"""\ Types: deb URIs: http://ports.ubuntu.com/ubuntu-ports Suites: noble noble-updates noble-backports Components: main universe restricted multiverse Architectures: {arch_foreign} Signed-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg Types: deb URIs: http://ports.ubuntu.com/ubuntu-ports Suites: noble-security Components: main universe restricted multiverse Architectures: {arch_foreign} Signed-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg """ # Write the file as root via 'sudo tee'; capture_output avoids tee echoing the content back to stdout. logger.info(f"~ Adding sources file: {sources_foreign_path}") run_command(["sudo", "tee", sources_foreign_path], input_data=sources_content.encode(), capture_output=True, dbg_mode=DebugMode.REPORT_ONLY) # Add the architecture. logger.info(f"~ Adding Architecture: {arch_foreign}") run_command(["sudo", "dpkg", "--add-architecture", arch_foreign], dbg_mode=DebugMode.REPORT_ONLY) @staticmethod def install_packages(target: str) -> None: """Installs the necessary packages depending on the environment Linux or Windows.""" logger.info(f"About to install required packages for ({target})...") # Prefix the target with the system name. target = sys.platform + '/' + target logger.info(f"# Target: {target}") # noinspection PyBroadException try: if target == "linux/dio": # noinspection PyDeprecation if shutil.which("docker"): logger.warning(f"# Command 'docker' command is already available.") return run_command( ["sudo", "apt-get", "--yes", "install", "docker.io"], dbg_mode=DebugMode.REPORT_ONLY) elif target == "linux/dce": # Check if the 'docker' command exists. # noinspection PyDeprecation if shutil.which("docker"): logger.warning(f"# Command 'docker' command is already available.") return # Destination of the Docker sources-file. sources_file = "/etc/apt/sources.list.d/docker-ce.sources" if os.path.exists(sources_file): logger.warning(f"# Docker source-file '{sources_file}' already exists?!") return # logger.info(f"# Installing docker target.") # Get distribution information distro = run_command(["lsb_release", "-is"], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode( "utf-8").strip().lower() codename = run_command(["lsb_release", "-cs"], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode( "utf-8").strip() arch = run_command(["dpkg", "--print-architecture"], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode("utf-8").strip() # Download GPG key gpg_url = f"https://download.docker.com/linux/{distro}/gpg" gpg_result = run_command(["wget", "-qO-", gpg_url], capture_output=True, dbg_mode=DebugMode.SILENT) if gpg_result.returncode != 0: logger.error(f"! Failed to download Docker GPG key from {gpg_url}") else: gpg_key = gpg_result.stdout.decode("utf-8") # Indent GPG key lines with a space gpg_key_indented = "\n".join(" " + line for line in gpg_key.splitlines()) # Create Docker sources file content docker_sources = f"""Types: deb URIs: https://download.docker.com/linux/{distro} Suites: {codename} Components: stable Architectures: {arch} Signed-By: {gpg_key_indented} """ run_command(["sudo", "tee", sources_file], input_data=docker_sources.encode(), dbg_mode=DebugMode.REPORT_ONLY) run_command(["sudo", "apt-get", "update"], dbg_mode=DebugMode.REPORT_ONLY) run_command(["sudo", "apt-get", "--yes", "upgrade"], dbg_mode=DebugMode.REPORT_ONLY) # Install Docker CE and add the user to the 'docker' group. run_command(["sudo", "apt", "install", "-y", "docker-ce"], dbg_mode=DebugMode.REPORT_ONLY) if username := os.environ.get("USER", ""): run_command(["sudo", "usermod", "-aG", "docker", username], dbg_mode=DebugMode.REPORT_ONLY) logger.info(f"# Added user '{username}' to the 'docker' group.") elif target == "linux/lnx": # Initial updates and prerequisites run_command(["sudo", "apt-get", "update"], dbg_mode=DebugMode.REPORT_ONLY) run_command(["sudo", "apt-get", "--yes", "upgrade"], dbg_mode=DebugMode.REPORT_ONLY) run_command( ["sudo", "apt-get", "--yes", "install", "wget", "curl", "gpg", "lsb-release", "software-properties-common", "ccache", "python3", "python3-venv", "python3-dev", "python3-pefile", "python3-pyelftools", "python-is-python3"], dbg_mode=DebugMode.REPORT_ONLY) # XCB and Qt6 dependencies # noinspection SpellCheckingInspection xcb_pkgs = ["xcb", "libxkbcommon-x11-0", "libxcb-xinput0", "libxcb-cursor0", "libxcb-shape0", "libxcb-icccm4", "libxcb-image0", "libxcb-keysyms1", "libxcb-render-util0", "libpcre2-16-0"] run_command(["sudo", "apt-get", "--yes", "install"] + xcb_pkgs, dbg_mode=DebugMode.REPORT_ONLY) # LLVM Repository check and add repo_list = run_command(["apt-add-repository", "--list"], shell=False, capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode("utf-8") # if not re.findall(r'^Suites:\s+llvm-toolchain', repo_list, re.MULTILINE): # # Use shell=True for complex pipe operations # run_command( # ["wget https://apt.llvm.org/llvm-snapshot.gpg.key -O - | sudo tee /etc/apt/trusted.gpg.d/apt.llvm.org.asc >/dev/null"], # shell=True) # codename = run_command(["lsb_release", "-sc"], capture_output=True, # dbg_mode=DebugMode.SILENT).stdout.decode("utf-8").strip() # repo_url = f"deb https://apt.llvm.org/{codename}/ llvm-toolchain-{codename} main" # run_command(["sudo", "apt-add-repository", "--yes", "--no-update", repo_url]) # Kitware Repository (Ubuntu only) if not len(re.findall(r'apt\.kitware\.com/ubuntu', repo_list, re.MULTILINE)): distro = run_command(["lsb_release", "-is"], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode( "utf-8").strip() if distro == 'Ubuntu': run_command( ["wget -O - https://apt.kitware.com/keys/kitware-archive-latest.asc 2>/dev/null | gpg --dearmor - | sudo tee /etc/apt/trusted.gpg.d/kitware.gpg >/dev/null"], shell=True, dbg_mode=DebugMode.REPORT_ONLY) codename_cs = run_command(["lsb_release", "-cs"], capture_output=True).stdout.decode("utf-8").strip() run_command(["sudo", "apt-add-repository", "--yes", "--no-update", f"deb https://apt.kitware.com/ubuntu/ {codename_cs} main"], dbg_mode=DebugMode.REPORT_ONLY) else: logger.info("# Cannot install latest 'cmake' from Kitware since this in no Ubuntu distribution.") # Final updates and main package installation run_command(["sudo", "apt-get", "update"], dbg_mode=DebugMode.REPORT_ONLY) run_command(["sudo", "apt-get", "--yes", "upgrade"], dbg_mode=DebugMode.REPORT_ONLY) # noinspection SpellCheckingInspection main_pkgs = ["make", "cmake", "ninja-build", "gcc", "g++", "doxygen", "graphviz", "libopengl0", "libgl1-mesa-dev", "libglu1-mesa-dev", "libxkbcommon-dev", "libxkbfile-dev", "libvulkan-dev", "libssl-dev", "dpkg-dev", "debhelper", "dh-make", "build-essential", "default-jre-headless", "chrpath", "clang-format"] run_command(["sudo", "apt-get", "--yes", "install"] + main_pkgs, dbg_mode=DebugMode.REPORT_ONLY) elif target == "linux/qemu": run_command(["sudo", "apt-get", "install", "-y", "qemu-user-static", "binfmt-support"], dbg_mode=DebugMode.REPORT_ONLY) elif target == "linux/win": run_command(["sudo", "apt-get", "install", "-y", "mingw-w64", "nsis"], dbg_mode=DebugMode.REPORT_ONLY) # Check if wine is installed using shutil.which (cleaner than command -v) # noinspection PyDeprecation if not shutil.which("wine"): run_command(["sudo", "apt-get", "--yes", "install", "wine"], dbg_mode=DebugMode.REPORT_ONLY) elif target == "linux/arm": # Sanity check. if platform.machine() != "x86_64": raise RuntimeError(f"Cannot install arm packages cross compiler on a non x86_64 machine!") SubCommandInstall.setup_cross_repo() run_command(["sudo", "apt-get", "--yes", "install", "gcc-aarch64-linux-gnu", "g++-aarch64-linux-gnu", "binutils-aarch64-linux-gnu"], dbg_mode=DebugMode.REPORT_ONLY) arch_check: str = run_command(["dpkg", "--print-foreign-architectures"], capture_output=True, dbg_mode=DebugMode.SILENT).stdout.decode("utf-8") if "arm64" in arch_check.splitlines(): # noinspection SpellCheckingInspection arm_pkgs = [ "gcc-aarch64-linux-gnu:amd64", "g++-aarch64-linux-gnu:amd64", "binutils-aarch64-linux-gnu:amd64", "libgles-dev:arm64", "libegl-dev:arm64", "libgl-dev:arm64", "libpcre2-16-0:arm64", "libglvnd-dev:arm64", "libpng16-16t64:arm64", "xcb:arm64", "libxkbcommon-x11-0:arm64", "libxcb-xinput0:arm64", "libxcb-cursor0:arm64", "libxcb-shape0:arm64", "libpulse0:arm64", "libxcb-icccm4:arm64", "libxcb-image0:arm64", "libxcb-keysyms1:arm64", "libxcb-render-util0:arm64", "libdbus-1-3:arm64", "libcairo-gobject2:arm64", "qemu-user-static:amd64", "libxkbcommon-dev:arm64", "libxkbfile-dev:arm64", "libglu1-mesa-dev:arm64", "libavformat60:arm64", "libxrandr2:arm64" ] run_command(["sudo", "apt-get", "--yes", "install"] + arm_pkgs, dbg_mode=DebugMode.REPORT_ONLY) else: logger.info("Architecture 'arm64' is not enabled and packages are therefore not installed!") elif target == "win32/win" or target == "cygwin/win": # Install only for Windows non-standard packages using pip. for pkg in ["pefile", "pyelftools", "requests"]: if subprocess.run([sys.executable, "-m", "pip", "show", pkg], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0: logger.info(f"- Pip package '{pkg}' is already installed.") else: logger.info(f"- Installing pip package '{pkg}'.") if subprocess.run([sys.executable, "-m", "pip", "install", pkg]).returncode > 0: logger.error(f"! Failed to install Python package '{pkg}'!") # WinGet packages to install. wg_pkgs = { "Git": "Git.Git", "7-Zip": "7zip.7zip", "CMake C++ build tool": "Kitware.CMake", "Ninja build system": "Ninja-build.Ninja", "Nullsoft Install System": "NSIS.NSIS", "Termurin JRE": "EclipseAdoptium.Temurin.21.JRE", "LLVM (Clang-Format)": "LLVM.LLVM", "Doxygen": "DimitriVanHeesch.Doxygen", "Graphviz": "Graphviz.Graphviz" } for name, pkg_id in wg_pkgs.items(): if name[:1] == "!": logger.warning(f": Install '{name[1:]}' manually from URL '{pkg_id}'!") continue # Check if installed - using a specific check to avoid exception on "not found" check_installed = run_command( ["winget", "list", "--disable-interactivity", "--accept-source-agreements", "--exact", "--id", pkg_id], capture_output=True, check=False, dbg_mode=DebugMode.SILENT) if check_installed.returncode == 0: logger.info(f"- WinGet Package '{name}' already installed.") else: logger.info(f"- Installing WinGet package '{name}' ...") run_command( ["winget", "install", "--disable-interactivity", "--accept-source-agreements", "--exact", "--id", pkg_id], dbg_mode=DebugMode.REPORT_ONLY) # When the apt-cyg shell script is found, install also the Cygwin packages. # noinspection PyDeprecation if target == "cygwin/win" and shutil.which("apt-cyg") is not None: # Cygwin Packages cg_pkgs = ["dialog", "recode", "perl-Image-ExifTool", "graphviz", "pcre", "jq", "unzip", "colordiff", "dos2unix"] for pkg in cg_pkgs: try: run_command(["apt-cyg", "install", pkg], dbg_mode=DebugMode.REPORT_ONLY) except subprocess.CalledProcessError: logger.exception( "Failed to install 1 or more Cygwin packages (Try the Cygwin setup tool when elevation is needed)!") sys.exit(1) else: logger.error(f"Invalid requirements target '{target}', see help for valid ones!") sys.exit(1) except subprocess.CalledProcessError as ex: # logger.exception automatically includes the stack trace (exc_info=True) logger.exception(f"! Failed to install 1 or more packages due to a command failure.") raise ex except Exception as ex: logger.exception(f"! An unexpected error occurred during package installation.") raise ex class SubCommandVersion(SubCommand): """Subcommand handler for the 'version' command for repository version reporting and bumping.""" @dataclass() class CommitCacheEntry: """Holds the commit message information.""" name: str date: datetime msg: str def __init__(self, commit_msg: str): """Constructor taking the formatted message string.""" lines = commit_msg.split("\n") self.name = lines[0] self.date = datetime.strptime(lines[1], "%a %b %d %H:%M:%S %Y %z") self.msg = "\n".join(lines[2:]) @staticmethod def get_format_string() -> str: """Get the commit message format string.""" return "%cn\n%cd\n%B" _header_regex = re.compile(r"^([a-z_\-]+)(\(([a-z_\-]+)\))?(!)?:\s(.*)$") _increment_order = {"none": 0, "patch": 1, "minor": 2, "major": 3} # Holds the type map from ini config file. _type_map: dict[str, tuple[str, str]] = {} # Cache for commit message retrieval. _cache_get_cmt_msg: dict[str, CommitCacheEntry] = {} # Cache for the git-describe-exact tag retrieval. _cache_git_describe_exact: dict[str, str] = {} # Module directory to work on. _module_dir: str = "." def __init__(self): super().__init__("version", ["v"]) # Read the type map from the ini config file. for key, value in get_config_section("config-ver-type-map", fail=True).items(): self._type_map[key] = (*value.split(',', 1),) def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: self.parser = subparsers.add_parser(self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help="Report or bump semantic version based on conventional commits.") self.parser.epilog = f""" examples: {self.script} version info {self.script} version bump -V {self.script} version bump --msg-file version-bump.msgs.json {self.script} version bump --module cmake/lib info -ad """ if self.parser is None: raise ValueError("Parser cannot be None") return self.parser def options(self, parser: argparse.ArgumentParser): """Adds options to the version subcommand.""" super().options(parser) parser.add_argument("-V", "--verbose", action="store_true", help="Output extra processing information.") parser.add_argument("-m", "--module", type=str, metavar="", help="Submodule directory to work on.") parser.add_argument("-c", "--commit", type=str, metavar="", help="Commit hash to tag as new version (defaults to last commit/HEAD when omitted).") parser.add_argument("-a", "--all", action="store_true", help="Use all commits and not only merge commits in the process.") parser.add_argument("-s", "--select", action="store_true", help="Interactively select the target commit since last release tag (respecting merges flag).") parser.add_argument("--msg-file", type=str, metavar="", help="JSON file containing replacement messages for commit hashes (for debugging).") parser.add_argument("-l", "--long", action="store_true", help="Use long (full) git hashes; default prints short hashes.") parser.add_argument("action", nargs="?", choices=["info", "bump"], help=""" info: Report repository version information on the merge commits. bump: Generate release notes from the last or given commit hash.""") parser.epilog = """ message header format: ()!: │ │ │ └─⫸ Summary in an imperative mood. │ │ └─⫸ Optional exclamation mark '!' indicating a breaking change. │ └─⫸ Commit Scope: common|compiler|config|cmake|changelog|docs-infra|pack|iface|etc... └─⫸ Commit Type: build|ci|chore|docs|feat|fix|perf|refactor|style|test|revert type to change map: """ for key, value in self._type_map.items(): parser.epilog += f" {key} {' ' * (10 - len(key))}{value[0]} {' ' * (8 - len(value[0]))}{value[1]}\n" parser.epilog += f""" examples: Report versioning information using the last commit to determine the new version: ./{self.script} {self.command} info Report versioning information using a by a dialog selected commit: ./{self.script} {self.command} info -s Create Markdown release-notes in 'doc/release' directory for specified hash or tag: ./{self.script} {self.command} bump -c b8d37e2 ./{self.script} {self.command} bump -c v0.1.0-rc.5 """ return parser def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """Handles the version reporting/bumping.""" super().handle(args, args_left, args_right) if args.module: self._module_dir = args.module merges_only = not args.all logger.info(f"# Commits processed: {"Merges only" if merges_only else "All"}") if args.select: commit_hash = self.select_commit(merges_only) elif args.commit: # Convert the short or tag to a full commit hash. commit_hash = self.git(["rev-parse", args.commit], check=True) else: commit_hash = self.git(["rev-parse", "--verify", "HEAD"], check=True) git_top_level = self.git(["rev-parse", "--show-toplevel"], check=True) cur_ver_tag, tag_found = self.get_latest_non_rc_tag() tag_annotation = self.get_tag_annotation(cur_ver_tag) if tag_found else "" msg_overrides = self.load_message_overrides(args.msg_file) long_hash = args.long if args.action == "info": self.report_version_tags(git_top_level, cur_ver_tag, tag_found, commit_hash, merges_only, msg_overrides, tag_annotation, long_hash=long_hash) self.bump_version(info_only=True, cur_ver_tag=cur_ver_tag, commit_hash=commit_hash, merges_only=merges_only, verbose=args.verbose, tag_found=tag_found, overrides=msg_overrides, git_top_level=git_top_level) elif args.action == "bump": self.bump_version(info_only=False, cur_ver_tag=cur_ver_tag, commit_hash=commit_hash, merges_only=merges_only, verbose=args.verbose, tag_found=tag_found, overrides=msg_overrides, git_top_level=git_top_level) else: logger.info("! Missing version command.") return 1 return 0 @staticmethod def escape_markdown(text: str | None) -> str: """Escapes basic Markdown characters.""" if text is None: return "" escaped = text replacements = { "\\": "\", "`": "`", "*": "*", "#": "#", "[": "[", "]": "]", } for src, dst in replacements.items(): escaped = escaped.replace(src, dst) return escaped # noinspection PyMethodMayBeStatic def git(self, args: List[str], check: bool = False) -> str: """Runs git command and returns stdout as string.""" result = run_command(["git", "-C", self._module_dir] + args, capture_output=True, check=check, dbg_mode=DebugMode.SILENT) return result.stdout.decode("utf-8").strip() def git_lines(self, args: List[str], check: bool = False) -> List[str]: """Runs git and returns non-empty lines.""" return [ln for ln in self.git(args, check=check).splitlines() if ln.strip()] # noinspection PyMethodMayBeStatic def git_describe_exact(self, commit_hash: str) -> str: """Returns exact matching tag or '...' when none.""" if commit_hash in self._cache_git_describe_exact: return self._cache_git_describe_exact[commit_hash] result = run_command(["git", "describe", "--exact-match", commit_hash], capture_output=True, check=False, dbg_mode=DebugMode.SILENT) result = result.stdout.decode("utf-8").strip() if result.returncode == 0 else "..." self._cache_git_describe_exact[commit_hash] = result return result def get_latest_non_rc_tag(self) -> Tuple[str, bool]: """Returns the latest non-RC tag and flag indicating existence.""" tags = [tag for tag in self.git_lines(["tag", "--list", "--format", "%(tag)"], check=False) if re.match(r"^v\d+\.\d+\.\d+$", tag)] if not tags: return "v0.0.0", False tags.sort(key=lambda t: tuple(int(x) for x in t[1:].split(".")), reverse=True) return tags[0], True # noinspection PyMethodMayBeStatic def get_tag_annotation(self, tag: str) -> str: """Returns annotation text for tag.""" result = run_command(["git", "cat-file", "tag", tag], capture_output=True, check=False, dbg_mode=DebugMode.SILENT) if result.returncode != 0: return "" raw = result.stdout.decode("utf-8", errors="ignore") parts = raw.split("\n\n", 1) return parts[1].strip() if len(parts) == 2 else "" # noinspection PyMethodMayBeStatic def get_git_tag_version(self) -> Tuple[str, str, str, str]: """Returns tuple (version, rc, commits, hash) parsed from git describe information.""" result = run_command(["git", "-C", self._module_dir, "describe", "--dirty", "--match", "v*.*.*"], capture_output=True, check=False, dbg_mode=DebugMode.SILENT) desc = result.stdout.decode("utf-8").strip() match = re.match( r"^v(?P[0-9]+\.[0-9]+\.[0-9]+)(-rc\.?(?P[0-9]+))?(-((?P[0-9]+)?(-(?P[a-z0-9]+))?)?)?(-dirty)?$", desc) if not match: return "0.0.0", "", "", "" return match.group("ver") or "0.0.0", match.group("rc") or "", match.group("commits") or "", match.group( "hash") or "" @staticmethod def git_tag_version_string(version_info: Tuple[str, str, str, str]) -> str: """Formats the git describe tuple into a version string.""" version, rc, commits, _ = version_info result = version if rc: result += f"~rc{rc}" if commits: result += f"+{commits}" return result # noinspection PyMethodMayBeStatic def load_message_overrides(self, msg_file: str | None) -> Dict[str, str]: """Loads message overrides from JSON or shell-format file.""" if msg_file is None: json_default = os.path.join(RUN_DIR, ".version-bump.json") if os.path.exists(json_default): msg_file = json_default else: return {} if not os.path.exists(msg_file): logger.warning(f": Message file '{msg_file}' not found, ignoring.") return {} with open(msg_file, "r", encoding="utf-8") as fh: data = json.load(fh) if isinstance(data, dict): return {k: str(v) for k, v in data.items()} logger.warning(f": Message file '{msg_file}' has unsupported JSON structure, ignoring.") return {} # noinspection PyMethodMayBeStatic def get_commit_message(self, commit_hash: str, overrides: Dict[str, str]) -> CommitCacheEntry: """Returns commit message with overrides applied.""" # Check the cache for the entry. if commit_hash in self._cache_get_cmt_msg: return self._cache_get_cmt_msg[commit_hash] # Get the formated message string. result = run_command( ["git", "-C", self._module_dir, "show", "--no-patch", f"--format={self.CommitCacheEntry.get_format_string()}", commit_hash], capture_output=True, check=False, dbg_mode=DebugMode.SILENT) result = result.stdout.decode("utf-8", errors="ignore").strip() entry = self.CommitCacheEntry(result) # When this commit msg is overridden, assign the message text. if commit_hash in overrides: entry.msg = overrides[commit_hash] # Cache the entry. self._cache_get_cmt_msg[commit_hash] = entry return entry # noinspection PyMethodMayBeStatic def collect_commits(self, cur_ver_tag: str, commit_hash: str, merges_only: bool, tag_found: bool) -> List[str]: """Collects commits between tag and commit hash (or from root when no tag).""" cmd = ["git", "-C", self._module_dir, "log"] if merges_only: cmd.append("--merges") cmd += ["--pretty=format:%H"] cmd.append(f"{cur_ver_tag}^..{commit_hash}" if tag_found else commit_hash) result = run_command(cmd, capture_output=True, check=False, dbg_mode=DebugMode.SILENT) if result.returncode != 0 and not result.stdout: return [] return [ln for ln in result.stdout.decode("utf-8").splitlines() if ln.strip()] def calculate_next_for_commit(self, cur_ver_tag: str, commit_hash: str, merges_only: bool, tag_found: bool, overrides: dict[str, str] ) -> tuple[str, str]: """Returns the prospective next version tag if bumped at commit_hash.""" effect_max = "minor" if not tag_found else "none" for commit in self.collect_commits(cur_ver_tag, commit_hash, merges_only, tag_found): msg_string = self.get_commit_message(commit, overrides).msg if not msg_string: continue msg_header = msg_string.split("\n", 1)[0] match = self._header_regex.match(msg_header) if not match: continue msg_type = match.group(1) msg_breaking = match.group(4) type_effect = self._type_map.get(msg_type, ("none", ""))[0] effect = "major" if msg_breaking else type_effect effect_max = self.compare_increments(effect, effect_max) if effect_max == "none": return cur_ver_tag, effect_max base_ver = cur_ver_tag[1:] if cur_ver_tag.startswith("v") else cur_ver_tag result = f"v{self.increment_version(base_ver, effect_max)}" return result, effect_max def select_commit(self, merges_only: bool) -> str: """Prompts the user to select a commit since the last tag.""" cur_ver_tag, tag_found = self.get_latest_non_rc_tag() commits = self.collect_commits(cur_ver_tag, "HEAD", merges_only, tag_found) if not commits: raise subprocess.CalledProcessError(returncode=1, cmd=["git", "-C", self._module_dir, "log"], output=b"", stderr=b"No commits found") options = {} for commit in commits: cmt_msg = self.get_commit_message(commit, {}) heading = cmt_msg.msg.split("\n", 1)[0].strip() display = f"{commit[:7]} | {cmt_msg.date} | {heading}" options[commit] = display choice = ask_selection(options=options, title="Select Commit", caption="Choose commit for version calculation") if choice is None: raise subprocess.CalledProcessError(returncode=1, cmd=["selection"], output=b"", stderr=b"No commit selected") return str(choice) @staticmethod def compare_increments(effect: str, effect_max: str) -> str: """Returns the maximum increment.""" return effect if SubCommandVersion._increment_order.get(effect, 0) > SubCommandVersion._increment_order.get( effect_max, 0) else effect_max @staticmethod def increment_version(version: str, effect: str) -> str: """Increments version string based on effect.""" parts = version.split(".") while len(parts) < 3: parts.append("0") major, minor, patch = [int(x) for x in parts[:3]] if effect == "patch": patch += 1 elif effect == "minor": minor += 1 patch = 0 elif effect == "major": major += 1 minor = 0 patch = 0 return f"{major}.{minor}.{patch}" def bump_version(self, info_only: bool, cur_ver_tag: str, commit_hash: str, merges_only: bool, verbose: bool, tag_found: bool, overrides: Dict[str, str], git_top_level: str ): """Calculates the next version and optionally writes release notes.""" effect_max = "minor" if not tag_found else "none" md_table_lines: List[str] = ["| # | Type | Effect | Scope | Change |", "|---:|:---|:---|:---|:---|"] md_change_lines: List[str] = [] counter = 0 commits = self.collect_commits(cur_ver_tag, commit_hash, merges_only, tag_found) if verbose: logger.info(f"\n# Conventional commits from version: {cur_ver_tag} to ({commit_hash})") for commit in commits: cmt_msg = self.get_commit_message(commit, overrides) msg_string = cmt_msg.msg if not msg_string: continue msg_header = msg_string.split("\n", 1)[0] match = self._header_regex.match(msg_header) if not match: if verbose: logger.info(f"~ Ignoring commit: {commit}") continue msg_type = match.group(1) msg_scope = match.group(3) or "" msg_breaking = match.group(4) msg_heading = match.group(5) msg_body = msg_string.split("\n", 1)[1] if "\n" in msg_string else "" type_effect = self._type_map.get(msg_type, ("none", ""))[0] effect = "major" if msg_breaking else type_effect effect_max = self.compare_increments(effect, effect_max) if verbose: logger.info(f"= Accepting commit: {commit}") logger.info( f"Heading\t{msg_heading}\nType\t{msg_type}\nScope\t{msg_scope}\nBreak\t[{msg_breaking}]\nVersion Effect\t{effect}") if msg_body.strip(): logger.info(msg_body) if not info_only: counter += 1 md_table_lines.append( f"| **{counter}** | {self._type_map.get(msg_type, ('', ''))[1]} | {effect} | {msg_scope} | {msg_heading} |") md_change_lines.append( f"#### {counter}) {self._type_map.get(msg_type, ('', ''))[1]}: {self.escape_markdown(msg_heading)}\n\n{msg_body}\n\n---\n") logger.info("\n# Version Bump") if effect_max == "none": logger.info(": Changes maximum effect (none) do not bump the version.") return base_ver = cur_ver_tag[1:] if cur_ver_tag.startswith("v") else cur_ver_tag next_ver_tag = f"v{self.increment_version(base_ver, effect_max)}" logger.info(f"Current ver/tag : {cur_ver_tag}") logger.info(f"Max-effect : {effect_max}") logger.info(f"Upto commit : {commit_hash}") logger.info(f"Next version/tag : {next_ver_tag}") if not info_only: notes_dir = os.path.join(git_top_level, "doc", "release") os.makedirs(notes_dir, exist_ok=True) md_file = os.path.join(notes_dir, f"notes-{next_ver_tag}.md") with open(md_file, "w", encoding="utf-8") as fh: fh.write(f"# Release-notes Version {next_ver_tag}\n\n") fh.write(f"## Changelist since version {cur_ver_tag}\n\n") fh.write("\n".join(md_table_lines)) fh.write("\n\n### Changes\n") fh.write("\n".join(md_change_lines)) logger.info(f"- Release notes written to {md_file}") def report_version_tags(self, git_top_level: str, cur_ver_tag: str, tag_found: bool, commit_hash: str, merges_only: bool, overrides: Dict[str, str], tag_annotation: str, long_hash: bool = False ): """Reports the versions, the annotated tags, and the commit ranges.""" def fmt_hash(h: str) -> str: """Formats the given hash to a character short one taking the first 7 characters.""" return h if long_hash else h[:7] hash_w = 40 if long_hash else 8 logger.info("\n# Top level directory") logger.info(git_top_level) logger.info("\n# Current versions") logger.info(f"Package (Git) : {self.git_tag_version_string(self.get_git_tag_version())}") logger.info(f'Last non-rc : {cur_ver_tag} "{tag_annotation}"') if not tag_found: logger.info(f": No current git version tag was found using '{cur_ver_tag}'.") logger.info("\n# Annotated version-tags") logger.info(f"~Tag{'':<11} | Hash{'':<{hash_w - 4}} | Time | Annotation") for line in reversed(self.git_lines(["tag", "--list", "--format", "%(tag)\t%(object)\t%(subject)"], check=False)): tag, obj, subject = (line.split("\t", 2) + ["", "", ""])[:3] if not re.match(r"^v\d+\.\d+\.\d+(-rc\.\d+)?$", tag): continue cmt_msg = self.get_commit_message(fmt_hash(obj), {}) logger.info(f"{tag:<15} | {fmt_hash(obj):<{hash_w}} | {cmt_msg.date.date()} | \"{subject}\"") if False: logger.info("\n# All merge commits") logger.info(f"~Tag{'':<11} | Hash{'':<{hash_w - 4}} | Time | Bump Version | Commit heading") for commit in self.git_lines(["log", "--merges", "--pretty=format:%H"], check=False): tag = self.git_describe_exact(commit) cmt_msg = self.get_commit_message(commit, overrides) heading = (cmt_msg.msg.split("\n", 1)[0]).strip() next_ver, effect = self.calculate_next_for_commit(cur_ver_tag, commit, merges_only=True, tag_found=tag_found, overrides=overrides) logger.info( f"{tag:<15} | {fmt_hash(commit):<{hash_w}} | {cmt_msg.date.date()} | {next_ver:<10}{effect[:3]} | {heading}") logger.info(f"\n# Commits since version: {cur_ver_tag} upto '{commit_hash}'") logger.info(f"~Tag{'':<11} | Hash{'':<{hash_w - 4}} | Time | Bump Version | Commit heading") commits = self.collect_commits(cur_ver_tag, commit_hash, merges_only, tag_found) for commit in commits: tag = self.git_describe_exact(commit) cmt_msg = self.get_commit_message(commit, overrides) heading = (cmt_msg.msg.split("\n", 1)[0]).strip() next_ver, effect = self.calculate_next_for_commit(cur_ver_tag, commit, merges_only=False, tag_found=tag_found, overrides=overrides) logger.info( f"{tag:<15} | {fmt_hash(commit):<{hash_w}} | {cmt_msg.date.date()} | {next_ver:<10} {effect[:3]} | {heading}") class SubCommandRun(SubCommand): """Subcommand handler for the 'run' command.""" def __init__(self): super().__init__("run", ["r"]) def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: self.parser = subparsers.add_parser(self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help="Runs an executable with the environment from given configure preset.") self.parser.epilog = f""" examples: Run executable in with the working directory as the binary: Linux: ./{self.script} {self.command} -p gnu-debug ./{self.script} {self.command} -p gnu-debug -- ./hello-world.bin Windows: ./{self.script} {self.command} -p msvc-debug -- hello-world.exe ./{self.script} {self.command} -p gw-debug -- cmd /c echo %PATH% Execute a command with the preset environment: {self.script} --preset gnu-debug -- cmd /c echo ^%PATH^% """ if self.parser is None: raise ValueError("Parser cannot be None") return self.parser def options(self, parser: argparse.ArgumentParser): """Adds options to the given parser for the create command.""" # Adds the standard help option. super().options(parser) # Configure the command line options. parser.add_argument("-p", "--preset", type=str, nargs="?", required=True, metavar="", help="The configure preset providing for the environment and working directory.") parser.add_argument("-v", "--verbose", action="store_true", help="Shows information when the command is executed for error analysis.") def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """ Handles the 'create' command execution of the script. :return: Exit code. """ # Call parent to handle the common dry-run option. super().handle(args, args_left, args_right) # When no command is given, list the files in the directory. if args_right is None or not len(args_right): # When executing from Linux and the Windows cross-compiler is used. if sys.platform != "win32" and get_compiler_type(args.preset, PresetTypes.CONFIGURE) == "gw": args_right = ["cmd", "/c", "dir", "/a"] elif sys.platform == "win32": args_right = ["cmd", "/c", "dir", "/a"] else: args_right = ["ls", "-la"] # Toolchain environment by preset and bailout when not set. if not set_environment_by_preset(args.preset, PresetTypes.CONFIGURE): return 1 # Get the preset object. preset = get_preset_by_name(PresetTypes.CONFIGURE, args.preset) # Populate RUN_ENV with environment variables from the preset to allow macro expansion. env = preset.get("environment", {}) for var in env: RUN_ENV[var] = expand_macros(preset, env.get(var), context=env) # Get the binary directory from preset expanding macros. bin_dir = expand_macros(preset, env.get("SF_EXECUTABLE_DIR"), True) cmake_bin_dir = expand_macros(preset, preset.get("cacheVariables", {}).get("CMAKE_RUNTIME_OUTPUT_DIRECTORY", {}).get("value"), True) logger.debug(f"~ Working directory set as 'CMAKE_RUNTIME_OUTPUT_DIRECTORY' to: {cmake_bin_dir}") if bin_dir is None: logger.error(f"! Field 'binaryDir' not found for configure preset '{args.preset}'.") return 1 compiler_type = get_compiler_type(args.preset, PresetTypes.CONFIGURE) # Get the library output directory for path setup. lib_dir = expand_macros(preset, preset.get("cacheVariables", {}).get("CMAKE_LIBRARY_OUTPUT_DIRECTORY", {}).get("value"), True) # Get the Qt bin subdirectory from the preset's cacheVariables. qt_bin_subdir = preset.get("cacheVariables", {}).get("SF_QT_BIN_SUBDIR", {}).get("value", "") # If RUN_QT_VER_DIR is set, copy to QT_VER_DIR (Wine workaround). if "RUN_QT_VER_DIR" in RUN_ENV: RUN_ENV["QT_VER_DIR"] = RUN_ENV["RUN_QT_VER_DIR"] logger.info(f"~ Environment 'RUN_QT_VER_DIR' overrides 'QT_VER_DIR'.") # Set up environment based on platform and compiler. if sys.platform != "win32" and compiler_type == "gw": # Linux running Windows cross-compiler: set up WINEPATH and prefix with wine. args_right = ["wine"] + args_right if "QT_VER_DIR" in RUN_ENV and qt_bin_subdir: qt_bin_dir = os.path.join(RUN_ENV["QT_VER_DIR"], qt_bin_subdir) wine_qt_bin = "Z:" + qt_bin_dir.replace("/", "\\") if lib_dir is not None: wine_lib_dir = "Z:" + lib_dir.replace("/", "\\") existing_winepath = RUN_ENV.get("WINEPATH", "") RUN_ENV["WINEPATH"] = f"{wine_qt_bin};{wine_lib_dir};{existing_winepath}" else: existing_winepath = RUN_ENV.get("WINEPATH", "") RUN_ENV["WINEPATH"] = f"{wine_qt_bin};{existing_winepath}" if args.verbose: logger.info(f"~ WINEPATH: {RUN_ENV['WINEPATH']}") elif sys.platform == "win32" and compiler_type in ("mingw", "msvc"): # Windows native: set up PATH with Qt bin directory. if "QT_VER_DIR" in RUN_ENV and qt_bin_subdir: qt_bin_dir = os.path.join(RUN_ENV["QT_VER_DIR"], qt_bin_subdir) if os.path.isdir(qt_bin_dir): existing_path = RUN_ENV.get("PATH", "") RUN_ENV["PATH"] = f"{existing_path};{qt_bin_dir}" if args.verbose: logger.info(f"~ PATH: {RUN_ENV['PATH']}") else: # Linux native: log LD_LIBRARY_PATH if verbose. if args.verbose and "LD_LIBRARY_PATH" in RUN_ENV: logger.info(f"~ LD_LIBRARY_PATH: {RUN_ENV['LD_LIBRARY_PATH']}") return run_command(args_right, cwd=bin_dir, dbg_mode=DebugMode.REPORT_ONLY).returncode class SubCommandPublish(SubCommand): """Subcommand handler for the 'publish' command.""" def __init__(self): super().__init__("publish", ["u"]) def create_parser(self, subparsers: argparse._SubParsersAction) -> argparse.ArgumentParser: self.parser = subparsers.add_parser(self.command, aliases=self.aliases, add_help=False, formatter_class=argparse.RawTextHelpFormatter, help="Uploads files to a Sonatype Nexus repository depending on their extension.") self.parser.epilog = f""" examples: Upload packages to Nexus: {self.script} {self.command} bin/pkg/*.deb bin/pkg/*.zip {self.script} u -a staging -s dist/staging bin/pkg/*.deb File starting with '-': {self.script} {self.command} -- -help-file.zip Using internal globbing: {self.script} {self.command} "bin/pkg/*.deb" Upload files to exchange directory: {self.script} {self.command} -x "gitlab-ci/shared/devops/pipeline/123" bin/gcov/report.* Download files from exchange directory: {self.script} {self.command} -x "gitlab-ci/shared/devops/pipeline/123" -l bin/gcov report.xml report.txt Dry-run upload: {self.script} {self.command} -d bin/pkg/*.deb Template ini-file for '.nexus-credentials.ini': [nexus-credentials] NEXUS_USER=your-username NEXUS_PASSWORD=your-password NEXUS_SERVER_URL=https://nexus.example.com NEXUS_RAW_REPO=shared NEXUS_EXCHANGE_REPO=exchange [nexus-develop] __inherit__=nexus-credentials NEXUS_APT_REPO=develop NEXUS_RAW_SUBDIR=dist/develop NEXUS_WINGET_REFRESH=winget/develop/admin/refresh [nexus-staging] __inherit__=nexus-credentials NEXUS_APT_REPO=staging NEXUS_RAW_SUBDIR=dist/staging NEXUS_WINGET_REFRESH=winget/staging/admin/refresh [nexus-stable] __inherit__=nexus-credentials NEXUS_APT_REPO=stable NEXUS_RAW_SUBDIR=dist/stable NEXUS_WINGET_REFRESH=winget/stable/admin/refresh """ if self.parser is None: raise ValueError("Parser cannot be None") return self.parser def options(self, parser: argparse.ArgumentParser): """Adds options to the given parser for the publish command.""" super().options(parser) parser.add_argument("-e", "--exch-repo", type=str, metavar="", help="Sets or overrules variable 'NEXUS_EXCHANGE_REPO' as the exchange-repository.") parser.add_argument("-a", "--apt-repo", type=str, metavar="", help="Sets or overrules variable 'NEXUS_APT_REPO' as the apt-repository name.") parser.add_argument("-r", "--raw-repo", type=str, metavar="", help="Sets or overrules variable 'NEXUS_RAW_REPO' as the raw-repository name.") parser.add_argument("-s", "--raw-sub", type=str, metavar="", help="Sets or overrules variable 'NEXUS_RAW_SUBDIR' as the subdirectory.") parser.add_argument("-S", "--section", type=str, default="develop", metavar="
", help="Sets the configuration section name (default: 'develop').") parser.add_argument("--server-url", type=str, metavar="", help="Sets or overrules variable 'NEXUS_SERVER_URL'.") parser.add_argument("--user", type=str, metavar="", help="Sets or overrules variable 'NEXUS_USER'.") parser.add_argument("--password", type=str, metavar="", help="Sets or overrules variable 'NEXUS_PASSWORD'.") parser.add_argument("-V", "--verbose", action="store_true", help="Enables verbose output for upload requests.") parser.add_argument("-x", "--exchange", type=str, metavar="", help="Sets the exchange sub-directory in the repository up- and download (--local).") parser.add_argument("-l", "--local", type=str, metavar="", help="Sets the exchange local-directory in the exchange repository (only with --exchange).") parser.add_argument("files", nargs="*", metavar="", help="Files to upload to Nexus repository (use '--' before filenames starting with '-').") return parser def handle(self, args: argparse.Namespace, args_left: List[str], args_right: List[str] | None) -> int: """ Handles the 'publish' command execution. :return: Exit code. """ super().handle(args, args_left, args_right) # Check for impossible use of standalone argument '--local'. if args.local and not args.exchange: raise RuntimeError("Option '--local' requires '--exchange' to be specified!") # Combine files parsed as positional arguments and arguments after '--' raw_files: List[str] = (args.files or []) + (args_right or []) # Check if files were passed when uploading (not downloading via --local). if not args.local and not raw_files: logger.error("! No files to upload!") self.print_help() return 1 # Get the config section name and check its existence. section_name: str = "nexus-" + args.section if not CONFIG.has_section(section_name): logger.info(f"# No Nexus configuration: {section_name}") # The section is allowed not to exist and do not fail. merged = get_merged_config_section(section_name, fail=False) # Resolve all variables with precedence: CLI arg -> Environment (RUN_ENV) -> INI Config nexus_user: str = args.user or RUN_ENV.get("NEXUS_USER") or merged.get("NEXUS_USER") or "" nexus_password: str = args.password or RUN_ENV.get("NEXUS_PASSWORD") or merged.get("NEXUS_PASSWORD") or "" nexus_server_url: str = args.server_url or RUN_ENV.get("NEXUS_SERVER_URL") or merged.get("NEXUS_SERVER_URL") or "" nexus_exchange_repo: str = args.exch_repo or RUN_ENV.get("NEXUS_EXCHANGE_REPO") or merged.get( "NEXUS_EXCHANGE_REPO") or "" nexus_apt_repo: str = args.apt_repo or RUN_ENV.get("NEXUS_APT_REPO") or merged.get("NEXUS_APT_REPO") or "" nexus_raw_repo: str = args.raw_repo or RUN_ENV.get("NEXUS_RAW_REPO") or merged.get("NEXUS_RAW_REPO") or "" nexus_raw_subdir: str = args.raw_sub or RUN_ENV.get("NEXUS_RAW_SUBDIR") or merged.get("NEXUS_RAW_SUBDIR") or "" nexus_winget_refresh: str = args.raw_sub or RUN_ENV.get("NEXUS_WINGET_REFRESH") or merged.get( "NEXUS_WINGET_REFRESH") or "" # Report when overruled by command line arguments. if args.verbose: if args.raw_repo: logger.info(f"# RAW repository set to '{nexus_raw_repo}'.") if args.raw_sub: logger.info(f"# RAW subdirectory set to '{nexus_raw_subdir}'.") if args.apt_repo: logger.info(f"# APT repository set to '{nexus_apt_repo}'.") if args.exch_repo: logger.info(f"# EXCHANGE repository set to '{nexus_exchange_repo}'.") if args.exchange: logger.info(f"# Exchange directory set to '{args.exchange}'.") if args.local: logger.info(f"# Local directory set to '{args.local}'.") # Check existence of required variables. cred_vars = { "NEXUS_USER": nexus_user, "NEXUS_PASSWORD": nexus_password, "NEXUS_SERVER_URL": nexus_server_url, } if args.exchange: # Only required if uploading or downloading from exchange repo. cred_vars |= { "NEXUS_EXCHANGE_REPO": nexus_exchange_repo, } else: # Only required if uploading to at or raw repo. cred_vars |= { "NEXUS_APT_REPO": nexus_apt_repo, "NEXUS_RAW_REPO": nexus_raw_repo, "NEXUS_RAW_SUBDIR": nexus_raw_subdir, "NEXUS_WINGET_REFRESH": nexus_winget_refresh, } # Check on the presence of required variables. flag_var = False # Flag indicating the Winget definition cache on the Nexus server needs to be refreshed. flag_cache_refresh = False for var_name, var_val in cred_vars.items(): if not var_val: logger.error( f"! Required credentials/config variable '{var_name}' " f"is not set by credentials file or by parent environment!") flag_var = True if flag_var: self.print_help() return 1 # Handle Exchange Download: --exchange and --local if args.exchange and args.local: exchange_sub = str(args.exchange).strip('/\\').replace('\\', '/') if DEBUG_FLAG: if raw_files: for pattern in raw_files: fn = Path(pattern).name dest_file = os.path.join(args.local, fn) logger.info( f"~ Not downloading: GET {str(nexus_server_url).rstrip('/')}/repository/" f"{quote(str(nexus_exchange_repo), safe='')}/{exchange_sub}/{fn} -> {dest_file}" ) else: logger.info(f"~ Not downloading all files from exchange '{exchange_sub}' to '{args.local}'") return 0 # search_base_url = ( f"{str(nexus_server_url).rstrip('/')}/service/rest/v1/search" f"?repository={quote(str(nexus_exchange_repo), safe='')}" f"&group={quote('/' + exchange_sub, safe='/')}" ) token = None assets: List[Dict[str, Any]] = [] while True: token_url = search_base_url if not token else f"{search_base_url}&continuationToken={token or ""}" status, data = query_nexus_search(token_url, nexus_user, nexus_password, verbose=args.verbose) if status < 200 or status >= 300 or not isinstance(data, dict): logger.error(f"! Search failed ({status}) on repository '{nexus_exchange_repo}' group '{exchange_sub}'") if isinstance(data, str) and data: logger.error(data) return 1 for item in data.get("items", []): for asset in item.get("assets", []): assets.append(asset) token = data.get("continuationToken") if not token: break req_base_names = {Path(f).name for f in raw_files} if raw_files else set() if req_base_names: for asset in assets: asset_path = asset.get("path", "") fn = Path(asset_path).name # if req_base_names and fn not in req_base_names: # continue if any(fnmatch.fnmatch(fn, pat) or fnmatch.fnmatch(asset_path, pat) for pat in req_base_names): logger.info(f"- Matching file: {fn}") else: continue download_url = asset.get("downloadUrl", "") if not download_url: continue dest_file = os.path.join(args.local, fn) logger.info(f"- Downloading file: {asset_path} -> {dest_file}") status, err = download_file_http(download_url, dest_file, nexus_user, nexus_password, verbose=args.verbose) if status < 200 or status >= 300: logger.error(f"! Download failed ({status}) from '{download_url}' to '{dest_file}'") if err: logger.error(err) return 1 return 0 # Handle Exchange Upload: --exchange without --local if args.exchange: expanded_files: List[str] = [] for pattern in raw_files: matches = glob.glob(pattern) if matches: expanded_files.extend(matches) else: expanded_files.append(pattern) for upload_file in expanded_files: if not os.path.isfile(upload_file): if args.verbose: logger.info(f"~ File(s) not found for: {upload_file}") continue exchange_sub = str(args.exchange).strip('/\\').replace('\\', '/') exchange_parts = [quote(part, safe="") for part in exchange_sub.split('/') if part] exchange_parts.append(quote(Path(upload_file).name, safe="")) exchange_target = "/".join(exchange_parts) upload_url = ( f"{str(nexus_server_url).rstrip('/')}/repository/" f"{quote(str(nexus_exchange_repo), safe='')}/{exchange_target}" ) logger.info(f"- Uploading EXCHANGE repo file: {upload_file}") try: response_code, output = upload_file_http(upload_url, upload_file, nexus_user, nexus_password, method="PUT", verbose=args.verbose) except (OSError, ValueError, http.client.HTTPException) as ex: logger.error(f"! Upload failed for file '{upload_file}': {ex}") return 1 if response_code < 200 or response_code >= 300: logger.error(f"! Upload EXCHANGE package failed ({response_code}) of file: {upload_file}") if output: logger.error(output) return 1 return 0 # Handle Regular Publish (neither --exchange nor --local) expanded_files: List[str] = [] for pattern in raw_files: matches = glob.glob(pattern) if matches: expanded_files.extend(matches) else: expanded_files.append(pattern) # Iterate over all files for upload_file in expanded_files: if not os.path.isfile(upload_file): if args.verbose: logger.info(f"~ File(s) not found with: {upload_file}") continue ext = upload_file.rsplit('.', 1)[-1].lower() if ext not in ["deb", "zip", "zip-def", "exe", "gz", "tgz", "bz2", "xz", "7z", "yaml", "yml", "appimage"]: logger.warning(f"! No upload method for extension '{ext}' file: {upload_file}") continue try: if ext == "deb": logger.info(f"- Uploading APT repo file: {upload_file} ({self.get_readable_file_size(upload_file)})") upload_url = ( f"{str(nexus_server_url).rstrip('/')}/service/rest/v1/components" f"?repository={quote(str(nexus_apt_repo), safe='')}" ) response_code, output = upload_file_http(upload_url, upload_file, nexus_user, nexus_password, method="POST", multipart_field="apt.asset", content_type="application/vnd.debian.binary-package", verbose=args.verbose) upload_kind = "APT package" else: # Trigger a cache refresh for a winget package. if ext == "zip-def": flag_cache_refresh = True logger.info(f"~ Uploading RAW repo file: {upload_file} ({self.get_readable_file_size(upload_file)})") raw_sub = str(nexus_raw_subdir).strip('/\\').replace('\\', '/') raw_parts = [quote(part, safe="") for part in raw_sub.split('/') if part] raw_parts.append(quote(Path(upload_file).name, safe="")) raw_target = "/".join(raw_parts) upload_url = ( f"{str(nexus_server_url).rstrip('/')}/repository/" f"{quote(str(nexus_raw_repo), safe='')}/{raw_target}" ) response_code, output = upload_file_http(upload_url, upload_file, nexus_user, nexus_password, verbose=args.verbose) upload_kind = "RAW package" except (OSError, ValueError, http.client.HTTPException) as ex: logger.error(f"! Upload failed for file '{upload_file}': {ex}") return 1 if response_code < 200 or response_code >= 300: logger.error(f"! Upload {upload_kind} failed ({response_code}) of file: {upload_file}") if output: logger.error(output) return 1 # Check if the Wingewt cache is needed to be refreshed. if flag_cache_refresh: logger.info(f"~ Refreshing Winget definition cache.") refresh_url = f"{nexus_server_url}/{nexus_winget_refresh}" response_code, output = post_request_http(url=refresh_url, username=nexus_user, password=nexus_password, verbose=args.verbose) if response_code < 200 or response_code >= 300: logger.error(f"! Refreshing Winget definition cache failed ({response_code})") if output: logger.error(output) return 1 return 0 @staticmethod def get_readable_file_size(filepath): """Returns a human-readable file size.""" try: size_bytes = os.path.getsize(filepath) except FileNotFoundError: return "File not found" for unit in ['', 'Ki', 'Mi', 'Gi', 'Ti']: precision: int = 2 if len(unit) != 0 else 0 if size_bytes < 1024.0: return f"{size_bytes:.{precision}f} {unit}B" size_bytes /= 1024.0 def split_arguments(arguments: List[str], split_arg: str = "--") -> tuple[List[str], List[str]]: """ Splits the arguments in a left and right :param arguments: :param split_arg: """ # Get the separator index of an argument. arg_sep_idx = arguments.index(split_arg) if split_arg in arguments else -1 args_left = arguments[:arg_sep_idx] if arg_sep_idx >= 0 else arguments args_right = arguments[arg_sep_idx + 1:] if len(arguments) > arg_sep_idx >= 0 else [] return args_left, args_right def _load_subcommand_modules(): """ Dynamically import all 'build-*.py' modules in the script's directory. Each module is expected to define and register a SubCommand subclass. The prefix 'build-' is derived from the main script name 'build.py'. Subcommand modules must NOT use 'from build import ...' because that would re-import build.py from the script's directory, re-running module-level code. Instead, they should use 'sys.modules["build"].SubCommand' to reference the already-loaded module. """ import importlib.util # Use the real directory of the script (resolves symlinks), since subcommand # modules live alongside the actual build.py, not the symlink. script_dir = os.path.dirname(os.path.realpath(__file__)) if not "PY_SCRIPT_DIR" in PARENT_ENV \ else PARENT_ENV["PY_SCRIPT_DIR"] # Under Wine, os.path.realpath() does not resolve Linux symlinks, so # script_dir may point to the project root instead of the actual script # directory (e.g., 'sbin/'). Collect all candidate directories to search # for 'build-*.py' modules, deduplicating them to avoid double-loading. directories_to_search = [script_dir] sbin_dir = os.path.join(script_dir, "sbin") if os.path.isdir(sbin_dir) and sbin_dir not in directories_to_search: directories_to_search.append(sbin_dir) pattern = "build-*.py" seen_files = set() for search_dir in directories_to_search: for module_file in sorted(glob.glob(os.path.join(search_dir, pattern))): # Skip duplicates (same file found in multiple search dirs) real_file = os.path.realpath(module_file) if real_file in seen_files: continue seen_files.add(real_file) # Skip the main build.py itself and __pycache__ module_name = os.path.splitext(os.path.basename(module_file))[0] if module_name == "build": continue # Skip if already imported if module_name in sys.modules: continue try: spec = importlib.util.spec_from_file_location(module_name, module_file) if spec is not None and spec.loader is not None: module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) logger.debug(f"# Loaded subcommand module '{module_name}' from '{module_file}'.") else: logger.warning(f"! Could not load module spec for '{module_file}'.") except Exception as ex: logger.error(f"! Failed to import subcommand module '{module_file}': {ex}") def main() -> int: """ Main entry point for the build script. :return: Exit code. """ # Get the script's name. script = os.path.basename(__file__) # Strip other first argument which is the script itself. arguments = sys.argv[1:] # Register the commands. std_cmd: SubCommand = SubCommandNative().register() # Only when not in Windows if sys.platform != "win32": # And also not in 'Wine' register the command only for an x86_64 architecture. if not is_wine() and platform.processor() == "x86_64": SubCommandWine().register() # And also not in Docker register the command. if not is_docker() and sys.platform != "win32": SubCommandDocker().register() # Register unconditional commands. SubCommandInstall().register() SubCommandVersion().register() SubCommandRun().register() SubCommandPublish().register() # Dynamically import and register subcommand modules named 'build-*.py'. _load_subcommand_modules() # parser = argparse.ArgumentParser(description="""Helper for running CMake , CTest, CPack commands using 'CMakePresets.json' and 'CMakeUserPresets.json'. Running Native, Docker, Wine and nested as in Docker > Wine. """, formatter_class=argparse.RawTextHelpFormatter, add_help=False) # Get the ini file. ini_file = str(os.path.splitext(str(script))[0] + ".ini") parser.epilog = f""" The script depends on the configuration file '{ini_file}' which contains sections for creating environments for each nested call of this script. To Build and test the example project: On Linux: ./{script} i -r lnx # Required packages for Linux (Debian only). ./{script} i --project # Install the skeleton project by Git cloning and sets up a git # repository with this repository as submodule. ./{script} i -p # Clone the cmake-lib repository and copy the sample project. ./{script} -bt gnu-debug # Build and test a preset local. ./{script} run -p gnu-debug -- ./ # Execute an application from the presets' output directory. ./{script} d -- w -- -bt msvc-debug # Preset make, build and test from Wine in Docker. ./{script} d -- w -- run -p msvc-debug hello-world.exe # Run an application form the output in Docker/Wine. ./{script} d versions # Report all versions within the Docker image. ./{script} d -- -b gnu-debug -N # Build a target select from a menu (e.g. 'document' for DoxyGen). ./{script} d -- -w gnu-debug # Run a preset configured workflow including packaging mostly used in pipelines. ./{script} d start/stop # Start or stop the Docker container as daemon to speed up. ./{script} u bin/pkg/*.deb # Upload packages to Nexus repository. On Windows: {script} i -r win # Required packages (WinGet/Pip) for Windows. {script} i --project # Install the skeleton project by Git cloning and sets up a git # repository with this repository as submodule. {script} i -p # Clone the cmake-lib repository and copy the sample project. {script} i -t msvc # Install the MSVC toolchain. {script} -bt msvc-debug # Preset make, build and test. {script} run -p msvc-debug -- hello-world.exe {script} run -p msvc-debug -- hello-world-qt.exe """ # Subparsers for command-specific arguments subparsers = parser.add_subparsers(dest="subcmd", help=f"Subcommand defaulting to '{std_cmd.command}' when omitted.") # Create subparsers for all registered commands. for sub_cmd in SubCommand.registry: cmd = SubCommand.registry[sub_cmd] cmd.options(cmd.create_parser(subparsers)) # Show help when no arguments are passed on the command line. if len(arguments) == 0: parser.print_help() return 0 # When the first argument is not a subcommand insert the standard command. if len(arguments) and arguments[0] not in parser._actions[0].choices: arguments = [std_cmd.command] + arguments # When only the command is given show its help. if len(arguments) == 1: if arguments[0] in SubCommand.registry: SubCommand.registry[arguments[0]].print_help() return 0 for cmd in SubCommand.registry: if arguments[0] in SubCommand.registry[cmd].aliases: SubCommand.registry[cmd].print_help() return 0 # To circumvent the exit by the parser. try: args_left, args_right = split_arguments(arguments) # Parse the command line arguments left of '--'. args = parser.parse_args(args_left) if args.subcmd in SubCommand.registry: return SubCommand.registry[args.subcmd].handle(args, args_left, args_right) for cmd in SubCommand.registry: if args.subcmd in SubCommand.registry[cmd].aliases: return SubCommand.registry[cmd].handle(args, args_left, args_right) except HelpAction.HelpException: return 0 return 0 if __name__ == "__main__": """Main entry point for the script.""" # Widen the console buffer early so output is not wrapped by Wine or a narrow Windows console. _widen_console_buffer() exitcode = 0 try: PARENT_ENV['RUN_QT_VER'] = get_config_section("config", fail=True).get("qt-ver", PARENT_ENV['RUN_QT_VER']) # Change to the directory of this script. os.chdir(RUN_DIR) exitcode = main() origin = [sys.platform] if is_docker(): origin.append(f"Docker({platform.processor()})") if is_wine(): origin.append("Wine") logger.info(f"- {os.path.basename(__file__)} ({'>'.join(origin)}), executed in {int(time.time() - start_time)}s.") except KeyboardInterrupt: logger.info("! Interrupted by user.") exitcode = 130 except subprocess.CalledProcessError as cmd_ex: if cmd_ex.returncode != 130: logger.error(f"! Command error({cmd_ex.returncode}): {' '.join(cmd_ex.cmd)}") if cmd_ex.stdout: logger.error(cmd_ex.stdout.decode("utf-8")) if cmd_ex.stderr: logger.error(cmd_ex.stderr.decode("utf-8")) exitcode = cmd_ex.returncode except Exception as any_ex: logger.error(f"! Exception({any_ex.__class__.__name__}): {any_ex}") if hasattr(any_ex, '__notes__'): for note in any_ex.__notes__: logger.error(f"Note: {note}") exitcode = 1 # Show the cursor again. sys.exit(exitcode)