22 Arch = Literal["x86_64", "aarch64"]  variable
23 ARCH_OPTIONS = typing.cast(Tuple[Arch], typing.get_args(Arch))
51 def data_dir(arch: Arch):  argument
52     return CACHE_DIR.joinpath("crosvm_tools").joinpath(arch)
55 def pid_path(arch: Arch):  argument
56     return data_dir(arch).joinpath("pid")
59 def ssh_port_path(arch: Arch):  argument
60     return data_dir(arch).joinpath("ssh_port")
63 def log_path(arch: Arch):  argument
64     return data_dir(arch).joinpath("vm_log")
67 def base_img_name(arch: Arch):  argument
68     return f"base-{arch}-{BASE_IMG_VERSION}.qcow2"
71 def base_img_url(arch: Arch):  argument
72     return f"{IMAGE_DIR_URL}/{base_img_name(arch)}"
75 def base_img_path(arch: Arch):  argument
76     return data_dir(arch).joinpath(base_img_name(arch))
79 def rootfs_img_path(arch: Arch):  argument
80     return data_dir(arch).joinpath(f"rootfs-{arch}-{BASE_IMG_VERSION}.qcow2")
83 def ssh_port(arch: Arch) -> int:  argument
86     if not ssh_port_path(arch).exists():
87         return SSH_PORTS[arch]
88     return int(ssh_port_path(arch).read_text())
96 SSH_PORTS: Dict[Arch, int] = {
110 ARCH_TO_QEMU: Dict[Arch, cmd] = {
128 def ssh_opts(arch: Arch) -> Dict[str, str]:  argument
130         "Port": str(ssh_port(arch)),
139 def ssh_cmd_args(arch: Arch):  argument
140     return [f"-o{k}={v}" for k, v in ssh_opts(arch).items()]
143 def ssh_exec(arch: Arch, cmd: Optional[str] = None):  argument
147         *ssh_cmd_args(arch),
152 def ping_vm(arch: Arch):  argument
156         *ssh_cmd_args(arch),
162 def write_pid_file(arch: Arch, pid: int):  argument
163     with open(pid_path(arch), "w") as pid_file:
167 def read_pid_file(arch: Arch):  argument
168     if not pid_path(arch).exists():
171     with open(pid_path(arch), "r") as pid_file:
189     arch: Arch,  argument
195     qemu = ARCH_TO_QEMU[arch]
197         serial = f"file:{data_dir(arch).joinpath('vm_log')}"
201     console.print(f"Booting {arch} VM with disk", hda)
228         ssh_port_path(arch).write_text(str(port))
229         write_pid_file(arch, process.pid)
234 def run_vm(arch: Arch, background: bool = False):  argument
236         arch,
237         rootfs_img_path(arch),
242 def is_running(arch: Arch):  argument
243     pid = read_pid_file(arch)
255 def kill_vm(arch: Arch):  argument
256     pid = read_pid_file(arch)
267 def build_if_needed(arch: Arch, reset: bool = False):  argument
268     if reset and is_running(arch):
269         print(f"Killing existing {arch} VM to perform reset...")
270         kill_vm(arch)
273     data_dir(arch).mkdir(parents=True, exist_ok=True)
275     base_img = base_img_path(arch)
277         print(f"Downloading {arch} base image ({base_img_url(arch)})...")
278         download_file(base_img_url(arch), base_img_path(arch))
280     rootfs_img = rootfs_img_path(arch)
284         print(f"Creating {arch} rootfs overlay...")
295 def up(arch: Arch, reset: bool = False, wait: bool = False, timeout: int = 120):  argument
299     if is_running(arch):
301             console.print(f"{arch} VM is running on port {ssh_port(arch)}")
303         if not wait_until_reachable(arch, timeout):
304             if is_running(arch):
305                 print(f"{arch} VM is not reachable. Restarting it.")
306                 kill_vm(arch)
308                 print(f"{arch} VM stopped. Starting it again.")
310             console.print(f"{arch} VM is running on port {ssh_port(arch)}")
313     build_if_needed(arch, reset)
315         arch,
316         rootfs_img_path(arch),
321         if wait_until_reachable(arch, timeout):
322             console.print(f"{arch} VM is running on port {ssh_port(arch)}")
324             raise Exception(f"Waiting for {arch} VM timed out.")
327 def wait_until_reachable(arch: Arch, timeout: int = 120):  argument
329     if not is_running(arch):
331     if ping_vm(arch):
335         rich.spinner.Spinner("point", f"Waiting for {arch} VM to become reachable...")
339             if not is_running(arch):
341             if ping_vm(arch):
352 def state(arch: Arch):  argument
353     if is_running(arch):
354         if ping_vm(arch):