"""All constants for Music Assistant."""
-__version__ = "0.2.12"
+__version__ = "0.2.13"
REQUIRED_PYTHON_VER = "3.8"
# configuration keys/attributes
"""Analyze track audio, for now we only calculate EBU R128 loudness."""
if streamdetails.loudness is not None:
- # only when needed we do the analyze stuff
+ # only when needed we do the analyze job
return
- # only when needed we do the analyze stuff
LOGGER.debug(
"Start analyzing track %s/%s",
streamdetails.provider,
and hasattr(item.artist, "metadata")
and item.artist.metadata.get("image")
):
- return item.album.metadata["image"]
+ return item.artist.metadata["image"]
if media_type == MediaType.TRACK and item.album:
# try album instead for tracks
return await get_image_url(
isinstance(val, (list, set, filter, tuple))
or val.__class__ == "dict_valueiterator"
):
- return [get_val(x) for x in val]
+ return [get_val(x) for x in val] if val else []
if isinstance(val, dict):
return {key: get_val(value) for key, value in val.items()}
try:
return val.to_dict()
except AttributeError:
return val
+ except Exception:
+ return val
return get_val(obj)
from music_assistant.helpers.typing import MusicAssistant
from music_assistant.helpers.util import create_task, try_load_json_file
from music_assistant.helpers.web import api_route
-from music_assistant.models.config_entry import ConfigEntry, ConfigEntryType
+from music_assistant.models.config_entry import (
+ ConfigEntry,
+ ConfigEntryType,
+ ConfigValueOption,
+)
from music_assistant.models.player import PlayerControlType
from music_assistant.models.provider import ProviderType
from passlib.hash import pbkdf2_sha256
LOGGER = logging.getLogger("config_manager")
+SAMPLERATE_OPTIONS = [
+ ConfigValueOption(text=str(val), value=val)
+ for val in (41000, 48000, 96000, 176000, 192000, 384000)
+]
+
DEFAULT_PLAYER_CONFIG_ENTRIES = [
ConfigEntry(
entry_key=CONF_ENABLED,
ConfigEntry(
entry_key=CONF_MAX_SAMPLE_RATE,
entry_type=ConfigEntryType.INT,
- values=[41000, 48000, 96000, 176000, 192000, 384000],
+ options=SAMPLERATE_OPTIONS,
default_value=96000,
label=CONF_MAX_SAMPLE_RATE,
description="desc_sample_rate",
self.loading = True
conf_file = os.path.join(self.data_path, "config.json")
data = try_load_json_file(conf_file)
- if not data:
+ if data is None:
# might be a corrupt config file, retry with backup file
conf_file_backup = os.path.join(self.data_path, "config.json.backup")
data = try_load_json_file(conf_file_backup)
)
if power_controls:
controls = [
- {"text": f"{item.provider}: {item.name}", "value": item.control_id}
+ ConfigValueOption(
+ text=f"{item.provider}: {item.name}", value=item.control_id
+ )
for item in power_controls
]
entries.append(
entry_type=ConfigEntryType.STRING,
label=CONF_POWER_CONTROL,
description="desc_power_control",
- values=controls,
+ options=controls,
)
)
# append volume control config entries
)
if volume_controls:
controls = [
- {"text": f"{item.provider}: {item.name}", "value": item.control_id}
+ ConfigValueOption(
+ text=f"{item.provider}: {item.name}", value=item.control_id
+ )
for item in volume_controls
]
entries.append(
entry_type=ConfigEntryType.STRING,
label=CONF_VOLUME_CONTROL,
description="desc_volume_control",
- values=controls,
+ options=controls,
)
)
# append special group player entries
"""Model and helpers for Config entries."""
-from dataclasses import dataclass, field
+from __future__ import annotations
+
+from dataclasses import dataclass
from enum import Enum
-from typing import Any, List, Tuple
+from typing import List, Optional, Tuple, Union
from mashumaro import DataClassDictMixin
DICT = "dict"
+ValueTypes = Union[str, int, float, bool, dict, None]
+
+
+@dataclass
+class ConfigValueOption(DataClassDictMixin):
+ """Model for a value with seperated name/value."""
+
+ text: str
+ value: ValueTypes
+
+
@dataclass
class ConfigEntry(DataClassDictMixin):
"""Model for a Config Entry."""
entry_key: str
entry_type: ConfigEntryType
- default_value: Any = ""
- values: List[Any] = field(default_factory=list) # select from list of values
- range: Tuple[Any] = () # select values within range
- label: str = "" # a friendly name for the setting
- description: str = "" # extended description of the setting.
- help_key: str = "" # key in the translations file
+ default_value: ValueTypes = None
+ # options: select from list of possible values/options
+ options: Optional[List[ConfigValueOption]] = None
+ range: Optional[Tuple[int, int]] = None # select values within range
+ label: Optional[str] = None # a friendly name for the setting
+ description: Optional[str] = None # extended description of the setting.
+ help_key: Optional[str] = None # key in the translations file
multi_value: bool = False # allow multiple values from the list
- depends_on: str = "" # needs to be set before this setting shows up in frontend
+ depends_on: Optional[
+ str
+ ] = None # needs to be set before this setting shows up in frontend
hidden: bool = False # hide from UI
- value: Any = None # set by the configuration manager
+ value: ValueTypes = None # set by the configuration manager
store_hashed: bool = False # value will be hashed, non reversible
active_queue=active_queue,
)
- def to_dict(self) -> dict:
- """Return playerstate for compatability with json serializer."""
- return self._calculated_state.to_dict()
-
def __init__(self, *args, **kwargs) -> None:
"""Initialize a Player instance."""
self.mass: Optional[MusicAssistant] = None
self.added_to_mass = False
self._calculated_state = CalculatedPlayerState()
+
+ def to_dict(self):
+ """Return playerstate for compatability with json serializer."""
+ return self._calculated_state.to_dict()
"""Models and helpers for the streamdetails of a MediaItem."""
-from dataclasses import dataclass
+from dataclasses import dataclass, field
from enum import Enum
-from typing import Any, Optional
+from typing import Any, Dict, Optional
from mashumaro.serializer.base.dict import DataClassDictMixin
from music_assistant.models.media_types import MediaType
path: str
content_type: ContentType
player_id: str = ""
- details: Any = None
+ details: Dict[str, Any] = field(default_factory=dict)
seconds_played: int = 0
gain_correct: float = 0
loudness: Optional[float] = None
bit_depth: Optional[int] = None
media_type: MediaType = MediaType.TRACK
- def to_dict(
- self,
- use_bytes: bool = False,
- use_enum: bool = False,
- use_datetime: bool = False,
- **kwargs,
- ):
- """Handle conversion to dict."""
- return {
- "provider": self.provider,
- "item_id": self.item_id,
- "content_type": self.content_type.value,
- "media_type": self.media_type.value,
- "sample_rate": self.sample_rate,
- "bit_depth": self.bit_depth,
- "gain_correct": self.gain_correct,
- "seconds_played": self.seconds_played,
- }
+ def __post_serialize__(self, d: Dict[Any, Any]) -> Dict[Any, Any]:
+ """Exclude internal fields from dict."""
+ d.pop("path")
+ d.pop("details")
+ return d
def __str__(self):
"""Return pretty printable string of object."""
from pychromecast.controllers.multizone import MultizoneManager
from .const import PROV_ID, PROV_NAME, PROVIDER_CONFIG_ENTRIES
-from .helpers import DEFAULT_PORT, ChromecastInfo
+from .helpers import ChromecastInfo
from .player import ChromecastPlayer
LOGGER = logging.getLogger(PROV_ID)
def _discover_chromecast(self, uuid, _):
"""Discover a Chromecast."""
- device_info = self._browser.devices[uuid]
+ cast_info: pychromecast.models.CastInfo = self._browser.devices[uuid]
info = ChromecastInfo(
- services=device_info.services,
- uuid=device_info.uuid,
- model_name=device_info.model_name,
- friendly_name=device_info.friendly_name,
- is_audio_group=device_info.port != DEFAULT_PORT,
+ services=cast_info.services,
+ uuid=cast_info.uuid,
+ model_name=cast_info.model_name,
+ friendly_name=cast_info.friendly_name,
+ cast_type=cast_info.cast_type,
+ manufacturer=cast_info.manufacturer,
)
if info.uuid is None:
player.set_cast_info(info)
create_task(self.mass.players.add_player(player))
- def _remove_chromecast(self, uuid, service, cast_info):
+ @staticmethod
+ def _remove_chromecast(uuid, service, cast_info):
"""Handle zeroconf discovery of a removed chromecast."""
# pylint: disable=unused-argument
player_id = str(service[1])
import attr
from pychromecast import dial
-from pychromecast.const import CAST_MANUFACTURERS
+from pychromecast.const import CAST_TYPE_GROUP
DEFAULT_PORT = 8009
@attr.s(slots=True, frozen=True)
class ChromecastInfo:
- """Class to hold all data about a chromecast for creating connections.
+ """
+ Class to hold all data about a chromecast for creating connections.
This also has the same attributes as the mDNS fields by zeroconf.
"""
services: set | None = attr.ib()
- uuid: str | None = attr.ib(
- converter=attr.converters.optional(str), default=None
- ) # always convert UUID to string if not None
- _manufacturer = attr.ib(type=Optional[str], default=None)
- model_name: str = attr.ib(default="")
- friendly_name: str | None = attr.ib(default=None)
- is_audio_group = attr.ib(type=Optional[bool], default=False)
+ uuid: str = attr.ib(converter=attr.converters.optional(str))
+ model_name: str = attr.ib()
+ friendly_name: str = attr.ib()
+ cast_type: str = attr.ib()
+ manufacturer: str = attr.ib()
is_dynamic_group = attr.ib(type=Optional[bool], default=None)
@property
- def is_information_complete(self) -> bool:
- """Return if all information is filled out."""
- want_dynamic_group = self.is_audio_group
- have_dynamic_group = self.is_dynamic_group is not None
- have_all_except_dynamic_group = all(
- attr.astuple(
- self,
- filter=attr.filters.exclude(
- attr.fields(ChromecastInfo).is_dynamic_group
- ),
- )
- )
- return have_all_except_dynamic_group and (
- not want_dynamic_group or have_dynamic_group
- )
-
- @property
- def manufacturer(self) -> str | None:
- """Return the manufacturer."""
- if self._manufacturer:
- return self._manufacturer
- if not self.model_name:
- return None
- return CAST_MANUFACTURERS.get(self.model_name.lower(), "Google Inc.")
+ def is_audio_group(self) -> bool:
+ """Return if the cast is an audio group."""
+ return self.cast_type == CAST_TYPE_GROUP
def fill_out_missing_chromecast_info(self, zconf) -> ChromecastInfo:
- """Return a new ChromecastInfo object with missing attributes filled in.
+ """
+ Return a new ChromecastInfo object with missing attributes filled in.
Uses blocking HTTP / HTTPS.
"""
- if self.is_information_complete:
+ if not self.is_audio_group or self.is_dynamic_group is not None:
# We have all information, no need to check HTTP API.
return self
# Fill out missing group information via HTTP API.
- if self.is_audio_group:
- is_dynamic_group = False
- http_group_status = None
- if self.uuid:
- http_group_status = dial.get_multizone_status(
- None,
- services=self.services,
- zconf=zconf,
- )
- if http_group_status is not None:
- is_dynamic_group = any(
- str(g.uuid) == self.uuid
- for g in http_group_status.dynamic_groups
- )
-
- return ChromecastInfo(
- services=self.services,
- uuid=self.uuid,
- friendly_name=self.friendly_name,
- model_name=self.model_name,
- is_audio_group=True,
- is_dynamic_group=is_dynamic_group,
- )
-
- # Fill out some missing information (friendly_name, uuid) via HTTP dial.
- http_device_status = dial.get_device_status(
- None, services=self.services, zconf=zconf
+ is_dynamic_group = False
+ http_group_status = None
+ http_group_status = dial.get_multizone_status(
+ None,
+ services=self.services,
+ zconf=zconf,
)
- if http_device_status is None:
- # HTTP dial didn't give us any new information.
- return self
+ if http_group_status is not None:
+ is_dynamic_group = any(
+ str(g.uuid) == self.uuid for g in http_group_status.dynamic_groups
+ )
return ChromecastInfo(
services=self.services,
- uuid=(self.uuid or http_device_status.uuid),
- friendly_name=(self.friendly_name or http_device_status.friendly_name),
- manufacturer=(self.manufacturer or http_device_status.manufacturer),
- model_name=(self.model_name or http_device_status.model_name),
+ uuid=self.uuid,
+ friendly_name=self.friendly_name,
+ model_name=self.model_name,
+ cast_type=self.cast_type,
+ manufacturer=self.manufacturer,
+ is_dynamic_group=is_dynamic_group,
)
def __str__(self):
class CastStatusListener:
- """Helper class to handle pychromecast status callbacks.
+ """
+ Helper class to handle pychromecast status callbacks.
Necessary because a CastDevice entity can create a new socket client
and therefore callbacks from multiple chromecast connections can
self._cast_device.multizone_new_media_status(group_uuid, media_status)
def invalidate(self):
- """Invalidate this status listener.
+ """
+ Invalidate this status listener.
All following callbacks won't be forwarded.
"""
"""Representation of a Cast device on the network."""
import asyncio
import logging
-import uuid
from typing import List, Optional
import pychromecast
self._cast_info = cast_info
self._player_id = cast_info.uuid
- self.services = cast_info.services
self._chromecast: Optional[pychromecast.Chromecast] = None
self.cast_status = None
self.media_status = None
)
# Chromecast does not support power so we (ab)use mute instead
- if self._chromecast.media_controller.is_active:
- return (
- self.cast_status.app_id
- in ["705D30C6", self._chromecast.media_controller.app_id]
- and not self.cast_status.volume_muted
- )
- return not self.cast_status.volume_muted
+ if self.cast_status.app_id is None:
+ return not self.cast_status.volume_muted
+ return (
+ self.cast_status.app_id in ["705D30C6", pychromecast.APP_MEDIA_RECEIVER]
+ and not self.cast_status.volume_muted
+ )
@property
def should_poll(self) -> bool:
and self._chromecast
and not self._is_speaker_group
):
- return [
- str(uuid.UUID(item)) for item in self._chromecast.mz_controller.members
- ]
+ return self._chromecast.mz_controller.members
return []
@property
None,
pychromecast.get_chromecast_from_cast_info,
pychromecast.discovery.CastInfo(
- self.services,
+ self._cast_info.services,
self._cast_info.uuid,
self._cast_info.model_name,
self._cast_info.friendly_name,
None,
None,
+ self._cast_info.cast_type,
+ self._cast_info.manufacturer,
),
self.mass.zeroconf,
)
from music_assistant.helpers.typing import MusicAssistant
from music_assistant.helpers.util import create_task
-from music_assistant.models.config_entry import ConfigEntry, ConfigEntryType
+from music_assistant.models.config_entry import (
+ ConfigEntry,
+ ConfigEntryType,
+ ConfigValueOption,
+)
from music_assistant.models.player import DeviceInfo, Player, PlayerState
from music_assistant.models.provider import PlayerProvider
def __get_config_entries(self):
"""Return config entries for this group player."""
all_players = [
- {"text": item.name, "value": item.player_id}
+ ConfigValueOption(text=item.name, value=item.player_id)
for item in self.mass.players
if item.player_id is not self._player_id
]
# selected_players_ids = []
selected_players = []
for player_id in selected_players_ids:
- player = self.mass.players.get_player(player_id)
- if player:
+ if player := self.mass.players.get_player(player_id):
selected_players.append(
- {"text": player.name, "value": player.player_id}
+ ConfigValueOption(text=player.name, value=player.player_id)
)
default_master = ""
if selected_players:
entry_key=CONF_PLAYERS,
entry_type=ConfigEntryType.STRING,
default_value=[],
- values=all_players,
+ options=all_players,
label=CONF_PLAYERS,
description="group_player_players_desc",
multi_value=True,
entry_key=CONF_MASTER,
entry_type=ConfigEntryType.STRING,
default_value=default_master,
- values=selected_players,
+ options=selected_players,
label=CONF_MASTER,
description="group_player_master_desc",
multi_value=False,
projects / music-assistant-server.git / commitdiff