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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
| # device_pool.py
import redis
import adbutils
from dataclasses import dataclass
from typing import Optional, List
import threading
import time
@dataclass
class Device:
serial: str
status: str # available, busy, offline
android_version: str
assigned_to: Optional[str] = None
class DevicePool:
"""设备池管理器"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.adb = adbutils.AdbClient()
self.lock = threading.Lock()
self._refresh_interval = 30
self._running = False
def discover_devices(self) -> List[Device]:
"""发现所有已连接设备"""
devices = []
for d in self.adb.device_list():
try:
version = d.shell("getprop ro.build.version.release").strip()
devices.append(Device(
serial=d.serial,
status="available",
android_version=version
))
except Exception as e:
print(f"Error discovering {d.serial}: {e}")
return devices
def register_devices(self, devices: List[Device]):
"""注册设备到 Redis"""
pipe = self.redis.pipeline()
for device in devices:
key = f"device:{device.serial}"
pipe.hset(key, mapping={
"serial": device.serial,
"status": device.status,
"android_version": device.android_version,
"assigned_to": device.assigned_to or ""
})
pipe.sadd("devices:all", device.serial)
if device.status == "available":
pipe.sadd("devices:available", device.serial)
pipe.execute()
def acquire_device(self, task_id: str,
min_version: str = None) -> Optional[Device]:
"""获取一个可用设备"""
with self.lock:
# 从可用设备集合中获取
serial = self.redis.spop("devices:available")
if not serial:
return None
serial = serial.decode() if isinstance(serial, bytes) else serial
key = f"device:{serial}"
# 检查版本要求
if min_version:
version = self.redis.hget(key, "android_version")
version = version.decode() if version else "0"
if version < min_version:
# 放回并继续找
self.redis.sadd("devices:available", serial)
return self.acquire_device(task_id, min_version)
# 标记为占用
self.redis.hset(key, mapping={
"status": "busy",
"assigned_to": task_id
})
data = self.redis.hgetall(key)
return Device(
serial=serial,
status="busy",
android_version=data.get(b"android_version", b"").decode(),
assigned_to=task_id
)
def release_device(self, serial: str):
"""释放设备"""
key = f"device:{serial}"
self.redis.hset(key, mapping={
"status": "available",
"assigned_to": ""
})
self.redis.sadd("devices:available", serial)
def get_stats(self) -> dict:
"""获取设备池统计"""
total = self.redis.scard("devices:all")
available = self.redis.scard("devices:available")
return {
"total": total,
"available": available,
"busy": total - available
}
def start_health_check(self):
"""启动健康检查线程"""
self._running = True
def check_loop():
while self._running:
try:
self._check_device_health()
except Exception as e:
print(f"Health check error: {e}")
time.sleep(self._refresh_interval)
thread = threading.Thread(target=check_loop, daemon=True)
thread.start()
def _check_device_health(self):
"""检查设备健康状态"""
all_serials = self.redis.smembers("devices:all")
connected = {d.serial for d in self.adb.device_list()}
for serial in all_serials:
serial = serial.decode() if isinstance(serial, bytes) else serial
key = f"device:{serial}"
if serial not in connected:
# 设备离线
self.redis.hset(key, "status", "offline")
self.redis.srem("devices:available", serial)
else:
# 检查是否需要恢复
status = self.redis.hget(key, "status")
if status == b"offline":
self.redis.hset(key, "status", "available")
self.redis.sadd("devices:available", serial)
# 使用示例
if __name__ == "__main__":
pool = DevicePool()
# 发现并注册设备
devices = pool.discover_devices()
pool.register_devices(devices)
print(f"Registered {len(devices)} devices")
# 启动健康检查
pool.start_health_check()
# 获取设备执行任务
device = pool.acquire_device("task-001", min_version="11")
if device:
print(f"Acquired device: {device.serial}")
try:
# 执行任务...
adb = adbutils.AdbClient()
d = adb.device(device.serial)
d.shell("pm list packages")
finally:
pool.release_device(device.serial)
print(pool.get_stats())
|
支付宝
微信