229 lines
6.9 KiB
Python
229 lines
6.9 KiB
Python
# Standard library imports
|
|
from dataclasses import dataclass
|
|
from typing import 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
|
|
from sqlalchemy.orm import scoped_session
|
|
|
|
# App imports
|
|
from config import Config
|
|
from models import PlaylistRows
|
|
import helpers
|
|
|
|
|
|
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:
|
|
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 PlaylistTrack:
|
|
"""
|
|
Used to provide a single reference point for specific playlist tracks,
|
|
typically the previous, current and next track.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
Only initialises data structure. Call set_plr to populate.
|
|
"""
|
|
|
|
self.artist: Optional[str] = None
|
|
self.duration: Optional[int] = None
|
|
self.end_time: Optional[dt.datetime] = None
|
|
self.fade_at: Optional[int] = None
|
|
self.fade_graph: Optional[FadeCurve] = None
|
|
self.fade_length: Optional[int] = None
|
|
self.path: Optional[str] = None
|
|
self.playlist_id: Optional[int] = None
|
|
self.plr_id: Optional[int] = None
|
|
self.plr_rownum: Optional[int] = None
|
|
self.resume_marker: Optional[float] = None
|
|
self.silence_at: Optional[int] = None
|
|
self.start_gap: Optional[int] = None
|
|
self.start_time: Optional[dt.datetime] = None
|
|
self.title: Optional[str] = None
|
|
self.track_id: Optional[int] = None
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<PlaylistTrack(title={self.title}, artist={self.artist}, "
|
|
f"plr_rownum={self.plr_rownum}, playlist_id={self.playlist_id}>"
|
|
)
|
|
|
|
def set_plr(self, session: scoped_session, plr: PlaylistRows) -> None:
|
|
"""
|
|
Update with new plr information
|
|
"""
|
|
|
|
session.add(plr)
|
|
self.plr_rownum = plr.plr_rownum
|
|
if not plr.track:
|
|
return
|
|
track = plr.track
|
|
|
|
self.artist = track.artist
|
|
self.duration = track.duration
|
|
self.end_time = None
|
|
self.fade_at = track.fade_at
|
|
self.path = track.path
|
|
self.playlist_id = plr.playlist_id
|
|
self.plr_id = plr.id
|
|
self.silence_at = track.silence_at
|
|
self.start_gap = track.start_gap
|
|
self.start_time = None
|
|
self.title = track.title
|
|
self.track_id = track.id
|
|
|
|
if track.silence_at and track.fade_at:
|
|
self.fade_length = track.silence_at - track.fade_at
|
|
|
|
# Initialise and add FadeCurve in a thread as it's slow
|
|
# Import in separate thread
|
|
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 start(self) -> None:
|
|
"""
|
|
Called when track starts playing
|
|
"""
|
|
|
|
self.start_time = dt.datetime.now()
|
|
if self.duration:
|
|
self.end_time = self.start_time + dt.timedelta(milliseconds=self.duration)
|
|
|
|
|
|
class AddFadeCurve(QObject):
|
|
"""
|
|
Initialising a fade curve introduces a noticeable delay so carry out in
|
|
a thread.
|
|
"""
|
|
|
|
finished = pyqtSignal()
|
|
|
|
def __init__(
|
|
self,
|
|
playlist_track: PlaylistTrack,
|
|
track_path: str,
|
|
track_fade_at: int,
|
|
track_silence_at: int,
|
|
):
|
|
super().__init__()
|
|
self.playlist_track = playlist_track
|
|
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
|
|
"""
|
|
|
|
self.playlist_track.fade_graph = FadeCurve(
|
|
self.track_path, self.track_fade_at, self.track_silence_at
|
|
)
|
|
self.finished.emit()
|
|
|
|
|
|
class TrackSequence:
|
|
next = PlaylistTrack()
|
|
now = PlaylistTrack()
|
|
previous = PlaylistTrack()
|
|
|
|
|
|
track_sequence = TrackSequence()
|