Function run_pio_tracked

ci/util/pio_process_killer.py:413–467

Run a PIO command with automatic process tracking. This is a drop-in replacement for subprocess.run() that tracks the process and ensures it's killed on program exit if still running. Args: args: Command to execute cwd: Working directory env: Environment variabl

(
    args: list[str],
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    shell: bool = False,
    capture_output: bool = False,
    text: bool = True,
    check: bool = False,
    timeout: float | None = None,
)

def run_pio_tracked(
    args: list[str],
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    shell: bool = False,
    capture_output: bool = False,
    text: bool = True,
    check: bool = False,
    timeout: float | None = None,
) -> subprocess.CompletedProcess[Any]:
    """Run a PIO command with automatic process tracking.

    This is a drop-in replacement for subprocess.run() that tracks the process
    and ensures it's killed on program exit if still running.

    Args:
        args: Command to execute
        cwd: Working directory
        env: Environment variables
        shell: Use shell execution
        capture_output: Capture stdout/stderr
        text: Text mode for I/O
        check: Raise CalledProcessError on non-zero exit
        timeout: Maximum time to wait in seconds

    Returns:
        subprocess.CompletedProcess instance
    """
    stdout_pipe = subprocess.PIPE if capture_output else None
    stderr_pipe = subprocess.PIPE if capture_output else None

    with TrackedPopen(
        args,
        cwd=cwd,
        env=env,
        shell=shell,
        stdout=stdout_pipe,
        stderr=stderr_pipe,
        text=text,
    ) as tracked:
        try:
            stdout, stderr = tracked.communicate(timeout=timeout)
        except subprocess.TimeoutExpired:
            tracked.kill()
            stdout, stderr = tracked.communicate()
            raise subprocess.TimeoutExpired(
                args, timeout or 0.0, output=stdout, stderr=stderr
            )

        retcode = tracked.returncode

    if check and retcode:
        raise subprocess.CalledProcessError(retcode, args, output=stdout, stderr=stderr)

    return subprocess.CompletedProcess(args, retcode or 0, stdout, stderr)