File size: 4,792 Bytes
0838626
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import os
import subprocess
import time
from flask import Flask, request, jsonify


def list_all_devices():
    adb_command = "adb devices"
    device_list = []
    result = EmulatorController.execute_adb(adb_command)
    if result != "ERROR":
        devices = result.split("\n")[1:]
        for d in devices:
            device_list.append(d.split()[0])

    return device_list

def get_adb_device_name(avd_name=None):
    device_list = list_all_devices()
    for device in device_list:
        command = f"adb -s {device} emu avd name"
        ret = EmulatorController.execute_adb(command)
        ret = ret.split("\n")[0]
        if ret == avd_name:
            return device
    return None


app = Flask(__name__)

class Config:
    avd_log_dir = "/logs"  # 请根据实际路径进行修改

class EmulatorController:
    def __init__(self):
        self.avd_log_dir = "logs"
        self.emulator_process = None
        self.out_file = None
    
    @classmethod
    def execute_adb(self, adb_command):
        print(f"Executing command: {adb_command}")
        assert adb_command.startswith("adb"), "Command must start with 'adb'"
        adb_command = "/root/.android/platform-tools/adb" + adb_command[3:]

        result = subprocess.run(adb_command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        print(f"Return code: {result}")
        if result.returncode == 0:
            return result.stdout.strip()
        print(f"Command execution failed: {adb_command}")
        print(result.stderr)
        return "ERROR"
    
    def start_emulator(self, avd_name):
        print(f"Starting Android Emulator with AVD name: {avd_name}")
        
        if not os.path.exists(self.avd_log_dir):
            os.makedirs(self.avd_log_dir, exist_ok=True)
        
        self.out_file = open(os.path.join(self.avd_log_dir, 'emulator_output.txt'), 'a')
        
        self.emulator_process = subprocess.Popen(
            ["/root/.android/emulator/emulator", "-avd", avd_name, "-no-snapshot-save", "-no-window", "-no-audio"], 
            stdout=self.out_file,
            stderr=self.out_file
        )
        
        print("Waiting for the emulator to start...")
        
        while True:
            time.sleep(1)
            try:
                device = get_adb_device_name(avd_name)
            except:
                import traceback
                traceback.print_exc()
                continue
            if device is not None:
                break
        
        print("Device name: ", device)
        print("AVD name: ", avd_name)
        
        while True:
            boot_complete = f"adb -s {device} shell getprop init.svc.bootanim"
            boot_complete = self.execute_adb(boot_complete)
            if boot_complete == 'stopped':
                print("Emulator started successfully")
                break
            time.sleep(1)
        
        time.sleep(1)
        
        device_list = list_all_devices()
        if len(device_list) == 1:
            device = device_list[0]
            print(f"Device selected: {device}")
        
        return device

    def stop_emulator(self, avd_name):
        print("Stopping Android Emulator...")
        if self.emulator_process:
            self.emulator_process.terminate()
        
        while True:
            try:
                device = get_adb_device_name(avd_name)
                command = f"adb -s {device} reboot -p"
                ret = self.execute_adb(command)
                self.emulator_process.terminate()
            except:
                device = None
            if device is None:
                print("Emulator stopped successfully")
                break
            time.sleep(1)
        
        if self.out_file:
            self.out_file.close()

emulator_controller = EmulatorController()

@app.route('/start', methods=['POST'])
def start():
    avd_name = request.json.get('avd_name')
    if not avd_name:
        return jsonify({"error": "No AVD name provided"}), 400

    device = emulator_controller.start_emulator(avd_name)
    return jsonify({"result": "Emulator started", "device": device})

@app.route('/stop', methods=['POST'])
def stop():
    avd_name = request.json.get('avd_name')
    if not avd_name:
        return jsonify({"error": "No AVD name provided"}), 400

    emulator_controller.stop_emulator(avd_name)
    return jsonify({"result": "Emulator stopped"})

@app.route('/execute', methods=['POST'])
def execute():
    adb_command = request.json.get('command')
    if not adb_command:
        return jsonify({"error": "No command provided"}), 400

    result = emulator_controller.execute_adb(adb_command)
    return jsonify({"result": result})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=6060)