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

1# -*- coding: utf-8 -*-

2# -- This file is part of the Apio project

4# -- Author Jesús Arroyo

5# -- License GPLv2

6# -- Derived from:

7# ---- Platformio project

9# ---- License Apache v2

10"""Misc utility functions and classes."""

12import sys

13import os

14from contextlib import contextmanager

15from enum import Enum, unique

16from dataclasses import dataclass

17from typing import Any

18import subprocess

19from threading import Thread

20from pathlib import Path

21import hashlib

22from tarfile import open as tarfile_open

23from rich.progress import track

24import apio

25from apio.utils import env_options

26from apio.common.apio_console import cout, cerror, console, fatal_error

27from apio.common.debug_util import is_debug

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

30# -- Constants

31# ----------------------------------------

34class AsyncPipe(Thread):

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

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

37 show its progress."""

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

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

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

42 and terminator is one of:

43 "\r" (CR)

44 "\n" (LF)

45 "" (EOF)

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

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

49 """

51 Thread.__init__(self)

52 self.outcallback = line_callback

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

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

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

59 self.start()

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

62 """DOC: TODO"""

64 return self._lines_buffer

66 def fileno(self) -> int:

67 """DOC: TODO"""

69 return self._fd_write

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

72 """Handle a new incoming line.

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

74 See __init__ for the description of terminator.

75 """

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

77 # -- chars with "�"

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

80 # -- Append to the lines log buffer.

81 self._lines_buffer.append(line)

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

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

85 self.outcallback(line, terminator)

87 def run(self) -> None:

88 """DOC: TODO"""

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

91 # -- its line terminator.

92 bfr = bytearray()

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

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

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

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

98 while True:

99 b: bytes = f.read(1)

100 assert len(b) <= 1

101

102 # -- Handle end of file

103 if not b:

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

105 self._handle_incoming_line(bfr, "")

106 return

107

108 # -- Handle \r terminator

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

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

111 bfr.clear()

112 continue

113

114 # -- Handle \n terminator

115 if b == b"\n":

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

117 bfr.clear()

118 continue

119

120 # -- Handle a regular character

121 bfr.append(b[0])

122

123 def close(self):

124 """DOC: TODO"""

125

126 os.close(self._fd_write)

127 self.join()

128

129

130@unique

131class TerminalMode(Enum):

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

133

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

135 # can have ansi colors.

136 TERMINAL = 1

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

138 # be avoided.

139 PIPE = 2

140

141

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

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

144 Inputs:

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

146 Use "" for root directory.

147

148 Returns:

149 * The absolute path as a PosixPath() object

150

151 Example: folder="commands"

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

153 """

154

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

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

157 current_python_file = Path(__file__)

158

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

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

161 path = current_python_file.parent.parent

162

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

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

165 path = path / subpath

166

167 # -- Return the path

168 return path

169

170

171@dataclass(frozen=True)

172class CommandResult:

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

174

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

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

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

178

179

180def exec_command(

181 cmd: list[str], stdout: AsyncPipe, stderr: AsyncPipe

182) -> CommandResult:

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

184

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

186 privilege elevation, to achieve that, use os.system() instead, as

187 done in drivers.py

188

189 INPUTS:

190 cmd: list of command token (strings)

191 stdout: the AsyncPipe to use for stdout

192 stderr: the AsyncPipe to use for stderr.

193

194 OUTPUT:

195 A CommandResult with the command results.

196 """

197

198 # -- Sanity check.

199 assert isinstance(cmd, list)

200 assert isinstance(cmd[0], str)

201 assert isinstance(stdout, AsyncPipe)

202 assert isinstance(stderr, AsyncPipe)

203

204 # -- Execute the command

205 try:

206 with subprocess.Popen(

207 cmd, stdout=stdout.fileno(), stderr=stderr.fileno(), shell=False

208 ) as proc:

209

210 # -- Wait for completion.

211 _, _ = proc.communicate()

212

213 # -- Get status code.

214 exit_code = proc.returncode

215

216 # -- Close the async pipes.

217 stdout.close()

218 stderr.close()

219

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

221 except KeyboardInterrupt:

222 cerror("Aborted by user")

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

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

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

226 os._exit(1)

227

228 # -- The command does not exist!

229 except FileNotFoundError:

230 fatal_error(

231 "Command not found:",

232 str(cmd),

233 )

234

235 # -- Extract stdout text

236 lines = stdout.get_buffer()

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

238

239 # -- Extract stderr text

240 lines = stderr.get_buffer()

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

242

243 # -- All done.

244 result = CommandResult(out_text, err_text, exit_code)

245 return result

246

247

248def user_directory_or_cwd(

249 dir_arg: Path | None,

250 *,

251 description: str,

252 must_exist: bool = False,

253 create_if_missing=False,

254) -> Path:

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

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

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

258 "Project" or "Destination".

259 """

260

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

262

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

264 if dir_arg:

265 project_dir = dir_arg

266

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

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

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

270

271 # -- If required, it must exist.

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

273 fatal_error(

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

275 )

276

277 # -- If requested, create

278 if create_if_missing and not project_dir.exists():

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

280 project_dir.mkdir(parents=True)

281

282 # -- All done.

283 return project_dir

284

285 # -- Case 2: Using current directory.

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

287 return Path(".")

288

289

290def get_python_version() -> str:

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

292

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

294

295

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

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

298 return sys.version_info[:3]

299

300

301def plurality(

302 obj: Any,

303 singular: str,

304 plural: str | None = None,

305 include_num: bool = True,

306) -> str:

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

308 # -- Figure out the size of the object

309 if isinstance(obj, int):

310 n = obj

311 else:

312 n = len(obj)

313

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

315 if n == 1:

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

317 return f"{n} {singular}"

318 return singular

319

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

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

322 plural = singular + "s"

323

324 if include_num:

325 return f"{n} {plural}"

326 return plural

327

328

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

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

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

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

333

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

335 if len(str_list) == 1:

336 return str_list[0]

337

338 # -- Handle the case of 2 items.

339 if len(str_list) == 2:

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

341

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

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

344

345

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

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

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

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

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

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

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

353 assert len(ver) == 3, ver

354 assert isinstance(ver[0], int)

355 assert isinstance(ver[1], int)

356 assert isinstance(ver[2], int)

357 return ver

358

359

360def get_apio_version_str() -> str:

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

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

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

364

365

366def get_apio_release_info() -> str:

367 """Returns the release info string."""

368 return apio.RELEASE_INFO

369

370

371def get_apio_version_message() -> str:

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

373 ver_str = get_apio_version_str()

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

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

376

377

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

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

380 requirements."""

381

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

383 assert isinstance(

384 apio_dir, Path

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

386

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

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

389 if not apio_dir.is_absolute():

390 fatal_error(

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

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

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

394 )

395

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

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

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

399 for ch in str(apio_dir):

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

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

402 if ch == " ":

403 ch_name = "space"

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

405 ch_name = ch

406 else:

407 ch_name = repr(ch)

408 fatal_error(

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

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

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

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

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

414 + f"{desc}.",

415 )

416

417

418def resolve_home_dir() -> Path:

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

420 profile is located and the packages are installed.

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

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

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

424 If the folders does not exist, they are created

425 """

426

427 # -- Get the optional apio home env.

428 apio_home_dir_env = env_options.get(env_options.APIO_HOME)

429

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

431 if apio_home_dir_env:

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

433 apio_home_dir_env = os.path.expanduser(apio_home_dir_env)

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

435 apio_home_dir_env = os.path.expandvars(apio_home_dir_env)

436 # -- Convert string to path.

437 home_dir = Path(apio_home_dir_env)

438 else:

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

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

441

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

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

444

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

446 try:

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

448 except OSError as e:

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

450 fatal_error(

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

452 cause=e,

453 )

454

455 # Return the home_dir as a Path

456 return home_dir

457

458

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

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

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

462 using the APIO_PACKAGES environment variable. If not set,

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

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

465 If the folders does not exist, they are created

466 """

467

468 # -- Get the optional apio packages env.

469 apio_packages_dir_env = env_options.get(env_options.APIO_PACKAGES)

470

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

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

473 # -- Verify that the env variable contains 'packages' to make sure we

474 # -- don't clobber system directories.

475 if "packages" not in apio_packages_dir_env: 475 ↛ 476line 475 didn't jump to line 476 because the condition on line 475 was never true

476 fatal_error(

477 "Apio packages dir APIO_PACKAGES should include the "

478 + "string 'packages'."

479 )

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

481 apio_packages_dir_env = os.path.expanduser(apio_packages_dir_env)

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

483 apio_packages_dir_env = os.path.expandvars(apio_packages_dir_env)

484 # -- Convert string to path.

485 packages_dir = Path(apio_packages_dir_env)

486 else:

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

488 packages_dir = apio_home_dir / "packages"

489

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

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

492

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

494 # try:

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

496 # except PermissionError:

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

498 # sys.exit(1)

499

500 # Return the packages as a Path

501 return packages_dir

502

503

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

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

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

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

508

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

510 # -- architectures.

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

512

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

514 # -- preferred order.

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

516

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

518 # -- lexicographic order.

519 return (primary_key, fpga_arch)

520

521

522def subprocess_call(

523 cmd: list[str],

524) -> int:

525 """A helper for running subprocess.call. Exit if an error."""

526

527 if is_debug(1): 527 ↛ 528line 527 didn't jump to line 528 because the condition on line 527 was never true

528 cout(f"subprocess_call: {cmd}")

529

530 # -- Invoke the command.

531 exit_code = subprocess.call(cmd, shell=False)

532

533 if is_debug(1): 533 ↛ 534line 533 didn't jump to line 534 because the condition on line 533 was never true

534 cout(f"subprocess_call: exit code is {exit_code}")

535

536 # -- If ok, return.

537 if exit_code == 0:

538 return exit_code

539

540 # -- Here when error

541 fatal_error(f"Command failed: {cmd}")

542

543

544@contextmanager

545def pushd(target_dir: Path):

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

547 prev_dir = os.getcwd()

548 os.chdir(target_dir)

549 try:

550 yield

551 finally:

552 os.chdir(prev_dir)

553

554

555def is_pyinstaller_app() -> bool:

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

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

558 """

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

560

561

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

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

564 # -- Sanity check

565 assert path.is_file(), path

566

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

568 # -- requirements.

569 h = hashlib.sha256()

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

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

572 h.update(chunk)

573 sha256 = h.hexdigest()

574

575 # -- All done OK.

576 return sha256

577

578

579def unpack_tgz(archive_file_path: Path, dest_dir: Path) -> None:

580 """Unpack a .tgz archive into the given dest directory."""

581

582 # -- Get the file name

583 archive_name = archive_file_path.name

584

585 # -- Check it's a ".tgz" file

586 if not archive_name.endswith(".tgz"): 586 ↛ 587line 586 didn't jump to line 587 because the condition on line 586 was never true

587 fatal_error(f"Cannot unarchive'{archive_name}', it's not a .tgz file.")

588

589 # -- Extract all items while animating the progress bar.

590 with tarfile_open(archive_file_path) as tar_file:

591

592 items = tar_file.getmembers()

593

594 for i in track(

595 range(len(items)),

596 description="Unpacking ",

597 console=console(),

598 ):

599 item = items[i]

600

601 # -- Skip .gitignore files.

602 if hasattr(item, "filename") and item.filename.endswith( 602 ↛ 605line 602 didn't jump to line 605 because the condition on line 602 was never true

603 ".gitignore"

604 ):

605 continue

606

607 # -- Extract the item

608 if get_python_ver_tuple() >= (3, 12, 0): 608 ↛ 615line 608 didn't jump to line 615 because the condition on line 608 was always true

609 # -- Special case for avoiding the tar deprecation warning.

610 # Search

611 # -- 'extraction_filter' in the page

612 # -- https://docs.python.org/3/library/tarfile.html

613 tar_file.extract(item, dest_dir, filter="fully_trusted")

614 else:

615 tar_file.extract(item, dest_dir)