from dataclasses import dataclass from datetime import datetime, timedelta from typing import Optional from PyQt6.QtCore import pyqtSignal, QObject, QThread import numpy as np import pyqtgraph as pg # type: ignore from config import Config from dbconfig import scoped_session, Session from models import PlaylistRows, Tracks 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 """ add_track_to_header_signal = pyqtSignal(int, int, int) add_track_to_playlist_signal = pyqtSignal(int, int, int, str) enable_escape_signal = pyqtSignal(bool) next_track_changed_signal = pyqtSignal() span_cells_signal = pyqtSignal(int, int, int, 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. Do NOT store row_number here - that changes if tracks are reordered in playlist (add, remove, drag/drop) and we shouldn't care about row number: that's the playlist's problem. """ self.artist: Optional[str] = None self.duration: Optional[int] = None self.end_time: Optional[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.silence_at: Optional[int] = None self.start_gap: Optional[int] = None self.start_time: Optional[datetime] = None self.title: Optional[str] = None self.track_id: Optional[int] = None def __repr__(self) -> str: return ( f"" ) def set_plr(self, session: scoped_session, plr: PlaylistRows) -> None: """ Update with new plr information """ if not plr.track: return session.add(plr) 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.plr_rownum = plr.plr_rownum 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 = datetime.now() if self.duration: self.end_time = self.start_time + 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() @helpers.singleton class CurrentTrack(PlaylistTrack): pass @helpers.singleton class NextTrack(PlaylistTrack): pass @helpers.singleton class PreviousTrack(PlaylistTrack): pass