# subprocess

Python API: `modelconverter.utils.subprocess`

Execution of external commands used by the conversions.

The conversions drive vendor toolchains through their command line tools. The helpers here run such a command, stream its output
to the log with ANSI escape sequences removed, and record the peak memory usage and the wall-clock run time of the whole process
tree.

## Classes

### SubprocessHandle

Context manager wrapping a subprocess with live psutil access and deferred result collection.

#### Methods

##### init

```python
def __init__(cmd: str | Sequence[PathType], *, silent: bool = False, timeout: float | None = None):
```

Initialize the subprocess handle.

Parameters

 * `cmd` (`str | Sequence[PathType]`): Command to execute. If a string is given, it will be split on whitespace. If a list is
   given, each element will be converted to a string.
 * `silent` (`bool`): If `True`, suppress all output from the command.
 * `timeout` (`float | None`): If given, the maximum time in seconds to allow the process to run. An exceeded timeout raises
   `subprocess.TimeoutExpired` -- from a truth test, which also terminates the process, or from `wait`, which leaves it running.

##### is_suspended

```python
def is_suspended(self) -> bool:
```

Return whether the process is currently suspended.

##### poll

```python
def poll(self) -> int | None:
```

Check whether the process has terminated.

Returns

 * `int | None`: The return code, or `None` if the process is still running.

##### result

```python
def result(self) -> SubprocessResult:
```

Collect the result of the finished process.

Waits for the reader threads to finish and logs a summary of the run unless the handle is silent.

Returns

 * `SubprocessResult`: Result of the command, carrying its captured output, the peak memory usage and the total run time.

Raises

 * `subprocess.SubprocessError`: If the command finished with a non-zero return code.

##### resume

```python
def resume(self):
```

Resume the process.

##### suspend

```python
def suspend(self):
```

Suspend the process.

##### wait

```python
def wait(timeout: float | None = None) -> int:
```

Wait for the process to terminate.

Parameters

 * `timeout` (`float | None`): Maximum time in seconds to wait. Ignored if a timeout was given when the handle was created.

Returns

 * `int`: The return code of the process.

Raises

 * `subprocess.TimeoutExpired`: If the process did not finish within the timeout. Unlike a truth test, this leaves the process
   running.

#### Attributes

##### proc

Return the underlying `subprocess.Popen` object.

Raises

 * `RuntimeError`: If the process has not been started yet.

##### ps_proc

Return the `psutil.Process` wrapping the process.

Raises

 * `RuntimeError`: If the process has not been started yet.

### SubprocessResult

Extension of `subprocess.CompletedProcess` that also carries peak memory usage.

#### Methods

##### init

```python
def __init__(*args, peak_memory: int, total_time: float, **kwargs):
```

Initialize the result.

Parameters

 * `*args`: Positional arguments passed to `subprocess.CompletedProcess`.
 * `peak_memory` (`int`): Peak memory usage of the process and its children, in bytes.
 * `total_time` (`float`): Wall-clock run time of the process, in seconds.
 * `**kwargs`: Keyword arguments passed to `subprocess.CompletedProcess`.

#### Attributes

##### peak_memory

Peak memory usage of the process and its children, in bytes.

##### total_time

Wall-clock run time of the process, in seconds.

## Functions

### strip_ansi

```python
def strip_ansi(s: str) -> str:
```

Remove ANSI escape sequences from a string.

> **Example**
> ```pycon
>>> strip_ansi("\x1b[31mred\x1b[0m")
'red'
```

Parameters

 * `s` (`str`): String to strip.

Returns

 * `str`: The string without ANSI escape sequences.

### subprocess_run

```python
def subprocess_run(cmd: str | Sequence[PathType], *, silent: bool = False, timeout: float | None = None) -> SubprocessResult:
```

Run a command and block until it finishes.

Backwards-compatible wrapper around [SubprocessHandle](https://docs.luxonis.com/software-v3/ai-inference/conversion/rvc-conversion/offline/modelconverter/modelconverter-api-reference/utils/subprocess.md).

Parameters

 * `cmd` (`str | Sequence[PathType]`): Command to execute. If a string is given, it will be split on whitespace. If a list is given, each element will be converted to a string.
 * `silent` (`bool`): If `True`, suppress all output from the command.
 * `timeout` (`float | None`): If given, the maximum time in seconds to allow the process to run. If the timeout is exceeded, the process is terminated and `subprocess.TimeoutExpired` is raised.

Returns

 * `SubprocessResult`: Result of the command.