sublime-music/sublime/config.py

import hashlib
import logging
import os
import pickle
from dataclasses import asdict, dataclass, field, fields
from enum import Enum
from pathlib import Path
from typing import List, Optional

import yaml

from sublime.ui.state import UIState


class ReplayGainType(Enum):
    NO = 0
    TRACK = 1
    ALBUM = 2

    def as_string(self) -> str:
        return ["no", "track", "album"][self.value]

    @staticmethod
    def from_string(replay_gain_type: str) -> "ReplayGainType":
        return {
            "no": ReplayGainType.NO,
            "disabled": ReplayGainType.NO,
            "track": ReplayGainType.TRACK,
            "album": ReplayGainType.ALBUM,
        }[replay_gain_type.lower()]


@dataclass
class ServerConfiguration:
    name: str = "Default"
    server_address: str = "http://yourhost"
    local_network_address: str = ""
    local_network_ssid: str = ""
    username: str = ""
    password: str = ""
    sync_enabled: bool = True
    disable_cert_verify: bool = False
    version: int = 0

    def migrate(self):
        self.version = 0

    _strhash: Optional[str] = None

    def strhash(self) -> str:
        # TODO (#197): make this configurable by the adapters the combination of the
        # hashes will be the hash dir
        """
        Returns the MD5 hash of the server's name, server address, and
        username. This should be used whenever it's necessary to uniquely
        identify the server, rather than using the name (which is not
        necessarily unique).

        >>> sc = ServerConfiguration(
        ...     name='foo',
        ...     server_address='bar',
        ...     username='baz',
        ... )
        >>> sc.strhash()
        '6df23dc03f9b54cc38a0fc1483df6e21'
        """
        if not self._strhash:
            server_info = self.name + self.server_address + self.username
            self._strhash = hashlib.md5(server_info.encode("utf-8")).hexdigest()
        return self._strhash


@dataclass
class AppConfiguration:
    version: int = 3
    cache_location: str = ""
    filename: Optional[Path] = None

    # Servers
    servers: List[ServerConfiguration] = field(default_factory=list)
    current_server_index: int = -1

    # Global Settings
    song_play_notification: bool = True
    offline_mode: bool = False
    serve_over_lan: bool = True
    port_number: int = 8282
    replay_gain: ReplayGainType = ReplayGainType.NO
    allow_song_downloads: bool = True
    download_on_stream: bool = True  # also download when streaming a song
    prefetch_amount: int = 3
    concurrent_download_limit: int = 5

    # TODO this should probably be moved to the cache adapter settings
    max_cache_size_mb: int = -1  # -1 means unlimited

    # Deprecated
    always_stream: bool = False  # always stream instead of downloading songs

    @staticmethod
    def load_from_file(filename: Path) -> "AppConfiguration":
        args = {}
        try:
            if filename.exists():
                with open(filename, "r") as f:
                    field_names = {f.name for f in fields(AppConfiguration)}
                    args = yaml.load(f, Loader=yaml.CLoader).items()
                    args = dict(filter(lambda kv: kv[0] in field_names, args))
        except Exception:
            pass

        config = AppConfiguration(**args)
        config.filename = filename

        return config

    def __post_init__(self):
        # Default the cache_location to ~/.local/share/sublime-music
        if not self.cache_location:
            path = Path(os.environ.get("XDG_DATA_HOME") or "~/.local/share")
            path = path.expanduser().joinpath("sublime-music").resolve()
            self.cache_location = path.as_posix()

        # Deserialize the YAML into the ServerConfiguration object.
        if len(self.servers) > 0 and type(self.servers[0]) != ServerConfiguration:
            self.servers = [ServerConfiguration(**sc) for sc in self.servers]

        self._state = None
        self._current_server_hash = None
        self.migrate()

    def migrate(self):
        for server in self.servers:
            server.migrate()

        if self.version < 4:
            self.allow_song_downloads = not self.always_stream

        self.version = 4
        self.state.migrate()

    @property
    def server(self) -> Optional[ServerConfiguration]:
        if 0 <= self.current_server_index < len(self.servers):
            return self.servers[self.current_server_index]

        return None

    @property
    def state(self) -> UIState:
        server = self.server
        if not server:
            return UIState()

        # If the server has changed, then retrieve the new server's state.
        if self._current_server_hash != server.strhash():
            self.load_state()

        return self._state

    def load_state(self):
        self._state = UIState()
        if not self.server:
            return

        self._current_server_hash = self.server.strhash()
        if self.state_file_location.exists():
            try:
                with open(self.state_file_location, "rb") as f:
                    self._state = pickle.load(f)
            except Exception:
                logging.warning(f"Couldn't load state from {self.state_file_location}")
                # Just ignore any errors, it is only UI state.
                self._state = UIState()

        # Do the import in the function to avoid circular imports.
        from sublime.adapters import AdapterManager

        AdapterManager.reset(self)

    @property
    def state_file_location(self) -> Path:
        assert self.server is not None
        server_hash = self.server.strhash()

        state_file_location = Path(os.environ.get("XDG_DATA_HOME") or "~/.local/share")
        return state_file_location.expanduser().joinpath(
            "sublime-music", server_hash, "state.pickle"
        )

    def save(self):
        # Save the config as YAML.
        self.filename.parent.mkdir(parents=True, exist_ok=True)
        with open(self.filename, "w+") as f:
            f.write(yaml.dump(asdict(self)))

        # Save the state for the current server.
        self.state_file_location.parent.mkdir(parents=True, exist_ok=True)
        with open(self.state_file_location, "wb+") as f:
            pickle.dump(self.state, f)