va_daemon.py
4.81 KB
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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
#!/data/data/com.termux/files/usr/bin/python3
"""VisualAgent Daemon — pure Python, no shell quoting issues.
Drop-in replacement for shell-based VisualAgent daemons. Eliminates all
shell-related problems: PATH resolution, JSON quoting, grep parsing,
su lifecycle, and code-version mismatch.
Start via Hermes background terminal (the ONLY reliable way on Android):
terminal(background=true, command='su -c "exec python3 /path/to/va_daemon.py"')
Do NOT start via `su -c 'nohup ... &'` — the daemon will die when su exits.
"""
import json, os, time, shlex
from datetime import datetime
CMD_DIR = "/sdcard/VisualAgent/commands"
RESULT_DIR = "/sdcard/VisualAgent/results"
PROCESSED_DIR = "/sdcard/VisualAgent/processed"
SCREENSHOT_DIR = "/sdcard/VisualAgent/screenshots"
PIDFILE = "/sdcard/VisualAgent/daemon.pid"
LOGFILE = "/sdcard/VisualAgent/daemon.log"
os.makedirs(RESULT_DIR, exist_ok=True)
os.makedirs(PROCESSED_DIR, exist_ok=True)
os.makedirs(SCREENSHOT_DIR, exist_ok=True)
with open(PIDFILE, 'w') as f:
f.write(str(os.getpid()))
def log(msg):
ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
with open(LOGFILE, 'a') as f:
f.write(f"[{ts}] {msg}\n")
log(f"VisualAgent Python daemon started (PID: {os.getpid()})")
def process_file(filepath):
fname = os.path.basename(filepath)
processed_path = os.path.join(PROCESSED_DIR, fname)
if os.path.exists(processed_path):
try: os.remove(filepath)
except: pass
return
ts = int(time.time())
try:
with open(filepath) as f:
data = json.load(f)
except Exception as e:
log(f"json parse error: {fname} — {e}")
try: os.rename(filepath, processed_path)
except: pass
return
cmd = data.get('command') or data.get('type') or ''
result = {}
try:
if cmd == 'screenshot':
ss = os.path.join(SCREENSHOT_DIR, f"screenshot_{ts}.png")
rc = os.system(f"screencap -p {shlex.quote(ss)}")
result = {"status": "success" if rc == 0 else "error", "type": "screenshot",
"path": ss, "timestamp": ts}
log(f"screenshot -> {ss} (rc={rc})")
elif cmd == 'tap':
x, y = data.get('x', 0), data.get('y', 0)
rc = os.system(f"input tap {x} {y}")
result = {"status": "success" if rc == 0 else "error", "type": "tap",
"x": x, "y": y, "timestamp": ts}
log(f"tap {x},{y} (rc={rc})")
elif cmd == 'swipe':
x1, y1 = data.get('x1', 0), data.get('y1', 0)
x2, y2 = data.get('x2', 0), data.get('y2', 0)
dur = data.get('duration', 300)
rc = os.system(f"input swipe {x1} {y1} {x2} {y2} {dur}")
result = {"status": "success" if rc == 0 else "error", "type": "swipe",
"timestamp": ts}
log(f"swipe {x1},{y1}->{x2},{y2} dur={dur} (rc={rc})")
elif cmd == 'input':
text = data.get('text') or data.get('input') or ''
if text:
rc = os.system(f"input text {shlex.quote(text)}")
result = {"status": "success" if rc == 0 else "error", "type": "input",
"timestamp": ts}
log(f"input: {text[:40]} (rc={rc})")
else:
result = {"status": "error", "type": "input", "message": "no text"}
elif cmd == 'start_app':
pkg = data.get('package', '')
if pkg:
rc = os.system(f"monkey -p {pkg} -c android.intent.category.LAUNCHER 1")
result = {"status": "success" if rc == 0 else "error", "type": "start_app",
"package": pkg, "timestamp": ts}
log(f"start_app: {pkg} (rc={rc})")
else:
result = {"status": "error", "type": "start_app", "message": "no package"}
else:
result = {"status": "error", "type": "unknown", "message": f"unknown: {cmd}",
"timestamp": ts}
log(f"unknown cmd: {cmd}")
except Exception as e:
result = {"status": "error", "type": "exception", "message": str(e), "timestamp": ts}
log(f"exception in {fname}: {e}")
# Write result
result_file = os.path.join(RESULT_DIR, f"result_{fname.replace('.json','')}_{ts}.json")
with open(result_file, 'w') as f:
json.dump(result, f)
# Move to processed
try:
os.rename(filepath, processed_path)
except:
pass
log(f"done: {fname} ({result.get('status','?')})")
# Main loop
while True:
try:
entries = os.listdir(CMD_DIR)
except:
entries = []
for entry in sorted(entries):
if entry.endswith('.json'):
filepath = os.path.join(CMD_DIR, entry)
if os.path.isfile(filepath):
process_file(filepath)
time.sleep(5)