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

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

16from dataclasses import dataclass

17from typing import Optional, Any, Tuple, List

18import subprocess

19from threading import Thread

20from pathlib import Path

21import apio

22from apio.utils import env_options

23from apio.common.apio_console import cout, cerror

24from apio.common.apio_styles import INFO

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

28# -- Constants

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

32class ApioException(Exception):

33 """Apio error"""

36class AsyncPipe(Thread):

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

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

39 show its progress."""

41 def __init__(self, line_callback=None):

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

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

44 and terminator is one of:

45 "\r" (CR)

46 "\n" (LF)

47 "" (EOF)

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

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

51 """

53 Thread.__init__(self)

54 self.outcallback = line_callback

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

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

59 self._lines_buffer = []

61 self.start()

63 def get_buffer(self):

64 """DOC: TODO"""

66 return self._lines_buffer

68 def fileno(self):

69 """DOC: TODO"""

71 return self._fd_write

73 def _handle_incoming_line(self, bfr: bytearray, terminator: str):

74 """Handle a new incoming line.

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

76 See __init__ for the description of terminator.

77 """

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

79 # -- chars with "�"

80 line = bfr.decode("utf-8", errors="replace")

82 # -- Append to the lines log buffer.

83 self._lines_buffer.append(line)

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

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

87 self.outcallback(line, terminator)

89 def run(self):

90 """DOC: TODO"""

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

93 # -- its line terminator.

94 bfr = bytearray()

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

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

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

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

100 while True:

101 b: bytes = f.read(1)

102 assert len(b) <= 1

103

104 # -- Handle end of file

105 if not b:

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

107 self._handle_incoming_line(bfr, "")

108 return

109

110 # -- Handle \r terminator

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

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

113 bfr.clear()

114 continue

115

116 # -- Handle \n terminator

117 if b == b"\n":

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

119 bfr.clear()

120 continue

121

122 # -- Handle a regular character

123 bfr.append(b[0])

124

125 def close(self):

126 """DOC: TODO"""

127

128 os.close(self._fd_write)

129 self.join()

130

131

132class TerminalMode(Enum):

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

134

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

136 # can have ansi colors.

137 TERMINAL = 1

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

139 # be avoided.

140 PIPE = 2

141

142

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

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

145 Inputs:

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

147 Use "" for root directory.

148

149 Returns:

150 * The absolute path as a PosixPath() object

151

152 Example: folder="commands"

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

154 """

155

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

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

158 current_python_file = Path(__file__)

159

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

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

162 path = current_python_file.parent.parent

163

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

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

166 path = path / subpath

167

168 # -- Return the path

169 return path

170

171

172@dataclass(frozen=True)

173class CommandResult:

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

175

176 out_text: Optional[str] = None # stdout multi-line text.

177 err_text: Optional[str] = None # stderr multi-line text.

178 exit_code: Optional[int] = None # Exit code, 0 = OK.

179

180

181def exec_command(

182 cmd: List[str], stdout: AsyncPipe, stderr: AsyncPipe

183) -> CommandResult:

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

185

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

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

188 done in drivers.py

189

190 INPUTS:

191 cmd: list of command token (strings)

192 stdout: the AsyncPipe to use for stdout

193 stderr: the AsyncPipe to use for stderr.

194

195 OUTPUT:

196 A CommandResult with the command results.

197 """

198

199 # -- Sanity check.

200 assert isinstance(cmd, list)

201 assert isinstance(cmd[0], str)

202 assert isinstance(stdout, AsyncPipe)

203 assert isinstance(stderr, AsyncPipe)

204

205 # -- Execute the command

206 try:

207 with subprocess.Popen(

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

209 ) as proc:

210

211 # -- Wait for completion.

212 out_text, err_text = proc.communicate()

213

214 # -- Get status code.

215 exit_code = proc.returncode

216

217 # -- Close the async pipes.

218 stdout.close()

219 stderr.close()

220

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

222 except KeyboardInterrupt:

223 cerror("Aborted by user")

224 # -- NOTE: If using here sys.exit(1), apio requires pressing ctl-c

225 # -- twice when running 'apio sim'. This form of exit is more direct

226 # -- and harder.

227 os._exit(1)

228

229 # -- The command does not exist!

230 except FileNotFoundError:

231 cerror("Command not found:", str(cmd))

232 sys.exit(1)

233

234 # -- Extract stdout text

235 lines = stdout.get_buffer()

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

237

238 # -- Extract stderr text

239 lines = stderr.get_buffer()

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

241

242 # -- All done.

243 result = CommandResult(out_text, err_text, exit_code)

244 return result

245

246

247def user_directory_or_cwd(

248 dir_arg: Optional[Path],

249 *,

250 description: str,

251 must_exist: bool = False,

252 create_if_missing=False,

253) -> Path:

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

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

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

257 "Project" or "Destination".

258 """

259

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

261

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

263 if dir_arg:

264 project_dir = dir_arg

265

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

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

268 cerror(f"{description} directory is a file: {project_dir}")

269 sys.exit(1)

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 cerror(f"{description} directory is missing: {str(project_dir)}")

274 sys.exit(1)

275

276 # -- If requested, create

277 if create_if_missing and not project_dir.exists():

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

279 project_dir.mkdir(parents=True)

280

281 # -- All done.

282 return project_dir

283

284 # -- Case 2: Using current directory.

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

286 return Path(".")

287

288

289def get_python_version() -> str:

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

291

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

293

294

295def get_python_ver_tuple() -> Tuple[int, int, int]:

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

297 return sys.version_info[:3]

298

299

300def plurality(

301 obj: Any,

302 singular: str,

303 plural: str | None = None,

304 include_num: bool = True,

305) -> str:

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

307 # -- Figure out the size of the object

308 if isinstance(obj, int):

309 n = obj

310 else:

311 n = len(obj)

312

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

314 if n == 1:

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

316 return f"{n} {singular}"

317 return singular

318

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

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

321 plural = singular + "s"

322

323 if include_num:

324 return f"{n} {plural}"

325 return plural

326

327

328def list_plurality(str_list: List[str], conjunction: str) -> str:

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

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

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

332

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

334 if len(str_list) == 1:

335 return str_list[0]

336

337 # -- Handle the case of 2 items.

338 if len(str_list) == 2:

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

340

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

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

343

344

345def debug_level() -> int:

346 """Returns the current debug level, with 0 as 'off'."""

347

348 # -- We get a fresh value so it can be adjusted dynamically when needed.

349 level_str = env_options.get(env_options.APIO_DEBUG, "0")

350 try:

351 level_int = int(level_str) # pyright: ignore[reportArgumentType]

352 except ValueError:

353 cerror(f"APIO_DEBUG value '{level_str}' is not an int.")

354 sys.exit(1)

355

356 # -- All done. We don't validate the value, assuming the user knows how

357 # -- to use it.

358 return level_int

359

360

361def is_debug(level: int) -> bool:

362 """Returns True if apio is in debug mode level 'level' or higher. Use

363 it to enable printing of debug information but not to modify the behavior

364 of the code. Also, all apio tests should be performed with debug

365 disabled. Important debug information should be at level 1 while

366 less important or spammy should be at higher levels."""

367 # -- Sanity check. A failure is indicates a programming error.

368 assert isinstance(level, int), type(level)

369 assert 1 <= level <= 10, level

370

371 return debug_level() >= level

372

373

374def get_apio_version_tuple() -> Tuple[int, int, int]:

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

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

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

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

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

380 ver: Tuple[int, int, int] = apio.APIO_VERSION

381 assert len(ver) == 3, ver

382 assert isinstance(ver[0], int)

383 assert isinstance(ver[1], int)

384 assert isinstance(ver[2], int)

385 return ver

386

387

388def get_apio_version_str() -> str:

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

390 ver: Tuple[int, int, int] = get_apio_version_tuple()

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

392

393

394def get_apio_release_info() -> str:

395 """Returns the release info string."""

396 return apio.RELEASE_INFO

397

398

399def get_apio_version_message() -> str:

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

401 ver_str = get_apio_version_str()

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

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

404

405

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

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

408 requirements."""

409

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

411 assert isinstance(

412 apio_dir, Path

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

414

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

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

417 if not apio_dir.is_absolute():

418 cerror(

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

420 )

421 cout(

422 f"You can use the system env var '{env_var}' to set "

423 f"a different apio {desc}.",

424 style=INFO,

425 )

426 sys.exit(1)

427

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

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

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

431 for ch in str(apio_dir):

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

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

434 if ch == " ":

435 ch_name = "space"

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

437 ch_name = ch

438 else:

439 ch_name = repr(ch)

440 cerror(

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

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

443 )

444 cout(

445 "Only the ASCII characters in the range 33 to 127 are "

446 "allowed, with no spaces. You can use the "

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

448 f"{desc}.",

449 style=INFO,

450 )

451 sys.exit(1)

452

453

454def resolve_home_dir() -> Path:

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

456 profile is located and the packages are installed.

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

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

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

460 If the folders does not exist, they are created

461 """

462

463 # -- Get the optional apio home env.

464 apio_home_dir_env = env_options.get(env_options.APIO_HOME)

465

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

467 if apio_home_dir_env:

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

469 apio_home_dir_env = os.path.expanduser(apio_home_dir_env)

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

471 apio_home_dir_env = os.path.expandvars(apio_home_dir_env)

472 # -- Convert string to path.

473 home_dir = Path(apio_home_dir_env)

474 else:

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

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

477

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

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

480

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

482 try:

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

484 except OSError as e:

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

486 cerror(f"No usable home directory {home_dir}", f"{e}")

487 sys.exit(1)

488

489 # Return the home_dir as a Path

490 return home_dir

491

492

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

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

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

496 using the APIO_PACKAGES environment variable. If not set,

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

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

499 If the folders does not exist, they are created

500 """

501

502 # -- Get the optional apio packages env.

503 apio_packages_dir_env = env_options.get(env_options.APIO_PACKAGES)

504

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

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

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

508 # -- don't clobber system directories.

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

510 cerror(

511 "Apio packages dir APIO_PACKAGES should include the "

512 "string 'packages'."

513 )

514 sys.exit(1)

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

516 apio_packages_dir_env = os.path.expanduser(apio_packages_dir_env)

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

518 apio_packages_dir_env = os.path.expandvars(apio_packages_dir_env)

519 # -- Convert string to path.

520 packages_dir = Path(apio_packages_dir_env)

521 else:

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

523 packages_dir = apio_home_dir / "packages"

524

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

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

527

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

529 # try:

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

531 # except PermissionError:

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

533 # sys.exit(1)

534

535 # Return the packages as a Path

536 return packages_dir

537

538

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

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

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

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

543

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

545 # -- architectures.

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

547

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

549 # -- preferred order.

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

551

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

553 # -- lexicographic order.

554 return (primary_key, fpga_arch)

555

556

557def subprocess_call(

558 cmd: List[str],

559) -> int:

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

561

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

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

564

565 # -- Invoke the command.

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

567

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

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

570

571 # -- If ok, return.

572 if exit_code == 0:

573 return exit_code

574

575 # -- Here when error

576 cerror(f"Command failed: {cmd}")

577 sys.exit(1)

578

579

580@contextmanager

581def pushd(target_dir: Path):

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

583 prev_dir = os.getcwd()

584 os.chdir(target_dir)

585 try:

586 yield

587 finally:

588 os.chdir(prev_dir)

589

590

591def is_pyinstaller_app() -> bool:

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

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

594 """

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

596

597

598def is_under_vscode_debugger() -> bool:

599 """Returns true if running under VSCode debugger."""

600 # if os.environ.get("TERM_PROGRAM") == "vscode":

601 # return True

602 if os.environ.get("DEBUGPY_RUNNING"): 602 ↛ 603line 602 didn't jump to line 603 because the condition on line 602 was never true

603 return True

604 return False