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