#!/usr/bin/env python3 # Standard library imports from __future__ import annotations from slugify import slugify # type: ignore from typing import Callable, Optional import argparse from dataclasses import dataclass import datetime as dt import os import subprocess import sys import urllib.parse import webbrowser import yaml # PyQt imports from PyQt6.QtCore import ( QDate, Qt, QTime, QTimer, QVariant, ) from PyQt6.QtGui import ( QAction, QCloseEvent, QColor, QFont, QIcon, QKeySequence, QPalette, QShortcut, ) from PyQt6.QtWidgets import ( QAbstractItemView, QApplication, QCheckBox, QComboBox, QDialog, QFileDialog, QHBoxLayout, QInputDialog, QLabel, QLineEdit, QListWidgetItem, QMainWindow, QMenu, QMessageBox, QPushButton, QSpinBox, QTableView, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, ) # Third party imports from pygame import mixer from sqlalchemy.orm.session import Session import stackprinter # type: ignore # App imports from classes import ( ApplicationError, Filter, MusicMusterSignals, TrackInfo, ) from config import Config from dialogs import TrackSelectDialog from file_importer import FileImporter from helpers import file_is_unreadable, get_name from log import log from models import db, Playdates, PlaylistRows, Playlists, Queries, Settings, Tracks from music_manager import RowAndTrack, track_sequence from playlistmodel import PlaylistModel, PlaylistProxyModel from playlists import PlaylistTab from querylistmodel import QuerylistModel from ui import icons_rc # noqa F401 from ui.dlg_SelectPlaylist_ui import Ui_dlgSelectPlaylist # type: ignore from ui.downloadcsv_ui import Ui_DateSelect # type: ignore from ui.main_window_header_ui import Ui_HeaderSection # type: ignore from ui.main_window_playlist_ui import Ui_PlaylistSection # type: ignore from ui.main_window_footer_ui import Ui_FooterSection # type: ignore from utilities import check_db, update_bitrates import helpers class Current: base_model: PlaylistModel proxy_model: PlaylistProxyModel playlist_id: int = 0 selected_rows: list[int] = [] def __repr__(self): return ( f"" ) class DownloadCSV(QDialog): def __init__(self, parent=None): super().__init__() self.ui = Ui_DateSelect() self.ui.setupUi(self) self.ui.dateTimeEdit.setDate(QDate.currentDate()) self.ui.dateTimeEdit.setTime(QTime(19, 59, 0)) self.ui.buttonBox.accepted.connect(self.accept) self.ui.buttonBox.rejected.connect(self.reject) class EditDeleteDialog(QDialog): def __init__(self, templates: list[tuple[str, int]]) -> None: super().__init__() self.templates = templates self.selection: tuple[str, int] = ("", -1) self.init_ui() def init_ui(self) -> None: # Create label label = QLabel("Select template:") # Create combo box self.combo_box = QComboBox() for text, id_ in self.templates: self.combo_box.addItem(text, id_) # Create buttons edit_button = QPushButton("Edit") delete_button = QPushButton("Delete") cancel_button = QPushButton("Cancel") # Connect buttons edit_button.clicked.connect(self.edit_clicked) delete_button.clicked.connect(self.delete_clicked) cancel_button.clicked.connect(self.cancel_clicked) # Layout setup top_layout = QHBoxLayout() top_layout.addWidget(label) top_layout.addWidget(self.combo_box) bottom_layout = QHBoxLayout() bottom_layout.addStretch() bottom_layout.addWidget(edit_button) bottom_layout.addWidget(delete_button) bottom_layout.addWidget(cancel_button) main_layout = QVBoxLayout() main_layout.addLayout(top_layout) main_layout.addLayout(bottom_layout) self.setLayout(main_layout) self.setWindowTitle("Edit or Delete Template") def edit_clicked(self) -> None: self.selection = ("Edit", self.combo_box.currentData()) self.accept() def delete_clicked(self) -> None: self.selection = ("Delete", self.combo_box.currentData()) self.accept() def cancel_clicked(self) -> None: self.selection = ("Cancelled", -1) self.reject() class FilterDialog(QDialog): def __init__(self, name: str, filter: Filter) -> None: super().__init__() self.filter = filter self.setWindowTitle("Filter Settings") layout = QVBoxLayout() # Name row name_layout = QHBoxLayout() name_label = QLabel("Name") self.name_text = QLineEdit() self.name_text.setText(name) name_layout.addWidget(name_label) name_layout.addWidget(self.name_text) layout.addLayout(name_layout) # Path row path_layout = QHBoxLayout() path_label = QLabel("Path") self.path_combo = QComboBox() self.path_combo.addItems( [Config.FILTER_PATH_CONTAINS, Config.FILTER_PATH_EXCLUDING] ) for idx in range(self.path_combo.count()): if self.path_combo.itemText(idx) == filter.path_type: self.path_combo.setCurrentIndex(idx) break self.path_text = QLineEdit() if filter.path: self.path_text.setText(filter.path) path_layout.addWidget(path_label) path_layout.addWidget(self.path_combo) path_layout.addWidget(self.path_text) layout.addLayout(path_layout) # Last played row last_played_layout = QHBoxLayout() last_played_label = QLabel("Last played") self.last_played_combo = QComboBox() self.last_played_combo.addItems( [ Config.FILTER_PLAYED_COMPARATOR_BEFORE, Config.FILTER_PLAYED_COMPARATOR_NEVER, Config.FILTER_PLAYED_COMPARATOR_ANYTIME, ] ) for idx in range(self.last_played_combo.count()): if self.last_played_combo.itemText(idx) == filter.last_played_comparator: self.last_played_combo.setCurrentIndex(idx) break self.last_played_spinbox = QSpinBox() self.last_played_spinbox.setMinimum(0) self.last_played_spinbox.setMaximum(100) self.last_played_spinbox.setValue(filter.last_played_number or 0) self.last_played_unit = QComboBox() self.last_played_unit.addItems( [ Config.FILTER_PLAYED_YEARS, Config.FILTER_PLAYED_MONTHS, Config.FILTER_PLAYED_WEEKS, Config.FILTER_PLAYED_DAYS, ] ) for idx in range(self.last_played_unit.count()): if self.last_played_unit.itemText(idx) == filter.last_played_unit: self.last_played_unit.setCurrentIndex(idx) break last_played_ago_label = QLabel("ago") last_played_layout.addWidget(last_played_label) last_played_layout.addWidget(self.last_played_combo) last_played_layout.addWidget(self.last_played_spinbox) last_played_layout.addWidget(self.last_played_unit) last_played_layout.addWidget(last_played_ago_label) layout.addLayout(last_played_layout) # Duration row duration_layout = QHBoxLayout() duration_label = QLabel("Duration") self.duration_combo = QComboBox() self.duration_combo.addItems( [Config.FILTER_DURATION_LONGER, Config.FILTER_DURATION_SHORTER] ) for idx in range(self.duration_combo.count()): if self.duration_combo.itemText(idx) == filter.duration_type: self.duration_combo.setCurrentIndex(idx) break self.duration_spinbox = QSpinBox() self.duration_spinbox.setMinimum(0) self.duration_spinbox.setMaximum(1000) self.duration_spinbox.setValue(filter.duration_number) self.duration_unit = QComboBox() self.duration_unit.addItems( [Config.FILTER_DURATION_MINUTES, Config.FILTER_DURATION_SECONDS] ) self.duration_unit.setCurrentText(Config.FILTER_DURATION_MINUTES) for idx in range(self.duration_unit.count()): if self.duration_unit.itemText(idx) == filter.duration_unit: self.duration_unit.setCurrentIndex(idx) break duration_layout.addWidget(duration_label) duration_layout.addWidget(self.duration_combo) duration_layout.addWidget(self.duration_spinbox) duration_layout.addWidget(self.duration_unit) layout.addLayout(duration_layout) # Buttons button_layout = QHBoxLayout() self.ok_button = QPushButton("OK") self.cancel_button = QPushButton("Cancel") self.cancel_button.clicked.connect(self.reject) self.ok_button.clicked.connect(self.ok_clicked) button_layout.addWidget(self.ok_button) button_layout.addWidget(self.cancel_button) layout.addLayout(button_layout) self.setLayout(layout) # Connect signals self.last_played_combo.currentIndexChanged.connect( self.toggle_last_played_controls ) self.toggle_last_played_controls() def toggle_last_played_controls(self): disabled = self.last_played_combo.currentText() == "never" self.last_played_spinbox.setDisabled(disabled) self.last_played_unit.setDisabled(disabled) def ok_clicked(self) -> None: """ Update filter to match selections """ self.filter.path_type = self.path_combo.currentText() self.filter.path = self.path_text.text() self.filter.last_played_number = self.last_played_spinbox.value() self.filter.last_played_comparator = self.last_played_combo.currentText() self.filter.last_played_unit = self.last_played_unit.currentText() self.filter.duration_type = self.duration_combo.currentText() self.filter.duration_number = self.duration_spinbox.value() self.filter.duration_unit = self.duration_unit.currentText() self.accept() @dataclass class ItemlistItem: id: int name: str favourite: bool = False class ItemlistManager(QDialog): def __init__(self) -> None: super().__init__() self.setWindowTitle("Item Manager") self.setMinimumSize(600, 400) layout = QVBoxLayout(self) self.table = QTableWidget(self) self.table.setColumnCount(2) self.table.setHorizontalHeaderLabels(["Item", "Actions"]) hh = self.table.horizontalHeader() if not hh: raise ApplicationError("ItemlistManager failed to create horizontalHeader") hh.setStretchLastSection(True) self.table.setColumnWidth(0, 288) self.table.setColumnWidth(1, 300) layout.addWidget(self.table) button_layout = QHBoxLayout() self.new_button = QPushButton("New") self.new_button.clicked.connect(self.new_item) button_layout.addWidget(self.new_button) self.close_button = QPushButton("Close") self.close_button.clicked.connect(self.close) button_layout.addWidget(self.close_button) layout.addLayout(button_layout) def populate_table(self, items: list[ItemlistItem]) -> None: """Populates the table with items and action buttons.""" self.items = items self.table.setRowCount(len(self.items)) for row, item in enumerate(self.items): item_text = QTableWidgetItem(item.name) if item.favourite: item_text.setFont(QFont("Arial", weight=QFont.Weight.Bold)) self.table.setItem(row, 0, item_text) # Action Buttons and Checkbox in a widget widget = QWidget() h_layout = QHBoxLayout(widget) h_layout.setContentsMargins(0, 0, 0, 0) h_layout.setSpacing(5) rename_button = QPushButton("Rename") rename_button.clicked.connect(lambda _, i=item.id: self.rename_item(i)) h_layout.addWidget(rename_button) edit_button = QPushButton("Edit") edit_button.clicked.connect(lambda _, i=item.id: self.edit_item(i)) h_layout.addWidget(edit_button) delete_button = QPushButton("Delete") delete_button.clicked.connect(lambda _, i=item.id: self.delete_item(i)) h_layout.addWidget(delete_button) fav_checkbox = QCheckBox() fav_checkbox.setChecked(item.favourite) fav_checkbox.stateChanged.connect( lambda state, cb=fav_checkbox, i=item.id: self.toggle_favourite( i, cb.isChecked() ) ) h_layout.addWidget(fav_checkbox) self.table.setCellWidget(row, 1, widget) def delete_item(self, item_id: int) -> None: """Subclass must implement this method""" raise NotImplementedError def edit_item(self, item_id: int) -> None: """Subclass must implement this method""" raise NotImplementedError def new_item(self) -> None: """Subclass must implement this method""" raise NotImplementedError def rename_item(self, item_id: int) -> None: """Subclass must implement this method""" raise NotImplementedError def change_text(self, item_id: int, new_text: str) -> None: """ Update text for one row """ for row in range(self.table.rowCount()): item = self.table.item(row, 0) if item and self.items[row].id == item_id: item.setText(new_text) self.items[row].name = new_text break def toggle_favourite(self, item_id: int, checked: bool) -> None: """Subclass must udpate database if required""" for row in range(self.table.rowCount()): item = self.table.item(row, 0) if item and self.items[row].id == item_id: font = QFont( "Arial", weight=QFont.Weight.Bold if checked else QFont.Weight.Normal, ) item.setFont(font) self.items[row].favourite = checked break class ManageQueries(ItemlistManager): """ Delete / edit queries """ def __init__(self, session: Session, musicmuster: Window) -> None: super().__init__() self.session = session self.musicmuster = musicmuster self.refresh_table() self.exec() def refresh_table(self) -> None: """ Update table in widget """ # Build a list of queries query_list: list[ItemlistItem] = [] for query in Queries.get_all(self.session): query_list.append( ItemlistItem(name=query.name, id=query.id, favourite=query.favourite) ) self.populate_table(query_list) def delete_item(self, query_id: int) -> None: """delete query""" query = self.session.get(Queries, query_id) if not query: raise ApplicationError( f"manage_template.delete({query_id=}) can't load query" ) if helpers.ask_yes_no( "Delete query", f"Delete query '{query.name}': " "Are you sure?", ): log.debug(f"manage_queries: delete {query=}") self.session.delete(query) self.session.commit() self.refresh_table() def _edit_item(self, query: Queries) -> None: """Edit query""" dlg = FilterDialog(query.name, query.filter) if dlg.exec(): query.filter = dlg.filter query.name = dlg.name_text.text() self.session.commit() def edit_item(self, query_id: int) -> None: """Edit query""" query = self.session.get(Queries, query_id) if not query: raise ApplicationError( f"manage_template.edit_item({query_id=}) can't load query" ) return self._edit_item(query) def toggle_favourite(self, query_id: int, favourite: bool) -> None: """Mark query as (not) favourite""" query = self.session.get(Queries, query_id) if not query: return query.favourite = favourite self.session.commit() def new_item(self) -> None: """Create new query""" query_name = get_name(prompt="New query name:") if not query_name: return query = Queries(self.session, query_name, Filter()) self._edit_item(query) self.refresh_table() def rename_item(self, query_id: int) -> None: """rename query""" query = self.session.get(Queries, query_id) if not query: raise ApplicationError( f"manage_template.delete({query_id=}) can't load query" ) new_name = get_name(prompt="New query name", default=query.name) if not new_name: return query.name = new_name self.session.commit() self.change_text(query_id, new_name) class ManageTemplates(ItemlistManager): """ Delete / edit templates """ def __init__(self, session: Session, musicmuster: Window) -> None: super().__init__() self.session = session self.musicmuster = musicmuster self.refresh_table() self.exec() def refresh_table(self) -> None: """ Update table in widget """ # Build a list of templates template_list: list[ItemlistItem] = [] for template in Playlists.get_all_templates(self.session): template_list.append( ItemlistItem( name=template.name, id=template.id, favourite=template.favourite ) ) self.populate_table(template_list) def delete_item(self, template_id: int) -> None: """delete template""" template = self.session.get(Playlists, template_id) if not template: raise ApplicationError( f"manage_template.delete({template_id=}) can't load template" ) if helpers.ask_yes_no( "Delete template", f"Delete template '{template.name}': Are you sure?", ): # If template is currently open, re-check open_idx = self.musicmuster.get_tab_index_for_playlist(template_id) if open_idx: if not helpers.ask_yes_no( "Delete open template", f"Template '{template.name}' is currently open. Really delete?", ): return else: self.musicmuster.playlist_section.tabPlaylist.removeTab(open_idx) log.debug(f"manage_templates: delete {template=}") self.session.delete(template) self.session.commit() def edit_item(self, template_id: int) -> None: """Edit template""" template = self.session.get(Playlists, template_id) if not template: raise ApplicationError( f"manage_template.edit({template_id=}) can't load template" ) # Simply load the template as a playlist. Any changes # made will persist self.musicmuster._open_playlist(template, is_template=True) def toggle_favourite(self, template_id: int, favourite: bool) -> None: """Mark template as (not) favourite""" template = self.session.get(Playlists, template_id) if not template: return template.favourite = favourite self.session.commit() def new_item( self, ) -> None: """Create new template""" # Get base template template_id = self.musicmuster.solicit_template_to_use( self.session, template_prompt="New template based upon:" ) if template_id is None: return # Get new template name name = self.musicmuster.get_playlist_name( self.session, default="", prompt="New template name:" ) if not name: return # Create playlist for template and mark is as a template template = self.musicmuster._create_playlist(self.session, name, template_id) template.is_template = True self.session.commit() # Open it for editing self.musicmuster._open_playlist(template, is_template=True) def rename_item(self, template_id: int) -> None: """rename template""" template = self.session.get(Playlists, template_id) if not template: raise ApplicationError( f"manage_template.delete({template_id=}) can't load template" ) new_name = self.musicmuster.get_playlist_name(self.session, template.name) if not new_name: return template.name = new_name self.session.commit() self.change_text(template_id, new_name) @dataclass class ItemlistManagerCallbacks: delete: Callable[[int], None] edit: Callable[[int], None] favourite: Callable[[int, bool], None] new_item: Callable[[], None] rename: Callable[[int], Optional[str]] class PreviewManager: """ Manage track preview player """ def __init__(self) -> None: mixer.init() self.intro: Optional[int] = None self.path: str = "" self.row_number: Optional[int] = None self.start_time: Optional[dt.datetime] = None self.track_id: int = 0 def back(self, ms: int) -> None: """ Move play position back by 'ms' milliseconds """ position = max(0, (self.get_playtime() - ms)) / 1000 mixer.music.set_pos(position) self.start_time = dt.datetime.now() - dt.timedelta(seconds=position) def forward(self, ms: int) -> None: """ Move play position forward by 'ms' milliseconds """ position = (self.get_playtime() + ms) / 1000 mixer.music.set_pos(position) self.start_time = dt.datetime.now() - dt.timedelta(seconds=position) def get_playtime(self) -> int: """ Return time since track started in milliseconds, 0 if not playing """ if not mixer.music.get_busy(): return 0 if not self.start_time: return 0 return int((dt.datetime.now() - self.start_time).total_seconds() * 1000) def is_playing(self) -> bool: return mixer.music.get_busy() def move_to_intro_end(self) -> None: """ Move play position to 'buffer' milliseconds before end of intro. If no intro defined, do nothing. """ if self.intro is None: return position = max(0, self.intro - Config.PREVIEW_END_BUFFER_MS) / 1000 mixer.music.set_pos(position) self.start_time = dt.datetime.now() - dt.timedelta(seconds=position) def play(self) -> None: mixer.music.play() self.start_time = dt.datetime.now() def restart(self) -> None: """ Restart player from beginning """ if not mixer.music.get_busy(): return mixer.music.rewind() self.start_time = dt.datetime.now() def set_intro(self, ms: int) -> None: """ Set intro time """ self.intro = ms def set_track_info(self, info: TrackInfo) -> None: self.track_id = info.track_id self.row_number = info.row_number with db.Session() as session: track = session.get(Tracks, self.track_id) if not track: raise ValueError( f"PreviewManager: unable to retreive track {self.track_id=}" ) self.intro = track.intro self.path = track.path # Check file readable if file_is_unreadable(self.path): raise ValueError(f"PreviewManager.__init__: {track.path=} unreadable") mixer.music.load(self.path) def stop(self) -> None: mixer.music.stop() mixer.music.unload() self.path = "" self.row_number = None self.track_id = 0 self.start_time = None class QueryDialog(QDialog): """Dialog box to handle selecting track from a query""" def __init__(self, session: Session, default: int = 0) -> None: super().__init__() self.session = session self.default = default # Build a list of (query-name, playlist-id) tuples self.selected_tracks: list[int] = [] self.query_list: list[tuple[str, int]] = [] self.query_list.append((Config.NO_QUERY_NAME, 0)) for query in Queries.get_all(self.session): self.query_list.append((query.name, query.id)) self.setWindowTitle("Query Selector") # Create label query_label = QLabel("Query:") # Top layout (Query label, combo box, and info label) top_layout = QHBoxLayout() # Query label query_label = QLabel("Query:") top_layout.addWidget(query_label) # Combo Box with fixed width self.combo_box = QComboBox() # self.combo_box.setFixedWidth(150) # Adjust as necessary for 20 characters for text, id_ in self.query_list: self.combo_box.addItem(text, id_) top_layout.addWidget(self.combo_box) # Table (middle part) self.table_view = QTableView() self.table_view.setSelectionMode( QAbstractItemView.SelectionMode.ExtendedSelection ) self.table_view.setSelectionBehavior( QAbstractItemView.SelectionBehavior.SelectRows ) self.table_view.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) self.table_view.setAlternatingRowColors(True) self.table_view.setVerticalScrollMode( QAbstractItemView.ScrollMode.ScrollPerPixel ) self.table_view.clicked.connect(self.handle_row_click) # Bottom layout (buttons) bottom_layout = QHBoxLayout() bottom_layout.addStretch() # Push buttons to the right self.add_tracks_button = QPushButton("Add tracks") self.add_tracks_button.setEnabled(False) # Disabled by default self.add_tracks_button.clicked.connect(self.add_tracks_clicked) bottom_layout.addWidget(self.add_tracks_button) self.cancel_button = QPushButton("Cancel") self.cancel_button.clicked.connect(self.cancel_clicked) bottom_layout.addWidget(self.cancel_button) # Main layout main_layout = QVBoxLayout() main_layout.addLayout(top_layout) main_layout.addWidget(self.table_view) main_layout.addLayout(bottom_layout) self.combo_box.currentIndexChanged.connect(self.query_changed) if self.default: default_idx = self.combo_box.findData(QVariant(self.default)) self.combo_box.setCurrentIndex(default_idx) self.path_text = QLineEdit() self.setLayout(main_layout) # Stretch last column *after* setting column widths which is # *much* faster h_header = self.table_view.horizontalHeader() if h_header: h_header.sectionResized.connect(self._column_resize) h_header.setStretchLastSection(True) # Resize on vertical header click v_header = self.table_view.verticalHeader() if v_header: v_header.setMinimumSectionSize(5) v_header.sectionHandleDoubleClicked.disconnect() v_header.sectionHandleDoubleClicked.connect( self.table_view.resizeRowToContents ) self.set_window_size() self.resizeRowsToContents() def add_tracks_clicked(self): self.selected_tracks = self.table_view.model().get_selected_track_ids() self.accept() def cancel_clicked(self): self.selected_tracks = [] self.reject() def closeEvent(self, event: QCloseEvent | None) -> None: """ Record size and columns """ self.save_sizes() super().closeEvent(event) def accept(self) -> None: self.save_sizes() super().accept() def reject(self) -> None: self.save_sizes() super().reject() def save_sizes(self) -> None: """ Save window size """ # Save dialog box attributes attributes_to_save = dict( querylist_height=self.height(), querylist_width=self.width(), querylist_x=self.x(), querylist_y=self.y(), ) for name, value in attributes_to_save.items(): record = Settings.get_setting(self.session, name) record.f_int = value header = self.table_view.horizontalHeader() if header is None: return column_count = header.count() if column_count < 2: return for column_number in range(column_count - 1): attr_name = f"querylist_col_{column_number}_width" record = Settings.get_setting(self.session, attr_name) record.f_int = self.table_view.columnWidth(column_number) self.session.commit() def _column_resize(self, column_number: int, _old: int, _new: int) -> None: """ Called when column width changes. """ header = self.table_view.horizontalHeader() if not header: return # Resize rows if necessary self.resizeRowsToContents() def resizeRowsToContents(self): header = self.table_view.verticalHeader() model = self.table_view.model() if model: for row in range(model.rowCount()): hint = self.table_view.sizeHintForRow(row) header.resizeSection(row, hint) def query_changed(self, idx: int) -> None: """ Called when user selects query """ # Get query id query_id = self.combo_box.currentData() query = self.session.get(Queries, query_id) if not query: return # Create model base_model = QuerylistModel(self.session, query.filter) # Create table self.table_view.setModel(base_model) self.set_column_sizes() def handle_row_click(self, index): self.table_view.model().toggle_row_selection(index.row()) self.table_view.clearSelection() # Enable 'Add tracks' button only when a row is selected selected = self.table_view.model().get_selected_track_ids() self.add_tracks_button.setEnabled(selected != []) def set_window_size(self) -> None: """Set window sizes""" x = Settings.get_setting(self.session, "querylist_x").f_int or 100 y = Settings.get_setting(self.session, "querylist_y").f_int or 100 width = Settings.get_setting(self.session, "querylist_width").f_int or 100 height = Settings.get_setting(self.session, "querylist_height").f_int or 100 self.setGeometry(x, y, width, height) def set_column_sizes(self) -> None: """Set column sizes""" header = self.table_view.horizontalHeader() if header is None: return column_count = header.count() if column_count < 2: return # Last column is set to stretch so ignore it here for column_number in range(column_count - 1): attr_name = f"querylist_col_{column_number}_width" record = Settings.get_setting(self.session, attr_name) if record.f_int is not None: self.table_view.setColumnWidth(column_number, record.f_int) else: self.table_view.setColumnWidth( column_number, Config.DEFAULT_COLUMN_WIDTH ) class SelectPlaylistDialog(QDialog): def __init__(self, parent=None, playlists=None, session=None): super().__init__() if playlists is None: return self.ui = Ui_dlgSelectPlaylist() self.ui.setupUi(self) self.ui.lstPlaylists.itemDoubleClicked.connect(self.list_doubleclick) self.ui.buttonBox.accepted.connect(self.open) self.ui.buttonBox.rejected.connect(self.close) self.session = session self.playlist = None record = Settings.get_setting(self.session, "select_playlist_dialog_width") width = record.f_int or 800 record = Settings.get_setting(self.session, "select_playlist_dialog_height") height = record.f_int or 600 self.resize(width, height) for playlist in playlists: p = QListWidgetItem() p.setText(playlist.name) p.setData(Qt.ItemDataRole.UserRole, playlist) self.ui.lstPlaylists.addItem(p) def __del__(self): # review record = Settings.get_setting(self.session, "select_playlist_dialog_height") record.f_int = self.height() record = Settings.get_setting(self.session, "select_playlist_dialog_width") record.f_int = self.width() self.session.commit() def list_doubleclick(self, entry): # review self.playlist = entry.data(Qt.ItemDataRole.UserRole) self.accept() def open(self): # review if self.ui.lstPlaylists.selectedItems(): item = self.ui.lstPlaylists.currentItem() self.playlist = item.data(Qt.ItemDataRole.UserRole) self.accept() class TemplateSelectorDialog(QDialog): """ Class to manage user selection of template """ def __init__( self, templates: list[tuple[str, int]], template_prompt: Optional[str] ) -> None: super().__init__() self.templates = templates self.template_prompt = template_prompt self.selected_id = None self.init_ui() def init_ui(self): # Create label if not self.template_prompt: self.template_prompt = "Select template:" label = QLabel(self.template_prompt) # Create combo box self.combo_box = QComboBox() for text, id_ in self.templates: self.combo_box.addItem(text, id_) # Create buttons ok_button = QPushButton("OK") cancel_button = QPushButton("Cancel") # Connect buttons ok_button.clicked.connect(self.ok_clicked) cancel_button.clicked.connect(self.cancel_clicked) # Layout setup top_layout = QHBoxLayout() top_layout.addWidget(label) top_layout.addWidget(self.combo_box) bottom_layout = QHBoxLayout() bottom_layout.addStretch() bottom_layout.addWidget(ok_button) bottom_layout.addWidget(cancel_button) main_layout = QVBoxLayout() main_layout.addLayout(top_layout) main_layout.addLayout(bottom_layout) self.setLayout(main_layout) self.setWindowTitle("Template Selector") def ok_clicked(self): self.selected_id = self.combo_box.currentData() self.accept() def cancel_clicked(self): self.selected_id = -1 self.reject() # Per-section UI files class HeaderSection(QWidget, Ui_HeaderSection): def __init__(self, parent=None): super().__init__(parent) self.setupUi(self) class PlaylistSection(QWidget, Ui_PlaylistSection): def __init__(self, parent=None): super().__init__(parent) self.setupUi(self) class FooterSection(QWidget, Ui_FooterSection): def __init__(self, parent=None): super().__init__(parent) self.setupUi(self) class Window(QMainWindow): def __init__( self, parent: Optional[QWidget] = None, *args: list, **kwargs: dict ) -> None: super().__init__(parent) # Build main window from per-section classes defined above central_widget = QWidget(self) self.setCentralWidget(central_widget) layout = QVBoxLayout(central_widget) self.header_section = HeaderSection() self.playlist_section = PlaylistSection() self.footer_section = FooterSection() layout.addWidget(self.header_section) layout.addWidget(self.playlist_section) layout.addWidget(self.footer_section) self.setWindowTitle(Config.MAIN_WINDOW_TITLE) # Add menu bar self.create_menu_bar() self.timer10: QTimer = QTimer() self.timer100: QTimer = QTimer() self.timer500: QTimer = QTimer() self.timer1000: QTimer = QTimer() self.set_main_window_size() self.lblSumPlaytime = QLabel("") self.statusbar = self.statusBar() if self.statusbar: self.statusbar.addPermanentWidget(self.lblSumPlaytime) self.txtSearch = QLineEdit() self.txtSearch.setHidden(True) self.statusbar.addWidget(self.txtSearch) self.hide_played_tracks = False self.preview_manager = PreviewManager() self.footer_section.widgetFadeVolume.hideAxis("bottom") self.footer_section.widgetFadeVolume.hideAxis("left") self.footer_section.widgetFadeVolume.setDefaultPadding(0) self.footer_section.widgetFadeVolume.setBackground(Config.FADE_CURVE_BACKGROUND) self.move_source_rows: Optional[list[int]] = None self.move_source_model: Optional[PlaylistModel] = None self.disable_selection_timing = False self.clock_counter = 0 self.timer10.start(10) self.timer100.start(100) self.timer500.start(500) self.timer1000.start(1000) self.signals = MusicMusterSignals() self.connect_signals_slots() self.catch_return_key = False self.importer: Optional[FileImporter] = None self.current = Current() webbrowser.register( "browser", None, webbrowser.BackgroundBrowser(Config.EXTERNAL_BROWSER_PATH), ) # Set up shortcut key for instant logging from keyboard self.action_quicklog = QShortcut(QKeySequence("Ctrl+L"), self) self.action_quicklog.activated.connect(self.quicklog) self.load_last_playlists() self.stop_autoplay = False # # # # # # # # # # Overrides # # # # # # # # # # def closeEvent(self, event: Optional[QCloseEvent]) -> None: """Handle attempt to close main window""" if not event: return # Don't allow window to close when a track is playing if track_sequence.current and track_sequence.current.is_playing(): event.ignore() helpers.show_warning( self, "Track playing", "Can't close application while track is playing" ) else: with db.Session() as session: # Save tab number of open playlists open_playlist_ids: dict[int, int] = {} for idx in range(self.playlist_section.tabPlaylist.count()): open_playlist_ids[ self.playlist_section.tabPlaylist.widget(idx).playlist_id ] = idx Playlists.clear_tabs(session, list(open_playlist_ids.keys())) for playlist_id, idx in open_playlist_ids.items(): playlist = session.get(Playlists, playlist_id) if playlist: log.debug(f"Set {playlist=} tab to {idx=}") playlist.tab = idx # Save window attributes attributes_to_save = dict( mainwindow_height=self.height(), mainwindow_width=self.width(), mainwindow_x=self.x(), mainwindow_y=self.y(), active_tab=self.playlist_section.tabPlaylist.currentIndex(), ) for name, value in attributes_to_save.items(): record = Settings.get_setting(session, name) record.f_int = value session.commit() event.accept() # # # # # # # # # # Internal utility functions # # # # # # # # # # def active_base_model(self) -> PlaylistModel: return self.current.base_model def active_tab(self) -> PlaylistTab: return self.playlist_section.tabPlaylist.currentWidget() # # # # # # # # # # Menu functions # # # # # # # # # # def create_action( self, text: str, handler: Callable, shortcut: Optional[str] = None ) -> QAction: """ Helper function to create an action, bind it to a method, and set a shortcut if provided. """ action = QAction(text, self) action.triggered.connect(handler) if shortcut: action.setShortcut(shortcut) # Adding the shortcut return action def create_menu_bar(self): """Dynamically creates the menu bar from a YAML file.""" menu_bar = self.menuBar() # Load menu structure from YAML file with open("app/menu.yaml", "r") as file: menu_data = yaml.safe_load(file) self.menu_actions = {} # Store reference for enabling/disabling actions self.dynamic_submenus = {} # Store submenus for dynamic population for menu_item in menu_data["menus"]: menu = menu_bar.addMenu(menu_item["title"]) for action_item in menu_item["actions"]: if "separator" in action_item and action_item["separator"]: menu.addSeparator() continue # Check whether this is a submenu first if action_item.get("submenu"): submenu = QMenu(action_item["text"], self) menu.addMenu(submenu) # Store submenu reference for dynamic population self.dynamic_submenus[action_item["handler"]] = submenu submenu.aboutToShow.connect(self.populate_dynamic_submenu) continue # Skip the rest of the loop (no handler needed) # Now check for a normal menu action handler = getattr(self, action_item["handler"], None) if handler is None: print(f"Warning: No handler found for {action_item['text']}") continue action = self.create_action( action_item["text"], handler, action_item.get("shortcut") ) # Store reference to "Clear Selection" so we can enable/disable it if action_item.get("store_reference"): self.menu_actions[action_item["handler"]] = action menu.addAction(action) def populate_dynamic_submenu(self): """Dynamically populates submenus when they are selected.""" submenu = self.sender() # Get the submenu that triggered the event # Find which submenu it is for key, stored_submenu in self.dynamic_submenus.items(): if submenu == stored_submenu: submenu.clear() # Dynamically call the correct function items = getattr(self, f"get_{key}_items")() for item in items: # Check for separator if "separator" in item and item["separator"]: submenu.addSeparator() continue action = QAction(item["text"], self) # Extract handler and arguments handler = getattr(self, item["handler"], None) args = item.get("args", ()) if handler: # Use a lambda to pass arguments to the function action.triggered.connect(lambda _, h=handler, a=args: h(*a)) submenu.addAction(action) break def get_new_playlist_dynamic_submenu_items( self, ) -> list[dict[str, str | tuple[Session, int] | bool]]: """ Return dynamically generated menu items, in this case templates marked as favourite from which to generate a new playlist. The handler is to call create_playlist with a session and template_id. """ with db.Session() as session: submenu_items: list[dict[str, str | tuple[Session, int] | bool]] = [ { "text": "Show all", "handler": "create_playlist_from_template", "args": (session, 0), }, { "separator": True, }, ] templates = Playlists.get_favourite_templates(session) for template in templates: submenu_items.append( { "text": template.name, "handler": "create_playlist_from_template", "args": ( session, template.id, ), } ) return submenu_items def get_query_dynamic_submenu_items( self, ) -> list[dict[str, str | tuple[Session, int] | bool]]: """ Return dynamically generated menu items, in this case templates marked as favourite from which to generate a new playlist. The handler is to call show_query with a session and query_id. """ with db.Session() as session: submenu_items: list[dict[str, str | tuple[Session, int] | bool]] = [ { "text": "Show all", "handler": "show_query", "args": (session, 0), }, { "separator": True, }, ] queries = Queries.get_favourites(session) for query in queries: submenu_items.append( { "text": query.name, "handler": "show_query", "args": ( session, query.id, ), } ) return submenu_items def show_query(self, session: Session, query_id: int) -> None: """ Show query dialog with query_id selected """ # Keep a reference else it will be gc'd self.query_dialog = QueryDialog(session, query_id) self.query_dialog.exec() # # # # # # # # # # Playlist management functions # # # # # # # # # # def _create_playlist( self, session: Session, name: str, template_id: int ) -> Playlists: """ Create a playlist in the database, populate it from the template if template_id > 0, and return the Playlists object. """ log.debug(f" _create_playlist({name=}, {template_id=})") return Playlists(session, name, template_id) def _open_playlist(self, playlist: Playlists, is_template: bool = False) -> int: """ With passed playlist: - create models - create tab - switch to tab - mark playist as open return: tab index """ log.debug(f" _open_playlist({playlist=}, {is_template=})") # Create base model and proxy model base_model = PlaylistModel(playlist.id, is_template) proxy_model = PlaylistProxyModel() proxy_model.setSourceModel(base_model) # Create tab playlist_tab = PlaylistTab(musicmuster=self, model=proxy_model) idx = self.playlist_section.tabPlaylist.addTab(playlist_tab, playlist.name) # Mark playlist as open playlist.mark_open() # Switch to new tab self.playlist_section.tabPlaylist.setCurrentIndex(idx) self.update_playlist_icons() return idx def create_playlist_from_template(self, session: Session, template_id: int) -> None: """ Prompt for new playlist name and create from passed template_id """ if template_id == 0: # Show all templates selected_template_id = self.solicit_template_to_use(session) if selected_template_id is None: return else: template_id = selected_template_id playlist_name = self.get_playlist_name(session) if not playlist_name: return playlist = self._create_playlist(session, playlist_name, template_id) self._open_playlist(playlist) session.commit() def delete_playlist(self) -> None: """ Delete current playlist """ with db.Session() as session: playlist_id = self.current.playlist_id playlist = session.get(Playlists, playlist_id) if playlist: if helpers.ask_yes_no( "Delete playlist", f"Delete playlist '{playlist.name}': " "Are you sure?", ): if self.close_playlist_tab(): session.delete(playlist) session.commit() else: log.error("Failed to retrieve playlist") def open_existing_playlist(self) -> None: """Open existing playlist""" with db.Session() as session: playlists = Playlists.get_closed(session) dlg = SelectPlaylistDialog(self, playlists=playlists, session=session) dlg.exec() playlist = dlg.playlist if playlist: self._open_playlist(playlist) session.commit() def save_as_template(self) -> None: """Save current playlist as template""" with db.Session() as session: template_names = [a.name for a in Playlists.get_all_templates(session)] while True: # Get name for new template dlg = QInputDialog(self) dlg.setInputMode(QInputDialog.InputMode.TextInput) dlg.setLabelText("Template name:") dlg.resize(500, 100) ok = dlg.exec() if not ok: return template_name = dlg.textValue() if template_name not in template_names: break helpers.show_warning( self, "Duplicate template", "Template name already in use" ) Playlists.save_as_template(session, self.current.playlist_id, template_name) session.commit() helpers.show_OK("Template", "Template saved", self) def get_playlist_name( self, session: Session, default: str = "", prompt: str = "Playlist name:" ) -> Optional[str]: """Get a name from the user""" dlg = QInputDialog(self) dlg.setInputMode(QInputDialog.InputMode.TextInput) dlg.setLabelText(prompt) while True: if default: dlg.setTextValue(default) dlg.resize(500, 100) ok = dlg.exec() if ok: proposed_name = dlg.textValue() if Playlists.name_is_available(session, proposed_name): return proposed_name else: helpers.show_warning( self, "Name in use", f"There's already a playlist called '{proposed_name}'", ) continue else: return None def solicit_template_to_use( self, session: Session, template_prompt: Optional[str] = None ) -> Optional[int]: """ Have user select a template. Return the template.id, or None if they cancel. template_id of zero means don't use a template. """ template_name_id_list: list[tuple[str, int]] = [] template_name_id_list.append((Config.NO_TEMPLATE_NAME, 0)) with db.Session() as session: for template in Playlists.get_all_templates(session): template_name_id_list.append((template.name, template.id)) dlg = TemplateSelectorDialog(template_name_id_list, template_prompt) if not dlg.exec() or dlg.selected_id is None: return None # User cancelled return dlg.selected_id # # # # # # # # # # Manage templates and queries # # # # # # # # # # def manage_queries_wrapper(self): """ Simply instantiate the manage_queries class """ with db.Session() as session: _ = ManageQueries(session, self) def manage_templates_wrapper(self): """ Simply instantiate the manage_queries class """ with db.Session() as session: _ = ManageTemplates(session, self) # # # # # # # # # # Miscellaneous functions # # # # # # # # # # def select_duplicate_rows(self) -> None: """Call playlist to select duplicate rows""" self.active_tab().select_duplicate_rows() def about(self) -> None: """Get git tag and database name""" try: git_tag = str( subprocess.check_output(["git", "describe"], stderr=subprocess.STDOUT) ).strip("'b\\n") except subprocess.CalledProcessError as exc_info: git_tag = str(exc_info.output) with db.Session() as session: if session.bind: dbname = session.bind.engine.url.database QMessageBox.information( self, "About", f"MusicMuster {git_tag}\n\nDatabase: {dbname}", QMessageBox.StandardButton.Ok, ) def clear_next(self) -> None: """ Clear next track """ track_sequence.set_next(None) self.update_headers() def clear_selection(self) -> None: """Clear row selection""" # Unselect any selected rows if self.active_tab(): self.active_tab().clear_selection() # Clear the search bar self.search_playlist_clear() def close_playlist_tab(self) -> bool: """ Close active playlist tab, called by menu item """ return self.close_tab(self.playlist_section.tabPlaylist.currentIndex()) def close_tab(self, tab_index: int) -> bool: """ Close playlist tab unless it holds the current or next track. Called from close_playlist_tab() or by clicking close button on tab. Return True if tab closed else False. """ closing_tab_playlist_id = self.playlist_section.tabPlaylist.widget( tab_index ).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 current_track_playlist_id: if closing_tab_playlist_id == current_track_playlist_id: helpers.show_OK( "Current track", "Can't close current track playlist", self ) 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 next_track_playlist_id: if closing_tab_playlist_id == next_track_playlist_id: helpers.show_OK( "Next track", "Can't close next track playlist", self ) return False # Record playlist as closed and update remaining playlist tabs with db.Session() as session: playlist = session.get(Playlists, closing_tab_playlist_id) if playlist: playlist.close(session) # Close playlist and remove tab self.playlist_section.tabPlaylist.widget(tab_index).close() self.playlist_section.tabPlaylist.removeTab(tab_index) return True def connect_signals_slots(self) -> None: # Menu bars connections are in create_menu_bar() self.footer_section.btnDrop3db.clicked.connect(self.drop3db) self.footer_section.btnFade.clicked.connect(self.fade) self.footer_section.btnHidePlayed.clicked.connect(self.hide_played) self.footer_section.btnPreviewArm.clicked.connect(self.preview_arm) self.footer_section.btnPreviewBack.clicked.connect(self.preview_back) self.footer_section.btnPreview.clicked.connect(self.preview) self.footer_section.btnPreviewEnd.clicked.connect(self.preview_end) self.footer_section.btnPreviewFwd.clicked.connect(self.preview_fwd) self.footer_section.btnPreviewMark.clicked.connect(self.preview_mark) self.footer_section.btnPreviewStart.clicked.connect(self.preview_start) self.footer_section.btnStop.clicked.connect(self.stop) self.header_section.hdrCurrentTrack.clicked.connect(self.show_current) self.header_section.hdrNextTrack.clicked.connect(self.show_next) self.playlist_section.tabPlaylist.currentChanged.connect(self.tab_change) self.playlist_section.tabPlaylist.tabCloseRequested.connect(self.close_tab) self.tabBar = self.playlist_section.tabPlaylist.tabBar() self.txtSearch.textChanged.connect(self.search_playlist_text_changed) self.signals.enable_escape_signal.connect(self.enable_escape) self.signals.next_track_changed_signal.connect(self.update_headers) self.signals.status_message_signal.connect(self.show_status_message) self.signals.show_warning_signal.connect(self.show_warning) self.signals.track_ended_signal.connect(self.end_of_track_actions) self.timer10.timeout.connect(self.tick_10ms) self.timer500.timeout.connect(self.tick_500ms) self.timer100.timeout.connect(self.tick_100ms) self.timer1000.timeout.connect(self.tick_1000ms) self.signals.search_songfacts_signal.connect(self.open_songfacts_browser) self.signals.search_wikipedia_signal.connect(self.open_wikipedia_browser) def current_row_or_end(self) -> int: """ If a row or rows are selected, return the row number of the first selected row otherwise return the row number for a new row at the of the playlist. """ if self.current.selected_rows: return self.current.selected_rows[0] return self.current.base_model.rowCount() def debug(self): """Invoke debugger""" import ipdb # type: ignore ipdb.set_trace() def download_played_tracks(self) -> None: """Download a CSV of played tracks""" dlg = DownloadCSV(self) if dlg.exec(): start_dt = dlg.ui.dateTimeEdit.dateTime().toPyDateTime() # Get output filename pathspec = QFileDialog.getSaveFileName( self, "Save CSV of tracks played", directory="/tmp/playlist.csv", filter="CSV files (*.csv)", ) if not pathspec: return path = pathspec[0] if not path.endswith(".csv"): path += ".csv" with open(path, "w") as f: with db.Session() as session: for playdate in Playdates.played_after(session, start_dt): f.write(f"{playdate.track.artist},{playdate.track.title}\n") 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()) def enable_escape(self, enabled: bool) -> None: """ Manage signal to enable/disable handling ESC character. Needed because we want to use ESC when editing playlist in place, so we need to disable it here while editing. """ log.debug(f"enable_escape({enabled=})") if "clear_selection" in self.menu_actions: self.menu_actions["clear_selection"].setEnabled(enabled) def end_of_track_actions(self) -> None: """ Actions required: - Reset track_sequence objects - Tell playlist track has finished - Reset clocks - Update headers - 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 # Tell playlist previous track has finished self.current.base_model.previous_track_ended() # Reset clocks self.footer_section.frame_fade.setStyleSheet("") self.footer_section.frame_silent.setStyleSheet("") self.footer_section.label_fade_timer.setText("00:00") self.footer_section.label_silent_timer.setText("00:00") # Update headers self.update_headers() # Enable controls self.catch_return_key = False self.show_status_message("Play controls: Enabled", 0) # autoplay # if not self.stop_autoplay: # self.play_next() def export_playlist_tab(self) -> None: """Export the current playlist to an m3u file""" playlist_id = self.current.playlist_id with db.Session() as session: # Get output filename playlist = session.get(Playlists, playlist_id) if not playlist: return pathspec = QFileDialog.getSaveFileName( self, "Save Playlist", directory=f"{playlist.name}.m3u", filter="M3U files (*.m3u);;All files (*.*)", ) if not pathspec: return path = pathspec[0] if not path.endswith(".m3u"): path += ".m3u" # Get list of track rows for this playlist plrs = PlaylistRows.get_rows_with_tracks(session, playlist_id) with open(path, "w") as f: # Required directive on first line f.write("#EXTM3U\n") for track in [a.track for a in plrs]: if track.duration is None: track.duration = 0 f.write( "#EXTINF:" f"{int(track.duration / 1000)}," f"{track.title} - " f"{track.artist}" "\n" f"{track.path}" "\n" ) def fade(self) -> None: """Fade currently playing track""" if track_sequence.current: track_sequence.current.fade() def get_tab_index_for_playlist(self, playlist_id: int) -> Optional[int]: """ Return the tab index for the passed playlist_id if it is displayed, else return None. """ for idx in range(self.playlist_section.tabPlaylist.count()): if self.playlist_section.tabPlaylist.widget(idx).playlist_id == playlist_id: return idx return None def hide_played(self): """Toggle hide played tracks""" if self.hide_played_tracks: self.hide_played_tracks = False self.current.base_model.hide_played_tracks(False) self.footer_section.btnHidePlayed.setText("Hide played") else: self.hide_played_tracks = True self.footer_section.btnHidePlayed.setText("Show played") if Config.HIDE_PLAYED_MODE == Config.HIDE_PLAYED_MODE_SECTIONS: self.active_tab().hide_played_sections() else: self.current.base_model.hide_played_tracks(True) # Reset row heights self.active_tab().resize_rows() def import_files_wrapper(self) -> None: """ Pass import files call to file_importer module """ # We need to keep a reference to the FileImporter else it will be # garbage collected while import threads are still running self.importer = FileImporter(self.current.base_model, self.current_row_or_end()) self.importer.start() def insert_header(self) -> None: """Show dialog box to enter header text and add to playlist""" # Get header text dlg: QInputDialog = QInputDialog(self) dlg.setInputMode(QInputDialog.InputMode.TextInput) dlg.setLabelText("Header text:") dlg.resize(500, 100) ok = dlg.exec() if ok: self.current.base_model.insert_row( proposed_row_number=self.current_row_or_end(), note=dlg.textValue(), ) 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() def load_last_playlists(self) -> None: """Load the playlists that were open when the last session closed""" playlist_ids = [] with db.Session() as session: for playlist in Playlists.get_open(session): if playlist: log.debug(f"load_last_playlists() loaded {playlist=}") # Create tab playlist_ids.append(self._open_playlist(playlist)) # Set active tab record = Settings.get_setting(session, "active_tab") if record.f_int is not None and record.f_int >= 0: self.playlist_section.tabPlaylist.setCurrentIndex(record.f_int) # Tabs may move during use. Rather than track where tabs # are, we record the tab index when we close the main # window. To avoid possible duplicate tab entries, we null # them all out now. Playlists.clear_tabs(session, playlist_ids) session.commit() def lookup_row_in_songfacts(self) -> None: """ Display songfacts page for title in highlighted row """ track_info = self.selected_or_next_track_info() if not track_info: return self.signals.search_songfacts_signal.emit(track_info.title) def lookup_row_in_wikipedia(self) -> None: """ Display Wikipedia page for title in highlighted row or next track """ track_info = self.selected_or_next_track_info() if not track_info: return self.signals.search_wikipedia_signal.emit(track_info.title) def mark_rows_for_moving(self) -> None: """ Cut rows ready for pasting. """ # Save the selected PlaylistRows items ready for a later # paste self.move_source_rows = self.current.selected_rows self.move_source_model = self.current.base_model log.debug( f"mark_rows_for_moving(): {self.move_source_rows=} {self.move_source_model=}" ) def move_playlist_rows(self, row_numbers: list[int]) -> None: """ Move passed playlist rows to another playlist """ # Identify destination playlist playlists = [] source_playlist_id = self.current.playlist_id with db.Session() as session: for playlist in Playlists.get_all(session): if playlist.id == source_playlist_id: continue else: playlists.append(playlist) dlg = SelectPlaylistDialog(self, playlists=playlists, session=session) dlg.exec() if not dlg.playlist: return to_playlist_id = dlg.playlist.id # Get row number in destination playlist last_row = PlaylistRows.get_last_used_row(session, to_playlist_id) if last_row is not None: to_row = last_row + 1 else: to_row = 0 # Move rows self.current.base_model.move_rows_between_playlists( row_numbers, to_row, to_playlist_id ) # 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) def move_selected(self) -> None: """ Move selected rows to another playlist """ selected_rows = self.current.selected_rows if not selected_rows: return self.move_playlist_rows(selected_rows) def move_unplayed(self) -> None: """ Move unplayed rows to another playlist """ unplayed_rows = self.current.base_model.get_unplayed_rows() if not unplayed_rows: return # We can get a race condition as selected rows change while # moving so disable selected rows timing for move self.disable_selection_timing = True self.move_playlist_rows(unplayed_rows) self.disable_selection_timing = False def open_songfacts_browser(self, title: str) -> None: """Search Songfacts for title""" slug = slugify(title, replacements=([["'", ""]])) log.info(f"Songfacts browser tab for {title=}") url = f"https://www.songfacts.com/search/songs/{slug}" webbrowser.get("browser").open_new_tab(url) def open_wikipedia_browser(self, title: str) -> None: """Search Wikipedia for title""" str = urllib.parse.quote_plus(title) log.info(f"Wikipedia browser tab for {title=}") url = f"https://www.wikipedia.org/w/index.php?search={str}" webbrowser.get("browser").open_new_tab(url) def paste_rows(self) -> None: """ Paste earlier cut rows. """ if not self.move_source_rows or not self.move_source_model: return to_playlist_model = self.current.base_model destination_row = self.current_row_or_end() # If we move a row to immediately under the current track, make # 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 ): set_next_row = destination_row if to_playlist_model.playlist_id == self.move_source_model.playlist_id: self.move_source_model.move_rows(self.move_source_rows, destination_row) else: self.move_source_model.move_rows_between_playlists( self.move_source_rows, destination_row, to_playlist_model.playlist_id ) self.active_tab().resize_rows() self.active_tab().clear_selection() if set_next_row: to_playlist_model.set_next_row(set_next_row) def play_next(self, position: Optional[float] = None) -> None: """ Play next track, optionally from passed position. Actions required: - If there is no next track set, return. - Check for inadvertent press of return - If there's currently a track playing, fade it. - Move next track to current track. - Clear next track - Restore volume if -3dB active - Play (new) current track. - Show fade graph - Notify playlist - Note that track is now playing - Disable play next controls - Update headers """ # If there is no next track set, return. if track_sequence.next is None: log.error("musicmuster.play_next(): no next track selected") return # Check for inadvertent press of 'return' if self.return_pressed_in_error(): return # Issue #223 concerns a very short pause (maybe 0.1s) sometimes # when starting to play at track. Resolution appears to be to # disable timer10 for a short time. Timer is re-enabled in # update_clocks. self.timer10.stop() 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() # 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() # 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) # 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 = ( self.footer_section.widgetFadeVolume ) track_sequence.current.fade_graph.clear() track_sequence.current.fade_graph.plot() # Disable play next controls self.catch_return_key = True self.show_status_message("Play controls: Disabled", 0) # Notify playlist self.active_tab().current_track_started() # Update headers self.update_headers() with db.Session() as session: last_played = Playdates.last_played_tracks(session) tracklist = [] for lp in last_played: track = session.get(Tracks, lp.track_id) tracklist.append(f"{track.title} ({track.artist})") tt = "
".join(tracklist) self.header_section.hdrPreviousTrack.setToolTip(tt) def preview(self) -> None: """ Preview selected or next track. We use a different mechanism to normal track playing so that the user can route the output audio differently (eg, to headphones). """ if self.footer_section.btnPreview.isChecked(): # Get track path for first selected track if there is one 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: track_info = TrackInfo( track_sequence.next.track_id, track_sequence.next.row_number ) else: return if not track_info: return self.preview_manager.set_track_info(track_info) self.preview_manager.play() else: self.preview_manager.stop() def preview_arm(self): """Manager arm button for setting intro length""" self.footer_section.btnPreviewMark.setEnabled(self.btnPreviewArm.isChecked()) def preview_back(self) -> None: """Wind back preview file""" self.preview_manager.back(5000) def preview_end(self) -> None: """Advance preview file to Config.PREVIEW_END_BUFFER_MS before end of intro""" if self.preview_manager: self.preview_manager.move_to_intro_end() def preview_fwd(self) -> None: """Advance preview file""" self.preview_manager.forward(5000) def preview_mark(self) -> None: """Set intro time""" if self.preview_manager.is_playing(): track_id = self.preview_manager.track_id row_number = self.preview_manager.row_number if not row_number: return with db.Session() as session: track = session.get(Tracks, track_id) if track: # Save intro as millisends rounded to nearest 0.1 # second because editor spinbox only resolves to 0.1 # seconds intro = round(self.preview_manager.get_playtime() / 100) * 100 track.intro = intro session.commit() self.preview_manager.set_intro(intro) self.current.base_model.refresh_row(session, row_number) self.current.base_model.invalidate_row(row_number) def preview_start(self) -> None: """Restart preview""" self.preview_manager.restart() def quicklog(self) -> None: """ Create log entry """ log.debug("quicklog timestamp; entry follows") # Get log text dlg: QInputDialog = QInputDialog(self) dlg.setInputMode(QInputDialog.InputMode.TextInput) dlg.setLabelText("Log text:") dlg.resize(500, 100) ok = dlg.exec() if ok: log.debug("quicklog: " + dlg.textValue()) def rename_playlist(self) -> None: """ Rename current playlist """ with db.Session() as session: playlist_id = self.current.playlist_id playlist = session.get(Playlists, playlist_id) if playlist: new_name = self.get_playlist_name(session, playlist.name) if new_name: playlist.rename(session, new_name) idx = self.tabBar.currentIndex() self.tabBar.setTabText(idx, new_name) session.commit() def return_pressed_in_error(self) -> bool: """ Check whether Return key has been pressed in error. Return True if it has, False if not """ if 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 + dt.timedelta(milliseconds=Config.RETURN_KEY_DEBOUNCE_MS) > dt.datetime.now() ): return True # 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() * 1000 > Config.PLAY_NEXT_GUARD_MS ) if default_yes: msg = "Hit return to play next track now" else: msg = "Press tab to select Yes and hit return to play next track" if not helpers.ask_yes_no( "Play next track", msg, default_yes=default_yes, parent=self, ): return True return False def resume(self) -> None: """ Resume playing last track. We may be playing the next track or none; take care of both eventualities. Actions required: - Return if no saved position - Resume last track - If a track is playing, make that the next track """ if not track_sequence.previous: return # Return if no saved position resume_marker = 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) # Now resume playing the now-next track self.play_next(resume_marker) # Adjust track info so that clocks and graph are correct. # 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 ): elapsed_ms = ( track_sequence.current.duration * track_sequence.current.resume_marker ) track_sequence.current.start_time -= dt.timedelta(milliseconds=elapsed_ms) def search_playlist(self) -> None: """Show text box to search playlist""" # Disable play controls so that 'return' in search box doesn't # play next track self.catch_return_key = True self.txtSearch.setHidden(False) self.txtSearch.setFocus() # Select any text that may already be there self.txtSearch.selectAll() def search_playlist_clear(self) -> None: """Tidy up and reset search bar""" # Clean up search bar self.txtSearch.setText("") self.txtSearch.setHidden(True) def search_playlist_text_changed(self) -> None: """ Incremental search of playlist """ self.current.proxy_model.set_incremental_search(self.txtSearch.text()) def selected_or_next_track_info(self) -> Optional[RowAndTrack]: """ Return RowAndTrack info for selected track. If no selected track, return for next track. If no next track, return None. """ row_number: Optional[int] = None if self.current.selected_rows: row_number = self.current.selected_rows[0] if row_number is None: if track_sequence.next: if track_sequence.next.track_id: row_number = track_sequence.next.row_number if row_number is None: return None track_info = self.current.base_model.get_row_info(row_number) if track_info is None: return None return track_info def set_main_window_size(self) -> None: """Set size of window from database""" with db.Session() as session: x = Settings.get_setting(session, "mainwindow_x").f_int or 100 y = Settings.get_setting(session, "mainwindow_y").f_int or 100 width = Settings.get_setting(session, "mainwindow_width").f_int or 100 height = Settings.get_setting(session, "mainwindow_height").f_int or 100 self.setGeometry(x, y, width, height) def set_selected_track_next(self) -> None: """ Set currently-selected row on visible playlist tab as next track """ playlist_tab = self.active_tab() if playlist_tab: playlist_tab.set_row_as_next_track() else: log.error("No active tab") def set_tab_colour(self, widget: PlaylistTab, colour: QColor) -> None: """ Find the tab containing the widget and set the text colour """ idx = self.playlist_section.tabPlaylist.indexOf(widget) self.playlist_section.tabPlaylist.tabBar().setTabTextColor(idx, colour) def show_current(self) -> None: """Scroll to show current track""" if track_sequence.current: self.show_track(track_sequence.current) def show_warning(self, title: str, body: str) -> None: """ Display a warning dialog """ print(f"show_warning({title=}, {body=})") QMessageBox.warning(self, title, body) def show_next(self) -> None: """Scroll to show next track""" if track_sequence.next: self.show_track(track_sequence.next) def show_status_message(self, message: str, timing: int) -> None: """ Show status message in status bar for timing milliseconds """ if self.statusbar: self.statusbar.showMessage(message, timing) def show_track(self, playlist_track: RowAndTrack) -> None: """Scroll to show track in plt""" # Switch to the correct tab playlist_id = playlist_track.playlist_id if not playlist_id: # No playlist return # Switch to correct tab if playlist_id != self.current.playlist_id: open_idx = self.get_tab_index_for_playlist(playlist_id) if open_idx: self.playlist_section.tabPlaylist.setCurrentIndex(open_idx) else: raise ApplicationError( f"show_track() can't find current playlist tab {playlist_id=}" ) self.active_tab().scroll_to_top(playlist_track.row_number) def stop(self) -> None: """Stop playing immediately""" self.stop_autoplay = True if track_sequence.current: track_sequence.current.stop() def tab_change(self) -> None: """Called when active tab changed""" self.active_tab().tab_live() def tick_10ms(self) -> None: """ Called every 10ms """ if track_sequence.current: track_sequence.current.update_fade_graph() def tick_100ms(self) -> None: """ Called every 100ms """ if track_sequence.current: try: 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() if intro_ms_remaining > 0: self.footer_section.label_intro_timer.setText( f"{intro_ms_remaining / 1000:.1f}" ) if intro_ms_remaining <= Config.INTRO_SECONDS_WARNING_MS: self.footer_section.label_intro_timer.setStyleSheet( f"background: {Config.COLOUR_WARNING_TIMER}" ) return else: if self.footer_section.label_intro_timer.styleSheet() != "": self.footer_section.label_intro_timer.setStyleSheet("") self.footer_section.label_intro_timer.setText("0.0") except AttributeError: # current track ended during servicing tick pass # Ensure preview button is reset if preview finishes playing # Update preview timer if self.footer_section.btnPreview.isChecked(): if self.preview_manager.is_playing(): self.footer_section.btnPreview.setChecked(True) minutes, seconds = divmod( self.preview_manager.get_playtime() / 1000, 60 ) self.footer_section.label_intro_timer.setText( f"{int(minutes)}:{seconds:04.1f}" ) else: self.footer_section.btnPreview.setChecked(False) self.footer_section.label_intro_timer.setText("0.0") self.footer_section.label_intro_timer.setStyleSheet("") self.footer_section.btnPreview.setChecked(False) def tick_1000ms(self) -> None: """ Called every 1000ms """ # Only update play clocks once a second so that their updates # are synchronised (otherwise it looks odd) self.update_clocks() def tick_500ms(self) -> None: """ Called every 500ms """ self.header_section.lblTOD.setText( dt.datetime.now().strftime(Config.TOD_TIME_FORMAT) ) def update_clocks(self) -> None: """ Update track clocks. """ # If track is playing, update track clocks time and colours if track_sequence.current and 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(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() self.footer_section.label_fade_timer.setText( helpers.ms_to_mmss(time_to_fade) ) # If silent in the next 5 seconds, put warning colour on # time to silence box and enable play controls if time_to_silence <= Config.WARNING_MS_BEFORE_SILENCE: css_silence = f"background: {Config.COLOUR_ENDING_TIMER}" if self.footer_section.frame_silent.styleSheet() != css_silence: self.footer_section.frame_silent.setStyleSheet(css_silence) self.catch_return_key = False self.show_status_message("Play controls: Enabled", 0) # Set warning colour on time to silence box when fade starts elif time_to_fade <= 500: css_fade = f"background: {Config.COLOUR_WARNING_TIMER}" if self.footer_section.frame_silent.styleSheet() != css_fade: self.footer_section.frame_silent.setStyleSheet(css_fade) # WARNING_MS_BEFORE_FADE milliseconds before fade starts, set # warning colour on time to silence box and enable play # controls. This is also a good time to re-enable the 10ms # timer (see play_next() and issue #223). elif time_to_fade <= Config.WARNING_MS_BEFORE_FADE: self.footer_section.frame_fade.setStyleSheet( f"background: {Config.COLOUR_WARNING_TIMER}" ) self.catch_return_key = False self.show_status_message("Play controls: Enabled", 0) # Re-enable 10ms timer (see above) if not self.timer10.isActive(): self.timer10.start(10) log.debug("issue223: update_clocks: 10ms timer enabled") else: self.footer_section.frame_silent.setStyleSheet("") self.footer_section.frame_fade.setStyleSheet("") self.footer_section.label_silent_timer.setText( helpers.ms_to_mmss(time_to_silence) ) def update_headers(self) -> None: """ Update last / current / next track headers """ if track_sequence.previous: self.header_section.hdrPreviousTrack.setText( f"{track_sequence.previous.title} - {track_sequence.previous.artist}" ) else: self.header_section.hdrPreviousTrack.setText("") if track_sequence.current: self.header_section.hdrCurrentTrack.setText( f"{track_sequence.current.title.replace('&', '&&')} - " f"{track_sequence.current.artist.replace('&', '&&')}" ) else: self.header_section.hdrCurrentTrack.setText("") if track_sequence.next: self.header_section.hdrNextTrack.setText( f"{track_sequence.next.title.replace('&', '&&')} - " f"{track_sequence.next.artist.replace('&', '&&')}" ) else: self.header_section.hdrNextTrack.setText("") self.update_playlist_icons() def update_playlist_icons(self) -> None: """ Set current / next playlist tab icons """ # 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 ): set_next = False for idx in range(self.tabBar.count()): widget = self.playlist_section.tabPlaylist.widget(idx) if ( track_sequence.next and set_next and widget.playlist_id == 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.playlist_section.tabPlaylist.setTabIcon( idx, QIcon(Config.PLAYLIST_ICON_CURRENT) ) elif ( self.playlist_section.tabPlaylist.widget(idx) .model() .sourceModel() .is_template ): self.playlist_section.tabPlaylist.setTabIcon( idx, QIcon(Config.PLAYLIST_ICON_TEMPLATE) ) else: self.playlist_section.tabPlaylist.setTabIcon(idx, QIcon()) if __name__ == "__main__": """ If command line arguments given, carry out requested function and exit. Otherwise run full application. """ p = argparse.ArgumentParser() # Only allow at most one option to be specified group = p.add_mutually_exclusive_group() group.add_argument( "-b", "--bitrates", action="store_true", dest="update_bitrates", default=False, help="Update bitrates in database", ) group.add_argument( "-c", "--check-database", action="store_true", dest="check_db", default=False, help="Check and report on database", ) args = p.parse_args() # Run as required if args.check_db: log.debug("Checking database") with db.Session() as session: check_db(session) elif args.update_bitrates: log.debug("Update bitrates") with db.Session() as session: update_bitrates(session) else: app = QApplication(sys.argv) try: # PyQt6 defaults to a grey for labels palette = app.palette() palette.setColor( QPalette.ColorRole.WindowText, QColor(Config.COLOUR_LABEL_TEXT) ) # Set colours that will be used by playlist row stripes palette.setColor( QPalette.ColorRole.Base, QColor(Config.COLOUR_EVEN_PLAYLIST) ) palette.setColor( QPalette.ColorRole.AlternateBase, QColor(Config.COLOUR_ODD_PLAYLIST) ) app.setPalette(palette) win = Window() win.show() status = app.exec() sys.exit(status) except Exception as exc: if os.environ["MM_ENV"] == "PRODUCTION": from helpers import send_mail msg = stackprinter.format(exc) send_mail( ",".join(Config.ERRORS_TO), ",".join(Config.ERRORS_FROM), "Exception from musicmuster.py", msg, ) log.debug(msg) else: print( stackprinter.format( exc, suppressed_paths=["/pypoetry/virtualenvs/"], style="darkbg" ) )