"""Recursively yield DirEntry objects for given directory."""
for entry in os.scandir(path):
if entry.is_dir(follow_symlinks=False):
- yield from scantree(entry.path) # see below for Python 2.x
+ yield from scantree(entry.path)
else:
yield entry
"""
Implementation of a musicprovider for local files.
- Assumes files are stored on disk in format <artist>/<album>/<track.ext>
- Reads ID3 tags from file and falls back to parsing filename
- Supports m3u files only for playlists
- Supports having URI's from streaming providers within m3u playlist
+ Reads ID3 tags from file and falls back to parsing filename.
+ Optionally reads metadata from nfo files and images in folder structure <artist>/<album>.
+ Supports m3u files only for playlists.
+ Supports having URI's from streaming providers within m3u playlist.
"""
_attr_name = "Filesystem"
self, playlist_path: str, checksum: Optional[str] = None
) -> Playlist | None:
"""Parse playlist from file."""
- if self.config.path not in playlist_path:
- playlist_path = os.path.join(self.config.path, playlist_path)
+ playlist_path = self._get_absolute_path(playlist_path)
playlist_path_base = self._get_relative_path(playlist_path)
playlist_item_id = self._get_item_id(playlist_path_base)
checksum = checksum or self._get_checksum(playlist_path)
def _get_relative_path(self, filename: str) -> str:
"""Get relative path for filename (without the base dir)."""
+ if self.config.path not in filename:
+ return filename
filename = filename.replace(self.config.path, "")
if filename.startswith(os.sep):
filename = filename[1:]
def _get_absolute_path(self, filename: str) -> str:
"""Get absolute path for filename (including the base dir)."""
+ if self.config.path in filename:
+ return filename
return os.path.join(self.config.path, filename)
def _get_item_id(self, filename: str) -> str:
await mass.music.mark_item_played(
streamdetails.item_id, streamdetails.provider
)
+ # send analyze job to background worker
+ if streamdetails.loudness is None and streamdetails.provider != "url":
+ uri = f"{streamdetails.provider}://{streamdetails.media_type.value}/{streamdetails.item_id}"
+ mass.add_job(
+ analyze_audio(mass, streamdetails), f"Analyze audio for {uri}"
+ )
finally:
mass.signal_event(
MassEvent(
data=streamdetails,
)
)
- # send analyze job to background worker
- if streamdetails.loudness is None and streamdetails.provider != "url":
- uri = f"{streamdetails.provider}://{streamdetails.media_type.value}/{streamdetails.item_id}"
- mass.add_job(
- analyze_audio(mass, streamdetails), f"Analyze audio for {uri}"
- )
async def check_audio_support(try_install: bool = False) -> Tuple[bool, bool, bool]:
from typing import TYPE_CHECKING, Optional
from PIL import Image
-from tinytag import TinyTag
if TYPE_CHECKING:
from music_assistant.mass import MusicAssistant
continue
if not prov.exists(path):
continue
- if TinyTag.is_supported(path):
- # embedded image in music file
- def get_embedded_image():
- tags = TinyTag.get(path, image=True)
- return tags.get_image()
-
- img_data = await mass.loop.run_in_executor(None, get_embedded_image)
- else:
- # regular image file on disk
+ # embedded image in music file
+ img_data = await prov.get_embedded_image(path)
+ # regular image file on disk
+ if not img_data:
async with prov.open_file(path) as _file:
img_data = await _file.read()
break