| ############################################################################ |
| # SPDX-License-Identifier: Apache-2.0 |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. The |
| # ASF licenses this file to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance with the |
| # License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| # License for the specific language governing permissions and limitations |
| # under the License. |
| # |
| ############################################################################ |
| |
| """Product core configuration handler.""" |
| |
| import os |
| from functools import cached_property |
| from typing import Any, Dict, Optional, Union |
| |
| from ntfc.lib.elf.app_bindir import AppBinDir, symbol_patterns |
| from ntfc.lib.elf.elf_parser import ElfParser |
| |
| |
| class CoreConfig: |
| """Product core configuration.""" |
| |
| def __init__(self, cfg: Dict[str, Any]) -> None: |
| """Initialzie product core configuration.""" |
| self._config = cfg |
| |
| self._kv_values: Dict[str, Any] = {} |
| self._elf: Optional[ElfParser] = None |
| |
| conf_path = self._config.get("conf_path", None) |
| if conf_path: |
| # load config values |
| self._load_core_config() |
| |
| elf_path = self._config.get("elf_path", None) |
| if elf_path and self.os == "nuttx": |
| # load ELF |
| self._elf = ElfParser(elf_path) |
| |
| @staticmethod |
| def _parse_config_value(val: str) -> Union[bool, str, int]: |
| """Parse a single Kconfig option value.""" |
| if val == "y": |
| return True |
| if val == "n": |
| return False |
| if val.startswith('"') and val.endswith('"'): |
| # quoted strings first: they may contain hex digits |
| return val[1:-1] |
| if val.startswith("0x"): |
| try: |
| return int(val, 16) |
| except ValueError: |
| return val |
| if val.isdigit(): |
| return int(val) |
| return val |
| |
| def _load_core_config(self) -> None: |
| """Load core configuration.""" |
| with open(self._config["conf_path"], "r", encoding="utf-8") as f: |
| for line in f: |
| line = line.rstrip("\n") |
| # ignore blank lines and commented lines |
| if not line or line.startswith("#"): |
| continue |
| |
| name, sep, val = line.partition("=") |
| if not sep: |
| # no '=' found — skip malformed line |
| continue |
| |
| self._kv_values[name] = self._parse_config_value(val) |
| |
| @property |
| def uptime(self) -> Any: |
| """Return core uptime.""" |
| return self._config.get("uptime", 3) |
| |
| @property |
| def read_poll_interval(self) -> float: |
| """Return console polling interval in seconds.""" |
| value = float(self._config.get("read_poll_interval", 0.1)) |
| if value <= 0: |
| raise ValueError("read_poll_interval must be positive") |
| return value |
| |
| @property |
| def device(self) -> Any: |
| """Return core device.""" |
| return self._config.get("device", None) |
| |
| @property |
| def os(self) -> str: |
| """Return target operating system name.""" |
| return str(self._config.get("os", "nuttx")).lower() |
| |
| @property |
| def name(self) -> Any: |
| """Return core name.""" |
| return self._config.get("name", "unknown_name") |
| |
| def _load_prompt_from_config(self) -> Optional[str]: |
| """Load the prompt string from the .config file.""" |
| return self._kv_values.get("CONFIG_NSH_PROMPT_STRING", None) |
| |
| @property |
| def prompt(self) -> Any: |
| """Return core prompt. |
| |
| First check YAML config, if not present, load from .config file. |
| """ |
| # First check if prompt is explicitly set in YAML config |
| yaml_prompt = self._config.get("prompt", None) |
| if yaml_prompt: |
| return yaml_prompt |
| |
| # Otherwise, load from .config file |
| return self._load_prompt_from_config() |
| |
| @property |
| def elf_path(self) -> Any: |
| """Return core elf path.""" |
| return self._config.get("elf_path", "") |
| |
| @property |
| def exec_path(self) -> Any: |
| """Return core exec path.""" |
| return self._config.get("exec_path", "") |
| |
| @property |
| def exec_args(self) -> Any: |
| """Return core exec args.""" |
| return self._config.get("exec_args", "") |
| |
| @property |
| def reboot(self) -> Any: |
| """Return core reboot command.""" |
| return self._config.get("reboot", "") |
| |
| @property |
| def poweroff(self) -> Any: |
| """Return core poweroff command.""" |
| return self._config.get("poweroff", "") |
| |
| @property |
| def flash_only(self) -> bool: |
| """Return whether this core is build/flash-only. |
| |
| Flash-only cores are still built/flashed by NTFC, but are skipped for |
| runtime test orchestration: no boot checks, no log collection, no test |
| requirements validation, and no command execution against them. |
| """ |
| return bool(self._config.get("flash_only", False)) |
| |
| @property |
| def is_kernel_build(self) -> bool: |
| """Return True when the core .config has CONFIG_BUILD_KERNEL=y.""" |
| return self.kv_check("CONFIG_BUILD_KERNEL") is True |
| |
| @property |
| def exec_cwd(self) -> Optional[str]: |
| """Return working directory for spawned sim/qemu processes. |
| |
| Kernel-mode hostfs mounts resolve relative to this directory. |
| """ |
| return self._config.get("exec_cwd", None) |
| |
| @property |
| def boot_timeout(self) -> int: |
| """Return seconds to wait for the first shell prompt after start.""" |
| return int(self._config.get("boot_timeout", 5)) |
| |
| @property |
| def app_bindir(self) -> Optional[str]: |
| """Return directory with kernel-mode application binaries. |
| |
| Defaults to the bin/ directory next to the NuttX ELF. |
| """ |
| bindir = self._config.get("app_bindir", None) |
| if bindir: |
| return str(bindir) |
| if self.is_kernel_build and self.elf_path: |
| return os.path.join(os.path.dirname(self.elf_path), "bin") |
| return None |
| |
| @property |
| def line_buffered(self) -> bool: |
| """Return whether commands are sent in one transport write.""" |
| return bool(self._config.get("line_buffered", False)) |
| |
| def kv_check(self, cfg: str) -> Any: |
| """Check Kconfig option and return its value. |
| |
| :param cfg: Kconfig option name (e.g., 'CONFIG_DEBUG') |
| :return: Config value if set (bool True, string value, etc), |
| False if not found |
| """ |
| if not self._kv_values: |
| # No config data loaded |
| return False |
| |
| return self._kv_values.get(cfg, False) |
| |
| @cached_property |
| def _appbin(self) -> Optional[AppBinDir]: |
| """Return the installed kernel-mode application binaries.""" |
| bindir = self.app_bindir |
| if not bindir or not os.path.isdir(bindir): |
| return None |
| |
| return AppBinDir(bindir) |
| |
| @property |
| def has_app_bindir(self) -> bool: |
| """Return True when kernel-mode application binaries are found.""" |
| return self._appbin is not None |
| |
| @property |
| def path_initial(self) -> str: |
| """Return the initial target PATH, see CONFIG_PATH_INITIAL.""" |
| return str(self.kv_check("CONFIG_PATH_INITIAL") or "/system/bin") |
| |
| def cmd_check(self, cmd: str, core: int = 0) -> bool: |
| """Check if command is available in binary. |
| |
| Kernel-mode cores resolve commands against the application binary |
| directory; otherwise the symbol must exist in the core ELF. |
| """ |
| if self._appbin: |
| if self._appbin.has_command(cmd): |
| return True |
| # not an application: NSH builtins and test entry points |
| # are symbols inside the application binaries |
| return self._appbin.has_symbol(cmd) |
| |
| if not self._elf: |
| raise AttributeError("no elf data") |
| |
| return any( |
| self._elf.has_symbol(symbol) for symbol in symbol_patterns(cmd) |
| ) |
