Coverage for apio/utils/util.py: 85%

1# This file is part of the Apio project

3# Project authors:

4# * Jesús Arroyo (Jesus89) (2016-2019)

5# * Juan Gonzalez (obijuan) (2019-2024)

6# * The Apio Team (2024-present)

7# License: GPLv2

9"""Misc utility functions and classes."""

11import sys

12import os

13from contextlib import contextmanager

14from enum import Enum, unique

15from dataclasses import dataclass

16from typing import Any

17import subprocess

18from threading import Thread

19from pathlib import Path

20import hashlib

21from typing_extensions import override

22import apio

23from apio.utils import env_options

24from apio.common.apio_console import cout, cerror, fatal_error

25from apio.common.common_util import cmd_in_env

27# ----------------------------------------

28# -- Constants

29# ----------------------------------------

32class AsyncPipe(Thread):

33 """A class that implements a pipe that calls back on each incoming line

34 from an internal thread. Used to process in real time scons output to

35 show its progress."""

37 def __init__(self, line_callback=None) -> None:

38 """If line_callback is not None, it is called for each line as

39 line_callback(line:str, terminator:str) where line is the line content

40 and terminator is one of:

41 "\r" (CR)

42 "\n" (LF)

43 "" (EOF)

45 The callback is done from a private Python thread of this pipe so make

46 sure to have the proper locking and synchronization as needed.

47 """

49 Thread.__init__(self)

50 self.outcallback = line_callback

52 self._fd_read, self._fd_write = os.pipe()

54 # -- A list of lines received so far.

55 self._lines_buffer: list[str] = []

57 self.start()

59 def get_buffer(self) -> list[str]:

60 """DOC: TODO"""

62 return self._lines_buffer

64 def fileno(self) -> int:

65 """DOC: TODO"""

67 return self._fd_write

69 def _handle_incoming_line(self, bfr: bytearray, terminator: str) -> None:

70 """Handle a new incoming line.

71 Bfr is a bytes with the line's content, possibly empty.

72 See __init__ for the description of terminator.

73 """

74 # -- Convert the line's bytes to a string. Replace invalid utf-8

75 # -- chars with "�"

76 line: str = bfr.decode("utf-8", errors="replace")

78 # -- Append to the lines log buffer.

79 self._lines_buffer.append(line)

81 # -- Report back if caller passed a callback.

82 if self.outcallback: 82 ↛ exitline 82 didn't return from function '_handle_incoming_line' because the condition on line 82 was always true

83 self.outcallback(line, terminator)

85 @override

86 def run(self) -> None:

87 """DOC: TODO"""

89 # -- Prepare a buffer for collecting the line chars, excluding

90 # -- its line terminator.

91 bfr = bytearray()

93 # -- We open in binary mode so we have access to the line terminators.

94 # -- This is important with progress bars which don't advance to the

95 # -- next line but redraw on the same line.

96 with os.fdopen(self._fd_read, "rb") as f:

97 while True:

98 b: bytes = f.read(1)

99 assert len(b) <= 1

100

101 # -- Handle end of file

102 if not b:

103 if bfr: 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true

104 self._handle_incoming_line(bfr, "")

105 return

106

107 # -- Handle \r terminator

108 if b == b"\r": 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 self._handle_incoming_line(bfr, "\r")

110 bfr.clear()

111 continue

112

113 # -- Handle \n terminator

114 if b == b"\n":

115 self._handle_incoming_line(bfr, "\n")

116 bfr.clear()

117 continue

118

119 # -- Handle a regular character

120 bfr.append(b[0])

121

122 def close(self):

123 """DOC: TODO"""

124

125 os.close(self._fd_write)

126 self.join()

127

128

129@unique

130class TerminalMode(Enum):

131 """Represents to two modes of stdout/err."""

132

133 # Output is sent to a terminal. Terminal width is available, and text

134 # can have ansi colors.

135 TERMINAL = 1

136 # Output is sent to a filter or a file. No width and ansi colors should

137 # be avoided.

138 PIPE = 2

139

140

141def get_path_in_apio_package(subpath: str) -> Path:

142 """Get the full path to the given folder in the apio package.

143 Inputs:

144 * subdir: String with a relative path within the apio package.

145 Use "" for root directory.

146

147 Returns:

148 * The absolute path as a PosixPath() object

149

150 Example: folder="commands"

151 Output: PosixPath('/home/obijuan/.../apio/commands')

152 """

153

154 # -- Get the full path of this file (util.py)

155 # -- Ex: /home/obijuan/.../site-packages/apio/util.py

156 current_python_file = Path(__file__)

157

158 # -- The parent folder is the apio root folder

159 # -- Ex: /home/obijuan/.../site-packages/apio

160 path = current_python_file.parent.parent

161

162 # -- Add the given folder to the path. If subpath = "" this

163 # -- does nothing, but fails if subpath is None.

164 path = path / subpath

165

166 # -- Return the path

167 return path

168

169

170@dataclass(frozen=True)

171class CommandResult:

172 """Contains the results of a command (subprocess) execution."""

173

174 out_text: str | None = None # stdout multi-line text.

175 err_text: str | None = None # stderr multi-line text.

176 exit_code: int | None = None # Exit code, 0 = OK.

177

178

179def exec_command(

180 cmd: list[str],

181 stdout: AsyncPipe,

182 stderr: AsyncPipe,

183 env: dict[str, str],

184) -> CommandResult:

185 """Execute the given command using async stdout/stderr..

186

187 NOTE: When running on windows, this function does not support

188 privilege elevation, to achieve that invoke the subprocess as a shell

189 command as done in drivers.py

190

191 INPUTS:

192 cmd: list of command token (strings)

193 stdout: the AsyncPipe to use for stdout

194 stderr: the AsyncPipe to use for stderr.

195

196 OUTPUT:

197 A CommandResult with the command results.

198 """

199

200 # -- Sanity check.

201 assert isinstance(cmd, list)

202 assert isinstance(cmd[0], str)

203 assert isinstance(stdout, AsyncPipe)

204 assert isinstance(stderr, AsyncPipe)

205 assert env is not None

206

207 # -- Execute the command

208 try:

209 with subprocess.Popen(

210 cmd_in_env(cmd, env),

211 stdout=stdout.fileno(),

212 stderr=stderr.fileno(),

213 shell=False,

214 env=env,

215 ) as proc:

216

217 # -- Wait for completion.

218 _, _ = proc.communicate()

219

220 # -- Get status code.

221 exit_code = proc.returncode

222

223 # -- Close the async pipes.

224 stdout.close()

225 stderr.close()

226

227 # -- User has pressed the Ctrl-C for aborting the command

228 except KeyboardInterrupt:

229 cerror("Aborted by user")

230 # -- NOTE: If using here sys.exit(1) (including indirectly via

231 # -- fatal_error()), apio requires pressing ctl-c twice when running

232 # -- 'apio sim'. This form of exit is more direct and harder.

233 os._exit(1)

234

235 # -- The command does not exist!

236 except FileNotFoundError:

237 fatal_error(

238 "Command not found:",

239 str(cmd),

240 )

241

242 # -- Extract stdout text

243 lines = stdout.get_buffer()

244 out_text = "\n".join(lines)

245

246 # -- Extract stderr text

247 lines = stderr.get_buffer()

248 err_text = "\n".join(lines)

249

250 # -- All done.

251 result = CommandResult(out_text, err_text, exit_code)

252 return result

253

254

255def user_directory_or_cwd(

256 dir_arg: Path | None,

257 *,

258 description: str,

259 must_exist: bool = False,

260 create_if_missing=False,

261) -> Path:

262 """Condition a directory arg with current directory as default. If dir_arg

263 is specified, it is return after validation, else cwd "." is returned.

264 Description is directory function to include in error messages, e.g.

265 "Project" or "Destination".

266 """

267

268 assert not (create_if_missing and must_exist), "Conflicting flags."

269

270 # -- Case 1: User provided dir path.

271 if dir_arg:

272 project_dir = dir_arg

273

274 # -- If exists, it must be a dir.

275 if project_dir.exists() and not project_dir.is_dir(): 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true

276 fatal_error(f"{description} directory is a file: {project_dir}")

277

278 # -- If required, it must exist.

279 if must_exist and not project_dir.exists(): 279 ↛ 280line 279 didn't jump to line 280 because the condition on line 279 was never true

280 fatal_error(

281 f"{description} directory is missing: {str(project_dir)}"

282 )

283

284 # -- If requested, create

285 if create_if_missing and not project_dir.exists():

286 cout(f"Creating folder: {project_dir}")

287 project_dir.mkdir(parents=True)

288

289 # -- All done.

290 return project_dir

291

292 # -- Case 2: Using current directory.

293 # -- We prefer the relative path "." over the absolute path Path.cwd().

294 return Path(".")

295

296

297def get_python_version() -> str:

298 """Return a string with the python version"""

299

300 return f"{sys.version_info[0]}.{sys.version_info[1]}"

301

302

303def get_python_ver_tuple() -> tuple[int, int, int]:

304 """Return a tuple with the python version. e.g. (3, 12, 1)."""

305 return sys.version_info[:3]

306

307

308def plurality(

309 obj: Any,

310 singular: str,

311 plural: str | None = None,

312 include_num: bool = True,

313) -> str:

314 """Returns singular or plural based on the size of the object."""

315 # -- Figure out the size of the object

316 if isinstance(obj, int):

317 n = obj

318 else:

319 n = len(obj)

320

321 # -- For value of 1 return the singular form.

322 if n == 1:

323 if include_num: 323 ↛ 325line 323 didn't jump to line 325 because the condition on line 323 was always true

324 return f"{n} {singular}"

325 return singular

326

327 # -- For all other values, return the plural form.

328 if plural is None: 328 ↛ 331line 328 didn't jump to line 331 because the condition on line 328 was always true

329 plural = singular + "s"

330

331 if include_num:

332 return f"{n} {plural}"

333 return plural

334

335

336def list_plurality(str_list: list[str], conjunction: str) -> str:

337 """Format a list as a human friendly string."""

338 # -- This is a programming error. Not a user error.

339 assert str_list, "list_plurality expect len() >= 1."

340

341 # -- Handle the case of a single item.

342 if len(str_list) == 1:

343 return str_list[0]

344

345 # -- Handle the case of 2 items.

346 if len(str_list) == 2:

347 return f"{str_list[0]} {conjunction} {str_list[1]}"

348

349 # -- Handle the case of three or more items.

350 return ", ".join(str_list[:-1]) + f", {conjunction} {str_list[-1]}"

351

352

353def get_apio_version_tuple() -> tuple[int, int, int]:

354 """Returns the version of the apio package as tuple of 3 ints."""

355 # -- Apio's version is defined in the __init__.py file of the apio package.

356 # -- Using the version from a file in the apio package rather than from

357 # -- the pip metadata makes apio more self contained, for example when

358 # -- installing with pyinstaller rather than with pip.

359 ver: tuple[int, int, int] = apio.APIO_VERSION

360 assert len(ver) == 3, ver

361 assert isinstance(ver[0], int)

362 assert isinstance(ver[1], int)

363 assert isinstance(ver[2], int)

364 return ver

365

366

367def get_apio_version_str() -> str:

368 """Returns the version of the apio package as a string like "1.22.3"."""

369 ver: tuple[int, int, int] = get_apio_version_tuple()

370 return f"{ver[0]}.{ver[1]}.{ver[2]}"

371

372

373def get_apio_release_info() -> str:

374 """Returns the release info string."""

375 return apio.RELEASE_INFO

376

377

378def get_apio_version_message() -> str:

379 """Returns the string to show on `apio --version`."""

380 ver_str = get_apio_version_str()

381 release_str = get_apio_release_info() or "no release info"

382 return f"Apio CLI version {ver_str} ({release_str})"

383

384

385def _check_apio_dir(apio_dir: Path, desc: str, env_var: str):

386 """Checks the apio home dir or packages dir path for the apio

387 requirements."""

388

389 # Sanity check. If this fails, it's a programming error.

390 assert isinstance(

391 apio_dir, Path

392 ), f"Error: {desc} is no a Path: {type(apio_dir)}, {apio_dir}"

393

394 # -- The path should be absolute, see discussion here:

395 # -- https://github.com/FPGAwars/apio/issues/522

396 if not apio_dir.is_absolute():

397 fatal_error(

398 f"Apio {desc} should be an absolute path " f"[{str(apio_dir)}].",

399 info=f"You can use the system env var '{env_var}' to set "

400 + f"a different apio {desc}.",

401 )

402

403 # -- We have problem with spaces and non ascii character above value

404 # -- 127, so we allow only ascii characters in the range [33, 127].

405 # -- See here https://github.com/FPGAwars/apio/issues/515

406 for ch in str(apio_dir):

407 if ord(ch) < 33 or ord(ch) > 127:

408 # -- Name the char if it has no visible glyph, e.g. space or tab.

409 if ch == " ":

410 ch_name = "space"

411 elif ch.isprintable(): 411 ↛ 414line 411 didn't jump to line 414 because the condition on line 411 was always true

412 ch_name = ch

413 else:

414 ch_name = repr(ch)

415 fatal_error(

416 f"Unsupported character [{ch_name}] in apio {desc}: "

417 + f"[{str(apio_dir)}].",

418 info="Only the ASCII characters in the range 33 to 127 are "

419 + "allowed, with no spaces. You can use the "

420 + f"system env var '{env_var}' to set a different apio "

421 + f"{desc}.",

422 )

423

424

425def resolve_home_dir() -> Path:

426 """Get the absolute apio home dir. This is the apio folder where the

427 profile is located and the packages are installed.

428 The apio home dir can be overridden using the APIO_HOME environment

429 variable. If not set, the user_home/.apio folder is used by default:

430 Ej. Linux: /home/obijuan/.apio

431 If the folders does not exist, they are created

432 """

433

434 # -- Get the optional apio home env.

435 apio_home_dir_env = env_options.get(env_options.APIO_HOME)

436

437 # -- If the env vars specified an home dir then use it.

438 if apio_home_dir_env:

439 # -- Expand user home '~' marker, if exists.

440 apio_home_dir_env = os.path.expanduser(apio_home_dir_env)

441 # -- Expand varas such as $HOME or %HOME% on windows.

442 apio_home_dir_env = os.path.expandvars(apio_home_dir_env)

443 # -- Convert string to path.

444 home_dir = Path(apio_home_dir_env)

445 else:

446 # -- Else, use the default home dir ~/.apio.

447 home_dir = Path.home() / ".apio"

448

449 # -- Verify that the home dir meets apio's requirements.

450 _check_apio_dir(home_dir, "home dir", "APIO_HOME")

451

452 # -- Create the folder if it does not exist

453 try:

454 home_dir.mkdir(parents=True, exist_ok=True)

455 except OSError as e:

456 # -- E.g. no permission, or the path exists as a plain file.

457 fatal_error(

458 f"No usable home directory {home_dir}",

459 cause=e,

460 )

461

462 # Return the home_dir as a Path

463 return home_dir

464

465

466def resolve_packages_dir(apio_home_dir: Path) -> Path:

467 """Get the absolute apio packages dir. This is the apio folder where the

468 packages are installed. The default apio packages dir can be overridden

469 using the APIO_PACKAGES environment variable. If not set,

470 the <apio-home>/packages folder is used by default:

471 Ej. Linux: /home/obijuan/.apio/packages

472 If the folders does not exist, they are created

473 """

474

475 # -- Get the optional apio packages env.

476 apio_packages_dir_env = env_options.get(env_options.APIO_PACKAGES)

477

478 # -- If the env vars specified an packages dir then use it.

479 if apio_packages_dir_env: 479 ↛ 505line 479 didn't jump to line 505 because the condition on line 479 was always true

480 # -- Expand user home '~' marker, if exists.

481 apio_packages_dir_env = os.path.expanduser(apio_packages_dir_env)

482 # -- Expand vars such as $HOME or %HOME% on windows.

483 apio_packages_dir_env = os.path.expandvars(apio_packages_dir_env)

484 # -- Disallow '..' path components since they can be used to bypass

485 # -- the safety check below, e.g. '/home/me/apio-packages/..'.

486 if ".." in Path(apio_packages_dir_env).parts:

487 fatal_error(

488 "Apio packages dir APIO_PACKAGES should not contain "

489 + f"'..' [{apio_packages_dir_env}]."

490 )

491 # -- Verify that the expanded value contains 'apio' and 'packages'

492 # -- (case insensitive) to make sure we don't clobber system

493 # -- directories. This is done after the expansion since the

494 # -- expanded value is the one we use.

495 lower_value = apio_packages_dir_env.lower()

496 if "apio" not in lower_value or "packages" not in lower_value:

497 fatal_error(

498 "Apio packages dir APIO_PACKAGES should include the "

499 + f"strings 'apio' and 'packages' [{apio_packages_dir_env}]."

500 )

501 # -- Convert string to path.

502 packages_dir = Path(apio_packages_dir_env)

503 else:

504 # -- Else, use the default <home_dir>/packages.

505 packages_dir = apio_home_dir / "packages"

506

507 # -- Verify that the home dir meets apio's requirements.

508 _check_apio_dir(packages_dir, "packages dir", "APIO_PACKAGES")

509

510 # -- Create the folder if it does not exist

511 # try:

512 # packages_dir.mkdir(parents=True, exist_ok=True)

513 # except PermissionError:

514 # cerror(f"No usable packages directory {packages_dir}")

515 # sys.exit(1)

516

517 # Return the packages as a Path

518 return packages_dir

519

520

521def fpga_arch_sort_key(fpga_arch: str) -> Any:

522 """Given an fpga arch name such as 'ice40', return a sort key

523 got force our preferred order of sorting by architecture. Used in

524 reports such as examples, fpgas, and boards."""

525

526 # -- The preferred order of architectures, Add more if adding new

527 # -- architectures.

528 archs = ["ice40", "ecp5", "gowin", "xilinx"]

529

530 # -- Primary key with preferred architecture first and in the

531 # -- preferred order.

532 primary_key = archs.index(fpga_arch) if fpga_arch in archs else len(archs)

533

534 # -- Construct the key, unknown architectures list at the end by

535 # -- lexicographic order.

536 return (primary_key, fpga_arch)

537

538

539@contextmanager

540def pushd(target_dir: Path):

541 """A context manager for temporary execution in a given directory."""

542 prev_dir = os.getcwd()

543 os.chdir(target_dir)

544 try:

545 yield

546 finally:

547 os.chdir(prev_dir)

548

549

550def is_pyinstaller_app() -> bool:

551 """Return true if this is a pyinstaller packaged app.

552 Base on https://pyinstaller.org/en/stable/runtime-information.html

553 """

554 return getattr(sys, "frozen", False) and hasattr(sys, "_MEIPASS")

555

556

557def compute_file_sha256(path: Path) -> str:

558 """Given a file path, compute and return sha256 string."""

559 # -- Sanity check

560 assert path.is_file(), path

561

562 # -- Compute sha256. We perform it in chunks to limit the memory

563 # -- requirements.

564 h = hashlib.sha256()

565 with path.open("rb") as f:

566 for chunk in iter(lambda: f.read(1024 * 1024), b""):

567 h.update(chunk)

568 sha256 = h.hexdigest()

569

570 # -- All done OK.

571 return sha256

572

573

574def json_dict_contains_substring(data, substr: str) -> bool:

575 """Returns True if any of the string values in the dict (at any level)

576 contains the given substring, case insensitive."""

577 substr = substr.casefold()

578

579 def walk(obj) -> bool:

580 if isinstance(obj, str):

581 return substr in obj.casefold()

582 if isinstance(obj, dict):

583 return any(walk(v) for v in obj.values())

584 if isinstance(obj, list):

585 return any(walk(v) for v in obj)

586 return False

587

588 return walk(data)