android_dev/asyncrun.py at main · dpc00/android_dev · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
import json
import subprocess

from csubproc import ContinuousSubprocess, Qi1, Qi2

# from snoop import pp
# from snoop import snoop


def colored(r, g, b, text):
    return f"\033[38;2;{r};{g};{b}m{text}\033[0m"


idel = 1


def a_run(shell_command, cwd=None):
    txt = ""
    p = subprocess.run(
        shell_command,
        shell=True,
        cwd=cwd,
        text=True,
        capture_output=True,
    )
    so = p.stdout
    if so:
        txt = so
        if txt:
            print(txt)
    return p.returncode, txt


def a_run1(shell_command, cwd=None):
    txt = ""
    p = subprocess.run(
        shell_command,
        shell=True,
        cwd=cwd,
        text=True,
        capture_output=True,
    )
    so = p.stdout
    if so:
        txt = so
    return p.returncode, txt


def a_run2(shell_command, cwd=None):
    p = subprocess.run(
        shell_command,
        shell=True,
        cwd=cwd,
        text=True,
    )
    return p.returncode


def a_run3(shell_command, cwd=None):
    txt = ""
    msglst = []
    csp = ContinuousSubprocess(shell_command)
    olg = csp.execute(path=cwd)
    try:
        for ln in olg:
            match ln:
                case Qi1():
                    txt += ln
                    print(colored(0, 255, 0, ln), end="")
                case Qi2():
                    if ln and len(ln):
                        msg = json.loads(ln)
                        msglst.append(msg)
                    # print(colored(255, 0, 0, msg))

    except subprocess.CalledProcessError as exc:
        error_output = json.loads(exc.output)
        message = error_output["message"]
        trace = error_output["trace"]
        print(message)
        print(trace)
        return exc.returncode, txt, msglst
    return 0, txt, msglst


def a_run4(shell_command, cwd=None):
    csp = ContinuousSubprocess(shell_command)
    olg = csp.execute(path=cwd)
    txt1 = ""
    txt2 = ""
    try:
        for ln in olg:
            match ln:
                case Qi1():
                    txt1 += ln
                    print(colored(0, 255, 0, ln), end="")
                case Qi2():
                    txt2 += ln
                    print(colored(0, 0, 255, ln), end="")

    except subprocess.CalledProcessError as exc:
        error_output = json.loads(exc.output)
        message = error_output["message"]
        trace = error_output["trace"]
        print(message)
        print(trace)
        return exc.returncode, txt1, txt2
    return 0, txt1, txt2


run = a_run
run1 = a_run1
run2 = a_run2
run3 = a_run3
run4 = a_run4