Android Emulator
Version proof-of-concept.
HF Free = 2GB RAM, pas de KVM.
Android sera tres lent a demarrer (10-30 min).
Chargement...
"""
def get_ram():
try:
with open("/proc/meminfo") as f:
lines = f.readlines()
total = int(lines[0].split()[1]) // 1024
avail = 0
for l in lines:
if l.startswith("MemAvailable:"):
avail = int(l.split()[1]) // 1024
return f"{total - avail}MB / {total}MB"
except:
return "?"
def download_iso():
iso = os.path.join(WORK, "android.iso")
if os.path.exists(iso) and os.path.getsize(iso) > 1000:
log.info(f"ISO deja la: {os.path.getsize(iso)//1048576}MB")
return True
os.makedirs(WORK, exist_ok=True)
log.info("Telechargement ISO Android-x86 9.0...")
url = "https://sourceforge.net/projects/android-x86/files/Release%209.0/android-x86_64-9.0-r2.iso/download"
r = subprocess.run(["wget","-q","-O",iso,url], capture_output=True, text=True)
if os.path.exists(iso) and os.path.getsize(iso) > 1000:
log.info(f"OK: {os.path.getsize(iso)//1048576}MB")
return True
else:
log.error("Echec telechargement")
log.error(r.stderr if r.stderr else "pas d'erreur")
return False
def create_disk():
disk = os.path.join(WORK, "disk.qcow2")
if os.path.exists(disk):
return True
log.info("Creation disque 4G...")
subprocess.run(["qemu-img","create","-f","qcow2",disk,"4G"], check=True)
return True
def start_qemu():
iso = os.path.join(WORK, "android.iso")
disk = os.path.join(WORK, "disk.qcow2")
kvm = os.path.exists("/dev/kvm")
cmd = ["qemu-system-x86_64"]
if kvm:
cmd += ["-enable-kvm", "-cpu", "host"]
else:
cmd += ["-cpu", "qemu64", "-accel", "tcg,thread=multi"]
cmd += [
"-m", "1024",
"-smp", "2",
"-drive", f"file={disk},format=qcow2,if=virtio",
"-cdrom", iso,
"-boot", "d",
"-display", "none",
"-net", "nic,model=virtio",
"-net", "user",
"-nographic",
]
log.info(f"QEMU: {' '.join(cmd)}")
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
def reader():
for line in iter(proc.stdout.readline, b""):
t = line.decode("utf-8", errors="replace").strip()
if t:
log.info(f"[QEMU] {t}")
threading.Thread(target=reader, daemon=True).start()
return proc
def pipeline():
try:
state["phase"] = "downloading"
if not download_iso():
state["phase"] = "error"
return
create_disk()
state["phase"] = "booting"
state["proc"] = start_qemu()
state["phase"] = "running"
log.info("QEMU demarre!")
log.info("Sans KVM = tres lent, patience...")
except Exception as e:
log.error(f"Erreur: {e}")
state["phase"] = "error"
def stop():
if state["proc"]:
state["proc"].terminate()
state["proc"] = None
state["phase"] = "idle"
class H(http.server.BaseHTTPRequestHandler):
def do_GET(self):
p = self.path.split("?")[0]
if p == "/":
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(HTML.encode())
elif p == "/api/status":
running = state["proc"] and state["proc"].poll() is None
d = {
"phase": state["phase"],
"kvm": os.path.exists("/dev/kvm"),
"ram": get_ram(),
"qemu": shutil.which("qemu-system-x86_64") is not None,
"iso": os.path.exists(os.path.join(WORK, "android.iso")),
"disk": os.path.exists(os.path.join(WORK, "disk.qcow2")),
"running": running,
"logs": LOG_BUFFER[-50:],
}
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(d).encode())
elif p == "/api/start":
if state["phase"] in ("idle", "error"):
threading.Thread(target=pipeline, daemon=True).start()
m = {"ok": True, "message": "Demarrage..."}
else:
m = {"ok": False, "message": f"Phase: {state['phase']}"}
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(m).encode())
elif p == "/api/stop":
stop()
m = {"ok": True, "message": "Arrete"}
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(m).encode())
else:
self.send_error(404)
def log_message(self, *a):
pass
def main():
log.info("=" * 40)
log.info(" Android Emulator - HF Spaces")
log.info("=" * 40)
log.info(f"KVM: {os.path.exists('/dev/kvm')}")
log.info(f"QEMU: {shutil.which('qemu-system-x86_64')}")
log.info(f"RAM: {get_ram()}")
s = socketserver.ThreadingTCPServer(("0.0.0.0", PORT), H)
s.allow_reuse_address = True
log.info(f"http://0.0.0.0:{PORT}")
s.serve_forever()
if __name__ == "__main__":
main()