Skip to content

muxtools.utils.subprogress

__all__ module-attribute 

__all__ = ['ProgressBarConfig', 'run_cmd_pb', 'PERCENTAGE_PATTERN', 'FFMPEG_TIME_PATTERN']

PERCENTAGE_PATTERN module-attribute 

PERCENTAGE_PATTERN = compile('(\\d+(?:\\.\\d+)?)%')

FFMPEG_TIME_PATTERN module-attribute 

FFMPEG_TIME_PATTERN = compile('time=(\\d+:\\d+:\\d+.\\d+)')

ProgressBarConfig dataclass 

ProgressBarConfig(description: str = '', target: int | timedelta = 100, regex: Pattern | str | None = None, groupnum: int = 1, success_return: int = 0)
Source code in muxtools/utils/subprogress.py 
17
18
19
20
21
22
23
@dataclass
class ProgressBarConfig:
    description: str = ""
    target: int | timedelta = 100
    regex: Pattern | str | None = None
    groupnum: int = 1
    success_return: int = 0

description class-attribute instance-attribute 

description: str = ''

target class-attribute instance-attribute 

target: int | timedelta = 100

regex class-attribute instance-attribute 

regex: Pattern | str | None = None

groupnum class-attribute instance-attribute 

groupnum: int = 1

success_return class-attribute instance-attribute 

success_return: int = 0

run_cmd_pb 

run_cmd_pb(cmd: str | list[str], silent: bool = True, pbc: ProgressBarConfig = ProgressBarConfig(), **kwargs: Any) -> int
Source code in muxtools/utils/subprogress.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def run_cmd_pb(cmd: str | list[str], silent: bool = True, pbc: ProgressBarConfig = ProgressBarConfig(), **kwargs: Any) -> int:
    from rich.progress import Progress, TextColumn, BarColumn, TaskProgressColumn, TimeRemainingColumn, TimeElapsedColumn

    args = splitcmd(cmd) if isinstance(cmd, str) else cmd
    process = Popen(args, stdout=PIPE, stderr=STDOUT, text=True, encoding="UTF-8", errors="ignore", bufsize=1, **kwargs)
    if not pbc.regex:
        pbc.regex = PERCENTAGE_PATTERN if isinstance(pbc.target, int) else FFMPEG_TIME_PATTERN
    if isinstance(pbc.regex, str):
        pbc.regex = compile(pbc.regex)

    with Progress(
        TextColumn("[progress.description]{task.description}"), BarColumn(), TaskProgressColumn(), TimeRemainingColumn(), TimeElapsedColumn()
    ) as pro:
        task = pro.add_task(pbc.description)
        prev = 0
        assert process.stdout
        for line in iter(process.stdout.readline, b""):
            if not silent:
                print(line, end="" if line.endswith("\n") else "\n")

            line = line.strip()
            matches = pbc.regex.search(line)

            if matches:
                if isinstance(pbc.target, int):
                    val = float(matches.group(pbc.groupnum))
                    rounded = round(val)
                    if prev < rounded:
                        pro.update(task, completed=rounded)
                        prev = rounded
                else:
                    current_string: str = matches[pbc.groupnum]
                    if current_string.count(":") == 1:
                        current_string = f"0:{current_string}"
                    current = timedelta_from_formatted(current_string)
                    percentage = current.total_seconds() / pbc.target.total_seconds() * 100
                    if prev < percentage:
                        pro.update(task, completed=ceil(percentage))
                        prev = percentage

            # Break this loop if process is done
            if process.poll() is not None:
                if process.returncode == pbc.success_return:
                    pro.update(task, completed=100)
                break

        pro.stop()
    return process.returncode