Trixy7/trixy_core/music/sources/base.py

# -*- coding: utf-8 -*-
"""
Basis-Klassen für Musikquellen.

Definiert die abstrakte Schnittstelle für alle Musikquellen.
"""

import asyncio
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Any, AsyncIterator

from trixy_core.music.track import Track, TrackMetadata


class SourceType(Enum):
    """Typ der Musikquelle."""
    LOCAL = "local"           # Lokale Dateien
    USB = "usb"               # USB-Speicher
    URL = "url"               # HTTP/HTTPS URLs
    SPOTIFY = "spotify"       # Spotify (zukünftig)
    AMAZON = "amazon"         # Amazon Music (zukünftig)
    YOUTUBE = "youtube"       # YouTube Music (zukünftig)
    WEBDAV = "webdav"         # WebDAV (zukünftig)
    SMB = "smb"               # SMB/CIFS Share (zukünftig)


class SourceState(Enum):
    """Zustand einer Musikquelle."""
    UNKNOWN = "unknown"       # Unbekannt
    AVAILABLE = "available"   # Verfügbar
    SCANNING = "scanning"     # Wird gescannt
    UNAVAILABLE = "unavailable"  # Nicht verfügbar
    ERROR = "error"           # Fehler
    MOUNTING = "mounting"     # Wird eingebunden
    UNMOUNTING = "unmounting"  # Wird ausgehängt


@dataclass
class SourceInfo:
    """Informationen über eine Musikquelle."""
    source_type: SourceType
    name: str
    description: str = ""
    path: Path | None = None
    uri: str = ""
    state: SourceState = SourceState.UNKNOWN

    # Statistiken
    track_count: int = 0
    total_size_bytes: int = 0
    last_scan: datetime | None = None

    # Capabilities
    supports_search: bool = True
    supports_metadata: bool = True
    supports_artwork: bool = True
    supports_streaming: bool = True

    # Zusätzliche Daten
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "source_type": self.source_type.value,
            "name": self.name,
            "description": self.description,
            "path": str(self.path) if self.path else None,
            "uri": self.uri,
            "state": self.state.value,
            "track_count": self.track_count,
            "total_size_bytes": self.total_size_bytes,
            "last_scan": self.last_scan.isoformat() if self.last_scan else None,
            "supports_search": self.supports_search,
            "supports_metadata": self.supports_metadata,
            "supports_artwork": self.supports_artwork,
            "supports_streaming": self.supports_streaming,
        }


class MusicSource(ABC):
    """
    Abstrakte Basis-Klasse für Musikquellen.

    Definiert die Schnittstelle für alle Arten von Musikquellen:
    - Lokale Dateien
    - USB-Speicher
    - HTTP/HTTPS URLs
    - Streaming-Dienste (Spotify, Amazon, etc.)
    """

    def __init__(
        self,
        name: str,
        source_type: SourceType,
    ) -> None:
        """
        Initialisiert die Musikquelle.

        Args:
            name: Name der Quelle
            source_type: Typ der Quelle
        """
        self.name = name
        self.source_type = source_type
        self._state = SourceState.UNKNOWN
        self._tracks: dict[str, Track] = {}
        self._scanning = False
        self._scan_lock = asyncio.Lock()

    @property
    def state(self) -> SourceState:
        """Aktueller Zustand."""
        return self._state

    @property
    def track_count(self) -> int:
        """Anzahl der gefundenen Tracks."""
        return len(self._tracks)

    @property
    def tracks(self) -> list[Track]:
        """Alle gefundenen Tracks."""
        return list(self._tracks.values())

    @property
    def is_available(self) -> bool:
        """Prüft ob Quelle verfügbar ist."""
        return self._state == SourceState.AVAILABLE

    @property
    def is_scanning(self) -> bool:
        """Prüft ob gerade gescannt wird."""
        return self._scanning

    # ==========================================================================
    # Abstrakte Methoden
    # ==========================================================================

    @abstractmethod
    async def connect(self) -> bool:
        """
        Verbindet zur Quelle.

        Returns:
            True wenn erfolgreich
        """
        pass

    @abstractmethod
    async def disconnect(self) -> None:
        """Trennt die Verbindung zur Quelle."""
        pass

    @abstractmethod
    async def scan(self) -> int:
        """
        Scannt die Quelle nach Tracks.

        Returns:
            Anzahl der gefundenen Tracks
        """
        pass

    @abstractmethod
    async def get_track(self, track_id: str) -> Track | None:
        """
        Holt Track nach ID.

        Args:
            track_id: Track-ID

        Returns:
            Track oder None
        """
        pass

    @abstractmethod
    async def get_audio_stream(self, track: Track) -> AsyncIterator[bytes]:
        """
        Liefert Audio-Daten als Stream.

        Args:
            track: Track zum Streamen

        Yields:
            Audio-Daten in Chunks
        """
        pass

    @abstractmethod
    async def get_metadata(self, track: Track) -> TrackMetadata | None:
        """
        Lädt Metadaten für einen Track.

        Args:
            track: Track

        Returns:
            Metadaten oder None
        """
        pass

    # ==========================================================================
    # Optionale Methoden (mit Default-Implementierung)
    # ==========================================================================

    async def search(
        self,
        query: str,
        limit: int = 50,
    ) -> list[Track]:
        """
        Sucht nach Tracks.

        Args:
            query: Suchstring
            limit: Maximale Ergebnisse

        Returns:
            Liste der gefundenen Tracks
        """
        query_lower = query.lower()
        results = []

        for track in self._tracks.values():
            if (query_lower in track.title.lower() or
                query_lower in track.artist.lower() or
                query_lower in track.metadata.album.lower()):
                results.append(track)
                if len(results) >= limit:
                    break

        return results

    async def get_artwork(self, track: Track) -> bytes | None:
        """
        Lädt Artwork für einen Track.

        Args:
            track: Track

        Returns:
            Artwork-Daten (JPEG/PNG) oder None
        """
        if track.metadata.artwork_data:
            return track.metadata.artwork_data
        return None

    async def refresh(self) -> None:
        """Aktualisiert die Track-Liste."""
        await self.scan()

    def get_info(self) -> SourceInfo:
        """Liefert Informationen über die Quelle."""
        return SourceInfo(
            source_type=self.source_type,
            name=self.name,
            state=self._state,
            track_count=self.track_count,
        )

    # ==========================================================================
    # Hilfsmethoden
    # ==========================================================================

    def _add_track(self, track: Track) -> None:
        """Fügt Track zur internen Liste hinzu."""
        self._tracks[track.id] = track

    def _remove_track(self, track_id: str) -> Track | None:
        """Entfernt Track aus interner Liste."""
        return self._tracks.pop(track_id, None)

    def _clear_tracks(self) -> None:
        """Leert die Track-Liste."""
        self._tracks.clear()

    def __repr__(self) -> str:
        return (
            f"{self.__class__.__name__}("
            f"name={self.name!r}, "
            f"state={self._state.value!r}, "
            f"tracks={self.track_count})"
        )