/
/
/
1"""Helpers to manage OS-level mounts of remote filesystems."""
2
3from __future__ import annotations
4
5import asyncio
6import os
7import platform
8from typing import TYPE_CHECKING
9
10from music_assistant_models.errors import SetupFailedError
11
12from music_assistant.helpers.process import check_output
13
14if TYPE_CHECKING:
15 from logging import Logger
16
17
18def error_summary(output: str) -> str:
19 """
20 Return the summary line of a (u)mount tool's output, for display to the user.
21
22 :param output: The decoded output of the (u)mount command.
23 """
24 # the mount tools state the actual problem on the first line and then append generic
25 # troubleshooting pointers (man pages, dmesg) that are noise in a UI message
26 for line in output.splitlines():
27 if stripped := line.strip():
28 return stripped
29 return ""
30
31
32async def unmount(path: str, logger: Logger) -> None:
33 """
34 Unmount the given path, ensuring it is free for a new mount afterwards.
35
36 Does nothing if the path is not a mountpoint.
37
38 :param path: The (local) mountpoint to unmount.
39 :param logger: Logger to report a failed (regular) unmount on.
40 :raises SetupFailedError: If the path could not be freed.
41 """
42 if not await asyncio.to_thread(os.path.ismount, path):
43 return
44 returncode, output = await check_output("umount", path)
45 if returncode == 0:
46 return
47 error = output.decode().strip()
48 logger.warning("Unmount of %s failed with error: %s", path, error)
49 # a busy mountpoint keeps blocking a new mount on the same path, so detach it anyway:
50 # lazy detach on Linux (frees the mountpoint immediately, even with files still open)
51 # and the forced variant on macOS, which has no lazy equivalent.
52 detach_flag = "-f" if platform.system() == "Darwin" else "-l"
53 returncode, output = await check_output("umount", detach_flag, path)
54 if returncode != 0 and await asyncio.to_thread(os.path.ismount, path):
55 error = output.decode().strip()
56 msg = f"Unable to unmount {path}: {error}"
57 raise SetupFailedError(
58 msg,
59 translation_key="unmount_failed",
60 translation_args=[error_summary(error)],
61 )
62