kae/musicmuster
1
0
Fork 0
Code Issues 28 Pull Requests Projects Releases 1 Activity

WIP: Can play tracks without errors

Browse Source
This commit is contained in:
Keith Edmunds 2025-03-21 12:10:46 +00:00
parent 7361086da5
commit a95aa918b1
13 changed files with 1134 additions and 1065 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
app/classes.py
View File

@ -21,7 +21,6 @@ from PyQt6.QtWidgets import (
)
# App imports
# from music_manager import FadeCurve
# Define singleton first as it's needed below
@ -163,6 +162,14 @@ class TrackInfo(NamedTuple):
row_number: int
# Classes for signals
@dataclass
class InsertTrack:
playlist_id: int
track_id: int | None
note: str
@singleton
@dataclass
class MusicMusterSignals(QObject):
@ -181,9 +188,13 @@ class MusicMusterSignals(QObject):
search_wikipedia_signal = pyqtSignal(str)
show_warning_signal = pyqtSignal(str, str)
signal_add_track_to_header = pyqtSignal(int, int)
signal_insert_track = pyqtSignal(InsertTrack)
signal_playlist_selected_rows = pyqtSignal(int, list)
signal_set_next_row = pyqtSignal(int)
# TODO: undestirable (and unresolvable) reference
# signal_set_next_track = pyqtSignal(PlaylistRow)
# signal_set_next_track takes a PlaylistRow as an argument. We can't
# specify that here as it requires us to import PlaylistRow from
# playlistrow.py, which itself imports MusicMusterSignals
signal_set_next_track = pyqtSignal(object)
span_cells_signal = pyqtSignal(int, int, int, int, int)
status_message_signal = pyqtSignal(str, int)
track_ended_signal = pyqtSignal()

292
app/dialogs.py
View File

@ -9,12 +9,22 @@ from PyQt6.QtWidgets import (
QListWidgetItem,
QMainWindow,
)
from PyQt6.QtWidgets import (
QDialog,
QHBoxLayout,
QLabel,
QLineEdit,
QListWidget,
QListWidgetItem,
QPushButton,
QVBoxLayout,
)
# Third party imports
from sqlalchemy.orm.session import Session
# App imports
from classes import MusicMusterSignals
from classes import ApplicationError, InsertTrack, MusicMusterSignals
from helpers import (
ask_yes_no,
get_relative_date,
@ -23,209 +33,153 @@ from helpers import (
from log import log
from models import Settings, Tracks
from playlistmodel import PlaylistModel
import repository
from ui import dlg_TrackSelect_ui
class TrackSelectDialog(QDialog):
"""Select track from database"""
class TrackInsertDialog(QDialog):
def __init__(
self,
parent: QMainWindow,
session: Session,
new_row_number: int,
base_model: PlaylistModel,
playlist_id: int,
add_to_header: Optional[bool] = False,
*args: Qt.WindowType,
**kwargs: Qt.WindowType,
) -> None:
"""
Subclassed QDialog to manage track selection
"""
super().__init__(parent, *args, **kwargs)
self.session = session
self.new_row_number = new_row_number
self.base_model = base_model
super().__init__(parent)
self.playlist_id = playlist_id
self.add_to_header = add_to_header
self.ui = dlg_TrackSelect_ui.Ui_Dialog()
self.ui.setupUi(self)
self.ui.btnAdd.clicked.connect(self.add_selected)
self.ui.btnAddClose.clicked.connect(self.add_selected_and_close)
self.ui.btnClose.clicked.connect(self.close)
self.ui.matchList.itemDoubleClicked.connect(self.add_selected)
self.ui.matchList.itemSelectionChanged.connect(self.selection_changed)
self.ui.radioTitle.toggled.connect(self.title_artist_toggle)
self.ui.searchString.textEdited.connect(self.chars_typed)
self.track: Optional[Tracks] = None
self.signals = MusicMusterSignals()
self.setWindowTitle("Insert Track")
record = Settings.get_setting(self.session, "dbdialog_width")
width = record.f_int or 800
record = Settings.get_setting(self.session, "dbdialog_height")
height = record.f_int or 600
self.resize(width, height)
# Title input on one line
self.title_label = QLabel("Title:")
self.title_edit = QLineEdit()
self.title_edit.textChanged.connect(self.update_list)
if add_to_header:
self.ui.lblNote.setVisible(False)
self.ui.txtNote.setVisible(False)
title_layout = QHBoxLayout()
title_layout.addWidget(self.title_label)
title_layout.addWidget(self.title_edit)
def add_selected(self) -> None:
"""Handle Add button"""
# Track list
self.track_list = QListWidget()
self.track_list.itemDoubleClicked.connect(self.add_clicked)
self.track_list.itemSelectionChanged.connect(self.selection_changed)
track = None
# Note input on one line
self.note_label = QLabel("Note:")
self.note_edit = QLineEdit()
if self.ui.matchList.selectedItems():
item = self.ui.matchList.currentItem()
if item:
track = item.data(Qt.ItemDataRole.UserRole)
note_layout = QHBoxLayout()
note_layout.addWidget(self.note_label)
note_layout.addWidget(self.note_edit)
note = self.ui.txtNote.text()
# Track path
self.path = QLabel()
path_layout = QHBoxLayout()
path_layout.addWidget(self.path)
if not (track or note):
# Buttons
self.add_btn = QPushButton("Add")
self.add_close_btn = QPushButton("Add and close")
self.close_btn = QPushButton("Close")
self.add_btn.clicked.connect(self.add_clicked)
self.add_close_btn.clicked.connect(self.add_and_close_clicked)
self.close_btn.clicked.connect(self.close)
btn_layout = QHBoxLayout()
btn_layout.addWidget(self.add_btn)
btn_layout.addWidget(self.add_close_btn)
btn_layout.addWidget(self.close_btn)
# Main layout
layout = QVBoxLayout()
layout.addLayout(title_layout)
layout.addWidget(self.track_list)
layout.addLayout(note_layout)
layout.addLayout(path_layout)
layout.addLayout(btn_layout)
self.setLayout(layout)
self.resize(800, 600)
# TODO
# record = Settings.get_setting(self.session, "dbdialog_width")
# width = record.f_int or 800
# record = Settings.get_setting(self.session, "dbdialog_height")
# height = record.f_int or 600
# self.resize(width, height)
def update_list(self, text: str) -> None:
self.track_list.clear()
if text.strip() == "":
# Do not search or populate list if input is empty
return
track_id = None
if track:
track_id = track.id
if text.startswith("a/") and len(text) > 2:
self.tracks = repository.tracks_like_artist(text[2:])
else:
self.tracks = repository.tracks_like_title(text)
if note and not track_id:
self.base_model.insert_row(self.new_row_number, track_id, note)
self.ui.txtNote.clear()
self.new_row_number += 1
for track in self.tracks:
duration_str = ms_to_mmss(track.duration)
last_played_str = get_relative_date(track.lastplayed)
item_str = (
f"{track.title} - {track.artist} [{duration_str}] {last_played_str}"
)
item = QListWidgetItem(item_str)
item.setData(Qt.ItemDataRole.UserRole, track.track_id)
self.track_list.addItem(item)
def get_selected_track_id(self) -> int | None:
selected_items = self.track_list.selectedItems()
if selected_items:
return selected_items[0].data(Qt.ItemDataRole.UserRole)
return None
def add_clicked(self):
track_id = self.get_selected_track_id()
note_text = self.note_edit.text()
if track_id is None and not note_text:
return
self.ui.txtNote.clear()
self.select_searchtext()
insert_track_data = InsertTrack(self.playlist_id, track_id, note_text)
self.signals.signal_insert_track.emit(insert_track_data)
if track_id is None:
log.error("track_id is None and should not be")
return
# Check whether track is already in playlist
move_existing = False
existing_prd = self.base_model.is_track_in_playlist(track_id)
if existing_prd is not None:
if ask_yes_no(
"Duplicate row",
"Track already in playlist. " "Move to new location?",
default_yes=True,
):
move_existing = True
self.title_edit.clear()
self.note_edit.clear()
self.track_list.clear()
self.title_edit.setFocus()
if self.add_to_header:
if move_existing and existing_prd: # "and existing_prd" for mypy's benefit
self.base_model.move_track_to_header(
self.new_row_number, existing_prd, note
)
else:
self.base_model.add_track_to_header(self.new_row_number, track_id)
# Close dialog - we can only add one track to a header
self.accept()
else:
# Adding a new track row
if move_existing and existing_prd: # "and existing_prd" for mypy's benefit
self.base_model.move_track_add_note(
self.new_row_number, existing_prd, note
)
else:
self.base_model.insert_row(self.new_row_number, track_id, note)
self.new_row_number += 1
def add_selected_and_close(self) -> None:
"""Handle Add and Close button"""
self.add_selected()
def add_and_close_clicked(self):
track_id = self.get_selected_track_id()
if track_id is not None:
note_text = self.note_edit.text()
insert_track_data = InsertTrack(
playlist_id=self.playlist_id, track_id=self.track_id, note=self.note
)
insert_track_data = InsertTrack(self.playlist_id, track_id, note_text)
self.signals.signal_insert_track.emit(insert_track_data)
self.accept()
def chars_typed(self, s: str) -> None:
"""Handle text typed in search box"""
self.ui.matchList.clear()
if len(s) > 0:
if s.startswith("a/") and len(s) > 2:
matches = Tracks.search_artists(self.session, "%" + s[2:])
elif self.ui.radioTitle.isChecked():
matches = Tracks.search_titles(self.session, "%" + s)
else:
matches = Tracks.search_artists(self.session, "%" + s)
if matches:
for track in matches:
last_played = None
last_playdate = max(
track.playdates, key=lambda p: p.lastplayed, default=None
)
if last_playdate:
last_played = last_playdate.lastplayed
t = QListWidgetItem()
track_text = (
f"{track.title} - {track.artist} "
f"[{ms_to_mmss(track.duration)}] "
f"({get_relative_date(last_played)})"
)
t.setText(track_text)
t.setData(Qt.ItemDataRole.UserRole, track)
self.ui.matchList.addItem(t)
def closeEvent(self, event: Optional[QEvent]) -> None:
"""
Override close and save dialog coordinates
"""
if not event:
return
record = Settings.get_setting(self.session, "dbdialog_height")
record.f_int = self.height()
record = Settings.get_setting(self.session, "dbdialog_width")
record.f_int = self.width()
self.session.commit()
event.accept()
def keyPressEvent(self, event: QKeyEvent | None) -> None:
"""
Clear selection on ESC if there is one
"""
if event and event.key() == Qt.Key.Key_Escape:
if self.ui.matchList.selectedItems():
self.ui.matchList.clearSelection()
return
super(TrackSelectDialog, self).keyPressEvent(event)
def select_searchtext(self) -> None:
"""Select the searchbox"""
self.ui.searchString.selectAll()
self.ui.searchString.setFocus()
def selection_changed(self) -> None:
"""Display selected track path in dialog box"""
if not self.ui.matchList.selectedItems():
self.path.setText("")
track_id = self.get_selected_track_id()
if track_id is None:
return
item = self.ui.matchList.currentItem()
track = item.data(Qt.ItemDataRole.UserRole)
last_playdate = max(track.playdates, key=lambda p: p.lastplayed, default=None)
if last_playdate:
last_played = last_playdate.lastplayed
else:
last_played = None
path_text = f"{track.path} ({get_relative_date(last_played)})"
tracklist = [t for t in self.tracks if t.track_id == track_id]
if not tracklist:
return
if len(tracklist) > 1:
raise ApplicationError("More than one track returned")
track = tracklist[0]
self.ui.dbPath.setText(path_text)
def title_artist_toggle(self) -> None:
"""
Handle switching between searching for artists and searching for
titles
"""
# Logic is handled already in chars_typed(), so just call that.
self.chars_typed(self.ui.searchString.text())
self.path.setText(track.path)

3
app/file_importer.py
View File

@ -41,7 +41,7 @@ from helpers import (
)
from log import log
from models import db, Tracks
from music_manager import track_sequence
from playlistrow import TrackSequence
from playlistmodel import PlaylistModel
import helpers
@ -701,6 +701,7 @@ class PickMatch(QDialog):
self.setWindowTitle("New or replace")
layout = QVBoxLayout()
track_sequence = TrackSequence()
# Add instructions
instructions = (

80
app/helpers.py
View File

@ -168,7 +168,7 @@ def get_name(prompt: str, default: str = "") -> str | None:
def get_relative_date(
past_date: Optional[dt.datetime], reference_date: Optional[dt.datetime] = None
past_date: Optional[dt.datetime], now: Optional[dt.datetime] = None
) -> str:
"""
Return how long before reference_date past_date is as string.
@ -182,31 +182,33 @@ def get_relative_date(
if not past_date or past_date == Config.EPOCH:
return "Never"
if not reference_date:
reference_date = dt.datetime.now()
if not now:
now = dt.datetime.now()
# Check parameters
if past_date > reference_date:
return "get_relative_date() past_date is after relative_date"
if past_date > now:
raise ApplicationError("get_relative_date() past_date is after relative_date")
days: int
days_str: str
weeks: int
weeks_str: str
delta = now - past_date
days = delta.days
weeks, days = divmod((reference_date.date() - past_date.date()).days, 7)
if weeks == days == 0:
# Same day so return time instead
return Config.LAST_PLAYED_TODAY_STRING + " " + past_date.strftime("%H:%M")
if weeks == 1:
weeks_str = "week"
else:
weeks_str = "weeks"
if days == 1:
days_str = "day"
else:
days_str = "days"
return f"{weeks} {weeks_str}, {days} {days_str}"
if days == 0:
return "(Today)"
elif days == 1:
return "(Yesterday)"
years, days_remain = divmod(days, 365)
months, days_final = divmod(days_remain, 30)
parts = []
if years:
parts.append(f"{years}y")
if months:
parts.append(f"{months}m")
if days_final:
parts.append(f"{days_final}d")
formatted = " ".join(parts)
return f"({formatted} ago)"
def get_tags(path: str) -> Tags:
@ -264,39 +266,15 @@ def leading_silence(
return min(trim_ms, len(audio_segment))
def ms_to_mmss(
ms: Optional[int],
decimals: int = 0,
negative: bool = False,
none: Optional[str] = None,
) -> str:
def ms_to_mmss(ms: int | None, none: str = "-") -> str:
"""Convert milliseconds to mm:ss"""
minutes: int
remainder: int
seconds: float
if ms is None:
return none
if not ms:
if none:
return none
else:
return "-"
sign = ""
if ms < 0:
if negative:
sign = "-"
else:
ms = 0
minutes, seconds = divmod(ms // 1000, 60)
minutes, remainder = divmod(ms, 60 * 1000)
seconds = remainder / 1000
# if seconds >= 59.5, it will be represented as 60, which looks odd.
# So, fake it under those circumstances
if seconds >= 59.5:
seconds = 59.0
return f"{sign}{minutes:.0f}:{seconds:02.{decimals}f}"
return f"{minutes}:{seconds:02d}"
def normalise_track(path: str) -> None:

509
app/music_manager.py
View File

@ -3,32 +3,23 @@ from __future__ import annotations
import datetime as dt
from time import sleep
from typing import Any, Optional
# Third party imports
# import line_profiler
import numpy as np
import pyqtgraph as pg # type: ignore
from sqlalchemy.orm.session import Session
import vlc # type: ignore
# PyQt imports
from PyQt6.QtCore import (
pyqtSignal,
QObject,
QThread,
)
from pyqtgraph import PlotWidget
from pyqtgraph.graphicsItems.PlotDataItem import PlotDataItem # type: ignore
from pyqtgraph.graphicsItems.LinearRegionItem import LinearRegionItem # type: ignore
# App imports
from classes import ApplicationError, MusicMusterSignals
from classes import singleton
from config import Config
import helpers
from log import log
from repository import PlaylistRowDTO
from vlcmanager import VLCManager
# Define the VLC callback function type
# import ctypes
@ -63,106 +54,6 @@ from vlcmanager import VLCManager
# libc.vsnprintf.restype = ctypes.c_int
class _AddFadeCurve(QObject):
"""
Initialising a fade curve introduces a noticeable delay so carry out in
a thread.
"""
finished = pyqtSignal()
def __init__(
self,
plr: PlaylistRow,
track_path: str,
track_fade_at: int,
track_silence_at: int,
) -> None:
super().__init__()
self.plr = plr
self.track_path = track_path
self.track_fade_at = track_fade_at
self.track_silence_at = track_silence_at
def run(self) -> None:
"""
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.plr.fade_graph = fc
self.finished.emit()
class FadeCurve:
GraphWidget: Optional[PlotWidget] = 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: int = max(
0, track_fade_at - Config.FADE_CURVE_MS_BEFORE_FADE - 1
)
self.end_ms: int = track_silence_at
audio_segment = audio[self.start_ms : self.end_ms]
self.graph_array = np.array(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.curve: Optional[PlotDataItem] = None
self.region: Optional[LinearRegionItem] = None
def clear(self) -> None:
"""Clear the current graph"""
if self.GraphWidget:
self.GraphWidget.clear()
def plot(self) -> None:
if self.GraphWidget:
self.curve = self.GraphWidget.plot(self.graph_array)
if self.curve:
self.curve.setPen(Config.FADE_CURVE_FOREGROUND)
else:
log.debug("_FadeCurve.plot: no curve")
else:
log.debug("_FadeCurve.plot: no GraphWidget")
def tick(self, play_time: int) -> 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
if self.region:
self.region.setRegion([0, ms_of_graph * self.ms_to_array_factor])
class _FadeTrack(QThread):
finished = pyqtSignal()
@ -196,21 +87,32 @@ class _FadeTrack(QThread):
self.finished.emit()
# TODO can we move this into the _Music class?
vlc_instance = VLCManager().vlc_instance
@singleton
class VLCManager:
"""
Singleton class to ensure we only ever have one vlc Instance
"""
def __init__(self) -> None:
self.vlc_instance = vlc.Instance()
def get_instance(self) -> vlc.Instance:
return self.vlc_instance
class _Music:
class Music:
"""
Manage the playing of music tracks
"""
def __init__(self, name: str) -> None:
vlc_instance.set_user_agent(name, name)
self.player: Optional[vlc.MediaPlayer] = None
self.name = name
vlc_manager = VLCManager()
self.vlc_instance = vlc_manager.get_instance()
self.vlc_instance.set_user_agent(name, name)
self.player: vlc.MediaPlayer | None = None
self.max_volume: int = Config.VLC_VOLUME_DEFAULT
self.start_dt: Optional[dt.datetime] = None
self.start_dt: dt.datetime | None = None
# Set up logging
# self._set_vlc_log()
@ -300,7 +202,7 @@ class _Music:
self,
path: str,
start_time: dt.datetime,
position: Optional[float] = None,
position: float | None = None,
) -> None:
"""
Start playing the track at path.
@ -317,7 +219,7 @@ class _Music:
log.error(f"play({path}): path not readable")
return None
self.player = vlc.MediaPlayer(vlc_instance, path)
self.player = vlc.MediaPlayer(self.vlc_instance, path)
if self.player is None:
log.error(f"_Music:play: failed to create MediaPlayer ({path=})")
helpers.show_warning(
@ -341,7 +243,7 @@ class _Music:
self.player.set_position(position)
def set_volume(
self, volume: Optional[int] = None, set_default: bool = True
self, volume: int | None = None, set_default: bool = True
) -> None:
"""Set maximum volume used for player"""
@ -381,370 +283,3 @@ class _Music:
self.player.stop()
self.player.release()
self.player = None
class PlaylistRow:
"""
Object to manage playlist row and track.
"""
def __init__(self, dto: PlaylistRowDTO) -> None:
"""
The dto object will include a Tracks object if this row has a track.
"""
self.dto = dto
self.music = _Music(name=Config.VLC_MAIN_PLAYER_NAME)
self.signals = MusicMusterSignals()
self.end_of_track_signalled: bool = False
self.end_time: dt.datetime | None = None
self.fade_graph: Any | None = None
self.fade_graph_start_updates: dt.datetime | None = None
self.forecast_end_time: dt.datetime | None = None
self.forecast_start_time: dt.datetime | None = None
self.note_bg: str | None = None
self.note_fg: str | None = None
self.resume_marker: float = 0.0
self.row_bg: str | None = None
self.row_fg: str | None = None
self.start_time: dt.datetime | None = None
def __repr__(self) -> str:
return (
f"<PlaylistRow(playlist_id={self.dto.playlist_id}, "
f"row_number={self.dto.row_number}, "
f"playlistrow_id={self.dto.playlistrow_id}, "
f"note={self.dto.note}, track_id={self.dto.track_id}>"
)
# Expose TrackDTO fields as properties
@property
def artist(self):
return self.dto.artist
@property
def bitrate(self):
return self.dto.bitrate
@property
def duration(self):
return self.dto.duration
@property
def fade_at(self):
return self.dto.fade_at
@property
def intro(self):
return self.dto.intro
@property
def lastplayed(self):
return self.dto.lastplayed
@property
def path(self):
return self.dto.path
@property
def silence_at(self):
return self.dto.silence_at
@property
def start_gap(self):
return self.dto.start_gap
@property
def title(self):
return self.dto.title
@property
def track_id(self):
return self.dto.track_id
@track_id.setter
def track_id(self, value: int) -> None:
"""
Adding a track_id should only happen to a header row.
"""
if self.track_id:
raise ApplicationError("Attempting to add track to row with existing track ({self=}")
# TODO: set up write access to track_id. Should only update if
# track_id == 0. Need to update all other track fields at the
# same time.
print("set track_id attribute for {self=}, {value=}")
pass
# Expose PlaylistRowDTO fields as properties
@property
def note(self):
return self.dto.note
@note.setter
def note(self, value: str) -> None:
# TODO set up write access to db
print("set note attribute for {self=}, {value=}")
# self.dto.note = value
@property
def played(self):
return self.dto.played
@played.setter
def played(self, value: bool = True) -> None:
# TODO set up write access to db
print("set played attribute for {self=}")
# self.dto.played = value
@property
def playlist_id(self):
return self.dto.playlist_id
@property
def playlistrow_id(self):
return self.dto.playlistrow_id
@property
def row_number(self):
return self.dto.row_number
@row_number.setter
def row_number(self, value: int) -> None:
# TODO do we need to set up write access to db?
self.dto.row_number = value
def check_for_end_of_track(self) -> None:
"""
Check whether track has ended. If so, emit track_ended_signal
"""
if self.start_time is None:
return
if self.end_of_track_signalled:
return
if self.music.is_playing():
return
self.start_time = None
if self.fade_graph:
self.fade_graph.clear()
# Ensure that player is released
self.music.fade(0)
self.signals.track_ended_signal.emit()
self.end_of_track_signalled = True
def drop3db(self, enable: bool) -> None:
"""
If enable is true, drop output by 3db else restore to full volume
"""
if enable:
self.music.set_volume(volume=Config.VLC_VOLUME_DROP3db, set_default=False)
else:
self.music.set_volume(volume=Config.VLC_VOLUME_DEFAULT, set_default=False)
def fade(self, fade_seconds: int = Config.FADEOUT_SECONDS) -> None:
"""Fade music"""
self.resume_marker = self.music.get_position()
self.music.fade(fade_seconds)
self.signals.track_ended_signal.emit()
def is_playing(self) -> bool:
"""
Return True if we're currently playing else False
"""
if self.start_time is None:
return False
return self.music.is_playing()
def play(self, position: Optional[float] = None) -> None:
"""Play track"""
now = dt.datetime.now()
self.start_time = now
# Initialise player
self.music.play(self.path, start_time=now, position=position)
self.end_time = now + 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 set_forecast_start_time(
self, modified_rows: list[int], start: Optional[dt.datetime]
) -> Optional[dt.datetime]:
"""
Set forecast start time for this row
Update passed modified rows list if we changed the row.
Return new start time
"""
changed = False
if self.forecast_start_time != start:
self.forecast_start_time = start
changed = True
if start is None:
if self.forecast_end_time is not None:
self.forecast_end_time = None
changed = True
new_start_time = None
else:
end_time = start + dt.timedelta(milliseconds=self.duration())
new_start_time = end_time
if self.forecast_end_time != end_time:
self.forecast_end_time = end_time
changed = True
if changed and self.row_number not in modified_rows:
modified_rows.append(self.row_number)
return new_start_time
def stop(self, fade_seconds: int = 0) -> None:
"""
Stop this track playing
"""
self.resume_marker = self.music.get_position()
self.fade(fade_seconds)
# Reset fade graph
if self.fade_graph:
self.fade_graph.clear()
def time_playing(self) -> int:
"""
Return time track has been playing in milliseconds, zero if not playing
"""
if self.start_time is None:
return 0
return self.music.get_playtime()
def time_remaining_intro(self) -> int:
"""
Return milliseconds of intro remaining. Return 0 if no intro time in track
record or if intro has finished.
"""
if not self.intro:
return 0
return max(0, self.intro - self.time_playing())
def time_to_fade(self) -> int:
"""
Return milliseconds until fade time. Return zero if we're not playing.
"""
if self.start_time is None:
return 0
return self.fade_at - self.time_playing()
def time_to_silence(self) -> int:
"""
Return milliseconds until silent. Return zero if we're not playing.
"""
if self.start_time is None:
return 0
return self.silence_at - self.time_playing()
def update_fade_graph(self) -> None:
"""
Update fade graph
"""
if (
not self.is_playing()
or not self.fade_graph_start_updates
or not self.fade_graph
):
return
now = dt.datetime.now()
if self.fade_graph_start_updates > now:
return
self.fade_graph.tick(self.time_playing())
def update_playlist_and_row(self, session: Session) -> None:
"""
Update local playlist_id and row_number from playlistrow_id
"""
# TODO: only seems to be used by track_sequence
return
# plr = session.get(PlaylistRows, self.playlistrow_id)
# if not plr:
# raise ApplicationError(f"(Can't retrieve PlaylistRows entry, {self=}")
# self.playlist_id = plr.playlist_id
# self.row_number = plr.row_number
class TrackSequence:
next: Optional[PlaylistRow] = None
current: Optional[PlaylistRow] = None
previous: Optional[PlaylistRow] = None
def set_next(self, rat: Optional[PlaylistRow]) -> None:
"""
Set the 'next' track to be passed rat. Clear
any previous next track. If passed rat is None
just clear existing next track.
"""
# Clear any existing fade graph
if self.next and self.next.fade_graph:
self.next.fade_graph.clear()
if rat is None:
self.next = None
else:
self.next = rat
self.create_fade_graph()
def create_fade_graph(self) -> None:
"""
Initialise and add FadeCurve in a thread as it's slow
"""
self.fadecurve_thread = QThread()
if self.next is None:
raise ApplicationError("hell in a handcart")
self.worker = _AddFadeCurve(
self.next,
track_path=self.next.path,
track_fade_at=self.next.fade_at,
track_silence_at=self.next.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()
track_sequence = TrackSequence()

208
app/musicmuster.py
View File

@ -70,12 +70,12 @@ from classes import (
TrackInfo,
)
from config import Config
from dialogs import TrackSelectDialog
from dialogs import TrackInsertDialog
from file_importer import FileImporter
from helpers import ask_yes_no, file_is_unreadable, get_name
from log import log
from models import db, Playdates, PlaylistRows, Playlists, Queries, Settings, Tracks
from music_manager import PlaylistRow, track_sequence
from playlistrow import PlaylistRow, TrackSequence
from playlistmodel import PlaylistModel, PlaylistProxyModel
from playlists import PlaylistTab
from querylistmodel import QuerylistModel
@ -94,12 +94,12 @@ class Current:
base_model: PlaylistModel
proxy_model: PlaylistProxyModel
playlist_id: int = 0
selected_rows: list[int] = []
selected_row_numbers: list[int] = []
def __repr__(self):
return (
f"<Current(base_model={self.base_model}, proxy_model={self.proxy_model}, "
f"playlist_id={self.playlist_id}, selected_rows={self.selected_rows}>"
f"playlist_id={self.playlist_id}, selected_rows={self.selected_row_numbers}>"
)
@ -1194,6 +1194,7 @@ class Window(QMainWindow):
self.catch_return_key = False
self.importer: Optional[FileImporter] = None
self.current = Current()
self.track_sequence = TrackSequence()
webbrowser.register(
"browser",
@ -1217,7 +1218,7 @@ class Window(QMainWindow):
return
# Don't allow window to close when a track is playing
if track_sequence.current and track_sequence.current.is_playing():
if self.track_sequence.current and self.track_sequence.current.is_playing():
event.ignore()
helpers.show_warning(
self, "Track playing", "Can't close application while track is playing"
@ -1671,7 +1672,7 @@ class Window(QMainWindow):
Clear next track
"""
track_sequence.set_next(None)
self.track_sequence.set_next(None)
self.update_headers()
def clear_selection(self) -> None:
@ -1704,8 +1705,8 @@ class Window(QMainWindow):
).playlist_id
# Don't close current track playlist
if track_sequence.current is not None:
current_track_playlist_id = track_sequence.current.playlist_id
if self.track_sequence.current is not None:
current_track_playlist_id = self.track_sequence.current.playlist_id
if current_track_playlist_id:
if closing_tab_playlist_id == current_track_playlist_id:
helpers.show_OK(
@ -1714,8 +1715,8 @@ class Window(QMainWindow):
return False
# Don't close next track playlist
if track_sequence.next is not None:
next_track_playlist_id = track_sequence.next.playlist_id
if self.track_sequence.next is not None:
next_track_playlist_id = self.track_sequence.next.playlist_id
if next_track_playlist_id:
if closing_tab_playlist_id == next_track_playlist_id:
helpers.show_OK(
@ -1777,8 +1778,8 @@ class Window(QMainWindow):
of the playlist.
"""
if self.current.selected_rows:
return self.current.selected_rows[0]
if self.current.selected_row_numbers:
return self.current.selected_row_numbers[0]
return self.current.base_model.rowCount()
def debug(self):
@ -1816,8 +1817,8 @@ class Window(QMainWindow):
def drop3db(self) -> None:
"""Drop music level by 3db if button checked"""
if track_sequence.current:
track_sequence.current.drop3db(self.footer_section.btnDrop3db.isChecked())
if self.track_sequence.current:
self.track_sequence.current.drop3db(self.footer_section.btnDrop3db.isChecked())
def enable_escape(self, enabled: bool) -> None:
"""
@ -1843,13 +1844,8 @@ class Window(QMainWindow):
- Enable controls
"""
if track_sequence.current:
# Dereference the fade curve so it can be garbage collected
track_sequence.current.fade_graph = None
# Reset track_sequence objects
track_sequence.previous = track_sequence.current
track_sequence.current = None
if self.track_sequence.current:
self.track_sequence.move_current_to_previous()
# Tell playlist previous track has finished
self.current.base_model.previous_track_ended()
@ -1915,8 +1911,8 @@ class Window(QMainWindow):
def fade(self) -> None:
"""Fade currently playing track"""
if track_sequence.current:
track_sequence.current.fade()
if self.track_sequence.current:
self.track_sequence.current.fade()
def get_tab_index_for_playlist(self, playlist_id: int) -> Optional[int]:
"""
@ -1976,15 +1972,11 @@ class Window(QMainWindow):
def insert_track(self) -> None:
"""Show dialog box to select and add track from database"""
with db.Session() as session:
dlg = TrackSelectDialog(
parent=self,
session=session,
new_row_number=self.current_row_or_end(),
base_model=self.current.base_model,
)
dlg.exec()
session.commit()
dlg = TrackInsertDialog(
parent=self,
playlist_id=self.active_tab().playlist_id
)
dlg.exec()
def load_last_playlists(self) -> None:
"""Load the playlists that were open when the last session closed"""
@ -2038,7 +2030,7 @@ class Window(QMainWindow):
# Save the selected PlaylistRows items ready for a later
# paste
self.move_source_rows = self.current.selected_rows
self.move_source_rows = self.current.selected_row_numbers
self.move_source_model = self.current.base_model
log.debug(
@ -2080,21 +2072,14 @@ class Window(QMainWindow):
)
# Reset track_sequences
with db.Session() as session:
for ts in [
track_sequence.next,
track_sequence.current,
track_sequence.previous,
]:
if ts:
ts.update_playlist_and_row(session)
self.track_sequence.update()
def move_selected(self) -> None:
"""
Move selected rows to another playlist
"""
selected_rows = self.current.selected_rows
selected_rows = self.current.selected_row_numbers
if not selected_rows:
return
@ -2147,9 +2132,9 @@ class Window(QMainWindow):
# that moved row the next track
set_next_row: Optional[int] = None
if (
track_sequence.current
and track_sequence.current.playlist_id == to_playlist_model.playlist_id
and destination_row == track_sequence.current.row_number + 1
self.track_sequence.current
and self.track_sequence.current.playlist_id == to_playlist_model.playlist_id
and destination_row == self.track_sequence.current.row_number + 1
):
set_next_row = destination_row
@ -2185,7 +2170,7 @@ class Window(QMainWindow):
"""
# If there is no next track set, return.
if track_sequence.next is None:
if self.track_sequence.next is None:
log.error("musicmuster.play_next(): no next track selected")
return
@ -2202,35 +2187,34 @@ class Window(QMainWindow):
log.debug("issue223: play_next: 10ms timer disabled")
# If there's currently a track playing, fade it.
if track_sequence.current:
track_sequence.current.fade()
if self.track_sequence.current:
self.track_sequence.current.fade()
# Move next track to current track.
# end_of_track_actions() will have saved current track to
# previous_track
track_sequence.current = track_sequence.next
# Clear next track
self.clear_next()
self.track_sequence.move_next_to_current()
if self.track_sequence.current is None:
raise ApplicationError("No current track")
# Restore volume if -3dB active
if self.footer_section.btnDrop3db.isChecked():
self.footer_section.btnDrop3db.setChecked(False)
# Play (new) current track
log.debug(f"Play: {track_sequence.current.title}")
track_sequence.current.play(position)
log.debug(f"Play: {self.track_sequence.current.title}")
self.track_sequence.current.play(position)
# Update clocks now, don't wait for next tick
self.update_clocks()
# Show closing volume graph
if track_sequence.current.fade_graph:
track_sequence.current.fade_graph.GraphWidget = (
if self.track_sequence.current.fade_graph:
self.track_sequence.current.fade_graph.GraphWidget = (
self.footer_section.widgetFadeVolume
)
track_sequence.current.fade_graph.clear()
track_sequence.current.fade_graph.plot()
self.track_sequence.current.fade_graph.clear()
self.track_sequence.current.fade_graph.plot()
# Disable play next controls
self.catch_return_key = True
@ -2263,10 +2247,10 @@ class Window(QMainWindow):
track_info = self.active_tab().get_selected_row_track_info()
if not track_info:
# Otherwise get track_id to next track to play
if track_sequence.next:
if track_sequence.next.track_id:
if self.track_sequence.next:
if self.track_sequence.next.track_id:
track_info = TrackInfo(
track_sequence.next.track_id, track_sequence.next.row_number
self.track_sequence.next.track_id, self.track_sequence.next.row_number
)
else:
return
@ -2384,12 +2368,12 @@ class Window(QMainWindow):
Return True if it has, False if not
"""
if track_sequence.current and self.catch_return_key:
if self.track_sequence.current and self.catch_return_key:
# Suppress inadvertent double press
if (
track_sequence.current
and track_sequence.current.start_time
and track_sequence.current.start_time
self.track_sequence.current
and self.track_sequence.current.start_time
and self.track_sequence.current.start_time
+ dt.timedelta(milliseconds=Config.RETURN_KEY_DEBOUNCE_MS)
> dt.datetime.now()
):
@ -2398,8 +2382,8 @@ class Window(QMainWindow):
# If return is pressed during first PLAY_NEXT_GUARD_MS then
# default to NOT playing the next track, else default to
# playing it.
default_yes: bool = track_sequence.current.start_time is not None and (
(dt.datetime.now() - track_sequence.current.start_time).total_seconds()
default_yes: bool = self.track_sequence.current.start_time is not None and (
(dt.datetime.now() - self.track_sequence.current.start_time).total_seconds()
* 1000
> Config.PLAY_NEXT_GUARD_MS
)
@ -2428,18 +2412,18 @@ class Window(QMainWindow):
- If a track is playing, make that the next track
"""
if not track_sequence.previous:
if not self.track_sequence.previous:
return
# Return if no saved position
resume_marker = track_sequence.previous.resume_marker
resume_marker = self.track_sequence.previous.resume_marker
if not resume_marker:
log.error("No previous track position")
return
# We want to use play_next() to resume, so copy the previous
# track to the next track:
track_sequence.set_next(track_sequence.previous)
self.track_sequence.move_previous_to_next()
# Now resume playing the now-next track
self.play_next(resume_marker)
@ -2448,15 +2432,15 @@ class Window(QMainWindow):
# We need to fake the start time to reflect where we resumed the
# track
if (
track_sequence.current
and track_sequence.current.start_time
and track_sequence.current.duration
and track_sequence.current.resume_marker
self.track_sequence.current
and self.track_sequence.current.start_time
and self.track_sequence.current.duration
and self.track_sequence.current.resume_marker
):
elapsed_ms = (
track_sequence.current.duration * track_sequence.current.resume_marker
self.track_sequence.current.duration * self.track_sequence.current.resume_marker
)
track_sequence.current.start_time -= dt.timedelta(milliseconds=elapsed_ms)
self.track_sequence.current.start_time -= dt.timedelta(milliseconds=elapsed_ms)
def search_playlist(self) -> None:
"""Show text box to search playlist"""
@ -2491,12 +2475,12 @@ class Window(QMainWindow):
row_number: Optional[int] = None
if self.current.selected_rows:
row_number = self.current.selected_rows[0]
if self.current.selected_row_numbers:
row_number = self.current.selected_row_numbers[0]
if row_number is None:
if track_sequence.next:
if track_sequence.next.track_id:
row_number = track_sequence.next.row_number
if self.track_sequence.next:
if self.track_sequence.next.track_id:
row_number = self.track_sequence.next.row_number
if row_number is None:
return None
@ -2540,8 +2524,8 @@ class Window(QMainWindow):
def show_current(self) -> None:
"""Scroll to show current track"""
if track_sequence.current:
self.show_track(track_sequence.current)
if self.track_sequence.current:
self.show_track(self.track_sequence.current)
def show_warning(self, title: str, body: str) -> None:
"""
@ -2554,8 +2538,8 @@ class Window(QMainWindow):
def show_next(self) -> None:
"""Scroll to show next track"""
if track_sequence.next:
self.show_track(track_sequence.next)
if self.track_sequence.next:
self.show_track(self.track_sequence.next)
def show_status_message(self, message: str, timing: int) -> None:
"""
@ -2594,8 +2578,8 @@ class Window(QMainWindow):
"""Stop playing immediately"""
self.stop_autoplay = True
if track_sequence.current:
track_sequence.current.stop()
if self.track_sequence.current:
self.track_sequence.current.stop()
def tab_change(self) -> None:
"""Called when active tab changed"""
@ -2607,22 +2591,22 @@ class Window(QMainWindow):
Called every 10ms
"""
if track_sequence.current:
track_sequence.current.update_fade_graph()
if self.track_sequence.current:
self.track_sequence.current.update_fade_graph()
def tick_100ms(self) -> None:
"""
Called every 100ms
"""
if track_sequence.current:
if self.track_sequence.current:
try:
track_sequence.current.check_for_end_of_track()
self.track_sequence.current.check_for_end_of_track()
# Update intro counter if applicable and, if updated, return
# because playing an intro takes precedence over timing a
# preview.
intro_ms_remaining = track_sequence.current.time_remaining_intro()
intro_ms_remaining = self.track_sequence.current.time_remaining_intro()
if intro_ms_remaining > 0:
self.footer_section.label_intro_timer.setText(
f"{intro_ms_remaining / 1000:.1f}"
@ -2682,17 +2666,17 @@ class Window(QMainWindow):
"""
# If track is playing, update track clocks time and colours
if track_sequence.current and track_sequence.current.is_playing():
if self.track_sequence.current and self.track_sequence.current.is_playing():
# Elapsed time
self.header_section.label_elapsed_timer.setText(
helpers.ms_to_mmss(track_sequence.current.time_playing())
helpers.ms_to_mmss(self.track_sequence.current.time_playing())
+ " / "
+ helpers.ms_to_mmss(track_sequence.current.duration)
+ helpers.ms_to_mmss(self.track_sequence.current.duration)
)
# Time to fade
time_to_fade = track_sequence.current.time_to_fade()
time_to_silence = track_sequence.current.time_to_silence()
time_to_fade = self.track_sequence.current.time_to_fade()
time_to_silence = self.track_sequence.current.time_to_silence()
self.footer_section.label_fade_timer.setText(
helpers.ms_to_mmss(time_to_fade)
)
@ -2741,25 +2725,25 @@ class Window(QMainWindow):
Update last / current / next track headers
"""
if track_sequence.previous:
if self.track_sequence.previous:
self.header_section.hdrPreviousTrack.setText(
f"{track_sequence.previous.title} - {track_sequence.previous.artist}"
f"{self.track_sequence.previous.title} - {self.track_sequence.previous.artist}"
)
else:
self.header_section.hdrPreviousTrack.setText("")
if track_sequence.current:
if self.track_sequence.current:
self.header_section.hdrCurrentTrack.setText(
f"{track_sequence.current.title.replace('&', '&&')} - "
f"{track_sequence.current.artist.replace('&', '&&')}"
f"{self.track_sequence.current.title.replace('&', '&&')} - "
f"{self.track_sequence.current.artist.replace('&', '&&')}"
)
else:
self.header_section.hdrCurrentTrack.setText("")
if track_sequence.next:
if self.track_sequence.next:
self.header_section.hdrNextTrack.setText(
f"{track_sequence.next.title.replace('&', '&&')} - "
f"{track_sequence.next.artist.replace('&', '&&')}"
f"{self.track_sequence.next.title.replace('&', '&&')} - "
f"{self.track_sequence.next.artist.replace('&', '&&')}"
)
else:
self.header_section.hdrNextTrack.setText("")
@ -2774,25 +2758,25 @@ class Window(QMainWindow):
# Do we need to set a 'next' icon?
set_next = True
if (
track_sequence.current
and track_sequence.next
and track_sequence.current.playlist_id == track_sequence.next.playlist_id
self.track_sequence.current
and self.track_sequence.next
and self.track_sequence.current.playlist_id == self.track_sequence.next.playlist_id
):
set_next = False
for idx in range(self.tabBar.count()):
widget = self.playlist_section.tabPlaylist.widget(idx)
if (
track_sequence.next
self.track_sequence.next
and set_next
and widget.playlist_id == track_sequence.next.playlist_id
and widget.playlist_id == self.track_sequence.next.playlist_id
):
self.playlist_section.tabPlaylist.setTabIcon(
idx, QIcon(Config.PLAYLIST_ICON_NEXT)
)
elif (
track_sequence.current
and widget.playlist_id == track_sequence.current.playlist_id
self.track_sequence.current
and widget.playlist_id == self.track_sequence.current.playlist_id
):
self.playlist_section.tabPlaylist.setTabIcon(
idx, QIcon(Config.PLAYLIST_ICON_CURRENT)

324
app/playlistmodel.py
View File

@ -48,7 +48,7 @@ from helpers import (
)
from log import log
from models import db, NoteColours, Playdates, PlaylistRows, Tracks
from music_manager import PlaylistRow, track_sequence
from playlistrow import PlaylistRow, TrackSequence
import repository
@ -83,6 +83,7 @@ class PlaylistModel(QAbstractTableModel):
self.playlist_id = playlist_id
self.is_template = is_template
self.track_sequence = TrackSequence()
self.playlist_rows: dict[int, PlaylistRow] = {}
self.selected_rows: list[PlaylistRow] = []
@ -93,13 +94,16 @@ class PlaylistModel(QAbstractTableModel):
self.signals.begin_reset_model_signal.connect(self.begin_reset_model)
self.signals.end_reset_model_signal.connect(self.end_reset_model)
self.signals.signal_add_track_to_header.connect(self.add_track_to_header)
self.signals.signal_playlist_selected_rows.connect(self.set_selected_rows)
self.signals.signal_set_next_row.connect(self.set_next_row)
with db.Session() as session:
# Ensure row numbers in playlist are contiguous
# TODO: remove this
PlaylistRows.fixup_rownumbers(session, playlist_id)
# Populate self.playlist_rows
self.load_data(session)
# Populate self.playlist_rows
self.load_data()
self.update_track_times()
def __repr__(self) -> str:
@ -131,7 +135,7 @@ class PlaylistModel(QAbstractTableModel):
# playing it. It's also possible that the track marked as
# next has already been played. Check for either of those.
for ts in [track_sequence.next, track_sequence.current]:
for ts in [self.track_sequence.next, self.track_sequence.current]:
if (
ts
and ts.row_number == row_number
@ -178,14 +182,14 @@ class PlaylistModel(QAbstractTableModel):
# Update local copy
self.refresh_row(selected_row.row_number)
# Repaint row
roles = [
roles_to_invalidate = [
Qt.ItemDataRole.BackgroundRole,
Qt.ItemDataRole.DisplayRole,
Qt.ItemDataRole.FontRole,
Qt.ItemDataRole.ForegroundRole,
]
# only invalidate required roles
self.invalidate_row(row_number, roles)
self.invalidate_row(selected_row.row_number, roles_to_invalidate)
self.signals.resize_rows_signal.emit(self.playlist_id)
@ -207,16 +211,16 @@ class PlaylistModel(QAbstractTableModel):
return QBrush(QColor(Config.COLOUR_UNREADABLE))
# Current track
if (
track_sequence.current
and track_sequence.current.playlist_id == self.playlist_id
and track_sequence.current.row_number == row
self.track_sequence.current
and self.track_sequence.current.playlist_id == self.playlist_id
and self.track_sequence.current.row_number == row
):
return QBrush(QColor(Config.COLOUR_CURRENT_PLAYLIST))
# Next track
if (
track_sequence.next
and track_sequence.next.playlist_id == self.playlist_id
and track_sequence.next.row_number == row
self.track_sequence.next
and self.track_sequence.next.playlist_id == self.playlist_id
and self.track_sequence.next.row_number == row
):
return QBrush(QColor(Config.COLOUR_NEXT_PLAYLIST))
@ -271,66 +275,69 @@ class PlaylistModel(QAbstractTableModel):
log.debug(f"{self}: current_track_started()")
if not track_sequence.current:
if not self.track_sequence.current:
return
row_number = track_sequence.current.row_number
row_number = self.track_sequence.current.row_number
playlist_dto = self.playlist_rows[row_number]
# Check for OBS scene change
self.obs_scene_change(row_number)
# Sanity check that we have a track_id
track_id = track_sequence.current.track_id
track_id = playlist_dto.track_id
if not track_id:
raise ApplicationError(
f"{self}: current_track_started() called with {track_id=}"
f"current_track_started() called with no track_id ({playlist_dto=})"
)
with db.Session() as session:
# Update Playdates in database
log.debug(f"{self}: update playdates {track_id=}")
Playdates(session, track_id)
session.commit()
# TODO: ensure Playdates is updated
# with db.Session() as session:
# # Update Playdates in database
# log.debug(f"{self}: update playdates {track_id=}")
# Playdates(session, track_id)
# session.commit()
# Mark track as played in playlist
log.debug(f"{self}: Mark track as played")
plr = session.get(PlaylistRows, track_sequence.current.playlistrow_id)
if plr:
plr.played = True
self.refresh_row(session, plr.row_number)
else:
log.error(
f"{self}: Can't retrieve plr, {track_sequence.current.playlistrow_id=}"
)
# Mark track as played in playlist
playlist_dto.played = True
# Update colour and times for current row
# Update colour and times for current row
roles_to_invalidate = [Qt.ItemDataRole.DisplayRole]
self.invalidate_row(row_number, roles_to_invalidate)
# Update previous row in case we're hiding played rows
if self.track_sequence.previous and self.track_sequence.previous.row_number:
# only invalidate required roles
roles = [Qt.ItemDataRole.DisplayRole]
self.invalidate_row(row_number, roles)
self.invalidate_row(self.track_sequence.previous.row_number, roles_to_invalidate)
# Update previous row in case we're hiding played rows
if track_sequence.previous and track_sequence.previous.row_number:
# only invalidate required roles
self.invalidate_row(track_sequence.previous.row_number, roles)
# Update all other track times
self.update_track_times()
# Update all other track times
self.update_track_times()
# Find next track
next_row = self.find_next_row_to_play(row_number)
if next_row:
self.signals.signal_set_next_track.emit(self.playlist_rows[next_row])
# Find next track
next_row = None
unplayed_rows = [
a
for a in self.get_unplayed_rows()
if not self.is_header_row(a)
and not file_is_unreadable(self.playlist_rows[a].path)
]
if unplayed_rows:
try:
next_row = min([a for a in unplayed_rows if a > row_number])
except ValueError:
next_row = min(unplayed_rows)
if next_row is not None:
self.set_next_row(next_row)
def find_next_row_to_play(self, from_row_number: int) -> int | None:
"""
Find the next row to play in this playlist. Return row number or
None if there's no next track.
"""
next_row = None
unplayed_rows = [
a
for a in self.get_unplayed_rows()
if not self.is_header_row(a)
and not file_is_unreadable(self.playlist_rows[a].path)
]
if unplayed_rows:
try:
next_row = min([a for a in unplayed_rows if a > from_row_number])
except ValueError:
next_row = min(unplayed_rows)
return next_row
def data(
self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole
@ -401,7 +408,7 @@ class PlaylistModel(QAbstractTableModel):
session.commit()
super().endRemoveRows()
self.reset_track_sequence_row_numbers()
self.track_sequence.update()
self.update_track_times()
def _display_role(self, row: int, column: int, rat: PlaylistRow) -> str:
@ -477,7 +484,8 @@ class PlaylistModel(QAbstractTableModel):
with db.Session() as session:
self.refresh_data(session)
super().endResetModel()
self.reset_track_sequence_row_numbers()
self.track_sequence.update()
self.update_track_times()
def _edit_role(self, row: int, column: int, rat: PlaylistRow) -> str | int:
"""
@ -742,16 +750,17 @@ class PlaylistModel(QAbstractTableModel):
note=note,
)
# Insert into self.playlist_rows
# Move rows down to make room
for destination_row in range(len(self.playlist_rows), new_row_number, -1):
self.playlist_rows[destination_row] = self.playlist_rows[destination_row - 1]
self.playlist_rows[new_row_number] = new_row
# Insert into self.playlist_rows
self.playlist_rows[new_row_number] = PlaylistRow(new_row)
super().endInsertRows()
self.signals.resize_rows_signal.emit(self.playlist_id)
# TODO check this what we want to do and how we want to do it
self.reset_track_sequence_row_numbers()
self.track_sequence.update()
self.update_track_times()
roles_to_invalidate = [
Qt.ItemDataRole.BackgroundRole,
Qt.ItemDataRole.DisplayRole,
@ -816,7 +825,7 @@ class PlaylistModel(QAbstractTableModel):
return None
def load_data(self, session: Session) -> None:
def load_data(self) -> None:
"""
Same as refresh data, but only used when creating playslit.
Distinguishes profile time between initial load and other
@ -843,8 +852,8 @@ class PlaylistModel(QAbstractTableModel):
# build a new playlist_rows
# shouldn't be PlaylistRow
new_playlist_rows: dict[int, PlaylistRow] = {}
for p in repository.get_playlist_rows(self.playlist_id):
new_playlist_rows[p.row_number] = PlaylistRow(p)
for dto in repository.get_playlist_rows(self.playlist_id):
new_playlist_rows[dto.row_number] = PlaylistRow(dto)
# Copy to self.playlist_rows
self.playlist_rows = new_playlist_rows
@ -854,16 +863,9 @@ class PlaylistModel(QAbstractTableModel):
Mark row as unplayed
"""
with db.Session() as session:
for row_number in row_numbers:
playlist_row = session.get(
PlaylistRows, self.playlist_rows[row_number].playlistrow_id
)
if not playlist_row:
return
playlist_row.played = False
session.commit()
self.refresh_row(session, row_number)
for row_number in row_numbers:
self.playlist_rows[row_number].played = False
self.refresh_row(row_number)
self.update_track_times()
# only invalidate required roles
@ -933,7 +935,7 @@ class PlaylistModel(QAbstractTableModel):
self.refresh_data(session)
# Update display
self.reset_track_sequence_row_numbers()
self.track_sequence.update()
self.update_track_times()
# only invalidate required roles
roles = [
@ -987,8 +989,8 @@ class PlaylistModel(QAbstractTableModel):
[self.playlist_rows[a].playlistrow_id for a in row_group],
):
if (
track_sequence.current
and playlist_row.id == track_sequence.current.playlistrow_id
self.track_sequence.current
and playlist_row.id == self.track_sequence.current.playlistrow_id
):
# Don't move current track
continue
@ -1004,35 +1006,32 @@ class PlaylistModel(QAbstractTableModel):
session.commit()
# Reset of model must come after session has been closed
self.reset_track_sequence_row_numbers()
self.track_sequence.update()
self.signals.end_reset_model_signal.emit(to_playlist_id)
self.update_track_times()
def move_track_add_note(
self, new_row_number: int, existing_rat: PlaylistRow, note: str
self, new_row_number: int, existing_plr: PlaylistRow, note: str
) -> None:
"""
Move existing_rat track to new_row_number and append note to any existing note
"""
log.debug(
f"{self}: move_track_add_note({new_row_number=}, {existing_rat=}, {note=}"
f"{self}: move_track_add_note({new_row_number=}, {existing_plr=}, {note=}"
)
if note:
with db.Session() as session:
playlist_row = session.get(PlaylistRows, existing_rat.playlistrow_id)
if playlist_row:
if playlist_row.note:
playlist_row.note += "\n" + note
else:
playlist_row.note = note
self.refresh_row(session, playlist_row.row_number)
session.commit()
playlist_row = self.playlist_rows[existing_plr.row_number]
if playlist_row.note:
playlist_row.note += "\n" + note
else:
playlist_row.note = note
self.refresh_row(existing_plr.row_number)
# Carry out the move outside of the session context to ensure
# database updated with any note change
self.move_rows([existing_rat.row_number], new_row_number)
self.move_rows([existing_plr.row_number], new_row_number)
self.signals.resize_rows_signal.emit(self.playlist_id)
def obs_scene_change(self, row_number: int) -> None:
@ -1083,15 +1082,15 @@ class PlaylistModel(QAbstractTableModel):
log.debug(f"{self}: previous_track_ended()")
# Sanity check
if not track_sequence.previous:
if not self.track_sequence.previous:
log.error(
f"{self}: playlistmodel:previous_track_ended called with no current track"
)
return
if track_sequence.previous.row_number is None:
if self.track_sequence.previous.row_number is None:
log.error(
f"{self}: previous_track_ended called with no row number "
f"({track_sequence.previous=})"
f"({self.track_sequence.previous=})"
)
return
@ -1100,7 +1099,7 @@ class PlaylistModel(QAbstractTableModel):
roles = [
Qt.ItemDataRole.BackgroundRole,
]
self.invalidate_row(track_sequence.previous.row_number, roles)
self.invalidate_row(self.track_sequence.previous.row_number, roles)
def refresh_data(self, session: Session) -> None:
"""
@ -1123,21 +1122,25 @@ class PlaylistModel(QAbstractTableModel):
# build a new playlist_rows
new_playlist_rows: dict[int, PlaylistRow] = {}
for p in repository.get_playlist_rows(self.playlist_id):
if p.playlistrow_id not in plrid_to_row:
new_playlist_rows[p.row_number] = PlaylistRow(p)
for dto in repository.get_playlist_rows(self.playlist_id):
if dto.playlistrow_id not in plrid_to_row:
new_playlist_rows[dto.row_number] = PlaylistRow(dto)
else:
new_playlist_row = self.playlist_rows[plrid_to_row[p.playlistrow_id]]
new_playlist_row.row_number = p.row_number
new_playlist_row = self.playlist_rows[plrid_to_row[dto.playlistrow_id]]
new_playlist_row.row_number = dto.row_number
# Copy to self.playlist_rows
self.playlist_rows = new_playlist_rows
def refresh_row(self, session, row_number):
def refresh_row(self, row_number: int) -> None:
"""Populate dict for one row from database"""
p = PlaylistRows.deep_row(session, self.playlist_id, row_number)
self.playlist_rows[row_number] = PlaylistRow(p)
plrid = self.playlist_rows[row_number].playlistrow_id
refreshed_row = repository.get_playlist_row(plrid)
if not refreshed_row:
raise ApplicationError(f"Failed to retrieve row {self.playlist_id=}, {row_number=}")
self.playlist_rows[row_number] = PlaylistRow(refreshed_row)
def remove_track(self, row_number: int) -> None:
"""
@ -1146,14 +1149,8 @@ class PlaylistModel(QAbstractTableModel):
log.debug(f"{self}: remove_track({row_number=})")
with db.Session() as session:
playlist_row = session.get(
PlaylistRows, self.playlist_rows[row_number].playlistrow_id
)
if playlist_row:
playlist_row.track_id = None
session.commit()
self.refresh_row(session, row_number)
self.playlist_rows[row_number].track_id = None
# only invalidate required roles
roles = [
Qt.ItemDataRole.DisplayRole,
@ -1170,7 +1167,7 @@ class PlaylistModel(QAbstractTableModel):
with db.Session() as session:
track = session.get(Tracks, track_id)
set_track_metadata(track)
self.refresh_row(session, row_number)
self.refresh_row(row_number)
self.update_track_times()
roles = [
Qt.ItemDataRole.BackgroundRole,
@ -1181,32 +1178,6 @@ class PlaylistModel(QAbstractTableModel):
self.signals.resize_rows_signal.emit(self.playlist_id)
session.commit()
def reset_track_sequence_row_numbers(self) -> None:
"""
Signal handler for when row ordering has changed.
Example: row 4 is marked as next. Row 2 is deleted. The PlaylistRows table will
be correctly updated with change of row number, but track_sequence.next will still
contain row_number==4. This function fixes up the track_sequence row numbers by
looking up the playlistrow_id and retrieving the row number from the database.
"""
log.debug(f"issue285: {self}: reset_track_sequence_row_numbers()")
# Check the track_sequence.next, current and previous plrs and
# update the row number
with db.Session() as session:
for ts in [
track_sequence.next,
track_sequence.current,
track_sequence.previous,
]:
if ts:
ts.update_playlist_and_row(session)
session.commit()
self.update_track_times()
def remove_comments(self, row_numbers: list[int]) -> None:
"""
Remove comments from passed rows
@ -1349,16 +1320,16 @@ class PlaylistModel(QAbstractTableModel):
# calculate end time when all tracks are played.
end_time_str = ""
if (
track_sequence.current
and track_sequence.current.end_time
self.track_sequence.current
and self.track_sequence.current.end_time
and (
row_number
< track_sequence.current.row_number
< self.track_sequence.current.row_number
< rat.row_number
)
):
section_end_time = (
track_sequence.current.end_time
self.track_sequence.current.end_time
+ dt.timedelta(milliseconds=duration)
)
end_time_str = (
@ -1411,12 +1382,16 @@ class PlaylistModel(QAbstractTableModel):
return True
def set_selected_rows(self, selected_rows: list[int]) -> None:
def set_selected_rows(self, playlist_id: int, selected_row_numbers: list[int]) -> None:
"""
Keep track of which rows are selected in the view
Handle signal_playlist_selected_rows to keep track of which rows
are selected in the view
"""
self.selected_rows = [self.playlist_rows[a] for a in selected_rows]
if playlist_id != self.playlist_id:
return
self.selected_rows = [self.playlist_rows[a] for a in selected_row_numbers]
def set_next_row(self, playlist_id: int) -> None:
"""
@ -1430,8 +1405,8 @@ class PlaylistModel(QAbstractTableModel):
if len(self.selected_rows) == 0:
# No row selected so clear next track
if track_sequence.next is not None:
track_sequence.set_next(None)
if self.track_sequence.next is not None:
self.track_sequence.set_next(None)
return
if len(self.selected_rows) > 1:
@ -1445,10 +1420,10 @@ class PlaylistModel(QAbstractTableModel):
raise ApplicationError(f"set_next_row: no track_id ({rat=})")
old_next_row: Optional[int] = None
if track_sequence.next:
old_next_row = track_sequence.next.row_number
if self.track_sequence.next:
old_next_row = self.track_sequence.next.row_number
track_sequence.set_next(rat)
self.track_sequence.set_next(rat)
roles = [
Qt.ItemDataRole.BackgroundRole,
@ -1513,7 +1488,7 @@ class PlaylistModel(QAbstractTableModel):
# commit changes before refreshing data
session.commit()
self.refresh_row(session, row_number)
self.refresh_row(row_number)
self.dataChanged.emit(index, index, [Qt.ItemDataRole.DisplayRole, role])
return True
@ -1636,9 +1611,8 @@ class PlaylistModel(QAbstractTableModel):
a.row_number for a in self.playlist_rows.values() if a.track_id == track_id
]
if track_rows:
with db.Session() as session:
for row in track_rows:
self.refresh_row(session, row)
for row in track_rows:
self.refresh_row(row)
# only invalidate required roles
roles = [
Qt.ItemDataRole.BackgroundRole,
@ -1664,19 +1638,19 @@ class PlaylistModel(QAbstractTableModel):
current_track_row = None
next_track_row = None
if (
track_sequence.current
and track_sequence.current.playlist_id == self.playlist_id
self.track_sequence.current
and self.track_sequence.current.playlist_id == self.playlist_id
):
current_track_row = track_sequence.current.row_number
current_track_row = self.track_sequence.current.row_number
# Update current track details now so that they are available
# when we deal with next track row which may be above current
# track row.
self.playlist_rows[current_track_row].set_forecast_start_time(
update_rows, track_sequence.current.start_time
update_rows, self.track_sequence.current.start_time
)
if track_sequence.next and track_sequence.next.playlist_id == self.playlist_id:
next_track_row = track_sequence.next.row_number
if self.track_sequence.next and self.track_sequence.next.playlist_id == self.playlist_id:
next_track_row = self.track_sequence.next.row_number
for row_number in range(row_count):
rat = self.playlist_rows[row_number]
@ -1700,11 +1674,11 @@ class PlaylistModel(QAbstractTableModel):
# Set start time for next row if we have a current track
if (
row_number == next_track_row
and track_sequence.current
and track_sequence.current.end_time
and self.track_sequence.current
and self.track_sequence.current.end_time
):
next_start_time = rat.set_forecast_start_time(
update_rows, track_sequence.current.end_time
update_rows, self.track_sequence.current.end_time
)
continue
@ -1738,6 +1712,8 @@ class PlaylistProxyModel(QSortFilterProxyModel):
# Search all columns
self.setFilterKeyColumn(-1)
self.track_sequence = TrackSequence()
def __repr__(self) -> str:
return f"<PlaylistProxyModel: sourceModel={self.sourceModel()}>"
@ -1753,37 +1729,37 @@ class PlaylistProxyModel(QSortFilterProxyModel):
if self.sourceModel().is_played_row(source_row):
# Don't hide current track
if (
track_sequence.current
and track_sequence.current.playlist_id
self.track_sequence.current
and self.track_sequence.current.playlist_id
== self.sourceModel().playlist_id
and track_sequence.current.row_number == source_row
and self.track_sequence.current.row_number == source_row
):
return True
# Don't hide next track
if (
track_sequence.next
and track_sequence.next.playlist_id
self.track_sequence.next
and self.track_sequence.next.playlist_id
== self.sourceModel().playlist_id
and track_sequence.next.row_number == source_row
and self.track_sequence.next.row_number == source_row
):
return True
# Handle previous track
if track_sequence.previous:
if self.track_sequence.previous:
if (
track_sequence.previous.playlist_id
self.track_sequence.previous.playlist_id
!= self.sourceModel().playlist_id
or track_sequence.previous.row_number != source_row
or self.track_sequence.previous.row_number != source_row
):
# This row isn't our previous track: hide it
return False
if track_sequence.current and track_sequence.current.start_time:
if self.track_sequence.current and self.track_sequence.current.start_time:
# This row is our previous track. Don't hide it
# until HIDE_AFTER_PLAYING_OFFSET milliseconds
# after current track has started
if track_sequence.current.start_time and dt.datetime.now() > (
track_sequence.current.start_time
if self.track_sequence.current.start_time and dt.datetime.now() > (
self.track_sequence.current.start_time
+ dt.timedelta(
milliseconds=Config.HIDE_AFTER_PLAYING_OFFSET
)

533
app/playlistrow.py Normal file
View File

@ -0,0 +1,533 @@
# Standard library imports
import datetime as dt
from typing import Any
# PyQt imports
from PyQt6.QtCore import (
pyqtSignal,
QObject,
QThread,
)
# Third party imports
from pyqtgraph import PlotWidget # type: ignore
from pyqtgraph.graphicsItems.PlotDataItem import PlotDataItem # type: ignore
from pyqtgraph.graphicsItems.LinearRegionItem import LinearRegionItem # type: ignore
import numpy as np
import pyqtgraph as pg # type: ignore
# App imports
from classes import ApplicationError, MusicMusterSignals, PlaylistRowDTO, singleton
from config import Config
import helpers
from log import log
from music_manager import Music
import repository
class PlaylistRow:
"""
Object to manage playlist row and track.
"""
def __init__(self, dto: PlaylistRowDTO) -> None:
"""
The dto object will include a Tracks object if this row has a track.
"""
self.dto = dto
self.music = Music(name=Config.VLC_MAIN_PLAYER_NAME)
self.signals = MusicMusterSignals()
self.end_of_track_signalled: bool = False
self.end_time: dt.datetime | None = None
self.fade_graph: Any | None = None
self.fade_graph_start_updates: dt.datetime | None = None
self.forecast_end_time: dt.datetime | None = None
self.forecast_start_time: dt.datetime | None = None
self.note_bg: str | None = None
self.note_fg: str | None = None
self.resume_marker: float = 0.0
self.row_bg: str | None = None
self.row_fg: str | None = None
self.start_time: dt.datetime | None = None
def __repr__(self) -> str:
return (
f"<PlaylistRow(playlist_id={self.dto.playlist_id}, "
f"row_number={self.dto.row_number}, "
f"playlistrow_id={self.dto.playlistrow_id}, "
f"note={self.dto.note}, track_id={self.dto.track_id}>"
)
# Expose TrackDTO fields as properties
@property
def artist(self):
return self.dto.artist
@property
def bitrate(self):
return self.dto.bitrate
@property
def duration(self):
return self.dto.duration
@property
def fade_at(self):
return self.dto.fade_at
@property
def intro(self):
return self.dto.intro
@property
def lastplayed(self):
return self.dto.lastplayed
@property
def path(self):
return self.dto.path
@property
def silence_at(self):
return self.dto.silence_at
@property
def start_gap(self):
return self.dto.start_gap
@property
def title(self):
return self.dto.title
@property
def track_id(self):
return self.dto.track_id
@track_id.setter
def track_id(self, value: int) -> None:
"""
Adding a track_id should only happen to a header row.
"""
if self.track_id:
raise ApplicationError("Attempting to add track to row with existing track ({self=}")
# TODO: set up write access to track_id. Should only update if
# track_id == 0. Need to update all other track fields at the
# same time.
print("set track_id attribute for {self=}, {value=}")
pass
# Expose PlaylistRowDTO fields as properties
@property
def note(self):
return self.dto.note
@note.setter
def note(self, value: str) -> None:
# TODO set up write access to db
print("set note attribute for {self=}, {value=}")
# self.dto.note = value
@property
def played(self):
return self.dto.played
@played.setter
def played(self, value: bool = True) -> None:
# TODO set up write access to db
print("set played attribute for {self=}")
# self.dto.played = value
@property
def playlist_id(self):
return self.dto.playlist_id
@property
def playlistrow_id(self):
return self.dto.playlistrow_id
@property
def row_number(self):
return self.dto.row_number
@row_number.setter
def row_number(self, value: int) -> None:
# TODO do we need to set up write access to db?
self.dto.row_number = value
def check_for_end_of_track(self) -> None:
"""
Check whether track has ended. If so, emit track_ended_signal
"""
if self.start_time is None:
return
if self.end_of_track_signalled:
return
if self.music.is_playing():
return
self.start_time = None
if self.fade_graph:
self.fade_graph.clear()
# Ensure that player is released
self.music.fade(0)
self.signals.track_ended_signal.emit()
self.end_of_track_signalled = True
def drop3db(self, enable: bool) -> None:
"""
If enable is true, drop output by 3db else restore to full volume
"""
if enable:
self.music.set_volume(volume=Config.VLC_VOLUME_DROP3db, set_default=False)
else:
self.music.set_volume(volume=Config.VLC_VOLUME_DEFAULT, set_default=False)
def fade(self, fade_seconds: int = Config.FADEOUT_SECONDS) -> None:
"""Fade music"""
self.resume_marker = self.music.get_position()
self.music.fade(fade_seconds)
self.signals.track_ended_signal.emit()
def is_playing(self) -> bool:
"""
Return True if we're currently playing else False
"""
if self.start_time is None:
return False
return self.music.is_playing()
def play(self, position: float | None = None) -> None:
"""Play track"""
now = dt.datetime.now()
self.start_time = now
# Initialise player
self.music.play(self.path, start_time=now, position=position)
self.end_time = now + 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 set_forecast_start_time(
self, modified_rows: list[int], start: dt.datetime | None
) -> dt.datetime | None:
"""
Set forecast start time for this row
Update passed modified rows list if we changed the row.
Return new start time
"""
changed = False
if self.forecast_start_time != start:
self.forecast_start_time = start
changed = True
if start is None:
if self.forecast_end_time is not None:
self.forecast_end_time = None
changed = True
new_start_time = None
else:
end_time = start + dt.timedelta(milliseconds=self.duration)
new_start_time = end_time
if self.forecast_end_time != end_time:
self.forecast_end_time = end_time
changed = True
if changed and self.row_number not in modified_rows:
modified_rows.append(self.row_number)
return new_start_time
def stop(self, fade_seconds: int = 0) -> None:
"""
Stop this track playing
"""
self.resume_marker = self.music.get_position()
self.fade(fade_seconds)
# Reset fade graph
if self.fade_graph:
self.fade_graph.clear()
def time_playing(self) -> int:
"""
Return time track has been playing in milliseconds, zero if not playing
"""
if self.start_time is None:
return 0
return self.music.get_playtime()
def time_remaining_intro(self) -> int:
"""
Return milliseconds of intro remaining. Return 0 if no intro time in track
record or if intro has finished.
"""
if not self.intro:
return 0
return max(0, self.intro - self.time_playing())
def time_to_fade(self) -> int:
"""
Return milliseconds until fade time. Return zero if we're not playing.
"""
if self.start_time is None:
return 0
return self.fade_at - self.time_playing()
def time_to_silence(self) -> int:
"""
Return milliseconds until silent. Return zero if we're not playing.
"""
if self.start_time is None:
return 0
return self.silence_at - self.time_playing()
def update_fade_graph(self) -> None:
"""
Update fade graph
"""
if (
not self.is_playing()
or not self.fade_graph_start_updates
or not self.fade_graph
):
return
now = dt.datetime.now()
if self.fade_graph_start_updates > now:
return
self.fade_graph.tick(self.time_playing())
class _AddFadeCurve(QObject):
"""
Initialising a fade curve introduces a noticeable delay so carry out in
a thread.
"""
finished = pyqtSignal()
def __init__(
self,
plr: PlaylistRow,
track_path: str,
track_fade_at: int,
track_silence_at: int,
) -> None:
super().__init__()
self.plr = plr
self.track_path = track_path
self.track_fade_at = track_fade_at
self.track_silence_at = track_silence_at
def run(self) -> None:
"""
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.plr.fade_graph = fc
self.finished.emit()
class FadeCurve:
GraphWidget: PlotWidget | None = 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: int = max(
0, track_fade_at - Config.FADE_CURVE_MS_BEFORE_FADE - 1
)
self.end_ms: int = track_silence_at
audio_segment = audio[self.start_ms : self.end_ms]
self.graph_array = np.array(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.curve: PlotDataItem | None = None
self.region: LinearRegionItem | None = None
def clear(self) -> None:
"""Clear the current graph"""
if self.GraphWidget:
self.GraphWidget.clear()
def plot(self) -> None:
if self.GraphWidget:
self.curve = self.GraphWidget.plot(self.graph_array)
if self.curve:
self.curve.setPen(Config.FADE_CURVE_FOREGROUND)
else:
log.debug("_FadeCurve.plot: no curve")
else:
log.debug("_FadeCurve.plot: no GraphWidget")
def tick(self, play_time: int) -> 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
if self.region:
self.region.setRegion([0, ms_of_graph * self.ms_to_array_factor])
@singleton
class TrackSequence:
"""
Maintain a list of which track (if any) is next, current and
previous. A track can only be previous after being current, and can
only be current after being next. If one of the tracks listed here
moves, the row_number and/or playlist_id will change.
"""
def __init__(self) -> None:
"""
Set up storage for the three monitored tracks
"""
self.next: PlaylistRow | None = None
self.current: PlaylistRow | None = None
self.previous: PlaylistRow | None = None
def set_next(self, plr: PlaylistRow | None) -> None:
"""
Set the 'next' track to be passed PlaylistRow. Clear any previous
next track. If passed PlaylistRow is None just clear existing
next track.
"""
# Clear any existing fade graph
if self.next and self.next.fade_graph:
self.next.fade_graph.clear()
if plr is None:
self.next = None
else:
self.next = plr
self.create_fade_graph()
def move_next_to_current(self) -> None:
"""
Make the next track the current track
"""
self.current = self.next
self.next = None
def move_current_to_previous(self) -> None:
"""
Make the current track the previous track
"""
if self.current is None:
raise ApplicationError("Tried to move non-existent track from current to previous")
# Dereference the fade curve so it can be garbage collected
self.current.fade_graph = None
self.previous = self.current
self.current = None
def move_previous_to_next(self) -> None:
"""
Make the previous track the next track
"""
self.next = self.previous
self.previous = None
def create_fade_graph(self) -> None:
"""
Initialise and add FadeCurve in a thread as it's slow
"""
self.fadecurve_thread = QThread()
if self.next is None:
raise ApplicationError("hell in a handcart")
self.worker = _AddFadeCurve(
self.next,
track_path=self.next.path,
track_fade_at=self.next.fade_at,
track_silence_at=self.next.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 update(self) -> None:
"""
If a PlaylistRow is edited (moved, title changed, etc), the
playlistrow_id won't change. We can retrieve the PlaylistRow
using the playlistrow_id and update the stored PlaylistRow.
"""
for ts in [self.next, self.current, self.previous]:
if not ts:
continue
playlist_row_dto = repository.get_playlist_row(ts.playlistrow_id)
if not playlist_row_dto:
raise ApplicationError(f"(Can't retrieve PlaylistRows entry, {self=}")
ts = PlaylistRow(playlist_row_dto)

41
app/playlists.py
View File

@ -37,7 +37,7 @@ from PyQt6.QtWidgets import (
from audacity_controller import AudacityController
from classes import ApplicationError, Col, MusicMusterSignals, PlaylistStyle, TrackInfo
from config import Config
from dialogs import TrackSelectDialog
from dialogs import TrackInsertDialog
from helpers import (
ask_yes_no,
ms_to_mmss,
@ -46,7 +46,7 @@ from helpers import (
)
from log import log
from models import db, Settings
from music_manager import track_sequence
from playlistrow import TrackSequence
from playlistmodel import PlaylistModel, PlaylistProxyModel
if TYPE_CHECKING:
@ -278,6 +278,7 @@ class PlaylistTab(QTableView):
self.musicmuster = musicmuster
self.playlist_id = model.sourceModel().playlist_id
self.track_sequence = TrackSequence()
# Set up widget
self.setItemDelegate(PlaylistDelegate(self, model.sourceModel()))
@ -408,8 +409,8 @@ class PlaylistTab(QTableView):
# that moved row the next track
set_next_row: Optional[int] = None
if (
track_sequence.current
and to_model_row == track_sequence.current.row_number + 1
self.track_sequence.current
and to_model_row == self.track_sequence.current.row_number + 1
):
set_next_row = to_model_row
@ -461,12 +462,14 @@ class PlaylistTab(QTableView):
Toggle drag behaviour according to whether rows are selected
"""
selected_rows = self.get_selected_rows()
self.musicmuster.current.selected_rows = selected_rows
self.get_base_model().set_selected_rows(selected_rows)
selected_row_numbers = self.get_selected_rows()
# Signal selected rows to model
self.signals.signal_playlist_selected_rows.emit(self.playlist_id, selected_row_numbers)
# Put sum of selected tracks' duration in status bar
# If no rows are selected, we have nothing to do
if len(selected_rows) == 0:
if len(selected_row_numbers) == 0:
self.musicmuster.lblSumPlaytime.setText("")
else:
if not self.musicmuster.disable_selection_timing:
@ -517,11 +520,9 @@ class PlaylistTab(QTableView):
return
with db.Session() as session:
dlg = TrackSelectDialog(
dlg = TrackInsertDialog(
parent=self.musicmuster,
session=session,
new_row_number=model_row_number,
base_model=self.get_base_model(),
playlist_id=self.playlist_id,
add_to_header=True,
)
dlg.exec()
@ -538,12 +539,12 @@ class PlaylistTab(QTableView):
header_row = self.get_base_model().is_header_row(model_row_number)
track_row = not header_row
if track_sequence.current:
this_is_current_row = model_row_number == track_sequence.current.row_number
if self.track_sequence.current:
this_is_current_row = model_row_number == self.track_sequence.current.row_number
else:
this_is_current_row = False
if track_sequence.next:
this_is_next_row = model_row_number == track_sequence.next.row_number
if self.track_sequence.next:
this_is_next_row = model_row_number == self.track_sequence.next.row_number
else:
this_is_next_row = False
track_path = base_model.get_row_info(model_row_number).path
@ -760,8 +761,8 @@ class PlaylistTab(QTableView):
# Don't delete current or next tracks
selected_row_numbers = self.selected_model_row_numbers()
for ts in [
track_sequence.next,
track_sequence.current,
self.track_sequence.next,
self.track_sequence.current,
]:
if ts:
if (
@ -1122,7 +1123,7 @@ class PlaylistTab(QTableView):
# Update musicmuster
self.musicmuster.current.playlist_id = self.playlist_id
self.musicmuster.current.selected_rows = self.get_selected_rows()
self.musicmuster.current.selected_row_numbers = self.get_selected_rows()
self.musicmuster.current.base_model = self.get_base_model()
self.musicmuster.current.proxy_model = self.model()
@ -1131,6 +1132,6 @@ class PlaylistTab(QTableView):
def _unmark_as_next(self) -> None:
"""Rescan track"""
track_sequence.set_next(None)
self.track_sequence.set_next(None)
self.clear_selection()
self.signals.next_track_changed_signal.emit()

2
app/querylistmodel.py
View File

@ -40,7 +40,7 @@ from helpers import (
)
from log import log
from models import db, Playdates, Tracks
from music_manager import PlaylistRow
from playlistrow import PlaylistRow
@dataclass

154
app/repository.py
View File

@ -11,11 +11,12 @@ from sqlalchemy import (
)
from sqlalchemy.orm import aliased
from sqlalchemy.orm.session import Session
from sqlalchemy.sql.elements import BinaryExpression
from classes import ApplicationError, PlaylistRowDTO
# App imports
from classes import PlaylistDTO, TrackDTO
from app import helpers
import helpers
from log import log
from models import (
db,
@ -23,6 +24,7 @@ from models import (
Playdates,
PlaylistRows,
Playlists,
Settings,
Tracks,
)
@ -65,7 +67,7 @@ def get_colour(text: str, foreground: bool = False) -> str:
# Track functions
def add_track_to_header(self, playlistrow_id: int, track_id: int) -> None:
def add_track_to_header(playlistrow_id: int, track_id: int) -> None:
"""
Add a track to this (header) row
"""
@ -87,13 +89,28 @@ def create_track(path: str) -> TrackDTO:
metadata = helpers.get_all_track_metadata(path)
with db.Session() as session:
try:
track = Tracks(session=session, **metadata)
track = Tracks(
session=session,
path=str(metadata["path"]),
title=str(metadata["title"]),
artist=str(metadata["artist"]),
duration=int(metadata["duration"]),
start_gap=int(metadata["start_gap"]),
fade_at=int(metadata["fade_at"]),
silence_at=int(metadata["silence_at"]),
bitrate=int(metadata["bitrate"]),
)
track_id = track.id
session.commit()
except Exception:
raise ApplicationError("Can't create Track")
return track_by_id(track_id)
new_track = track_by_id(track_id)
if not new_track:
raise ApplicationError("Unable to create new track")
return new_track
def track_by_id(track_id: int) -> TrackDTO | None:
@ -154,21 +171,79 @@ def track_by_id(track_id: int) -> TrackDTO | None:
return dto
def _tracks_like(where: BinaryExpression) -> list[TrackDTO]:
"""
Return tracks selected by where
"""
# Alias PlaydatesTable for subquery
LatestPlaydate = aliased(Playdates)
# Subquery: latest playdate for each track
latest_playdate_subq = (
select(
LatestPlaydate.track_id,
func.max(LatestPlaydate.lastplayed).label("lastplayed"),
)
.group_by(LatestPlaydate.track_id)
.subquery()
)
stmt = (
select(
Tracks.id.label("track_id"),
Tracks.artist,
Tracks.bitrate,
Tracks.duration,
Tracks.fade_at,
Tracks.intro,
Tracks.path,
Tracks.silence_at,
Tracks.start_gap,
Tracks.title,
latest_playdate_subq.c.lastplayed,
)
.outerjoin(latest_playdate_subq, Tracks.id == latest_playdate_subq.c.track_id)
.where(where)
)
results: list[TrackDTO] = []
with db.Session() as session:
records = session.execute(stmt).all()
for record in records:
dto = TrackDTO(
artist=record.artist,
bitrate=record.bitrate,
duration=record.duration,
fade_at=record.fade_at,
intro=record.intro,
lastplayed=record.lastplayed,
path=record.path,
silence_at=record.silence_at,
start_gap=record.start_gap,
title=record.title,
track_id=record.track_id,
)
results.append(dto)
return results
def tracks_like_artist(filter_str: str) -> list[TrackDTO]:
"""
Return tracks where artist is like filter
"""
return _tracks_like(Tracks.artist.ilike(f"%{filter_str}%"))
def tracks_like_title(filter_str: str) -> list[TrackDTO]:
"""
Return tracks where title is like filter
"""
# TODO: add in playdates as per Tracks.search_titles
with db.Session() as session:
stmt = select(Tracks).where(Tracks.title.ilike(f"%{filter_str}%"))
results = (
session.execute(stmt).scalars().unique().all()
) # `scalars()` extracts ORM objects
return [
TrackDTO(**{k: v for k, v in vars(t).items() if not k.startswith("_")})
for t in results
]
return _tracks_like(Tracks.title.ilike(f"%{filter_str}%"))
# Playlist functions
@ -244,10 +319,14 @@ def create_playlist(name: str, template_id: int) -> PlaylistDTO:
except Exception:
raise ApplicationError("Can't create Playlist")
return playlist_by_id(playlist_id)
new_playlist = playlist_by_id(playlist_id)
if not new_playlist:
raise ApplicationError("Can't retrieve new Playlist")
return new_playlist
def get_playlist_row(playlist_row_id: int) -> PlaylistRowDTO | None:
def get_playlist_row(playlistrow_id: int) -> PlaylistRowDTO | None:
"""
Return specific row DTO
"""
@ -286,7 +365,7 @@ def get_playlist_row(playlist_row_id: int) -> PlaylistRowDTO | None:
)
.outerjoin(Tracks, PlaylistRows.track_id == Tracks.id)
.outerjoin(latest_playdate_subq, Tracks.id == latest_playdate_subq.c.track_id)
.where(PlaylistRows.id == playlist_row_id)
.where(PlaylistRows.id == playlistrow_id)
.order_by(PlaylistRows.row_number)
)
@ -428,7 +507,9 @@ def get_playlist_rows(playlist_id: int) -> list[PlaylistRowDTO]:
return dto_list
def insert_row(playlist_id: int, row_number: int, track_id: int, note: str) -> PlaylistRowDTO:
def insert_row(
playlist_id: int, row_number: int, track_id: int | None, note: str
) -> PlaylistRowDTO:
"""
Insert a new row into playlist and return new row DTO
"""
@ -455,7 +536,11 @@ def insert_row(playlist_id: int, row_number: int, track_id: int, note: str) -> P
# Sanity check
_check_row_number_sequence(session=session, playlist_id=playlist_id, fix=False)
return get_playlist_row(playlist_row_id=playlist_row_id)
new_playlist_row = get_playlist_row(playlistrow_id=playlist_row_id)
if not new_playlist_row:
raise ApplicationError("Can't retrieve new playlist row")
return new_playlist_row
def playlist_by_id(playlist_id: int) -> PlaylistDTO | None:
@ -485,3 +570,34 @@ def playlist_by_id(playlist_id: int) -> PlaylistDTO | None:
)
return dto
# Misc
def get_setting(name: str) -> int | None:
"""
Get int setting
"""
with db.Session() as session:
record = session.execute(select(Settings).where(Settings.name == name)).one_or_none()
if not record:
return None
return record.f_int
def set_setting(name: str, value: int) -> None:
"""
Add int setting
"""
with db.Session() as session:
record = session.execute(select(Settings).where(Settings.name == name)).one_or_none()
if not record:
record = Settings(session=session, name=name)
if not record:
raise ApplicationError("Can't create Settings record")
record.f_int = value
session.commit()

22
app/vlcmanager.py
View File

@ -1,22 +0,0 @@
# Standard library imports
# PyQt imports
# Third party imports
import vlc # type: ignore
# App imports
from classes import singleton
@singleton
class VLCManager:
"""
Singleton class to ensure we only ever have one vlc Instance
"""
def __init__(self) -> None:
self.vlc_instance = vlc.Instance()
def get_instance(self) -> vlc.Instance:
return self.vlc_instance

14
tests/test_db_updates.py
View File

@ -46,19 +46,19 @@ class MyTestCase(unittest.TestCase):
self.track2 = repository.create_track(track2_path)
# Add tracks and header to playlist
repository.insert_row(
self.row0 = repository.insert_row(
self.playlist.playlist_id,
row_number=0,
track_id=self.track1.track_id,
note="track 1",
)
repository.insert_row(
self.row1 = repository.insert_row(
self.playlist.playlist_id,
row_number=1,
track_id=0,
note="Header row",
)
repository.insert_row(
self.row2 = repository.insert_row(
self.playlist.playlist_id,
row_number=2,
track_id=self.track2.track_id,
@ -70,7 +70,9 @@ class MyTestCase(unittest.TestCase):
db.drop_all()
def test_xxx(self):
"""Comment"""
def test_add_track_to_header(self):
"""Add a track to a header row"""
pass
repository.add_track_to_header(self.row1.playlistrow_id, self.track2.track_id)
result = repository.get_playlist_row(self.row1.playlistrow_id)
assert result.track_id == self.track2.track_id