always_run: true
args:
- --branch=main
- # - id: pylint
- # name: 🌟 Starring code with pylint
- # language: system
- # types: [python]
- # entry: scripts/run-in-env.sh pylint
- # - id: trailing-whitespace
- # name: ✄ Trim Trailing Whitespace
- # language: system
- # types: [text]
- # entry: scripts/run-in-env.sh trailing-whitespace-fixer
- # stages: [commit, push, manual]
+ - id: trailing-whitespace
+ name: ✄ Trim Trailing Whitespace
+ language: system
+ types: [text]
+ entry: scripts/run-in-env.sh trailing-whitespace-fixer
+ stages: [commit, push, manual]
- id: mypy
name: mypy
entry: scripts/run-in-env.sh mypy
--no-cache \
--find-links "https://wheels.home-assistant.io/musllinux/" \
"music-assistant[server]@dist/music_assistant-${MASS_VERSION}-py3-none-any.whl"
-
+
# Set some labels
LABEL \
org.opencontainers.image.title="Music Assistant Server" \
io.hass.platform="${TARGETPLATFORM}" \
io.hass.type="addon"
-RUN rm -rf dist
+RUN rm -rf dist
VOLUME [ "/data" ]
EXPOSE 8095
def _enable_posix_spawn() -> None:
"""Enable posix_spawn on Alpine Linux."""
- # pylint: disable=protected-access
if subprocess._USE_POSIX_SPAWN:
return
from urllib.parse import urlparse
from uuid import UUID
-# pylint: disable=invalid-name
T = TypeVar("T")
CALLBACK_TYPE = Callable[[], None]
-# pylint: enable=invalid-name
keyword_pattern = re.compile("title=|artist=")
title_pattern = re.compile(r"title=\"(?P<title>.*?)\"")
def _get_ip() -> str:
"""Get primary IP-address for this host."""
- # pylint: disable=broad-except,no-member
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
# doesn't even have to be reachable
def _resolve() -> str | None:
try:
return socket.gethostbyname(dns_name)
- except Exception: # pylint: disable=broad-except
+ except Exception:
# fail gracefully!
return None
"""Return socket pton for local ip."""
if ip_string is None:
ip_string = await get_ip()
- # pylint:disable=no-member
try:
return await asyncio.to_thread(socket.inet_pton, socket.AF_INET, ip_string)
except OSError:
def get_folder_size(folderpath: str) -> float:
"""Return folder size in gb."""
total_size = 0
- # pylint: disable=unused-variable
for dirpath, _dirnames, filenames in os.walk(folderpath):
for _file in filenames:
_fp = os.path.join(dirpath, _file)
total_size += os.path.getsize(_fp)
- # pylint: enable=unused-variable
return total_size / float(1 << 30)
from music_assistant.common.helpers.json import get_serializable_value
from music_assistant.common.models.event import MassEvent
-# pylint: disable=unnecessary-lambda
-
@dataclass
class CommandMessage(DataClassORJSONMixin):
object_id: str | None = None # player_id, queue_id or uri
data: Any = field(
default=None,
- metadata={
- "serialize": lambda v: get_serializable_value(v) # pylint: disable=unnecessary-lambda
- },
+ metadata={"serialize": lambda v: get_serializable_value(v)},
)
) and (not checksum or db_row["checksum"] == checksum and db_row["expires"] >= cur_time):
try:
data = await asyncio.to_thread(json_loads, db_row["data"])
- except Exception as exc: # pylint: disable=broad-except
+ except Exception as exc:
LOGGER.error(
"Error parsing cache data for %s: %s",
memory_key,
return
except FileNotFoundError:
pass
- except JSON_DECODE_EXCEPTIONS: # pylint: disable=catching-non-exception
+ except JSON_DECODE_EXCEPTIONS:
LOGGER.exception("Error while reading persistent storage file %s", filename)
LOGGER.debug("Started with empty storage: No persistent storage file found.")
player.available = False
player.state = PlayerState.IDLE
player.powered = False
- except Exception as err: # pylint: disable=broad-except
+ except Exception as err:
self.logger.warning(
"Error while requesting latest state from player %s: %s",
player.display_name,
FLOW_DEFAULT_BIT_DEPTH = 24
-# pylint:disable=too-many-locals
-
isfile = wrap(os.path.isfile)
except asyncio.CancelledError:
self._logger.debug("Connection closed by client")
- except Exception: # pylint: disable=broad-except
+ except Exception:
self._logger.exception("Unexpected error inside websocket API")
finally:
elif asyncio.iscoroutine(result):
result = await result
self._send_message(SuccessResultMessage(msg.message_id, result))
- except Exception as err: # pylint: disable=broad-except
+ except Exception as err:
if self._logger.isEnabledFor(logging.DEBUG):
self._logger.exception("Error handling message: %s", msg)
else:
logging.getLogger(__name__).warning(err)
return None
if origin is type:
- return eval(value) # pylint: disable=eval-used
+ return eval(value)
if value_type is Any:
return value
if value is None and value_type is not NoneType:
-# pylint: skip-file
# fmt: off
# flake8: noqa
# ruff: noqa
from music_assistant.server import MusicAssistant
LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.audio")
-# pylint:disable=consider-using-f-string,too-many-locals,too-many-statements
+
# ruff: noqa: PLR0915
HTTP_HEADERS = {"User-Agent": "Lavf/60.16.100.MusicAssistant"}
def create_wave_header(samplerate=44100, channels=2, bitspersample=16, duration=None):
"""Generate a wave header from given params."""
- # pylint: disable=no-member
file = BytesIO()
# Generate format chunk
except UnidentifiedImageError:
raise FileNotFoundError(f"Invalid image: {path_or_url}")
if size:
- img.thumbnail((size, size), Image.LANCZOS) # pylint: disable=no-member
+ img.thumbnail((size, size), Image.LANCZOS)
mode = "RGBA" if image_format == "PNG" else "RGB"
img.convert(mode).save(data, image_format, optimize=True)
"""Catch and log exception."""
try:
await async_func(*args)
- except Exception: # pylint: disable=broad-except
+ except Exception:
log_exception(format_err, *args)
wrapper_func = async_wrapper
"""Catch and log exception."""
try:
func(*args)
- except Exception: # pylint: disable=broad-except
+ except Exception:
log_exception(format_err, *args)
wrapper_func = wrapper
"""Catch and log exception."""
try:
return await target
- except Exception: # pylint: disable=broad-except
+ except Exception:
log_exception(format_err, *args)
return None
DEFAULT_CHUNKSIZE = 64000
-# pylint: disable=invalid-name
-
class AsyncProcess:
"""
msg = "Dynamic routes are not enabled"
raise RuntimeError(msg)
key = f"{method}.{path}"
- if key in self._dynamic_routes: # pylint: disable=unsupported-membership-test
+ if key in self._dynamic_routes:
msg = f"Route {path} already registered."
raise RuntimeError(msg)
self._dynamic_routes[key] = handler
@property
def players(self) -> list[Player]:
"""Return all players belonging to this provider."""
- # pylint: disable=no-member
return [
player
for player in self.mass.players
)
from music_assistant.common.models.streamdetails import StreamDetails
from music_assistant.constants import CONF_PASSWORD
-
-# pylint: disable=no-name-in-module
from music_assistant.server.helpers.app_vars import app_var
from music_assistant.server.helpers.audio import get_hls_substream
from music_assistant.server.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
All following callbacks won't be forwarded.
"""
- # pylint: disable=protected-access
if self.castplayer.cast_info.is_audio_group:
self._mz_mgr.remove_multizone(self._uuid)
else:
)
from music_assistant.common.models.provider import ProviderManifest
from music_assistant.common.models.streamdetails import StreamDetails
-
-# pylint: disable=no-name-in-module
from music_assistant.server.helpers.app_vars import app_var
-
-# pylint: enable=no-name-in-module
from music_assistant.server.helpers.auth import AuthenticationHelper
from music_assistant.server.models import ProviderInstanceType
from music_assistant.server.models.music_provider import MusicProvider
async def get_config_entries(
mass: MusicAssistant,
- instance_id: str | None = None, # noqa: ARG001 pylint: disable=W0613
+ instance_id: str | None = None, # noqa: ARG001
action: str | None = None,
values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
)
-class DeezerProvider(MusicProvider): # pylint: disable=W0223
+class DeezerProvider(MusicProvider):
"""Deezer provider support."""
client: deezer.Client
from music_assistant.common.models.enums import ConfigEntryType, ExternalID, ProviderFeature
from music_assistant.common.models.media_items import ImageType, MediaItemImage, MediaItemMetadata
from music_assistant.server.controllers.cache import use_cache
-from music_assistant.server.helpers.app_vars import app_var # pylint: disable=no-name-in-module
+from music_assistant.server.helpers.app_vars import app_var
from music_assistant.server.helpers.throttle_retry import Throttler
from music_assistant.server.models.metadata_provider import MetadataProvider
playlist,
overwrite_existing=prev_checksum is not None,
)
- except Exception as err: # pylint: disable=broad-except
+ except Exception as err:
# we don't want the whole sync to crash on one file so we catch all exceptions here
self.logger.error(
"Error processing %s - %s",
track.position = idx
result.append(track)
- except Exception as err: # pylint: disable=broad-except
+ except Exception as err:
self.logger.warning(
"Error while parsing playlist %s: %s",
prov_playlist_id,
async def get_config_entries(
mass: MusicAssistant,
- instance_id: str | None = None, # pylint: disable=W0613
+ instance_id: str | None = None,
action: str | None = None,
values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
VARIOUS_ARTISTS_MBID,
VARIOUS_ARTISTS_NAME,
)
-
-# pylint: disable=no-name-in-module
from music_assistant.server.helpers.app_vars import app_var
-
-# pylint: enable=no-name-in-module
from music_assistant.server.helpers.throttle_retry import ThrottlerManager, throttle_with_retries
from music_assistant.server.helpers.util import lock
from music_assistant.server.models.music_provider import MusicProvider
async def _parse_track(self, track_obj: dict) -> Track:
"""Parse qobuz track object to generic layout."""
- # pylint: disable=too-many-branches
name, version = parse_title_and_version(track_obj["title"], track_obj.get("version"))
track = Track(
item_id=str(track_obj["id"]),
@throttle_with_retries
async def _get_data(self, endpoint, sign_request=False, **kwargs):
"""Get data from api."""
- # pylint: disable=too-many-branches
self.logger.debug("Handling GET request to %s", endpoint)
url = f"http://www.qobuz.com/api.json/0.2/{endpoint}"
headers = {"X-App-Id": app_var(0)}
if soco := getattr(instance, "soco", fallback_soco):
# Holds a SoCo instance attribute
# Only use attributes with no I/O
- return soco._player_name or soco.ip_address # pylint: disable=protected-access
+ return soco._player_name or soco.ip_address
return None
Track,
)
from music_assistant.common.models.streamdetails import StreamDetails
-
-# pylint: disable=no-name-in-module
from music_assistant.constants import VERBOSE_LOG_LEVEL
from music_assistant.server.helpers.app_vars import app_var
-
-# pylint: enable=no-name-in-module
from music_assistant.server.helpers.audio import get_chunksize
from music_assistant.server.helpers.auth import AuthenticationHelper
from music_assistant.server.helpers.process import AsyncProcess, check_output
try:
await self.load_provider_config(prov_conf)
- # pylint: disable=broad-except
except Exception as exc:
# if loading failed, we store the error in the config object
# so we can show something useful to the user
provider_manifest.icon_svg_dark = await get_icon_string(icon_path)
self._provider_manifests[provider_manifest.domain] = provider_manifest
LOGGER.debug("Loaded manifest for provider %s", provider_manifest.name)
- except Exception as exc: # pylint: disable=broad-except
+ except Exception as exc:
LOGGER.exception(
"Error while loading manifest for provider %s",
provider_domain,
def _on_mdns_service_state_change(
self,
- zeroconf: Zeroconf, # pylint: disable=unused-argument
+ zeroconf: Zeroconf,
service_type: str,
name: str,
state_change: ServiceStateChange,
"mypy==1.11.2",
"pre-commit==3.8.0",
"pre-commit-hooks==4.6.0",
- "pylint==3.2.7",
"pytest==8.3.3",
"pytest-aiohttp==1.0.5",
"pytest-cov==5.0.0",
# Force Linux/MacOS line endings
line-ending = "lf"
-[tool.pylint.MASTER]
-extension-pkg-whitelist = ["orjson"]
-ignore = ["tests"]
-
-[tool.pylint.BASIC]
-good-names = ["_", "id", "on", "Run", "T"]
-
-[tool.pylint.DESIGN]
-max-attributes = 8
-
-[tool.pylint."MESSAGES CONTROL"]
-disable = [
- "duplicate-code",
- "format",
- "unsubscriptable-object",
- "unused-argument", # handled by ruff
- "unspecified-encoding", # handled by ruff
- "isinstance-second-argument-not-valid-type", # conflict with ruff
- "fixme", # we're still developing # TEMPORARY DISABLED rules # The below rules must be enabled later one-by-one !
- "too-many-return-statements",
- "unsupported-assignment-operation",
- "invalid-name",
- "redefined-outer-name",
- "too-many-statements",
- "deprecated-method",
- "logging-fstring-interpolation",
- "attribute-defined-outside-init",
- "broad-exception-caught",
- "expression-not-assigned",
- "consider-using-f-string",
- "consider-using-with",
- "arguments-renamed",
- "protected-access",
- "too-many-boolean-expressions",
- "raise-missing-from",
- "too-many-locals",
- "abstract-method",
- "unnecessary-lambda",
- "stop-iteration-return",
- "no-else-return",
- "no-else-raise",
- "undefined-loop-variable",
- "too-many-nested-blocks",
- "too-many-public-methods", # unavoidable?
- "too-many-arguments", # unavoidable?
- "too-many-branches", # unavoidable?
- "too-many-instance-attributes", # unavoidable?
-]
-
-[tool.pylint.SIMILARITIES]
-ignore-imports = true
-
-[tool.pylint.FORMAT]
-max-line-length = 100
-
[tool.pytest.ini_options]
addopts = "--cov music_assistant"
asyncio_mode = "auto"
projects / music-assistant-server.git / commitdiff