Coverage for pyTooling/CLIAbstraction/__init__.py: 77%
254 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-25 12:33 +0000
1# ==================================================================================================================== #
2# _____ _ _ ____ _ ___ _ _ _ _ _ #
3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ / ___| | |_ _| / \ | |__ ___| |_ _ __ __ _ ___| |_(_) ___ _ __ #
4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` || | | | | | / _ \ | '_ \/ __| __| '__/ _` |/ __| __| |/ _ \| '_ \ #
5# | |_) | |_| || | (_) | (_) | | | | | | (_| || |___| |___ | | / ___ \| |_) \__ \ |_| | | (_| | (__| |_| | (_) | | | | #
6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)____|_____|___/_/ \_\_.__/|___/\__|_| \__,_|\___|\__|_|\___/|_| |_| #
7# |_| |___/ |___/ #
8# ==================================================================================================================== #
9# Authors: #
10# Patrick Lehmann #
11# #
12# License: #
13# ==================================================================================================================== #
14# Copyright 2017-2026 Patrick Lehmann - Bötzingen, Germany #
15# Copyright 2014-2016 Technische Universität Dresden - Germany, Chair of VLSI-Design, Diagnostics and Architecture #
16# #
17# Licensed under the Apache License, Version 2.0 (the "License"); #
18# you may not use this file except in compliance with the License. #
19# You may obtain a copy of the License at #
20# #
21# http://www.apache.org/licenses/LICENSE-2.0 #
22# #
23# Unless required by applicable law or agreed to in writing, software #
24# distributed under the License is distributed on an "AS IS" BASIS, #
25# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
26# See the License for the specific language governing permissions and #
27# limitations under the License. #
28# #
29# SPDX-License-Identifier: Apache-2.0 #
30# ==================================================================================================================== #
31#
32"""
33Basic abstraction layer for executables.
35.. seealso::
37 :mod:`pyTooling.Attributes.ArgParse`
38 |rarr| The other direction: describing the command line a program accepts.
39 :mod:`pyTooling.TerminalUI`
40 |rarr| Writing the program's own messages to the terminal.
41 :mod:`pyTooling.Platform`
42 |rarr| Deciding which executable name and path style the current platform uses.
43"""
44from __future__ import annotations
46# __keywords__ = ["abstract", "executable", "cli", "cli arguments"]
48from os import environ as os_environ
49from pathlib import Path
50from platform import system
51from shutil import which as shutil_which
52from subprocess import Popen as Subprocess_Popen, PIPE as Subprocess_Pipe, STDOUT as Subprocess_StdOut, TimeoutExpired
53from typing import Optional as Nullable, ClassVar, Iterator, Generator, Any, Mapping, Iterable
55from pyTooling.Decorators import export, readonly
56from pyTooling.MetaClasses import ExtendedType
57from pyTooling.Exceptions import ToolingException, PlatformNotSupportedError
58from pyTooling.Common import getFullyQualifiedName, getTypeName
59from pyTooling.Attributes import Attribute
60from pyTooling.CLIAbstraction.Argument import CommandLineArgument, formatCommandLine, NamedAndValuedArgument, \
61 ValuedArgument, PathArgument, PathListArgument, NamedTupledArgument
62from pyTooling.CLIAbstraction.ValuedFlag import ValuedFlag
63from pyTooling.Platform import Platform
66@export
67class CLIAbstractionError(ToolingException):
68 """Base-exception of all exceptions raised by :mod:`pyTooling.CLIAbstraction`."""
71@export
72class DryRunError(CLIAbstractionError):
73 """This exception is raised if an executable is launched while in dry-run mode."""
76@export
77class CLIArgument(Attribute):
78 """An attribute to annotate nested classes as an CLI argument."""
81@export
82class Environment(metaclass=ExtendedType, slots=True):
83 """
84 A class describing the environment of an executable.
86 .. topic:: Content of the environment
88 * Environment variables
89 """
90 _variables: dict[str, str] #: Dictionary of active environment variables.
92 # TODO: derive environment from existing environment object.
93 def __init__(
94 self, *,
95 environment: Nullable[Environment] = None,
96 newVariables: Nullable[Mapping[str, str]] = None,
97 addVariables: Nullable[Mapping[str, str]] = None,
98 delVariables: Nullable[Iterable[str]] = None
99 ) -> None:
100 """
101 Initializes an environment class managing.
103 .. topic:: Algorithm
105 1. Create a new dictionary of environment variables (name-value pairs) from either:
107 * an existing :class:`Environment` instance.
108 * current executable's environment by reading environment variables from :func:`os.environ`.
109 * a dictionary of name-value pairs.
111 2. Remove variables from environment.
112 3. Add new or update existing variables.
114 :param environment: Optional, existing Environment instance to derive a new environment.
115 :param newVariables: Optional, dictionary of new environment variables. |br|
116 If ``None``, read current environment variables from :func:`os.environ`.
117 :param addVariables: Optional, dictionary of variables to be added or modified in the environment.
118 :param delVariables: Optional, list of variable names to be removed from the environment.
119 """
120 if environment is not None: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 newVariables = environment._variables
122 elif newVariables is None:
123 newVariables = os_environ
125 self._variables = dict(newVariables)
127 if delVariables is not None: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 for variableName in delVariables:
129 del self._variables[variableName]
131 if addVariables is not None:
132 self._variables.update(addVariables)
134 def __len__(self) -> len:
135 """
136 Returns the number of set environment variables.
138 :returns: Number of environment variables.
139 """
140 return len(self._variables)
142 def __contains__(self, name: str) -> bool:
143 """
144 Checks if the variable is set in the environment.
146 :param name: The variable name to check.
147 :returns: ``True``, if the variable is set in the environment.
148 """
149 return name in self._variables
151 def __getitem__(self, name: str) -> str:
152 """
153 Access an environment variable in the environment by name.
155 :param name: Name of the environment variable.
156 :returns: The environment variable's value.
157 :raises KeyError: If Variable name is not set in the environment.
158 """
159 return self._variables[name]
161 def __setitem__(self, name: str, value: str) -> None:
162 """
163 Add or set an environment variable in the environment by name.
165 :param name: Name of the environment variable.
166 :param value: Value of the environment variable to be set.
167 """
168 self._variables[name] = value
170 def __delitem__(self, name: str) -> None:
171 """
172 Remove an environment variable from the environment by name.
174 :param name: The name of the environment variable to remove.
175 :raises KeyError: If name doesn't exist in the environment.
176 """
177 del self._variables[name]
180@export
181class Program(metaclass=ExtendedType, slots=True):
182 """
183 Represent a simple command line interface (CLI) executable (program or script).
185 CLI options are collected in a ``__cliOptions__`` dictionary.
186 """
188 _platform: str #: Current platform the executable runs on (Linux, Windows, ...)
189 _EXECUTABLE_NAMES: ClassVar[dict[str, str]] #: Dictionary of platform specific executable names.
190 _executablePath: Path #: The path to the executable (binary, script, ...).
191 _dryRun: bool #: True, if program shall run in *dry-run mode*.
192 _dryRunMessages: list[str] #: Actions skipped in *dry-run mode*.
193 __cliOptions__: ClassVar[dict[type[CommandLineArgument], int]] #: List of all possible CLI options.
194 __cliParameters__: dict[type[CommandLineArgument], CommandLineArgument] #: List of all CLI parameters.
196 def __init_subclass__(cls, *args: Any, **kwargs: Any) -> None:
197 """
198 Whenever a subclass is derived from :class:``Program``, all nested classes declared within ``Program`` and which are
199 marked with attribute ``CLIArgument`` are collected and then listed in the ``__cliOptions__`` dictionary.
201 :param args: Any positional arguments.
202 :param kwargs: Any keyword arguments.
203 """
204 super().__init_subclass__(*args, **kwargs)
206 # register all available CLI options (nested classes marked with attribute 'CLIArgument')
207 options: dict[type[CommandLineArgument], int] = {
208 option: order
209 for order, option in enumerate(CLIArgument.GetClasses(scope=cls))
210 }
211 cls.__cliOptions__ = options
213 def __init__(
214 self,
215 executablePath: Nullable[Path] = None,
216 binaryDirectoryPath: Nullable[Path] = None,
217 dryRun: bool = False
218 ) -> None:
219 """
220 Initializes a program instance.
222 .. todo:: Document algorithm
224 :param executablePath: Optional, path to the executable.
225 :param binaryDirectoryPath: Optional, path to the executable's directory.
226 :param dryRun: Optional, ``True``, when the program should run in dryrun mode.
227 :raises TypeError: If parameter 'executablePath' is not of type :class:`~pathlib.Path`.
228 :raises CLIAbstractionError: If the executable doesn't exist at the given path.
229 """
230 self._platform = system()
231 self._dryRun = dryRun
232 self._dryRunMessages = []
234 if executablePath is not None:
235 if isinstance(executablePath, Path):
236 if not executablePath.exists():
237 if dryRun:
238 self.LogDryRun(f"File check for '{executablePath}' failed. [SKIPPING]")
239 else:
240 raise CLIAbstractionError(f"Program '{executablePath}' not found.") from FileNotFoundError(executablePath)
241 else:
242 ex = TypeError("Parameter 'executablePath' is not of type 'Path'.")
243 ex.add_note(f"Got type '{getFullyQualifiedName(executablePath)}'.")
244 raise ex
245 elif binaryDirectoryPath is not None:
246 if isinstance(binaryDirectoryPath, Path):
247 if not binaryDirectoryPath.exists():
248 if dryRun:
249 self.LogDryRun(f"Directory check for '{binaryDirectoryPath}' failed. [SKIPPING]")
250 else:
251 raise CLIAbstractionError(f"Binary directory '{binaryDirectoryPath}' not found.") from FileNotFoundError(binaryDirectoryPath)
253 try:
254 executablePath = binaryDirectoryPath / self.__class__._EXECUTABLE_NAMES[self._platform]
255 except KeyError:
256 raise CLIAbstractionError(f"Program is not supported on platform '{self._platform}'.") from PlatformNotSupportedError(self._platform)
258 if not executablePath.exists():
259 if dryRun:
260 self.LogDryRun(f"File check for '{executablePath}' failed. [SKIPPING]")
261 else:
262 raise CLIAbstractionError(f"Program '{executablePath}' not found.") from FileNotFoundError(executablePath)
263 else:
264 ex = TypeError("Parameter 'binaryDirectoryPath' is not of type 'Path'.")
265 ex.add_note(f"Got type '{getFullyQualifiedName(binaryDirectoryPath)}'.")
266 raise ex
267 else:
268 try:
269 executablePath = Path(self._EXECUTABLE_NAMES[self._platform])
270 except KeyError:
271 raise CLIAbstractionError(f"Program is not supported on platform '{self._platform}'.") from PlatformNotSupportedError(self._platform)
273 resolvedExecutable = shutil_which(str(executablePath))
274 if dryRun:
275 if resolvedExecutable is None: 275 ↛ 278line 275 didn't jump to line 278 because the condition on line 275 was always true
276 self.LogDryRun(f"Search for '{executablePath}' in PATH failed. [SKIPPING]")
277 else:
278 fullExecutablePath = Path(resolvedExecutable)
279 if not fullExecutablePath.exists():
280 self.LogDryRun(f"File check for '{fullExecutablePath}' failed. [SKIPPING]")
281 else:
282 if resolvedExecutable is None:
283 raise CLIAbstractionError("Program could not be found in PATH.") from FileNotFoundError(executablePath)
285 fullExecutablePath = Path(resolvedExecutable)
286 if not fullExecutablePath.exists(): 286 ↛ 287line 286 didn't jump to line 287 because the condition on line 286 was never true
287 raise CLIAbstractionError(f"Program '{fullExecutablePath}' not found.") from FileNotFoundError(fullExecutablePath)
289 if resolvedExecutable is not None:
290 executablePath = Path(resolvedExecutable)
292 # TODO: log found executable in PATH
293 # TODO: check if found executable has execute permissions
294 # raise ValueError(f"Neither parameter 'executablePath' nor 'binaryDirectoryPath' was set.")
296 self._executablePath = executablePath
297 self.__cliParameters__ = {}
299 @readonly
300 def DryRunMessages(self) -> list[str]:
301 """
302 Read-only property to access the actions skipped in *dry-run mode* (:attr:`_dryRunMessages`).
304 :returns: The messages :meth:`LogDryRun` recorded, in the order the actions were skipped.
305 """
306 return self._dryRunMessages
308 def LogDryRun(self, message: str) -> None:
309 """
310 Record an action that was skipped, because the program runs in *dry-run mode*.
312 The message is appended to :attr:`DryRunMessages`. A derived class may override this method to write it, e.g.
313 to a terminal.
315 :param message: Description of the skipped action.
316 """
317 self._dryRunMessages.append(message)
319 @staticmethod
320 def _NeedsParameterInitialization(key: type) -> bool:
321 """
322 Check if an argument class needs a value when it is set.
324 :param key: Class of the command line argument.
325 :returns: ``True``, if the argument carries a value.
326 """
327 return issubclass(key, (ValuedFlag, ValuedArgument, NamedAndValuedArgument, NamedTupledArgument, PathArgument, PathListArgument))
329 def __getitem__(self, key: type[CommandLineArgument]) -> CommandLineArgument:
330 """
331 Access to a CLI parameter by CLI option, which is already used.
333 :param key: Class of the command line argument to read.
334 :returns: The command line argument object registered for that class.
335 :raises TypeError: If the key is not a subclass of :class:`~pyTooling.CLIAbstraction.Argument.CommandLineArgument`.
336 """
337 if not issubclass(key, CommandLineArgument): 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 ex = TypeError(f"Key '{key}' is not a subclass of 'CommandLineArgument'.")
339 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")
340 raise ex
342 # TODO: is nested check
343 return self.__cliParameters__[key]
345 def __setitem__(self, key: type[CommandLineArgument], value: CommandLineArgument) -> None:
346 """
347 Set a command line argument of this program by its argument class.
349 :param key: Class of the command line argument to set.
350 :param value: Value of that argument; ignored for arguments needing no value.
351 :raises TypeError: If the key is not a subclass of :class:`~pyTooling.CLIAbstraction.Argument.CommandLineArgument`.
352 :raises KeyError: If the argument isn't allowed on this program, or was set before.
353 """
354 if not issubclass(key, CommandLineArgument): 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true
355 ex = TypeError(f"Key '{key}' is not a subclass of 'CommandLineArgument'.")
356 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")
357 raise ex
358 elif key not in self.__cliOptions__:
359 raise KeyError(f"Option '{key}' is not allowed on executable '{getTypeName(self)}'")
360 elif key in self.__cliParameters__:
361 raise KeyError(f"Option '{key}' is already set to a value.")
363 if self._NeedsParameterInitialization(key):
364 self.__cliParameters__[key] = key(value)
365 else:
366 self.__cliParameters__[key] = key()
368 def _CopyParameters(self, tool: Program) -> None:
369 """
370 Copy every command line argument set on this program to another program instance.
372 A method deriving a program variant calls it on the new instance, then sets the arguments it is given explicitly.
374 :param tool: The program instance receiving the arguments.
375 :raises ValueError: If parameter 'tool' is None.
376 :raises TypeError: If parameter 'tool' is not of type :class:`Program`.
377 :raises KeyError: If an argument isn't allowed on 'tool', or is set there already.
378 """
379 if tool is None:
380 raise ValueError("Parameter 'tool' is None.")
381 elif not isinstance(tool, Program):
382 ex = TypeError("Parameter 'tool' is not of type 'Program'.")
383 ex.add_note(f"Got type '{getFullyQualifiedName(tool)}'.")
384 raise ex
386 for key, argument in self.__cliParameters__.items():
387 if self._NeedsParameterInitialization(key):
388 tool[key] = argument.Value
389 else:
390 tool[key] = True
392 @readonly
393 def Path(self) -> Path:
394 """
395 Read-only property to access the program's path (:attr:`_executablePath`).
397 :returns: The program's path. For a program searched in ``PATH``, the path it was found at.
398 """
399 return self._executablePath
401 def ToArgumentList(self) -> list[str]:
402 """
403 Convert a program and used CLI options to a list of CLI argument strings in correct order, unescaped.
405 The list can be passed directly to :func:`subprocess.run` or :class:`subprocess.Popen` (without ``shell=True``):
406 each string reaches the program as one argument, so nothing needs escaping or quoting.
408 :returns: List of CLI arguments
409 :raises TypeError: If an argument is neither a string nor a sequence of strings. |br|
410 An argument's :meth:`~pyTooling.CLIAbstraction.Argument.CommandLineArgument.AsArgument` has to
411 return a string, a tuple or a list.
412 """
413 result: list[str] = []
415 result.append(str(self._executablePath))
417 def predicate(item: tuple[type[CommandLineArgument], int]) -> int:
418 """
419 Nested function used as sort key.
421 :param item: Pair of an argument class and its argument object.
422 :returns: The position the argument class was registered at, so the command line keeps the declared order.
423 """
424 return self.__cliOptions__[item[0]]
426 for key, value in sorted(self.__cliParameters__.items(), key=predicate):
427 param = value.AsArgument()
428 if isinstance(param, str):
429 result.append(param)
430 elif isinstance(param, (tuple, list)): 430 ↛ 433line 430 didn't jump to line 433 because the condition on line 430 was always true
431 result += param
432 else:
433 ex = TypeError(f"Argument '{key.__name__}' was rendered to neither a string nor a sequence of strings.")
434 ex.add_note(f"Got type '{getFullyQualifiedName(param)}'.")
435 ex.add_note("'AsArgument()' has to return a 'str', a 'tuple' or a 'list'.")
436 raise ex
438 return result
440 def __repr__(self) -> str:
441 """
442 Return the argument list as a Python literal, with strings in double quotes.
444 Example: :pycode:`["/usr/bin/git", "--version"]`
446 :returns: :meth:`ToArgumentList`'s result as a literal.
447 """
448 literal = CommandLineArgument._DoubleQuotedLiteral
449 return "[" + ", ".join([literal(item) for item in self.ToArgumentList()]) + "]" # WORKAROUND: Python <3.12
450 # return f"[{", ".join([literal(item) for item in self.ToArgumentList()])}]"
452 def __str__(self) -> str:
453 """
454 Return the command line, escaped for the current platform.
456 Example: :pycode:`/usr/bin/git commit -m 'Bumped dependencies.'`
458 :returns: The argument list joined by ``formatCommandLine``: :func:`shlex.join`, on Windows
459 ``subprocess.list2cmdline()``.
460 """
461 return formatCommandLine(self.ToArgumentList())
464@export
465class Executable(Program): # (ILogable):
466 """Represent a CLI executable derived from :class:`Program`, that adds an abstraction of :class:`subprocess.Popen`."""
468 _BOUNDARY: ClassVar[str] = "====== BOUNDARY pyTooling.CLIAbstraction BOUNDARY ======" #: Marker line printed between the program's own output and the executable's output.
470 _workingDirectory: Nullable[Path] #: Path to the working directory
471 _environment: Nullable[Environment] #: Environment to use when executing.
472 _process: Nullable[Subprocess_Popen[str]] #: Reference to the running process.
473 _exitCode: Nullable[int] #: The child's process exit code.
474 _killed: Nullable[bool] #: True, if the child-process got killed (e.g. by a timeout).
475 _iterator: Nullable[Iterator[str]] #: Iterator for reading STDOUT.
477 def __init__(
478 self,
479 executablePath: Nullable[Path] = None,
480 binaryDirectoryPath: Nullable[Path] = None,
481 workingDirectory: Nullable[Path] = None,
482 environment: Nullable[Environment] = None,
483 dryRun: bool = False
484 ) -> None:
485 """
486 Initializes an executable instance.
488 :param executablePath: Optional, path to the executable.
489 :param binaryDirectoryPath: Optional, path to the executable's directory.
490 :param workingDirectory: Optional, path to the working directory.
491 :param environment: Optional, environment that should be setup when launching the executable.
492 :param dryRun: Optional, ``True``, when the program should run in dryrun mode.
493 """
494 super().__init__(executablePath, binaryDirectoryPath, dryRun)
496 self._workingDirectory = None
497 self._environment = environment
498 self._process = None
499 self._exitCode = None
500 self._killed = None
501 self._iterator = None
503 def StartProcess(self, environment: Nullable[Environment] = None) -> None:
504 """
505 Start the executable as a child-process.
507 :param environment: Optional, environment that should be setup when launching the executable. |br|
508 If ``None``, the :attr:`_environment` is used.
509 :raises CLIAbstractionError: When an :exc:`OSError` occurs while launching the child-process.
510 """
511 if self._dryRun:
512 self.LogDryRun(f"Start process: {self}")
513 return
515 if environment is not None: 515 ↛ 517line 515 didn't jump to line 517 because the condition on line 515 was always true
516 envVariables = environment._variables
517 elif self._environment is not None:
518 envVariables = self._environment._variables
519 else:
520 envVariables = None
522 # FIXME: verbose log start process
523 # FIXME: debug log - parameter list
524 try:
525 self._process = Subprocess_Popen(
526 self.ToArgumentList(),
527 stdin=Subprocess_Pipe,
528 stdout=Subprocess_Pipe,
529 stderr=Subprocess_StdOut,
530 cwd=self._workingDirectory,
531 env=envVariables,
532 universal_newlines=True,
533 bufsize=256
534 )
536 except OSError as ex:
537 raise CLIAbstractionError(f"Error while launching a process for '{self._executablePath}'.") from ex
539 def Send(self, line: str, end: str = "\n") -> None:
540 """
541 Send a string to STDIN of the running child-process.
543 :param line: Line to send.
544 :param end: Optional, line end character.
545 :raises CLIAbstractionError: If the child-process was not started, or has no standard input.
546 :raises CLIAbstractionError: When any error occurs while sending data to the child-process.
547 """
548 if self._process is None or self._process.stdin is None:
549 raise CLIAbstractionError(
550 f"The child-process '{self._executablePath}' was not started, or has no standard input."
551 )
553 try:
554 self._process.stdin.write(line + end)
555 self._process.stdin.flush()
556 except Exception as ex:
557 raise CLIAbstractionError(f"Error while sending data to the child-process '{self._executablePath}'.") from ex
559 # This is TCL specific ...
560 # def SendBoundary(self):
561 # self.Send("puts \"{0}\"".format(self._pyIPCMI_BOUNDARY))
563 def GetLineReader(self) -> Generator[str, None, None]:
564 """
565 Return a line-reader for STDOUT.
567 :returns: A generator object to read from STDOUT line-by-line.
568 :raises DryRunError: In case dryrun mode is active.
569 :raises CLIAbstractionError: When any error occurs while reading outputs from the child-process.
570 """
571 if self._dryRun: 571 ↛ 572line 571 didn't jump to line 572 because the condition on line 571 was never true
572 raise DryRunError(f"Can't read from the child-process '{self._executablePath}' in dry-run mode.")
574 if self._process is None or self._process.stdout is None: 574 ↛ 575line 574 didn't jump to line 575 because the condition on line 574 was never true
575 raise CLIAbstractionError(
576 f"The child-process '{self._executablePath}' was not started, or has no standard output."
577 )
579 try:
580 for line in iter(self._process.stdout.readline, ""): # FIXME: can it be improved?
581 yield line[:-1]
582 except Exception as ex:
583 raise CLIAbstractionError(f"Error while reading from the child-process '{self._executablePath}'.") from ex
584 # finally:
585 # self._process.terminate()
587 def Wait(self, timeout: Nullable[float] = None, kill: bool = False) -> Nullable[int]:
588 """
589 Wait on the child-process with an optional timeout.
591 When the timeout period exceeds, the child-process can be forcefully terminated.
593 :param timeout: Optional, timeout in seconds. |br|
594 Default: infinitely wait on the child-process.
595 :param kill: Optional, if ``True``, terminate (kill) the child-process if it didn't terminate by
596 itself within the timeout period.
597 :returns: ``None`` when the child-process is still running, otherwise the exit code.
598 :raises CLIAbstractionError: When the child-process is not started yet.
600 .. topic:: Usecases
602 :pycode:`executable.Wait()`
603 Infinitely wait on the child-process. When the child-process terminates by itself, the exit code is returned.
605 This is a blocking call.
607 :pycode:`executable.Wait(timeout=5.4)`
608 Wait for a specified time on the child-process' termination. If it terminated by itself within the specified
609 timeout period, the exit code is returned; otherwise ``None``.
611 Thus :pycode:`.Wait(timeout=0.0)` returning ``None`` indicates a running process.
613 :pycode:`executable.Wait(timeout=20.0, kill=True)`
614 Wait for a specified time on the child-process' termination. If it terminated by itself within the specified
615 timeout period, the exit code is returned; otherwise the child-process gets killed and it's exit code is returned.
617 :pycode:`executable.Wait(timeout=0.0, kill=True)`
618 Kill immediately.
620 .. seealso::
622 :meth:`Terminate` - Terminate the child-process.
623 """
624 if self._process is None:
625 raise CLIAbstractionError("Process not yet started.")
627 try:
628 self._exitCode = self._process.wait(timeout=timeout)
629 except TimeoutExpired:
630 # when timed out, the process isn't terminated/killed automatically
631 if kill:
632 self._killed = True
633 self._process.terminate()
634 # After killing, wait to clean up the "zombie" process
635 self._exitCode = self._process.wait()
637 return self._exitCode
639 def Terminate(self) -> Nullable[int]:
640 """
641 Terminate the child-process.
643 :returns: The child-process' exit code.
644 :raises CLIAbstractionError: When the child-process is not started yet.
646 .. seealso::
648 :meth:`Wait` - Wait on the child-process with an optional timeout.
649 """
650 return self.Wait(timeout=0.0, kill=True)
652 @readonly
653 def ExitCode(self) -> int:
654 """
655 Read-only property accessing the child-process' exit code.
657 :returns: Child-process' exit code or ``None`` if it's still running.
659 .. seealso::
661 :meth:`Wait`
662 |rarr| Wait on the child-process with an optional timeout.
663 :meth:`Terminate`
664 |rarr| Terminate the child-process.
665 """
666 return self._exitCode
668 # This is TCL specific
669 # def ReadUntilBoundary(self, indent=0):
670 # __indent = " " * indent
671 # if self._iterator is None:
672 # self._iterator = iter(self.GetReader())
673 #
674 # for line in self._iterator:
675 # print(__indent + line)
676 # if self._pyIPCMI_BOUNDARY in line:
677 # break
678 # self.LogDebug("Quartus II is ready")
681@export
682class OutputFilteredExecutable(Executable):
683 """Represent a CLI executable derived from :class:`Executable`, whose outputs are filtered."""
684 _hasOutput: bool #: ``True``, if the executable wrote any output.
685 _hasWarnings: bool #: ``True``, if the output filter classified a line as a warning.
686 _hasErrors: bool #: ``True``, if the output filter classified a line as an error.
687 _hasFatals: bool #: ``True``, if the output filter classified a line as a fatal error.
689 def __init__(self, platform: Platform, dryrun: bool, executablePath: Path) -> None: #, environment=None, logger=None) -> None:
690 """
691 Initialize an executable whose output is filtered, with all filter results cleared.
693 :param platform: Platform the executable is called on.
694 :param dryrun: If ``True``, the executable is not started, only the command line is assembled.
695 :param executablePath: Path to the executable.
696 """
697 super().__init__(platform, dryrun, executablePath) #, environment=environment, logger=logger)
699 self._hasOutput = False
700 self._hasWarnings = False
701 self._hasErrors = False
702 self._hasFatals = False
704 @readonly
705 def HasWarnings(self) -> bool:
706 # TODO: update doc-string
707 """
708 Check if warnings were found while processing the output stream.
710 :returns: ``True``, if at least one warning was found.
711 """
712 return self._hasWarnings
714 @readonly
715 def HasErrors(self) -> bool:
716 # TODO: update doc-string
717 """
718 Check if errors were found while processing the output stream.
720 :returns: ``True``, if at least one error was found.
721 """
722 return self._hasErrors
724 @readonly
725 def HasFatals(self) -> bool:
726 # TODO: update doc-string
727 """
728 Check if fatal errors were found while processing the output stream.
730 :returns: ``True``, if at least one fatal error was found.
731 """
732 return self._hasErrors