1# SPDX-License-Identifier: BSD-3-Clause 2# Copyright(c) 2010-2014 Intel Corporation 3# Copyright(c) 2022-2023 PANTHEON.tech s.r.o. 4# Copyright(c) 2022-2023 University of New Hampshire 5# Copyright(c) 2024 Arm Limited 6 7"""Various utility classes and functions. 8 9These are used in multiple modules across the framework. They're here because 10they provide some non-specific functionality, greatly simplify imports or just don't 11fit elsewhere. 12 13Attributes: 14 REGEX_FOR_PCI_ADDRESS: The regex representing a PCI address, e.g. ``0000:00:08.0``. 15""" 16 17import atexit 18import json 19import os 20import subprocess 21from enum import Enum 22from pathlib import Path 23from subprocess import SubprocessError 24 25from scapy.packet import Packet # type: ignore[import-untyped] 26 27from .exception import ConfigurationError 28 29REGEX_FOR_PCI_ADDRESS: str = "/[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}.[0-9]{1}/" 30 31 32def expand_range(range_str: str) -> list[int]: 33 """Process `range_str` into a list of integers. 34 35 There are two possible formats of `range_str`: 36 37 * ``n`` - a single integer, 38 * ``n-m`` - a range of integers. 39 40 The returned range includes both ``n`` and ``m``. Empty string returns an empty list. 41 42 Args: 43 range_str: The range to expand. 44 45 Returns: 46 All the numbers from the range. 47 """ 48 expanded_range: list[int] = [] 49 if range_str: 50 range_boundaries = range_str.split("-") 51 # will throw an exception when items in range_boundaries can't be converted, 52 # serving as type check 53 expanded_range.extend(range(int(range_boundaries[0]), int(range_boundaries[-1]) + 1)) 54 55 return expanded_range 56 57 58def get_packet_summaries(packets: list[Packet]) -> str: 59 """Format a string summary from `packets`. 60 61 Args: 62 packets: The packets to format. 63 64 Returns: 65 The summary of `packets`. 66 """ 67 if len(packets) == 1: 68 packet_summaries = packets[0].summary() 69 else: 70 packet_summaries = json.dumps(list(map(lambda pkt: pkt.summary(), packets)), indent=4) 71 return f"Packet contents: \n{packet_summaries}" 72 73 74def get_commit_id(rev_id: str) -> str: 75 """Given a Git revision ID, return the corresponding commit ID. 76 77 Args: 78 rev_id: The Git revision ID. 79 80 Raises: 81 ConfigurationError: The ``git rev-parse`` command failed, suggesting 82 an invalid or ambiguous revision ID was supplied. 83 """ 84 result = subprocess.run( 85 ["git", "rev-parse", "--verify", rev_id], 86 text=True, 87 capture_output=True, 88 ) 89 if result.returncode != 0: 90 raise ConfigurationError( 91 f"{rev_id} is not a valid git reference.\n" 92 f"Command: {result.args}\n" 93 f"Stdout: {result.stdout}\n" 94 f"Stderr: {result.stderr}" 95 ) 96 return result.stdout.strip() 97 98 99class StrEnum(Enum): 100 """Enum with members stored as strings.""" 101 102 @staticmethod 103 def _generate_next_value_(name: str, start: int, count: int, last_values: object) -> str: 104 return name 105 106 def __str__(self) -> str: 107 """The string representation is the name of the member.""" 108 return self.name 109 110 111class MesonArgs(object): 112 """Aggregate the arguments needed to build DPDK.""" 113 114 _default_library: str 115 116 def __init__(self, default_library: str | None = None, **dpdk_args: str | bool): 117 """Initialize the meson arguments. 118 119 Args: 120 default_library: The default library type, Meson supports ``shared``, ``static`` and 121 ``both``. Defaults to :data:`None`, in which case the argument won't be used. 122 dpdk_args: The arguments found in ``meson_options.txt`` in root DPDK directory. 123 Do not use ``-D`` with them. 124 125 Example: 126 :: 127 128 meson_args = MesonArgs(enable_kmods=True). 129 """ 130 self._default_library = f"--default-library={default_library}" if default_library else "" 131 self._dpdk_args = " ".join( 132 ( 133 f"-D{dpdk_arg_name}={dpdk_arg_value}" 134 for dpdk_arg_name, dpdk_arg_value in dpdk_args.items() 135 ) 136 ) 137 138 def __str__(self) -> str: 139 """The actual args.""" 140 return " ".join(f"{self._default_library} {self._dpdk_args}".split()) 141 142 143class _TarCompressionFormat(StrEnum): 144 """Compression formats that tar can use. 145 146 Enum names are the shell compression commands 147 and Enum values are the associated file extensions. 148 """ 149 150 gzip = "gz" 151 compress = "Z" 152 bzip2 = "bz2" 153 lzip = "lz" 154 lzma = "lzma" 155 lzop = "lzo" 156 xz = "xz" 157 zstd = "zst" 158 159 160class DPDKGitTarball(object): 161 """Compressed tarball of DPDK from the repository. 162 163 The class supports the :class:`os.PathLike` protocol, 164 which is used to get the Path of the tarball:: 165 166 from pathlib import Path 167 tarball = DPDKGitTarball("HEAD", "output") 168 tarball_path = Path(tarball) 169 """ 170 171 _git_ref: str 172 _tar_compression_format: _TarCompressionFormat 173 _tarball_dir: Path 174 _tarball_name: str 175 _tarball_path: Path | None 176 177 def __init__( 178 self, 179 git_ref: str, 180 output_dir: str, 181 tar_compression_format: _TarCompressionFormat = _TarCompressionFormat.xz, 182 ): 183 """Create the tarball during initialization. 184 185 The DPDK version is specified with `git_ref`. The tarball will be compressed with 186 `tar_compression_format`, which must be supported by the DTS execution environment. 187 The resulting tarball will be put into `output_dir`. 188 189 Args: 190 git_ref: A git commit ID, tag ID or tree ID. 191 output_dir: The directory where to put the resulting tarball. 192 tar_compression_format: The compression format to use. 193 """ 194 self._git_ref = git_ref 195 self._tar_compression_format = tar_compression_format 196 197 self._tarball_dir = Path(output_dir, "tarball") 198 199 self._create_tarball_dir() 200 201 self._tarball_name = ( 202 f"dpdk-tarball-{self._git_ref}.tar.{self._tar_compression_format.value}" 203 ) 204 self._tarball_path = self._check_tarball_path() 205 if not self._tarball_path: 206 self._create_tarball() 207 208 def _create_tarball_dir(self) -> None: 209 os.makedirs(self._tarball_dir, exist_ok=True) 210 211 def _check_tarball_path(self) -> Path | None: 212 if self._tarball_name in os.listdir(self._tarball_dir): 213 return Path(self._tarball_dir, self._tarball_name) 214 return None 215 216 def _create_tarball(self) -> None: 217 self._tarball_path = Path(self._tarball_dir, self._tarball_name) 218 219 atexit.register(self._delete_tarball) 220 221 result = subprocess.run( 222 'git -C "$(git rev-parse --show-toplevel)" archive ' 223 f'{self._git_ref} --prefix="dpdk-tarball-{self._git_ref + os.sep}" | ' 224 f"{self._tar_compression_format} > {Path(self._tarball_path.absolute())}", 225 shell=True, 226 text=True, 227 capture_output=True, 228 ) 229 230 if result.returncode != 0: 231 raise SubprocessError( 232 f"Git archive creation failed with exit code {result.returncode}.\n" 233 f"Command: {result.args}\n" 234 f"Stdout: {result.stdout}\n" 235 f"Stderr: {result.stderr}" 236 ) 237 238 atexit.unregister(self._delete_tarball) 239 240 def _delete_tarball(self) -> None: 241 if self._tarball_path and os.path.exists(self._tarball_path): 242 os.remove(self._tarball_path) 243 244 def __fspath__(self) -> str: 245 """The os.PathLike protocol implementation.""" 246 return str(self._tarball_path) 247