diff options
author | Dmitrii Agibov <dmitrii.agibov@arm.com> | 2023-01-27 09:12:50 +0000 |
---|---|---|
committer | Benjamin Klimczak <benjamin.klimczak@arm.com> | 2023-02-08 15:25:11 +0000 |
commit | 3e3dcb9bd5abb88adcd85b4f89e8a81e7f6fa293 (patch) | |
tree | 020eee6abef093113de5b49c135c915c37173843 /src/mlia/utils | |
parent | 836efd40317a397761ec8b66e3f4398faac43ad0 (diff) | |
download | mlia-3e3dcb9bd5abb88adcd85b4f89e8a81e7f6fa293.tar.gz |
MLIA-595 Remove old backend configuration mechanism
- Remove old backend configuration code
- Install backends into directory ~/.mlia
- Rename targets/backends in registry to make it consistent
across codebase.
Change-Id: I9c8b012fe863280f1c692940c0dcad3ef638aaae
Diffstat (limited to 'src/mlia/utils')
-rw-r--r-- | src/mlia/utils/proc.py | 55 |
1 files changed, 55 insertions, 0 deletions
diff --git a/src/mlia/utils/proc.py b/src/mlia/utils/proc.py new file mode 100644 index 0000000..d11bfc5 --- /dev/null +++ b/src/mlia/utils/proc.py @@ -0,0 +1,55 @@ +# SPDX-FileCopyrightText: Copyright 2023, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Module for process management.""" +from __future__ import annotations + +import logging +import subprocess # nosec +from dataclasses import dataclass +from pathlib import Path +from typing import Callable +from typing import Generator + + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class Command: + """Command information.""" + + cmd: list[str] + cwd: Path = Path.cwd() + env: dict[str, str] | None = None + + +def command_output(command: Command) -> Generator[str, None, None]: + """Get command output.""" + logger.debug("Running command: %s", command) + + with subprocess.Popen( # nosec + command.cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + universal_newlines=True, + bufsize=1, + cwd=command.cwd, + env=command.env, + ) as process: + yield from process.stdout or [] + + if process.returncode: + raise subprocess.CalledProcessError(process.returncode, command.cmd) + + +OutputConsumer = Callable[[str], None] + + +def process_command_output( + command: Command, + consumers: list[OutputConsumer], +) -> None: + """Execute command and process output.""" + for line in command_output(command): + for consumer in consumers: + consumer(line) |