319 lines
9.4 KiB
Python
319 lines
9.4 KiB
Python
# Standard library imports
|
|
from dataclasses import dataclass, field
|
|
from enum import auto, Enum
|
|
from typing import Any, Optional
|
|
import datetime as dt
|
|
|
|
# PyQt imports
|
|
from PyQt6.QtCore import pyqtSignal, QObject, QThread
|
|
|
|
# Third party imports
|
|
import numpy as np
|
|
import pyqtgraph as pg # type: ignore
|
|
|
|
# App imports
|
|
from config import Config
|
|
from log import log
|
|
from models import db, PlaylistRows, Tracks
|
|
from music import Music
|
|
import helpers
|
|
|
|
|
|
class Col(Enum):
|
|
START_GAP = 0
|
|
TITLE = auto()
|
|
ARTIST = auto()
|
|
INTRO = auto()
|
|
DURATION = auto()
|
|
START_TIME = auto()
|
|
END_TIME = auto()
|
|
LAST_PLAYED = auto()
|
|
BITRATE = auto()
|
|
NOTE = auto()
|
|
|
|
|
|
class FadeCurve:
|
|
GraphWidget = None
|
|
|
|
def __init__(
|
|
self, track_path: str, track_fade_at: int, track_silence_at: int
|
|
) -> None:
|
|
"""
|
|
Set up fade graph array
|
|
"""
|
|
|
|
audio = helpers.get_audio_segment(track_path)
|
|
if not audio:
|
|
log.error(f"FadeCurve: could not get audio for {track_path=}")
|
|
return None
|
|
|
|
# Start point of curve is Config.FADE_CURVE_MS_BEFORE_FADE
|
|
# milliseconds before fade starts to silence
|
|
self.start_ms = max(0, track_fade_at - Config.FADE_CURVE_MS_BEFORE_FADE - 1)
|
|
self.end_ms = track_silence_at
|
|
self.audio_segment = audio[self.start_ms : self.end_ms]
|
|
self.graph_array = np.array(self.audio_segment.get_array_of_samples())
|
|
|
|
# Calculate the factor to map milliseconds of track to array
|
|
self.ms_to_array_factor = len(self.graph_array) / (self.end_ms - self.start_ms)
|
|
|
|
self.region = None
|
|
|
|
def clear(self) -> None:
|
|
"""Clear the current graph"""
|
|
|
|
if self.GraphWidget:
|
|
self.GraphWidget.clear()
|
|
|
|
def plot(self):
|
|
self.curve = self.GraphWidget.plot(self.graph_array)
|
|
self.curve.setPen(Config.FADE_CURVE_FOREGROUND)
|
|
|
|
def tick(self, play_time) -> None:
|
|
"""Update volume fade curve"""
|
|
|
|
if not self.GraphWidget:
|
|
return
|
|
|
|
ms_of_graph = play_time - self.start_ms
|
|
if ms_of_graph < 0:
|
|
return
|
|
|
|
if self.region is None:
|
|
# Create the region now that we're into fade
|
|
self.region = pg.LinearRegionItem([0, 0], bounds=[0, len(self.graph_array)])
|
|
self.GraphWidget.addItem(self.region)
|
|
|
|
# Update region position
|
|
self.region.setRegion([0, ms_of_graph * self.ms_to_array_factor])
|
|
|
|
|
|
@helpers.singleton
|
|
@dataclass
|
|
class MusicMusterSignals(QObject):
|
|
"""
|
|
Class for all MusicMuster signals. See:
|
|
- https://zetcode.com/gui/pyqt5/eventssignals/
|
|
- https://stackoverflow.com/questions/62654525/
|
|
emit-a-signal-from-another-class-to-main-class
|
|
and Singleton class at
|
|
https://refactoring.guru/design-patterns/singleton/python/example#example-0
|
|
"""
|
|
|
|
begin_reset_model_signal = pyqtSignal(int)
|
|
enable_escape_signal = pyqtSignal(bool)
|
|
end_reset_model_signal = pyqtSignal(int)
|
|
next_track_changed_signal = pyqtSignal()
|
|
resize_rows_signal = pyqtSignal(int)
|
|
row_order_changed_signal = pyqtSignal(int)
|
|
search_songfacts_signal = pyqtSignal(str)
|
|
search_wikipedia_signal = pyqtSignal(str)
|
|
show_warning_signal = pyqtSignal(str, str)
|
|
span_cells_signal = pyqtSignal(int, int, int, int, int)
|
|
status_message_signal = pyqtSignal(str, int)
|
|
|
|
def __post_init__(self):
|
|
super().__init__()
|
|
|
|
|
|
class _TrackPlayer:
|
|
"""
|
|
Object to manage active playlist tracks,
|
|
typically the previous, current and next track.
|
|
"""
|
|
|
|
def __init__(self, session: db.Session, player_name: str, track_id: int) -> None:
|
|
"""
|
|
Initialises data structure.
|
|
Define a player.
|
|
Raise ValueError if no track in passed plr.
|
|
"""
|
|
|
|
track = session.get(Tracks, track_id)
|
|
if not track:
|
|
raise ValueError(f"_TrackPlayer: unable to retreived {track_id=}")
|
|
self.player_name = player_name
|
|
|
|
self.artist = track.artist
|
|
self.bitrate = track.bitrate
|
|
self.duration = track.duration
|
|
self.fade_at = track.fade_at
|
|
self.intro = track.intro
|
|
self.path = track.path
|
|
self.silence_at = track.silence_at
|
|
self.start_gap = track.start_gap
|
|
self.title = track.title
|
|
self.track_id = track.id
|
|
|
|
self.end_time: Optional[dt.datetime] = None
|
|
self.fade_graph: Optional[FadeCurve] = None
|
|
self.fade_graph_start_updates: Optional[dt.datetime] = None
|
|
self.resume_marker: Optional[float]
|
|
self.start_time: Optional[dt.datetime] = None
|
|
|
|
self.player = Music(name=player_name)
|
|
|
|
# Initialise and add FadeCurve in a thread as it's slow
|
|
self.fadecurve_thread = QThread()
|
|
self.worker = AddFadeCurve(
|
|
self,
|
|
track_path=track.path,
|
|
track_fade_at=track.fade_at,
|
|
track_silence_at=track.silence_at,
|
|
)
|
|
self.worker.moveToThread(self.fadecurve_thread)
|
|
self.fadecurve_thread.started.connect(self.worker.run)
|
|
self.worker.finished.connect(self.fadecurve_thread.quit)
|
|
self.worker.finished.connect(self.worker.deleteLater)
|
|
self.fadecurve_thread.finished.connect(self.fadecurve_thread.deleteLater)
|
|
self.fadecurve_thread.start()
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<_TrackPlayer(title={self.title}, artist={self.artist}, "
|
|
f"player_name={self.player_name}>"
|
|
)
|
|
|
|
def fade(self, fade_seconds: int = Config.FADEOUT_SECONDS) -> None:
|
|
"""Fade music"""
|
|
|
|
self.player.fade(fade_seconds)
|
|
|
|
def play(self, position: Optional[float] = None) -> None:
|
|
"""Play track"""
|
|
|
|
now = dt.datetime.now()
|
|
self.start_time = now
|
|
self.player.play(self.path, position)
|
|
|
|
self.end_time = self.start_time + dt.timedelta(milliseconds=self.duration)
|
|
|
|
# Calculate time fade_graph should start updating
|
|
if self.fade_at:
|
|
update_graph_at_ms = max(
|
|
0, self.fade_at - Config.FADE_CURVE_MS_BEFORE_FADE - 1
|
|
)
|
|
self.fade_graph_start_updates = now + dt.timedelta(
|
|
milliseconds=update_graph_at_ms
|
|
)
|
|
|
|
def stop_playing(self, fade_seconds: int = 0) -> None:
|
|
"""
|
|
Stop this track playing
|
|
"""
|
|
|
|
self.resume_marker = self.player.get_position()
|
|
self.player.fade(fade_seconds)
|
|
|
|
# Reset fade graph
|
|
if self.fade_graph:
|
|
self.fade_graph.clear()
|
|
|
|
|
|
class MainTrackPlayer(_TrackPlayer):
|
|
def __init__(self, session: db.Session, track_id: int) -> None:
|
|
super().__init__(
|
|
session=session, player_name=Config.VLC_MAIN_PLAYER_NAME, track_id=track_id
|
|
)
|
|
|
|
|
|
class PreviewTrackPlayer(_TrackPlayer):
|
|
def __init__(self, session: db.Session, track_id: int) -> None:
|
|
super().__init__(
|
|
session=session,
|
|
player_name=Config.VLC_PREVIEW_PLAYER_NAME,
|
|
track_id=track_id,
|
|
)
|
|
|
|
|
|
class PlaylistTrack:
|
|
"""
|
|
Used to provide a single reference point for specific playlist tracks,
|
|
typically the previous, current and next track.
|
|
"""
|
|
|
|
def __init__(self, plrid: int) -> None:
|
|
"""
|
|
Initialise
|
|
"""
|
|
|
|
with db.Session() as session:
|
|
# Ensure we have a track
|
|
plr = session.get(PlaylistRows, plrid)
|
|
if not plr:
|
|
raise ValueError(f"PlaylistTrack: unable to retreive plr {plrid=}")
|
|
|
|
self.track_id: int = plr.track_id
|
|
|
|
# Save non-track plr info
|
|
self.row_number: int = plr.plr_rownum
|
|
self.playlist_id: int = plr.playlist_id
|
|
self.plr_id: int = plr.id
|
|
|
|
# Initialise player
|
|
self.track_player = MainTrackPlayer(session=session, track_id=self.track_id)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<PlaylistTrack(title={self.track_player.title}, "
|
|
f"artist={self.track_player.artist}, "
|
|
f"plr_rownum={self.row_number}, playlist_id={self.playlist_id}>"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class TrackFileData:
|
|
"""
|
|
Simple class to track details changes to a track file
|
|
"""
|
|
|
|
new_file_path: str
|
|
track_id: int = 0
|
|
track_path: Optional[str] = None
|
|
obsolete_path: Optional[str] = None
|
|
tags: dict[str, Any] = field(default_factory=dict)
|
|
audio_metadata: dict[str, str | int | float] = field(default_factory=dict)
|
|
|
|
|
|
class AddFadeCurve(QObject):
|
|
"""
|
|
Initialising a fade curve introduces a noticeable delay so carry out in
|
|
a thread.
|
|
"""
|
|
|
|
finished = pyqtSignal()
|
|
|
|
def __init__(
|
|
self,
|
|
track_player: _TrackPlayer,
|
|
track_path: str,
|
|
track_fade_at: int,
|
|
track_silence_at: int,
|
|
):
|
|
super().__init__()
|
|
self.track_player = track_player
|
|
self.track_path = track_path
|
|
self.track_fade_at = track_fade_at
|
|
self.track_silence_at = track_silence_at
|
|
|
|
def run(self):
|
|
"""
|
|
Create fade curve and add to PlaylistTrack object
|
|
"""
|
|
|
|
fc = FadeCurve(self.track_path, self.track_fade_at, self.track_silence_at)
|
|
if not fc:
|
|
log.error(f"Failed to create FadeCurve for {self.track_path=}")
|
|
else:
|
|
self.track_player.fade_graph = fc
|
|
self.finished.emit()
|
|
|
|
|
|
class TrackSequence:
|
|
next: Optional[PlaylistTrack] = None
|
|
current: Optional[PlaylistTrack] = None
|
|
previous: Optional[PlaylistTrack] = None
|
|
|
|
|
|
track_sequence = TrackSequence()
|