music-assistant-server

1"""Run the Music Assistant Server."""
2
3from __future__ import annotations
4
5import argparse
6import asyncio
7import logging
8import os
9import resource
10import signal
11import subprocess
12import sys
13import threading
14import traceback
15from concurrent.futures import ThreadPoolExecutor
16from contextlib import suppress
17from logging.handlers import RotatingFileHandler
18from pathlib import Path
19from typing import Any, Final
20
21from colorlog import ColoredFormatter
22
23from music_assistant.constants import MASS_LOGGER_NAME, VERBOSE_LOG_LEVEL
24from music_assistant.helpers.diagnostics import install_diagnostics_log_handler
25from music_assistant.helpers.json import json_loads
26from music_assistant.helpers.logging import activate_log_queue_handler
27from music_assistant.helpers.util import cap_native_thread_pools
28from music_assistant.mass import MusicAssistant
29
30FORMAT_DATE: Final = "%Y-%m-%d"
31FORMAT_TIME: Final = "%H:%M:%S"
32FORMAT_DATETIME: Final = f"{FORMAT_DATE} {FORMAT_TIME}"
33MAX_LOG_FILESIZE = 1000000 * 10  # 10 MB
34ALPINE_RELEASE_FILE = "/etc/alpine-release"
35
36LOGGER = logging.getLogger(MASS_LOGGER_NAME)
37
38
39def get_arguments() -> argparse.Namespace:
40    """Arguments handling."""
41    parser = argparse.ArgumentParser(description="MusicAssistant")
42
43    # determine default data directory
44    if xdg_data_home := os.getenv("XDG_DATA_HOME"):
45        default_data_dir = os.path.join(xdg_data_home, "music-assistant")
46    else:
47        default_data_dir = os.path.join(Path("~").expanduser(), ".musicassistant")
48    # determine default cache directory
49    if xdg_cache_home := os.getenv("XDG_CACHE_HOME"):
50        default_cache_dir = os.path.join(xdg_cache_home, "music-assistant")
51    else:
52        default_cache_dir = os.path.join(default_data_dir, ".cache")
53
54    parser.add_argument(
55        "--data-dir",
56        "-c",
57        "--config",
58        metavar="path_to_data_dir",
59        default=default_data_dir,
60        help="Directory that contains MusicAssistant persistent data",
61    )
62    parser.add_argument(
63        "--cache-dir",
64        metavar="path_to_cache_dir",
65        default=default_cache_dir,
66        help="Directory that contains MusicAssistant cache data [optional]",
67    )
68    parser.add_argument(
69        "--log-level",
70        type=str,
71        default=os.environ.get("LOG_LEVEL", "info"),
72        help="Provide logging level. Example --log-level debug, "
73        "default=info, possible=(critical, error, warning, info, debug, verbose)",
74    )
75    parser.add_argument(
76        "--safe-mode",
77        action=argparse.BooleanOptionalAction,
78        help="Start in safe mode (core controllers only, no providers)",
79    )
80
81    return parser.parse_args()
82
83
84def setup_logger(data_path: str, level: str = "DEBUG") -> logging.Logger:
85    """Initialize logger."""
86    # define log formatter
87    log_fmt = "%(asctime)s.%(msecs)03d %(levelname)s (%(threadName)s) [%(name)s] %(message)s"
88
89    # base logging config for the root logger.
90    # The root level doubles as the gate for third-party libraries that never get an
91    # explicit level of their own, so it is kept separate from the Music Assistant log
92    # level below: a verbose MA (or provider) level stays scoped to MA's own loggers.
93    logging.basicConfig(level=logging.INFO)
94
95    colorfmt = f"%(log_color)s{log_fmt}%(reset)s"
96    logging.getLogger().handlers[0].setFormatter(
97        ColoredFormatter(
98            colorfmt,
99            datefmt=FORMAT_DATETIME,
100            reset=True,
101            log_colors={
102                "VERBOSE": "light_black",
103                "DEBUG": "cyan",
104                "INFO": "green",
105                "WARNING": "yellow",
106                "ERROR": "red",
107                "CRITICAL": "red",
108            },
109        )
110    )
111
112    # Capture warnings.warn(...) and friends messages in logs.
113    # The standard destination for them is stderr, which may end up unnoticed.
114    # This way they're where other messages are, and can be filtered as usual.
115    logging.captureWarnings(True)
116
117    # install the always-on diagnostics capture handler as early as possible
118    # so boot-time warnings/errors end up in the diagnostics report
119    install_diagnostics_log_handler()
120
121    # setup file handler
122    log_filename = os.path.join(data_path, "musicassistant.log")
123    file_handler = RotatingFileHandler(log_filename, maxBytes=MAX_LOG_FILESIZE, backupCount=1)
124    # rotate log at each start
125    with suppress(OSError):
126        file_handler.doRollover()
127    file_handler.setFormatter(logging.Formatter(log_fmt, datefmt=FORMAT_DATETIME))
128
129    logger = logging.getLogger()
130    logger.addHandler(file_handler)
131    logging.addLevelName(VERBOSE_LOG_LEVEL, "VERBOSE")
132
133    # apply the configured global log level to the (root) music assistant logger
134    logging.getLogger(MASS_LOGGER_NAME).setLevel(level)
135
136    # silence some noisy loggers
137    logging.getLogger("asyncio").setLevel(logging.WARNING)
138    logging.getLogger("aiosqlite").setLevel(logging.WARNING)
139    logging.getLogger("databases").setLevel(logging.WARNING)
140    logging.getLogger("requests").setLevel(logging.WARNING)
141    logging.getLogger("urllib3").setLevel(logging.WARNING)
142    logging.getLogger("aiohttp.access").setLevel(logging.WARNING)
143    logging.getLogger("httpx").setLevel(logging.WARNING)
144    logging.getLogger("charset_normalizer").setLevel(logging.WARNING)
145    logging.getLogger("urllib3.connectionpool").setLevel(logging.ERROR)
146    logging.getLogger("numba").setLevel(logging.WARNING)
147    logging.getLogger("torio._extension.utils").setLevel(logging.WARNING)
148    logging.getLogger("quic").setLevel(logging.WARNING)
149    logging.getLogger("http3").setLevel(logging.WARNING)
150
151    # Add a filter to suppress slow callback warnings from buffered audio streaming
152    # These warnings are expected when audio buffers fill up and producers wait for consumers
153    class BufferedGeneratorFilter(logging.Filter):
154        """Filter out expected slow callback warnings from buffered audio generators."""
155
156        def filter(self, record: logging.LogRecord) -> bool:
157            """Return False to suppress the log record."""
158            if record.levelno != logging.WARNING:
159                return True
160            # Check the formatted message, not the format string
161            msg = record.getMessage()
162            return "buffered.<locals>.producer()" not in msg
163
164    logging.getLogger("asyncio").addFilter(BufferedGeneratorFilter())
165
166    sys.excepthook = lambda *args: logging.getLogger(None).exception(
167        "Uncaught exception",
168        exc_info=args,
169    )
170    threading.excepthook = lambda args: logging.getLogger(None).exception(
171        "Uncaught thread exception",
172        exc_info=(  # type: ignore[arg-type]
173            args.exc_type,
174            args.exc_value,
175            args.exc_traceback,
176        ),
177    )
178
179    return logger
180
181
182def _enable_posix_spawn() -> None:
183    """Enable posix_spawn on Alpine Linux."""
184    if subprocess._USE_POSIX_SPAWN:
185        return
186
187    # The subprocess module does not know about Alpine Linux/musl
188    # and will use fork() instead of posix_spawn() which significantly
189    # less efficient. This is a workaround to force posix_spawn()
190    # on Alpine Linux which is supported by musl.
191    subprocess._USE_POSIX_SPAWN = Path(ALPINE_RELEASE_FILE).exists()  # type: ignore[misc]
192
193
194def _global_loop_exception_handler(_: Any, context: dict[str, Any]) -> None:
195    """Handle all exception inside the core loop."""
196    kwargs = {}
197    if exception := context.get("exception"):
198        kwargs["exc_info"] = (type(exception), exception, exception.__traceback__)
199
200    logger = logging.getLogger(__package__)
201    if source_traceback := context.get("source_traceback"):
202        stack_summary = "".join(traceback.format_list(source_traceback))
203        logger.error(
204            "Error doing job: %s: %s",
205            context["message"],
206            stack_summary,
207            **kwargs,  # type: ignore[arg-type]
208        )
209        return
210
211    logger.error(
212        "Error doing task: %s",
213        context["message"],
214        **kwargs,  # type: ignore[arg-type]
215    )
216
217
218def main() -> None:
219    """Start MusicAssistant."""
220    # parse arguments
221    args = get_arguments()
222
223    data_dir = args.data_dir
224    cache_dir = args.cache_dir
225
226    Path(data_dir).mkdir(parents=True, exist_ok=True)
227    Path(cache_dir).mkdir(parents=True, exist_ok=True)
228
229    # Override options though hass add-on config file
230    hass_options_file = os.path.join(data_dir, "options.json")
231    if Path(hass_options_file).is_file():
232        # we are running as a hass add-on
233        with open(hass_options_file, "rb") as _file:
234            hass_options = json_loads(_file.read())
235    else:
236        hass_options = {}
237
238    # prefer value in hass_options
239    log_level = hass_options.get("log_level", args.log_level).upper()
240    dev_mode = os.environ.get("PYTHONDEVMODE", "0") == "1"
241    safe_mode = bool(
242        args.safe_mode or hass_options.get("safe_mode") or os.environ.get("MASS_SAFE_MODE")
243    )
244
245    # setup logger
246    logger = setup_logger(data_dir, log_level)
247
248    # Size the native BLAS/OpenMP pools before any provider imports a math library,
249    # because those pools read the environment once at load time.
250    blas_budget = cap_native_thread_pools()
251    LOGGER.debug("Native BLAS/OpenMP thread pools capped to %d thread(s)", blas_budget)
252
253    # Raise the open-file soft limit to the hard limit so the concurrent provider
254    # imports at startup can't exhaust it (default soft=1024 in HAOS add-on containers).
255    # Skip when the hard limit is unlimited (e.g. macOS), which setrlimit won't apply.
256    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
257    if hard != resource.RLIM_INFINITY and soft < hard:
258        try:
259            resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard))
260        except (ValueError, OSError) as err:
261            LOGGER.warning("Could not raise open-file limit: %s", err)
262
263    mass = MusicAssistant(data_dir, cache_dir, safe_mode)
264
265    # enable alpine subprocess workaround
266    _enable_posix_spawn()
267
268    async def run_mass() -> None:
269        loop = asyncio.get_running_loop()
270        loop.set_default_executor(ThreadPoolExecutor(max_workers=32))
271        activate_log_queue_handler()
272        if dev_mode or log_level == "DEBUG":
273            loop.set_debug(True)
274            loop.slow_callback_duration = 0.2
275        loop.set_exception_handler(_global_loop_exception_handler)
276
277        stop_event = asyncio.Event()
278
279        def _set_stop() -> None:
280            stop_event.set()
281
282        for sig in (signal.SIGINT, signal.SIGTERM):
283            with suppress(NotImplementedError):
284                loop.add_signal_handler(sig, _set_stop)
285
286        try:
287            # a startup that fails part-way must be cleaned up too, or the databases it
288            # already opened keep their worker threads alive and the process never exits
289            await mass.start()
290            await stop_event.wait()
291        finally:
292            logger.info("shutdown requested!")
293            await mass.stop()
294
295    try:
296        asyncio.run(run_mass())
297    except KeyboardInterrupt:
298        logger.info("shutdown requested by keyboard interrupt")
299
300
301if __name__ == "__main__":
302    main()
303