__init__.py
Summary
- Lines:
201 of 256 covered (78.5%)
- Branches:
62 of 84 taken (73.8%), 10 lines partially
- Coverage:
77.4%
- Directory:
- Missed lines:
127, 134-135, 174, 183, 284-286, 293, 344-346, 361-363, 439-442, 523-526, 542-543, 554-563, 578, 581, 588-589, 630-643, 656, 672, 703-708, 718, 728, 738
Source
pyTooling/CLIAbstraction/__init__.pycoveredpartialuncoveredexcludednot executable
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.
34
35.. seealso::
36
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
45
46# __keywords__ = ["abstract", "executable", "cli", "cli arguments"]
47
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
54
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
64
65
66_Path = Path
67"""
68Alias of :class:`~pathlib.Path` for annotations in a class with a property named ``Path``, which the class body binds.
69"""
70
71
72@export
73class CLIAbstractionError(ToolingException):
74 """Base-exception of all exceptions raised by :mod:`pyTooling.CLIAbstraction`."""
75
76
77@export
78class DryRunError(CLIAbstractionError):
79 """This exception is raised if an executable is launched while in dry-run mode."""
80
81
82@export
83class CLIArgument(Attribute):
84 """An attribute to annotate nested classes as an CLI argument."""
85
86
87@export
88class Environment(metaclass=ExtendedType, slots=True):
89 """
90 A class describing the environment of an executable.
91
92 .. topic:: Content of the environment
93
94 * Environment variables
95 """
96 _variables: dict[str, str] #: Dictionary of active environment variables.
97
98 # TODO: derive environment from existing environment object.
99 def __init__(
100 self, *,
101 environment: Nullable[Environment] = None,
102 newVariables: Nullable[Mapping[str, str]] = None,
103 addVariables: Nullable[Mapping[str, str]] = None,
104 delVariables: Nullable[Iterable[str]] = None
105 ) -> None:
106 """
107 Initializes an environment class managing.
108
109 .. topic:: Algorithm
110
111 1. Create a new dictionary of environment variables (name-value pairs) from either:
112
113 * an existing :class:`Environment` instance.
114 * current executable's environment by reading environment variables from :func:`os.environ`.
115 * a dictionary of name-value pairs.
116
117 2. Remove variables from environment.
118 3. Add new or update existing variables.
119
120 :param environment: Optional, existing Environment instance to derive a new environment.
121 :param newVariables: Optional, dictionary of new environment variables. |br|
122 If ``None``, read current environment variables from :func:`os.environ`.
123 :param addVariables: Optional, dictionary of variables to be added or modified in the environment.
124 :param delVariables: Optional, list of variable names to be removed from the environment.
125 """
126 if environment is not None:
127 newVariables = environment._variables
128 elif newVariables is None:
129 newVariables = os_environ
130
131 self._variables = dict(newVariables)
132
133 if delVariables is not None:
134 for variableName in delVariables:
135 del self._variables[variableName]
136
137 if addVariables is not None:
138 self._variables.update(addVariables)
139
140 def __len__(self) -> len:
141 """
142 Returns the number of set environment variables.
143
144 :returns: Number of environment variables.
145 """
146 return len(self._variables)
147
148 def __contains__(self, name: str) -> bool:
149 """
150 Checks if the variable is set in the environment.
151
152 :param name: The variable name to check.
153 :returns: ``True``, if the variable is set in the environment.
154 """
155 return name in self._variables
156
157 def __getitem__(self, name: str) -> str:
158 """
159 Access an environment variable in the environment by name.
160
161 :param name: Name of the environment variable.
162 :returns: The environment variable's value.
163 :raises KeyError: If Variable name is not set in the environment.
164 """
165 return self._variables[name]
166
167 def __setitem__(self, name: str, value: str) -> None:
168 """
169 Add or set an environment variable in the environment by name.
170
171 :param name: Name of the environment variable.
172 :param value: Value of the environment variable to be set.
173 """
174 self._variables[name] = value
175
176 def __delitem__(self, name: str) -> None:
177 """
178 Remove an environment variable from the environment by name.
179
180 :param name: The name of the environment variable to remove.
181 :raises KeyError: If name doesn't exist in the environment.
182 """
183 del self._variables[name]
184
185
186@export
187class Program(metaclass=ExtendedType, slots=True):
188 """
189 Represent a simple command line interface (CLI) executable (program or script).
190
191 CLI options are collected in a ``__cliOptions__`` dictionary.
192 """
193
194 _platform: str #: Current platform the executable runs on (Linux, Windows, ...)
195 _EXECUTABLE_NAMES: ClassVar[dict[str, str]] #: Dictionary of platform specific executable names.
196 _executablePath: _Path #: The path to the executable (binary, script, ...).
197 _dryRun: bool #: True, if program shall run in *dry-run mode*.
198 _dryRunMessages: list[str] #: Actions skipped in *dry-run mode*.
199 __cliOptions__: ClassVar[dict[type[CommandLineArgument], int]] #: List of all possible CLI options.
200 __cliParameters__: dict[type[CommandLineArgument], CommandLineArgument] #: List of all CLI parameters.
201
202 def __init_subclass__(cls, *args: Any, **kwargs: Any) -> None:
203 """
204 Whenever a subclass is derived from :class:``Program``, all nested classes declared within ``Program`` and which are
205 marked with attribute ``CLIArgument`` are collected and then listed in the ``__cliOptions__`` dictionary.
206
207 :param args: Any positional arguments.
208 :param kwargs: Any keyword arguments.
209 """
210 super().__init_subclass__(*args, **kwargs)
211
212 # register all available CLI options (nested classes marked with attribute 'CLIArgument')
213 options: dict[type[CommandLineArgument], int] = {
214 option: order
215 for order, option in enumerate(CLIArgument.GetClasses(scope=cls))
216 }
217 cls.__cliOptions__ = options
218
219 def __init__(
220 self,
221 executablePath: Nullable[Path] = None,
222 binaryDirectoryPath: Nullable[Path] = None,
223 dryRun: bool = False
224 ) -> None:
225 """
226 Initializes a program instance.
227
228 .. todo:: Document algorithm
229
230 :param executablePath: Optional, path to the executable.
231 :param binaryDirectoryPath: Optional, path to the executable's directory.
232 :param dryRun: Optional, ``True``, when the program should run in dryrun mode.
233 :raises TypeError: If parameter 'executablePath' is not of type :class:`~pathlib.Path`.
234 :raises CLIAbstractionError: If the executable doesn't exist at the given path.
235 """
236 self._platform = system()
237 self._dryRun = dryRun
238 self._dryRunMessages = []
239
240 if executablePath is not None:
241 if isinstance(executablePath, Path):
242 if not executablePath.exists():
243 if dryRun:
244 self.LogDryRun(f"File check for '{executablePath}' failed. [SKIPPING]")
245 else:
246 raise CLIAbstractionError(f"Program '{executablePath}' not found.") from FileNotFoundError(executablePath)
247 else:
248 ex = TypeError("Parameter 'executablePath' is not of type 'Path'.")
249 ex.add_note(f"Got type '{getFullyQualifiedName(executablePath)}'.")
250 raise ex
251 elif binaryDirectoryPath is not None:
252 if isinstance(binaryDirectoryPath, Path):
253 if not binaryDirectoryPath.exists():
254 if dryRun:
255 self.LogDryRun(f"Directory check for '{binaryDirectoryPath}' failed. [SKIPPING]")
256 else:
257 raise CLIAbstractionError(f"Binary directory '{binaryDirectoryPath}' not found.") from FileNotFoundError(binaryDirectoryPath)
258
259 try:
260 executablePath = binaryDirectoryPath / self.__class__._EXECUTABLE_NAMES[self._platform]
261 except KeyError:
262 raise CLIAbstractionError(f"Program is not supported on platform '{self._platform}'.") from PlatformNotSupportedError(self._platform)
263
264 if not executablePath.exists():
265 if dryRun:
266 self.LogDryRun(f"File check for '{executablePath}' failed. [SKIPPING]")
267 else:
268 raise CLIAbstractionError(f"Program '{executablePath}' not found.") from FileNotFoundError(executablePath)
269 else:
270 ex = TypeError("Parameter 'binaryDirectoryPath' is not of type 'Path'.")
271 ex.add_note(f"Got type '{getFullyQualifiedName(binaryDirectoryPath)}'.")
272 raise ex
273 else:
274 try:
275 executablePath = Path(self._EXECUTABLE_NAMES[self._platform])
276 except KeyError:
277 raise CLIAbstractionError(f"Program is not supported on platform '{self._platform}'.") from PlatformNotSupportedError(self._platform)
278
279 resolvedExecutable = shutil_which(str(executablePath))
280 if dryRun:
281 if resolvedExecutable is None:
282 self.LogDryRun(f"Search for '{executablePath}' in PATH failed. [SKIPPING]")
283 else:
284 fullExecutablePath = Path(resolvedExecutable)
285 if not fullExecutablePath.exists():
286 self.LogDryRun(f"File check for '{fullExecutablePath}' failed. [SKIPPING]")
287 else:
288 if resolvedExecutable is None:
289 raise CLIAbstractionError("Program could not be found in PATH.") from FileNotFoundError(executablePath)
290
291 fullExecutablePath = Path(resolvedExecutable)
292 if not fullExecutablePath.exists():
293 raise CLIAbstractionError(f"Program '{fullExecutablePath}' not found.") from FileNotFoundError(fullExecutablePath)
294
295 if resolvedExecutable is not None:
296 executablePath = Path(resolvedExecutable)
297
298 # TODO: log found executable in PATH
299 # TODO: check if found executable has execute permissions
300 # raise ValueError(f"Neither parameter 'executablePath' nor 'binaryDirectoryPath' was set.")
301
302 self._executablePath = executablePath
303 self.__cliParameters__ = {}
304
305 @readonly
306 def DryRunMessages(self) -> list[str]:
307 """
308 Read-only property to access the actions skipped in *dry-run mode* (:attr:`_dryRunMessages`).
309
310 :returns: The messages :meth:`LogDryRun` recorded, in the order the actions were skipped.
311 """
312 return self._dryRunMessages
313
314 def LogDryRun(self, message: str) -> None:
315 """
316 Record an action that was skipped, because the program runs in *dry-run mode*.
317
318 The message is appended to :attr:`DryRunMessages`. A derived class may override this method to write it, e.g.
319 to a terminal.
320
321 :param message: Description of the skipped action.
322 """
323 self._dryRunMessages.append(message)
324
325 @staticmethod
326 def _NeedsParameterInitialization(key: type) -> bool:
327 """
328 Check if an argument class needs a value when it is set.
329
330 :param key: Class of the command line argument.
331 :returns: ``True``, if the argument carries a value.
332 """
333 return issubclass(key, (ValuedFlag, ValuedArgument, NamedAndValuedArgument, NamedTupledArgument, PathArgument, PathListArgument))
334
335 def __getitem__(self, key: type[CommandLineArgument]) -> CommandLineArgument:
336 """
337 Access to a CLI parameter by CLI option, which is already used.
338
339 :param key: Class of the command line argument to read.
340 :returns: The command line argument object registered for that class.
341 :raises TypeError: If the key is not a subclass of :class:`~pyTooling.CLIAbstraction.Argument.CommandLineArgument`.
342 """
343 if not issubclass(key, CommandLineArgument):
344 ex = TypeError(f"Key '{key}' is not a subclass of 'CommandLineArgument'.")
345 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")
346 raise ex
347
348 # TODO: is nested check
349 return self.__cliParameters__[key]
350
351 def __setitem__(self, key: type[CommandLineArgument], value: CommandLineArgument) -> None:
352 """
353 Set a command line argument of this program by its argument class.
354
355 :param key: Class of the command line argument to set.
356 :param value: Value of that argument; ignored for arguments needing no value.
357 :raises TypeError: If the key is not a subclass of :class:`~pyTooling.CLIAbstraction.Argument.CommandLineArgument`.
358 :raises KeyError: If the argument isn't allowed on this program, or was set before.
359 """
360 if not issubclass(key, CommandLineArgument):
361 ex = TypeError(f"Key '{key}' is not a subclass of 'CommandLineArgument'.")
362 ex.add_note(f"Got type '{getFullyQualifiedName(key)}'.")
363 raise ex
364 elif key not in self.__cliOptions__:
365 raise KeyError(f"Option '{key}' is not allowed on executable '{getTypeName(self)}'")
366 elif key in self.__cliParameters__:
367 raise KeyError(f"Option '{key}' is already set to a value.")
368
369 if self._NeedsParameterInitialization(key):
370 self.__cliParameters__[key] = key(value)
371 else:
372 self.__cliParameters__[key] = key()
373
374 def _CopyParameters(self, tool: Program) -> None:
375 """
376 Copy every command line argument set on this program to another program instance.
377
378 A method deriving a program variant calls it on the new instance, then sets the arguments it is given explicitly.
379
380 :param tool: The program instance receiving the arguments.
381 :raises ValueError: If parameter 'tool' is None.
382 :raises TypeError: If parameter 'tool' is not of type :class:`Program`.
383 :raises KeyError: If an argument isn't allowed on 'tool', or is set there already.
384 """
385 if tool is None:
386 raise ValueError("Parameter 'tool' is None.")
387 elif not isinstance(tool, Program):
388 ex = TypeError("Parameter 'tool' is not of type 'Program'.")
389 ex.add_note(f"Got type '{getFullyQualifiedName(tool)}'.")
390 raise ex
391
392 for key, argument in self.__cliParameters__.items():
393 if self._NeedsParameterInitialization(key):
394 tool[key] = argument.Value
395 else:
396 tool[key] = True
397
398 @readonly
399 def Path(self) -> Path:
400 """
401 Read-only property to access the program's path (:attr:`_executablePath`).
402
403 :returns: The program's path. For a program searched in ``PATH``, the path it was found at.
404 """
405 return self._executablePath
406
407 def ToArgumentList(self) -> list[str]:
408 """
409 Convert a program and used CLI options to a list of CLI argument strings in correct order, unescaped.
410
411 The list can be passed directly to :func:`subprocess.run` or :class:`subprocess.Popen` (without ``shell=True``):
412 each string reaches the program as one argument, so nothing needs escaping or quoting.
413
414 :returns: List of CLI arguments
415 :raises TypeError: If an argument is neither a string nor a sequence of strings. |br|
416 An argument's :meth:`~pyTooling.CLIAbstraction.Argument.CommandLineArgument.AsArgument` has to
417 return a string, a tuple or a list.
418 """
419 result: list[str] = []
420
421 result.append(str(self._executablePath))
422
423 def predicate(item: tuple[type[CommandLineArgument], int]) -> int:
424 """
425 Nested function used as sort key.
426
427 :param item: Pair of an argument class and its argument object.
428 :returns: The position the argument class was registered at, so the command line keeps the declared order.
429 """
430 return self.__cliOptions__[item[0]]
431
432 for key, value in sorted(self.__cliParameters__.items(), key=predicate):
433 param = value.AsArgument()
434 if isinstance(param, str):
435 result.append(param)
436 elif isinstance(param, (tuple, list)):
437 result += param
438 else:
439 ex = TypeError(f"Argument '{key.__name__}' was rendered to neither a string nor a sequence of strings.")
440 ex.add_note(f"Got type '{getFullyQualifiedName(param)}'.")
441 ex.add_note("'AsArgument()' has to return a 'str', a 'tuple' or a 'list'.")
442 raise ex
443
444 return result
445
446 def __repr__(self) -> str:
447 """
448 Return the argument list as a Python literal, with strings in double quotes.
449
450 Example: :pycode:`["/usr/bin/git", "--version"]`
451
452 :returns: :meth:`ToArgumentList`'s result as a literal.
453 """
454 literal = CommandLineArgument._DoubleQuotedLiteral
455 return "[" + ", ".join([literal(item) for item in self.ToArgumentList()]) + "]" # WORKAROUND: Python <3.12
456 # return f"[{", ".join([literal(item) for item in self.ToArgumentList()])}]"
457
458 def __str__(self) -> str:
459 """
460 Return the command line, escaped for the current platform.
461
462 Example: :pycode:`/usr/bin/git commit -m 'Bumped dependencies.'`
463
464 :returns: The argument list joined by ``formatCommandLine``: :func:`shlex.join`, on Windows
465 ``subprocess.list2cmdline()``.
466 """
467 return formatCommandLine(self.ToArgumentList())
468
469
470@export
471class Executable(Program): # (ILogable):
472 """Represent a CLI executable derived from :class:`Program`, that adds an abstraction of :class:`subprocess.Popen`."""
473
474 _BOUNDARY: ClassVar[str] = "====== BOUNDARY pyTooling.CLIAbstraction BOUNDARY ======" #: Marker line printed between the program's own output and the executable's output.
475
476 _workingDirectory: Nullable[Path] #: Path to the working directory
477 _environment: Nullable[Environment] #: Environment to use when executing.
478 _process: Nullable[Subprocess_Popen[str]] #: Reference to the running process.
479 _exitCode: Nullable[int] #: The child's process exit code.
480 _killed: Nullable[bool] #: True, if the child-process got killed (e.g. by a timeout).
481 _iterator: Nullable[Iterator[str]] #: Iterator for reading STDOUT.
482
483 def __init__(
484 self,
485 executablePath: Nullable[Path] = None,
486 binaryDirectoryPath: Nullable[Path] = None,
487 workingDirectory: Nullable[Path] = None,
488 environment: Nullable[Environment] = None,
489 dryRun: bool = False
490 ) -> None:
491 """
492 Initializes an executable instance.
493
494 :param executablePath: Optional, path to the executable.
495 :param binaryDirectoryPath: Optional, path to the executable's directory.
496 :param workingDirectory: Optional, path to the working directory.
497 :param environment: Optional, environment that should be setup when launching the executable.
498 :param dryRun: Optional, ``True``, when the program should run in dryrun mode.
499 """
500 super().__init__(executablePath, binaryDirectoryPath, dryRun)
501
502 self._workingDirectory = None
503 self._environment = environment
504 self._process = None
505 self._exitCode = None
506 self._killed = None
507 self._iterator = None
508
509 def StartProcess(self, environment: Nullable[Environment] = None) -> None:
510 """
511 Start the executable as a child-process.
512
513 :param environment: Optional, environment that should be setup when launching the executable. |br|
514 If ``None``, the :attr:`_environment` is used.
515 :raises CLIAbstractionError: When an :exc:`OSError` occurs while launching the child-process.
516 """
517 if self._dryRun:
518 self.LogDryRun(f"Start process: {self}")
519 return
520
521 if environment is not None:
522 envVariables = environment._variables
523 elif self._environment is not None:
524 envVariables = self._environment._variables
525 else:
526 envVariables = None
527
528 # FIXME: verbose log start process
529 # FIXME: debug log - parameter list
530 try:
531 self._process = Subprocess_Popen(
532 self.ToArgumentList(),
533 stdin=Subprocess_Pipe,
534 stdout=Subprocess_Pipe,
535 stderr=Subprocess_StdOut,
536 cwd=self._workingDirectory,
537 env=envVariables,
538 universal_newlines=True,
539 bufsize=256
540 )
541
542 except OSError as ex:
543 raise CLIAbstractionError(f"Error while launching a process for '{self._executablePath}'.") from ex
544
545 def Send(self, line: str, end: str = "\n") -> None:
546 """
547 Send a string to STDIN of the running child-process.
548
549 :param line: Line to send.
550 :param end: Optional, line end character.
551 :raises CLIAbstractionError: If the child-process was not started, or has no standard input.
552 :raises CLIAbstractionError: When any error occurs while sending data to the child-process.
553 """
554 if self._process is None or self._process.stdin is None:
555 raise CLIAbstractionError(
556 f"The child-process '{self._executablePath}' was not started, or has no standard input."
557 )
558
559 try:
560 self._process.stdin.write(line + end)
561 self._process.stdin.flush()
562 except Exception as ex:
563 raise CLIAbstractionError(f"Error while sending data to the child-process '{self._executablePath}'.") from ex
564
565 # This is TCL specific ...
566 # def SendBoundary(self):
567 # self.Send("puts \"{0}\"".format(self._pyIPCMI_BOUNDARY))
568
569 def GetLineReader(self) -> Generator[str, None, None]:
570 """
571 Return a line-reader for STDOUT.
572
573 :returns: A generator object to read from STDOUT line-by-line.
574 :raises DryRunError: In case dryrun mode is active.
575 :raises CLIAbstractionError: When any error occurs while reading outputs from the child-process.
576 """
577 if self._dryRun:
578 raise DryRunError(f"Can't read from the child-process '{self._executablePath}' in dry-run mode.")
579
580 if self._process is None or self._process.stdout is None:
581 raise CLIAbstractionError(
582 f"The child-process '{self._executablePath}' was not started, or has no standard output."
583 )
584
585 try:
586 for line in iter(self._process.stdout.readline, ""): # FIXME: can it be improved?
587 yield line[:-1]
588 except Exception as ex:
589 raise CLIAbstractionError(f"Error while reading from the child-process '{self._executablePath}'.") from ex
590 # finally:
591 # self._process.terminate()
592
593 def Wait(self, timeout: Nullable[float] = None, kill: bool = False) -> Nullable[int]:
594 """
595 Wait on the child-process with an optional timeout.
596
597 When the timeout period exceeds, the child-process can be forcefully terminated.
598
599 :param timeout: Optional, timeout in seconds. |br|
600 Default: infinitely wait on the child-process.
601 :param kill: Optional, if ``True``, terminate (kill) the child-process if it didn't terminate by
602 itself within the timeout period.
603 :returns: ``None`` when the child-process is still running, otherwise the exit code.
604 :raises CLIAbstractionError: When the child-process is not started yet.
605
606 .. topic:: Usecases
607
608 :pycode:`executable.Wait()`
609 Infinitely wait on the child-process. When the child-process terminates by itself, the exit code is returned.
610
611 This is a blocking call.
612
613 :pycode:`executable.Wait(timeout=5.4)`
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 ``None``.
616
617 Thus :pycode:`.Wait(timeout=0.0)` returning ``None`` indicates a running process.
618
619 :pycode:`executable.Wait(timeout=20.0, kill=True)`
620 Wait for a specified time on the child-process' termination. If it terminated by itself within the specified
621 timeout period, the exit code is returned; otherwise the child-process gets killed and it's exit code is returned.
622
623 :pycode:`executable.Wait(timeout=0.0, kill=True)`
624 Kill immediately.
625
626 .. seealso::
627
628 :meth:`Terminate` - Terminate the child-process.
629 """
630 if self._process is None:
631 raise CLIAbstractionError("Process not yet started.")
632
633 try:
634 self._exitCode = self._process.wait(timeout=timeout)
635 except TimeoutExpired:
636 # when timed out, the process isn't terminated/killed automatically
637 if kill:
638 self._killed = True
639 self._process.terminate()
640 # After killing, wait to clean up the "zombie" process
641 self._exitCode = self._process.wait()
642
643 return self._exitCode
644
645 def Terminate(self) -> Nullable[int]:
646 """
647 Terminate the child-process.
648
649 :returns: The child-process' exit code.
650 :raises CLIAbstractionError: When the child-process is not started yet.
651
652 .. seealso::
653
654 :meth:`Wait` - Wait on the child-process with an optional timeout.
655 """
656 return self.Wait(timeout=0.0, kill=True)
657
658 @readonly
659 def ExitCode(self) -> int:
660 """
661 Read-only property accessing the child-process' exit code.
662
663 :returns: Child-process' exit code or ``None`` if it's still running.
664
665 .. seealso::
666
667 :meth:`Wait`
668 |rarr| Wait on the child-process with an optional timeout.
669 :meth:`Terminate`
670 |rarr| Terminate the child-process.
671 """
672 return self._exitCode
673
674 # This is TCL specific
675 # def ReadUntilBoundary(self, indent=0):
676 # __indent = " " * indent
677 # if self._iterator is None:
678 # self._iterator = iter(self.GetReader())
679 #
680 # for line in self._iterator:
681 # print(__indent + line)
682 # if self._pyIPCMI_BOUNDARY in line:
683 # break
684 # self.LogDebug("Quartus II is ready")
685
686
687@export
688class OutputFilteredExecutable(Executable):
689 """Represent a CLI executable derived from :class:`Executable`, whose outputs are filtered."""
690 _hasOutput: bool #: ``True``, if the executable wrote any output.
691 _hasWarnings: bool #: ``True``, if the output filter classified a line as a warning.
692 _hasErrors: bool #: ``True``, if the output filter classified a line as an error.
693 _hasFatals: bool #: ``True``, if the output filter classified a line as a fatal error.
694
695 def __init__(self, platform: Platform, dryrun: bool, executablePath: Path) -> None: #, environment=None, logger=None) -> None:
696 """
697 Initialize an executable whose output is filtered, with all filter results cleared.
698
699 :param platform: Platform the executable is called on.
700 :param dryrun: If ``True``, the executable is not started, only the command line is assembled.
701 :param executablePath: Path to the executable.
702 """
703 super().__init__(platform, dryrun, executablePath) #, environment=environment, logger=logger)
704
705 self._hasOutput = False
706 self._hasWarnings = False
707 self._hasErrors = False
708 self._hasFatals = False
709
710 @readonly
711 def HasWarnings(self) -> bool:
712 # TODO: update doc-string
713 """
714 Check if warnings were found while processing the output stream.
715
716 :returns: ``True``, if at least one warning was found.
717 """
718 return self._hasWarnings
719
720 @readonly
721 def HasErrors(self) -> bool:
722 # TODO: update doc-string
723 """
724 Check if errors were found while processing the output stream.
725
726 :returns: ``True``, if at least one error was found.
727 """
728 return self._hasErrors
729
730 @readonly
731 def HasFatals(self) -> bool:
732 # TODO: update doc-string
733 """
734 Check if fatal errors were found while processing the output stream.
735
736 :returns: ``True``, if at least one fatal error was found.
737 """
738 return self._hasErrors