+++ /dev/null
-# The default ``config.py``
-# flake8: noqa
-
-
-def set_prefs(prefs):
- """This function is called before opening the project"""
-
- # Specify which files and folders to ignore in the project.
- # Changes to ignored resources are not added to the history and
- # VCSs. Also they are not returned in `Project.get_files()`.
- # Note that ``?`` and ``*`` match all characters but slashes.
- # '*.pyc': matches 'test.pyc' and 'pkg/test.pyc'
- # 'mod*.pyc': matches 'test/mod1.pyc' but not 'mod/1.pyc'
- # '.svn': matches 'pkg/.svn' and all of its children
- # 'build/*.o': matches 'build/lib.o' but not 'build/sub/lib.o'
- # 'build//*.o': matches 'build/lib.o' and 'build/sub/lib.o'
- prefs["ignored_resources"] = [
- "*.pyc",
- "*~",
- ".ropeproject",
- ".hg",
- ".svn",
- "_svn",
- ".git",
- ".tox",
- ]
-
- # Specifies which files should be considered python files. It is
- # useful when you have scripts inside your project. Only files
- # ending with ``.py`` are considered to be python files by
- # default.
- # prefs['python_files'] = ['*.py']
-
- # Custom source folders: By default rope searches the project
- # for finding source folders (folders that should be searched
- # for finding modules). You can add paths to that list. Note
- # that rope guesses project source folders correctly most of the
- # time; use this if you have any problems.
- # The folders should be relative to project root and use '/' for
- # separating folders regardless of the platform rope is running on.
- # 'src/my_source_folder' for instance.
- # prefs.add('source_folders', 'src')
-
- # You can extend python path for looking up modules
- # prefs.add('python_path', '~/python/')
-
- # Should rope save object information or not.
- prefs["save_objectdb"] = True
- prefs["compress_objectdb"] = False
-
- # If `True`, rope analyzes each module when it is being saved.
- prefs["automatic_soa"] = True
- # The depth of calls to follow in static object analysis
- prefs["soa_followed_calls"] = 0
-
- # If `False` when running modules or unit tests "dynamic object
- # analysis" is turned off. This makes them much faster.
- prefs["perform_doa"] = True
-
- # Rope can check the validity of its object DB when running.
- prefs["validate_objectdb"] = True
-
- # How many undos to hold?
- prefs["max_history_items"] = 32
-
- # Shows whether to save history across sessions.
- prefs["save_history"] = True
- prefs["compress_history"] = False
-
- # Set the number spaces used for indenting. According to
- # :PEP:`8`, it is best to use 4 spaces. Since most of rope's
- # unit-tests use 4 spaces it is more reliable, too.
- prefs["indent_size"] = 4
-
- # Builtin and c-extension modules that are allowed to be imported
- # and inspected by rope.
- prefs["extension_modules"] = []
-
- # Add all standard c-extensions to extension_modules list.
- prefs["import_dynload_stdmods"] = True
-
- # If `True` modules with syntax errors are considered to be empty.
- # The default value is `False`; When `False` syntax errors raise
- # `rope.base.exceptions.ModuleSyntaxError` exception.
- prefs["ignore_syntax_errors"] = False
-
- # If `True`, rope ignores unresolvable imports. Otherwise, they
- # appear in the importing namespace.
- prefs["ignore_bad_imports"] = False
-
- # If `True`, rope will insert new module imports as
- # `from <package> import <module>` by default.
- prefs["prefer_module_from_imports"] = False
-
- # If `True`, rope will transform a comma list of imports into
- # multiple separate import statements when organizing
- # imports.
- prefs["split_imports"] = False
-
- # If `True`, rope will remove all top-level import statements and
- # reinsert them at the top of the module when making changes.
- prefs["pull_imports_to_top"] = True
-
- # If `True`, rope will sort imports alphabetically by module name instead
- # of alphabetically by import statement, with from imports after normal
- # imports.
- prefs["sort_imports_alphabetically"] = False
-
- # Location of implementation of
- # rope.base.oi.type_hinting.interfaces.ITypeHintingFactory In general
- # case, you don't have to change this value, unless you're an rope expert.
- # Change this value to inject you own implementations of interfaces
- # listed in module rope.base.oi.type_hinting.providers.interfaces
- # For example, you can add you own providers for Django Models, or disable
- # the search type-hinting in a class hierarchy, etc.
- prefs["type_hinting_factory"] = "rope.base.oi.type_hinting.factory.default_type_hinting_factory"
-
-
-def project_opened(project):
- """This function is called after opening the project"""
- # Do whatever you like here!
from async_upnp_client.client import UpnpRequester, UpnpService, UpnpStateVariable
from async_upnp_client.utils import CaseInsensitiveDict
- from music_assistant.common.models.config_entries import PlayerConfig, ProviderConfig
+ from music_assistant.common.models.config_entries import (
+ PlayerConfig,
+ ProviderConfig,
+ )
from music_assistant.common.models.provider import ProviderManifest
from music_assistant.common.models.queue_item import QueueItem
from music_assistant.server import MusicAssistant
"""Catch UpnpError errors."""
@functools.wraps(func)
- async def wrapper(self: _DLNAPlayerProviderT, *args: _P.args, **kwargs: _P.kwargs) -> _R | None:
+ async def wrapper(
+ self: _DLNAPlayerProviderT, *args: _P.args, **kwargs: _P.kwargs
+ ) -> _R | None:
"""Catch UpnpError errors and check availability before and after request."""
player_id = kwargs["player_id"] if "player_id" in kwargs else args[0]
dlna_player = self.dlnaplayers[player_id]
dlna_player.player.display_name,
)
if not dlna_player.available:
- self.logger.warning("Device disappeared when trying to call %s", func.__name__)
+ self.logger.warning(
+ "Device disappeared when trying to call %s", func.__name__
+ )
return None
try:
return await func(self, *args, **kwargs)
"""Handle async initialization of the provider."""
self.dlnaplayers = {}
self.lock = asyncio.Lock()
- self.requester = AiohttpSessionRequester(self.mass.http_session, with_sleep=True)
+ self.requester = AiohttpSessionRequester(
+ self.mass.http_session, with_sleep=True
+ )
self.upnp_factory = UpnpFactory(self.requester, non_strict=True)
self.notify_server = DLNANotifyServer(self.requester, self.mass)
self.mass.create_task(self._run_discovery())
"""
use_flow_mode = await self.mass.config.get_player_config_value(
player_id, CONF_FLOW_MODE
- ) or await self.mass.config.get_player_config_value(player_id, CONF_CROSSFADE)
- enforce_mp3 = await self.mass.config.get_player_config_value(player_id, CONF_ENFORCE_MP3)
+ )
+ enforce_mp3 = await self.mass.config.get_player_config_value(
+ player_id, CONF_ENFORCE_MP3
+ )
url = await self.mass.streams.resolve_stream_url(
queue_item=queue_item,
output_codec=ContentType.MP3 if enforce_mp3 else ContentType.FLAC,
await self.poll_player(dlna_player.udn)
@catch_request_errors
- async def play_stream(self, player_id: str, stream_job: MultiClientStreamJob) -> None:
+ async def play_stream(
+ self, player_id: str, stream_job: MultiClientStreamJob
+ ) -> None:
"""Handle PLAY STREAM on given player.
This is a special feature from the Universal Group provider.
"""
- enforce_mp3 = await self.mass.config.get_player_config_value(player_id, CONF_ENFORCE_MP3)
+ enforce_mp3 = await self.mass.config.get_player_config_value(
+ player_id, CONF_ENFORCE_MP3
+ )
output_codec = ContentType.MP3 if enforce_mp3 else ContentType.FLAC
url = stream_job.resolve_stream_url(player_id, output_codec)
dlna_player = self.dlnaplayers[player_id]
if dlna_player.device.can_stop:
await self.cmd_stop(player_id)
didl_metadata = create_didl_metadata(self.mass, url, None)
- await dlna_player.device.async_set_transport_uri(url, "Music Assistant", didl_metadata)
+ await dlna_player.device.async_set_transport_uri(
+ url, "Music Assistant", didl_metadata
+ )
# Play it
await dlna_player.device.async_wait_for_can_play(10)
# optimistically set this timestamp to help in case of a player
await self.poll_player(dlna_player.udn)
@catch_request_errors
- async def enqueue_next_queue_item(self, player_id: str, queue_item: QueueItem) -> None:
+ async def enqueue_next_queue_item(
+ self, player_id: str, queue_item: QueueItem
+ ) -> None:
"""Handle enqueuing of the next queue item on the player."""
dlna_player = self.dlnaplayers[player_id]
url = await self.mass.streams.resolve_stream_url(
# we iterate between using a regular and multicast search (if enabled)
if allow_network_scan and use_multicast:
- await async_search(on_response, target=(str(IPv4Address("255.255.255.255")), 1900))
+ await async_search(
+ on_response, target=(str(IPv4Address("255.255.255.255")), 1900)
+ )
else:
await async_search(on_response)
async with self.lock:
if dlna_player := self.dlnaplayers.get(udn):
# existing player
- if dlna_player.description_url == description_url and dlna_player.player.available:
+ if (
+ dlna_player.description_url == description_url
+ and dlna_player.player.available
+ ):
# nothing to do, device is already connected
return
# update description url to newly discovered one
return
# Connect to the base UPNP device
- upnp_device = await self.upnp_factory.async_create_device(dlna_player.description_url)
+ upnp_device = await self.upnp_factory.async_create_device(
+ dlna_player.description_url
+ )
# Create profile wrapper
- dlna_player.device = DmrDevice(upnp_device, self.notify_server.event_handler)
+ dlna_player.device = DmrDevice(
+ upnp_device, self.notify_server.event_handler
+ )
# Subscribe to event notifications
try:
# Don't leave the device half-constructed
dlna_player.device.on_event = None
dlna_player.device = None
- self.logger.debug("Error while subscribing during device connect: %r", err)
+ self.logger.debug(
+ "Error while subscribing during device connect: %r", err
+ )
raise
else:
# connect was successful, update device info
for state_variable in state_variables:
# Force a state refresh when player begins or pauses playback
# to update the position info.
- if state_variable.name == "TransportState" and state_variable.value in (
- TransportState.PLAYING,
- TransportState.PAUSED_PLAYBACK,
+ if (
+ state_variable.name == "TransportState"
+ and state_variable.value
+ in (
+ TransportState.PLAYING,
+ TransportState.PAUSED_PLAYBACK,
+ )
):
dlna_player.force_poll = True
self.mass.create_task(self.poll_player(dlna_player.udn))
"""Set Player Features based on config values and capabilities."""
dlna_player.player.supported_features = BASE_PLAYER_FEATURES
player_id = dlna_player.player.player_id
- if self.mass.config.get_raw_player_config_value(player_id, CONF_ENQUEUE_NEXT, False):
+ if self.mass.config.get_raw_player_config_value(
+ player_id, CONF_ENQUEUE_NEXT, False
+ ):
dlna_player.player.supported_features = (
*dlna_player.player.supported_features,
PlayerFeature.ENQUEUE_NEXT,
projects / music-assistant-server.git / commitdiff