1151 lines
39 KiB
Python
1151 lines
39 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from enum import auto, Enum
|
|
from operator import attrgetter
|
|
from os.path import basename
|
|
from typing import List, Optional
|
|
|
|
from PyQt6.QtCore import (
|
|
pyqtSignal,
|
|
QAbstractTableModel,
|
|
QModelIndex,
|
|
QObject,
|
|
Qt,
|
|
QThread,
|
|
QVariant,
|
|
)
|
|
from PyQt6.QtGui import (
|
|
QBrush,
|
|
QColor,
|
|
QFont,
|
|
)
|
|
|
|
from classes import track_sequence, MusicMusterSignals, PlaylistTrack
|
|
from config import Config
|
|
from dbconfig import scoped_session, Session
|
|
from helpers import (
|
|
file_is_unreadable,
|
|
get_embedded_time,
|
|
get_file_metadata,
|
|
get_relative_date,
|
|
open_in_audacity,
|
|
normalise_track,
|
|
ms_to_mmss,
|
|
set_track_metadata,
|
|
)
|
|
from log import log
|
|
from models import Playdates, PlaylistRows, Tracks
|
|
|
|
|
|
HEADER_NOTES_COLUMN = 1
|
|
|
|
|
|
class Col(Enum):
|
|
START_GAP = 0
|
|
TITLE = auto()
|
|
ARTIST = auto()
|
|
DURATION = auto()
|
|
START_TIME = auto()
|
|
END_TIME = auto()
|
|
LAST_PLAYED = auto()
|
|
BITRATE = auto()
|
|
NOTE = auto()
|
|
|
|
|
|
class PlaylistRowData:
|
|
def __init__(self, plr: PlaylistRows) -> None:
|
|
"""
|
|
Populate PlaylistRowData from database PlaylistRows record
|
|
"""
|
|
|
|
self.artist: str = ""
|
|
self.bitrate = 0
|
|
self.duration: int = 0
|
|
self.lastplayed: datetime = Config.EPOCH
|
|
self.path = ""
|
|
self.played = False
|
|
self.start_gap: Optional[int] = None
|
|
self.title: str = ""
|
|
|
|
self.plrid: int = plr.id
|
|
self.plr_rownum: int = plr.plr_rownum
|
|
self.note: str = plr.note
|
|
self.track_id = plr.track_id
|
|
if plr.track:
|
|
self.start_gap = plr.track.start_gap
|
|
self.title = plr.track.title
|
|
self.artist = plr.track.artist
|
|
self.duration = plr.track.duration
|
|
self.played = plr.played
|
|
if plr.track.playdates:
|
|
self.lastplayed = max([a.lastplayed for a in plr.track.playdates])
|
|
else:
|
|
self.lastplayed = Config.EPOCH
|
|
self.bitrate = plr.track.bitrate or 0
|
|
self.path = plr.track.path
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<PlaylistRowData: plrid={self.plrid}, plr_rownum={self.plr_rownum}, "
|
|
f"note='{self.note}', title='{self.title}', artist='{self.artist}'>"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class StartEndTimes:
|
|
start_time: Optional[datetime] = None
|
|
end_time: Optional[datetime] = None
|
|
|
|
|
|
class PlaylistModel(QAbstractTableModel):
|
|
"""
|
|
The Playlist Model
|
|
|
|
Update strategy: update the database and then refresh the
|
|
row-indexed cached copy (self.playlist_rows). Do not edit
|
|
self.playlist_rows directly because keeping it and the
|
|
database in sync is uncessarily challenging.
|
|
|
|
refresh_row() will populate one row of playlist_rows from the
|
|
database
|
|
|
|
refresh_data() will repopulate all of playlist_rows from the
|
|
database
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
playlist_id: int,
|
|
*args,
|
|
**kwargs,
|
|
):
|
|
self.playlist_id = playlist_id
|
|
super().__init__(*args, **kwargs)
|
|
|
|
self.playlist_rows: dict[int, PlaylistRowData] = {}
|
|
self.start_end_times: dict[int, StartEndTimes] = {}
|
|
self.signals = MusicMusterSignals()
|
|
|
|
self.signals.add_track_to_header_signal.connect(self.add_track_to_header)
|
|
self.signals.add_track_to_playlist_signal.connect(self.add_track)
|
|
|
|
with Session() as session:
|
|
# Ensure row numbers in playlist are contiguous
|
|
PlaylistRows.fixup_rownumbers(session, playlist_id)
|
|
# Populate self.playlist_rows
|
|
self.refresh_data(session)
|
|
self.update_track_times()
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<PlaylistModel: playlist_id={self.playlist_id}, {self.rowCount()} rows>"
|
|
)
|
|
|
|
def add_track(
|
|
self,
|
|
playlist_id: int,
|
|
new_row_number: int,
|
|
track_id: Optional[int],
|
|
note: Optional[str],
|
|
) -> None:
|
|
"""
|
|
Add track if it's for our playlist
|
|
"""
|
|
|
|
# Ignore if it's not for us
|
|
if playlist_id != self.playlist_id:
|
|
return
|
|
|
|
self.insert_row(
|
|
proposed_row_number=new_row_number, track_id=track_id, note=note
|
|
)
|
|
|
|
def add_track_to_header(
|
|
self,
|
|
playlist_id: int,
|
|
row_number: int,
|
|
track_id: int,
|
|
) -> None:
|
|
"""
|
|
Add track to existing header row if it's for our playlist
|
|
"""
|
|
|
|
# Ignore if it's not for us
|
|
if playlist_id != self.playlist_id:
|
|
return
|
|
|
|
# Get existing row
|
|
try:
|
|
prd = self.playlist_rows[row_number]
|
|
except KeyError:
|
|
log.error(
|
|
f"KeyError in PlaylistModel:add_track_to_header ({playlist_id=}, "
|
|
f"{row_number=}, {track_id=}, {len(self.playlist_rows)=}"
|
|
)
|
|
return
|
|
if prd.path:
|
|
log.error(
|
|
f"Error in PlaylistModel:add_track_to_header ({prd=}, "
|
|
"Header row already has track associated"
|
|
)
|
|
return
|
|
with Session() as session:
|
|
plr = session.get(PlaylistRows, prd.plrid)
|
|
if plr:
|
|
# Add track to PlaylistRows
|
|
plr.track_id = track_id
|
|
# Reset header row spanning
|
|
self.signals.span_cells_signal.emit(
|
|
row_number, HEADER_NOTES_COLUMN, 1, 1
|
|
)
|
|
# Update local copy
|
|
self.refresh_row(session, row_number)
|
|
# Repaint row
|
|
self.invalidate_row(row_number)
|
|
|
|
def background_role(self, row: int, column: int, prd: PlaylistRowData) -> QBrush:
|
|
"""Return background setting"""
|
|
|
|
# Handle entire row colouring
|
|
# Header row
|
|
if not prd.path:
|
|
return QBrush(QColor(Config.COLOUR_NOTES_PLAYLIST))
|
|
# Unreadable track file
|
|
if file_is_unreadable(prd.path):
|
|
return QBrush(QColor(Config.COLOUR_UNREADABLE))
|
|
# Current track
|
|
if prd.plrid == track_sequence.now.plr_id:
|
|
return QBrush(QColor(Config.COLOUR_CURRENT_PLAYLIST))
|
|
# Next track
|
|
if prd.plrid == track_sequence.next.plr_id:
|
|
return QBrush(QColor(Config.COLOUR_NEXT_PLAYLIST))
|
|
|
|
# Individual cell colouring
|
|
if column == Col.START_GAP.value:
|
|
if prd.start_gap and prd.start_gap >= Config.START_GAP_WARNING_THRESHOLD:
|
|
return QBrush(QColor(Config.COLOUR_LONG_START))
|
|
if column == Col.BITRATE.value:
|
|
if prd.bitrate < Config.BITRATE_LOW_THRESHOLD:
|
|
return QBrush(QColor(Config.COLOUR_BITRATE_LOW))
|
|
elif prd.bitrate and prd.bitrate < Config.BITRATE_OK_THRESHOLD:
|
|
return QBrush(QColor(Config.COLOUR_BITRATE_MEDIUM))
|
|
else:
|
|
return QBrush(QColor(Config.COLOUR_BITRATE_OK))
|
|
|
|
return QBrush()
|
|
|
|
def columnCount(self, parent: QModelIndex = QModelIndex()) -> int:
|
|
"""Standard function for view"""
|
|
|
|
return 9
|
|
|
|
def current_track_started(self) -> None:
|
|
"""
|
|
Notification from musicmuster that the current track has just
|
|
started playing
|
|
|
|
Actions required:
|
|
- sanity check
|
|
- update display
|
|
- update track times
|
|
- update Playdates in database
|
|
- update PlaylistRows in database
|
|
- find next track
|
|
"""
|
|
|
|
row_number = track_sequence.now.plr_rownum
|
|
|
|
# Sanity check
|
|
if not track_sequence.now.track_id:
|
|
log.error(
|
|
"playlistmodel:current_track_started called with no current track"
|
|
)
|
|
return
|
|
if row_number is None:
|
|
log.error(
|
|
"playlistmodel:current_track_started called with no row number "
|
|
f"({track_sequence.now=})"
|
|
)
|
|
return
|
|
|
|
# Update Playdates in database
|
|
with Session() as session:
|
|
Playdates(session, track_sequence.now.track_id)
|
|
plr = session.get(PlaylistRows, track_sequence.now.plr_id)
|
|
if plr:
|
|
plr.played = True
|
|
self.refresh_row(session, plr.plr_rownum)
|
|
|
|
# Update track times
|
|
self.start_end_times[row_number].start_time = track_sequence.now.start_time
|
|
self.start_end_times[row_number].end_time = track_sequence.now.end_time
|
|
|
|
# Update colour and times for current row
|
|
self.invalidate_row(row_number)
|
|
# Update all other track times
|
|
self.update_track_times()
|
|
|
|
# Find next track
|
|
# Get all unplayed track rows
|
|
next_row = None
|
|
unplayed_rows = [
|
|
a.plr_rownum
|
|
for a in PlaylistRows.get_unplayed_rows(session, self.playlist_id)
|
|
]
|
|
|
|
if unplayed_rows:
|
|
try:
|
|
# Find next row after current track
|
|
next_row = min([a for a in unplayed_rows if a > row_number])
|
|
except ValueError:
|
|
# Find first unplayed track
|
|
next_row = min(unplayed_rows)
|
|
if next_row is not None:
|
|
self.set_next_row(next_row)
|
|
|
|
def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole):
|
|
"""Return data to view"""
|
|
|
|
if not index.isValid() or not (0 <= index.row() < len(self.playlist_rows)):
|
|
return QVariant()
|
|
|
|
row = index.row()
|
|
column = index.column()
|
|
# prd for playlist row data as it's used a lot
|
|
prd = self.playlist_rows[row]
|
|
|
|
# Dispatch to role-specific functions
|
|
if role == Qt.ItemDataRole.DisplayRole:
|
|
return self.display_role(row, column, prd)
|
|
elif role == Qt.ItemDataRole.DecorationRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.EditRole:
|
|
return self.edit_role(row, column, prd)
|
|
elif role == Qt.ItemDataRole.ToolTipRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.StatusTipRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.WhatsThisRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.SizeHintRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.FontRole:
|
|
return self.font_role(row, column, prd)
|
|
elif role == Qt.ItemDataRole.TextAlignmentRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.BackgroundRole:
|
|
return self.background_role(row, column, prd)
|
|
elif role == Qt.ItemDataRole.ForegroundRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.CheckStateRole:
|
|
pass
|
|
elif role == Qt.ItemDataRole.InitialSortOrderRole:
|
|
pass
|
|
|
|
# Fall through to no-op
|
|
return QVariant()
|
|
|
|
def delete_rows(self, row_numbers: List[int]) -> None:
|
|
"""
|
|
Delete passed rows from model
|
|
|
|
Need to delete them in contiguous groups wrapped in beginRemoveRows / endRemoveRows
|
|
calls. To keep it simple, if inefficient, delete rows one by one.
|
|
"""
|
|
|
|
with Session() as session:
|
|
for row_number in row_numbers:
|
|
super().beginRemoveRows(QModelIndex(), row_number, row_number)
|
|
PlaylistRows.delete_row(session, self.playlist_id, row_number)
|
|
super().endRemoveRows()
|
|
|
|
PlaylistRows.fixup_rownumbers(session, self.playlist_id)
|
|
self.refresh_data(session)
|
|
self.update_track_times()
|
|
|
|
def display_role(self, row: int, column: int, prd: PlaylistRowData) -> QVariant:
|
|
"""
|
|
Return text for display
|
|
"""
|
|
|
|
if not prd.path:
|
|
# No track so this is a header row
|
|
if column == HEADER_NOTES_COLUMN:
|
|
self.signals.span_cells_signal.emit(
|
|
row, HEADER_NOTES_COLUMN, 1, self.columnCount() - 1
|
|
)
|
|
return QVariant(self.header_text(prd))
|
|
else:
|
|
return QVariant()
|
|
|
|
if column == Col.START_GAP.value:
|
|
return QVariant(prd.start_gap)
|
|
if column == Col.TITLE.value:
|
|
return QVariant(prd.title)
|
|
if column == Col.ARTIST.value:
|
|
return QVariant(prd.artist)
|
|
if column == Col.DURATION.value:
|
|
return QVariant(ms_to_mmss(prd.duration))
|
|
if column == Col.START_TIME.value:
|
|
if row in self.start_end_times:
|
|
start_time = self.start_end_times[row].start_time
|
|
if start_time:
|
|
return QVariant(start_time.strftime(Config.TRACK_TIME_FORMAT))
|
|
return QVariant()
|
|
if column == Col.END_TIME.value:
|
|
if row in self.start_end_times:
|
|
end_time = self.start_end_times[row].end_time
|
|
if end_time:
|
|
return QVariant(end_time.strftime(Config.TRACK_TIME_FORMAT))
|
|
return QVariant()
|
|
if column == Col.LAST_PLAYED.value:
|
|
return QVariant(get_relative_date(prd.lastplayed))
|
|
if column == Col.BITRATE.value:
|
|
return QVariant(prd.bitrate)
|
|
if column == Col.NOTE.value:
|
|
return QVariant(prd.note)
|
|
|
|
return QVariant()
|
|
|
|
def get_duplicate_rows(self) -> List[int]:
|
|
"""
|
|
Return a list of duplicate rows. If track appears in rows 2, 3 and 4, return [3, 4]
|
|
(ie, ignore the first, not-yet-duplicate, track).
|
|
"""
|
|
|
|
found = []
|
|
result = []
|
|
|
|
for i in range(len(self.playlist_rows)):
|
|
track_id = self.playlist_rows[i].track_id
|
|
if track_id is None:
|
|
continue
|
|
if track_id in found:
|
|
result.append(i)
|
|
else:
|
|
found.append(track_id)
|
|
|
|
return result
|
|
|
|
def edit_role(self, row: int, column: int, prd: PlaylistRowData) -> QVariant:
|
|
"""
|
|
Return text for editing
|
|
"""
|
|
|
|
# If this is a header row and we're being asked for the
|
|
# HEADER_NOTES_COLUMN, return the note value
|
|
if self.is_header_row(row) and column == HEADER_NOTES_COLUMN:
|
|
return QVariant(prd.note)
|
|
|
|
if column == Col.TITLE.value:
|
|
return QVariant(prd.title)
|
|
if column == Col.ARTIST.value:
|
|
return QVariant(prd.artist)
|
|
if column == Col.NOTE.value:
|
|
return QVariant(prd.note)
|
|
|
|
return QVariant()
|
|
|
|
def flags(self, index: QModelIndex) -> Qt.ItemFlag:
|
|
"""
|
|
Standard model flags
|
|
"""
|
|
|
|
if not index.isValid():
|
|
return Qt.ItemFlag.ItemIsDropEnabled
|
|
|
|
default = (
|
|
Qt.ItemFlag.ItemIsEnabled
|
|
| Qt.ItemFlag.ItemIsSelectable
|
|
| Qt.ItemFlag.ItemIsDragEnabled
|
|
)
|
|
if index.column() in [Col.TITLE.value, Col.ARTIST.value, Col.NOTE.value]:
|
|
return default | Qt.ItemFlag.ItemIsEditable
|
|
|
|
return default
|
|
|
|
def font_role(self, row: int, column: int, prd: PlaylistRowData) -> QVariant:
|
|
"""
|
|
Return font
|
|
"""
|
|
|
|
# Notes column is never bold
|
|
if column == Col.NOTE.value:
|
|
return QVariant()
|
|
|
|
boldfont = QFont()
|
|
boldfont.setBold(not self.playlist_rows[row].played)
|
|
|
|
return QVariant(boldfont)
|
|
|
|
def _get_new_row_number(self, proposed_row_number: Optional[int]) -> int:
|
|
"""
|
|
Sanitises proposed new row number.
|
|
|
|
If proposed_row_number given, ensure it is valid.
|
|
If not given, return row number to add to end of model.
|
|
"""
|
|
|
|
if proposed_row_number is None or proposed_row_number > len(self.playlist_rows):
|
|
# We are adding to the end of the list
|
|
new_row_number = len(self.playlist_rows)
|
|
elif proposed_row_number < 0:
|
|
# Add to start of list
|
|
new_row_number = 0
|
|
else:
|
|
new_row_number = proposed_row_number
|
|
|
|
return new_row_number
|
|
|
|
def get_row_track_path(self, row_number: int) -> str:
|
|
"""
|
|
Return path of track associated with row or empty string if no track associated
|
|
"""
|
|
|
|
return self.playlist_rows[row_number].path
|
|
|
|
def get_rows_duration(self, row_numbers: List[int]) -> int:
|
|
"""
|
|
Return the total duration of the passed rows
|
|
"""
|
|
|
|
duration = 0
|
|
for row_number in row_numbers:
|
|
duration += self.playlist_rows[row_number].duration
|
|
|
|
return duration
|
|
|
|
def get_row_info(self, row_number: int) -> PlaylistRowData:
|
|
"""
|
|
Return info about passed row
|
|
"""
|
|
|
|
return self.playlist_rows[row_number]
|
|
|
|
def headerData(
|
|
self,
|
|
section: int,
|
|
orientation: Qt.Orientation,
|
|
role: int = Qt.ItemDataRole.DisplayRole,
|
|
) -> QVariant:
|
|
"""
|
|
Return text for headers
|
|
"""
|
|
|
|
if role == Qt.ItemDataRole.DisplayRole:
|
|
if orientation == Qt.Orientation.Horizontal:
|
|
if section == Col.START_GAP.value:
|
|
return QVariant(Config.HEADER_START_GAP)
|
|
elif section == Col.TITLE.value:
|
|
return QVariant(Config.HEADER_TITLE)
|
|
elif section == Col.ARTIST.value:
|
|
return QVariant(Config.HEADER_ARTIST)
|
|
elif section == Col.DURATION.value:
|
|
return QVariant(Config.HEADER_DURATION)
|
|
elif section == Col.START_TIME.value:
|
|
return QVariant(Config.HEADER_START_TIME)
|
|
elif section == Col.END_TIME.value:
|
|
return QVariant(Config.HEADER_END_TIME)
|
|
elif section == Col.LAST_PLAYED.value:
|
|
return QVariant(Config.HEADER_LAST_PLAYED)
|
|
elif section == Col.BITRATE.value:
|
|
return QVariant(Config.HEADER_BITRATE)
|
|
elif section == Col.NOTE.value:
|
|
return QVariant(Config.HEADER_NOTE)
|
|
else:
|
|
if Config.ROWS_FROM_ZERO:
|
|
return QVariant(str(section))
|
|
else:
|
|
return QVariant(str(section + 1))
|
|
|
|
elif role == Qt.ItemDataRole.FontRole:
|
|
boldfont = QFont()
|
|
boldfont.setBold(True)
|
|
return QVariant(boldfont)
|
|
|
|
return QVariant()
|
|
|
|
def header_text(self, prd: PlaylistRowData) -> str:
|
|
"""
|
|
Process possible section timing directives embeded in header
|
|
"""
|
|
|
|
count: int = 0
|
|
duration: int = 0
|
|
|
|
if prd.note.endswith("+"):
|
|
# This header is the start of a timed section
|
|
for row_number in range(prd.plr_rownum + 1, len(self.playlist_rows)):
|
|
row_prd = self.playlist_rows[row_number]
|
|
if self.is_header_row(row_number):
|
|
if row_prd.note.endswith("-"):
|
|
return (
|
|
f"{prd.note[:-1].strip()} "
|
|
f"[{count} tracks, {ms_to_mmss(duration)} unplayed]"
|
|
)
|
|
else:
|
|
continue
|
|
else:
|
|
count += 1
|
|
if not row_prd.played:
|
|
duration += row_prd.duration
|
|
return (
|
|
f"{prd.note[:-1].strip()} "
|
|
f"[{count} tracks, {ms_to_mmss(duration, none='none')} "
|
|
"unplayed (to end of playlist)]"
|
|
)
|
|
elif prd.note.endswith("="):
|
|
# Show subtotal
|
|
for row_number in range(prd.plr_rownum - 1, -1, -1):
|
|
row_prd = self.playlist_rows[row_number]
|
|
if self.is_header_row(row_number):
|
|
if row_prd.note.endswith("-"):
|
|
# There was no start of section
|
|
return prd.note
|
|
if row_prd.note.endswith(("+", "=")):
|
|
# If we are playing this section, also
|
|
# calculate end time if all tracks are played.
|
|
end_time_str = ""
|
|
if (
|
|
track_sequence.now.plr_rownum
|
|
and track_sequence.now.end_time
|
|
and (
|
|
row_number
|
|
< track_sequence.now.plr_rownum
|
|
< prd.plr_rownum
|
|
)
|
|
):
|
|
section_end_time = track_sequence.now.end_time + timedelta(
|
|
milliseconds=duration
|
|
)
|
|
end_time_str = (
|
|
", section end time "
|
|
+ section_end_time.strftime(Config.TRACK_TIME_FORMAT)
|
|
)
|
|
stripped_note = prd.note[:-1].strip()
|
|
if stripped_note:
|
|
return (
|
|
f"{stripped_note} [{count} track{'s' if count > 1 else ''}, "
|
|
f"{ms_to_mmss(duration)} unplayed{end_time_str}]"
|
|
)
|
|
else:
|
|
return (
|
|
f"[Subtotal: {count} track{'s' if count > 1 else ''}, "
|
|
f"{ms_to_mmss(duration, none='none')} unplayed{end_time_str}]"
|
|
)
|
|
else:
|
|
continue
|
|
else:
|
|
count += 1
|
|
if not row_prd.played:
|
|
duration += row_prd.duration
|
|
|
|
return prd.note
|
|
|
|
def import_tracks(
|
|
self, new_tracks: List[str], proposed_row_number: Optional[int]
|
|
) -> None:
|
|
"""
|
|
Import the file paths listed in new_tracks. The files have already been sanitised
|
|
so no further checks needed. Import in a separate thread as this is slow.
|
|
"""
|
|
|
|
# Import in separate thread
|
|
self.import_thread = QThread()
|
|
self.worker = ImportTrack(
|
|
self,
|
|
new_tracks,
|
|
proposed_row_number,
|
|
)
|
|
self.worker.moveToThread(self.import_thread)
|
|
self.import_thread.started.connect(self.worker.run)
|
|
self.worker.import_finished.connect(self.import_thread.quit)
|
|
self.worker.import_finished.connect(self.worker.deleteLater)
|
|
self.import_thread.finished.connect(self.import_thread.deleteLater)
|
|
self.import_thread.start()
|
|
|
|
def is_header_row(self, row_number: int) -> bool:
|
|
"""
|
|
Return True if row is a header row, else False
|
|
"""
|
|
|
|
return self.playlist_rows[row_number].path == ""
|
|
|
|
def is_unplayed_row(self, row_number: int) -> bool:
|
|
"""
|
|
Return True if row is an unplayed track row, else False
|
|
"""
|
|
|
|
return self.playlist_rows[row_number].played
|
|
|
|
def insert_row(
|
|
self,
|
|
proposed_row_number: Optional[int],
|
|
track_id: Optional[int] = None,
|
|
note: Optional[str] = None,
|
|
) -> PlaylistRows:
|
|
"""
|
|
Insert a track row.
|
|
"""
|
|
|
|
new_row_number = self._get_new_row_number(proposed_row_number)
|
|
|
|
with Session() as session:
|
|
super().beginInsertRows(QModelIndex(), new_row_number, new_row_number)
|
|
plr = PlaylistRows.insert_row(session, self.playlist_id, new_row_number)
|
|
|
|
plr.track_id = track_id
|
|
if note:
|
|
plr.note = note
|
|
|
|
self.refresh_data(session)
|
|
super().endInsertRows()
|
|
|
|
self.invalidate_rows(list(range(plr.plr_rownum, len(self.playlist_rows))))
|
|
|
|
return plr
|
|
|
|
def invalidate_row(self, modified_row: int) -> None:
|
|
"""
|
|
Signal to view to refresh invlidated row
|
|
"""
|
|
|
|
self.dataChanged.emit(
|
|
self.index(modified_row, 0), self.index(modified_row, self.columnCount())
|
|
)
|
|
|
|
def invalidate_rows(self, modified_rows: List[int]) -> None:
|
|
"""
|
|
Signal to view to refresh invlidated rows
|
|
"""
|
|
|
|
for modified_row in modified_rows:
|
|
self.invalidate_row(modified_row)
|
|
|
|
def mark_unplayed(self, row_numbers: List[int]) -> None:
|
|
"""
|
|
Mark row as unplayed
|
|
"""
|
|
|
|
with Session() as session:
|
|
for row_number in row_numbers:
|
|
plr = session.get(PlaylistRows, self.playlist_rows[row_number].plrid)
|
|
if not plr:
|
|
return
|
|
plr.played = False
|
|
self.refresh_row(session, row_number)
|
|
|
|
self.invalidate_rows(row_numbers)
|
|
|
|
def move_rows(self, from_rows: List[int], to_row_number: int) -> None:
|
|
"""
|
|
Move the playlist rows given to to_row and below.
|
|
"""
|
|
|
|
# Build a {current_row_number: new_row_number} dictionary
|
|
row_map: dict[int, int] = {}
|
|
|
|
# Put the from_row row numbers into the row_map. Ultimately the
|
|
# total number of elements in the playlist doesn't change, so
|
|
# check that adding the moved rows starting at to_row won't
|
|
# overshoot the end of the playlist.
|
|
if to_row_number + len(from_rows) > len(self.playlist_rows):
|
|
next_to_row = len(self.playlist_rows) - len(from_rows)
|
|
else:
|
|
next_to_row = to_row_number
|
|
|
|
for from_row, to_row in zip(
|
|
from_rows, range(next_to_row, next_to_row + len(from_rows))
|
|
):
|
|
row_map[from_row] = to_row
|
|
|
|
# Move the remaining rows to the row_map. We want to fill it
|
|
# before (if there are gaps) and after (likewise) the rows that
|
|
# are moving.
|
|
# This iterates old_row and new_row simultaneously.
|
|
for old_row, new_row in zip(
|
|
[x for x in self.playlist_rows.keys() if x not in from_rows],
|
|
[y for y in range(len(self.playlist_rows)) if y not in row_map.values()],
|
|
):
|
|
# Optimise: only add to map if there is a change
|
|
if old_row != new_row:
|
|
row_map[old_row] = new_row
|
|
if self.is_header_row(old_row):
|
|
# Reset column span
|
|
self.signals.span_cells_signal.emit(
|
|
old_row, HEADER_NOTES_COLUMN, 1, 1
|
|
)
|
|
|
|
# Check to see whether any rows in track_sequence have moved
|
|
if track_sequence.previous.plr_rownum in row_map:
|
|
track_sequence.previous.plr_rownum = row_map[
|
|
track_sequence.previous.plr_rownum
|
|
]
|
|
if track_sequence.now.plr_rownum in row_map:
|
|
track_sequence.now.plr_rownum = row_map[track_sequence.now.plr_rownum]
|
|
if track_sequence.next.plr_rownum in row_map:
|
|
track_sequence.next.plr_rownum = row_map[track_sequence.next.plr_rownum]
|
|
|
|
# For SQLAlchemy, build a list of dictionaries that map plrid to
|
|
# new row number:
|
|
sqla_map: List[dict[str, int]] = []
|
|
for oldrow, newrow in row_map.items():
|
|
plrid = self.playlist_rows[oldrow].plrid
|
|
sqla_map.append({"plrid": plrid, "plr_rownum": newrow})
|
|
|
|
with Session() as session:
|
|
PlaylistRows.update_plr_rownumbers(session, self.playlist_id, sqla_map)
|
|
# Update playlist_rows
|
|
self.refresh_data(session)
|
|
|
|
# Update display
|
|
self.update_track_times()
|
|
self.invalidate_rows(list(row_map.keys()))
|
|
|
|
def open_in_audacity(self, row_number: int) -> None:
|
|
"""
|
|
Open track at passed row number in Audacity
|
|
"""
|
|
|
|
path = self.playlist_rows[row_number].path
|
|
if path:
|
|
open_in_audacity(path)
|
|
|
|
def previous_track_ended(self) -> None:
|
|
"""
|
|
Notification from musicmuster that the previous track has ended.
|
|
|
|
Actions required:
|
|
- sanity check
|
|
- update display
|
|
- update track times
|
|
"""
|
|
|
|
# Sanity check
|
|
if not track_sequence.previous.track_id:
|
|
log.error("playlistmodel:previous_track_ended called with no current track")
|
|
return
|
|
if track_sequence.previous.plr_rownum is None:
|
|
log.error(
|
|
"playlistmodel:previous_track_ended called with no row number "
|
|
f"({track_sequence.previous=})"
|
|
)
|
|
return
|
|
|
|
# Update display
|
|
self.invalidate_row(track_sequence.previous.plr_rownum)
|
|
|
|
# Update track times
|
|
# TODO
|
|
|
|
def refresh_data(self, session: scoped_session):
|
|
"""Populate dicts for data calls"""
|
|
|
|
# Populate self.playlist_rows with playlist data
|
|
self.playlist_rows.clear()
|
|
for p in PlaylistRows.deep_rows(session, self.playlist_id):
|
|
self.playlist_rows[p.plr_rownum] = PlaylistRowData(p)
|
|
|
|
def refresh_row(self, session, row_number):
|
|
"""Populate dict for one row from database"""
|
|
|
|
p = PlaylistRows.deep_row(session, self.playlist_id, row_number)
|
|
self.playlist_rows[row_number] = PlaylistRowData(p)
|
|
|
|
def remove_track(self, row_number: int) -> None:
|
|
"""
|
|
Remove track from row
|
|
"""
|
|
|
|
# TODO
|
|
print(f"remove_track({row_number=})")
|
|
|
|
def rescan_track(self, row_number: int) -> None:
|
|
"""
|
|
Rescan track at passed row number
|
|
"""
|
|
|
|
track_id = self.playlist_rows[row_number].track_id
|
|
if track_id:
|
|
with Session() as session:
|
|
track = session.get(Tracks, track_id)
|
|
set_track_metadata(track)
|
|
self.refresh_row(session, row_number)
|
|
self.invalidate_row(row_number)
|
|
|
|
def rowCount(self, index: QModelIndex = QModelIndex()) -> int:
|
|
"""Standard function for view"""
|
|
|
|
return len(self.playlist_rows)
|
|
|
|
def selection_is_sortable(self, row_numbers: List[int]) -> bool:
|
|
"""
|
|
Return True if the selection is sortable. That means:
|
|
- at least two rows selected
|
|
- selected rows are contiguous
|
|
- selected rows do not include any header rows
|
|
"""
|
|
|
|
# at least two rows selected
|
|
if len(row_numbers) < 2:
|
|
return False
|
|
|
|
# selected rows are contiguous
|
|
if sorted(row_numbers) != list(range(min(row_numbers), max(row_numbers) + 1)):
|
|
return False
|
|
|
|
# selected rows do not include any header rows
|
|
for row_number in row_numbers:
|
|
if self.is_header_row(row_number):
|
|
return False
|
|
|
|
return True
|
|
|
|
def set_next_row(self, row_number: Optional[int]) -> None:
|
|
"""
|
|
Set row_number as next track. If row_number is None, clear next track.
|
|
"""
|
|
|
|
if row_number is None:
|
|
next_row_was = track_sequence.next.plr_rownum
|
|
if next_row_was is None:
|
|
return
|
|
track_sequence.next = PlaylistTrack()
|
|
self.invalidate_row(next_row_was)
|
|
return
|
|
|
|
# Update playing_trtack
|
|
with Session() as session:
|
|
track_sequence.next = PlaylistTrack()
|
|
try:
|
|
plrid = self.playlist_rows[row_number].plrid
|
|
except IndexError:
|
|
log.error(
|
|
f"playlistmodel.set_next_track({row_number=}, "
|
|
f"{self.playlist_id=}"
|
|
)
|
|
return
|
|
plr = session.get(PlaylistRows, plrid)
|
|
if plr:
|
|
# Check this isn't a header row
|
|
if self.is_header_row(row_number):
|
|
return
|
|
# Check track is readable
|
|
if file_is_unreadable(plr.track.path):
|
|
return
|
|
track_sequence.next.set_plr(session, plr)
|
|
self.signals.next_track_changed_signal.emit()
|
|
self.invalidate_row(row_number)
|
|
self.update_track_times()
|
|
|
|
def setData(
|
|
self, index: QModelIndex, value: QVariant, role: int = Qt.ItemDataRole.EditRole
|
|
) -> bool:
|
|
"""
|
|
Update model with edited data
|
|
"""
|
|
|
|
if index.isValid() and role == Qt.ItemDataRole.EditRole:
|
|
row_number = index.row()
|
|
column = index.column()
|
|
|
|
with Session() as session:
|
|
plr = session.get(PlaylistRows, self.playlist_rows[row_number].plrid)
|
|
if not plr:
|
|
print(
|
|
f"Error saving data: {row_number=}, {column=}, "
|
|
f"{value=}, {self.playlist_id=}"
|
|
)
|
|
return False
|
|
|
|
if plr.track_id:
|
|
if column == Col.TITLE.value or column == Col.ARTIST.value:
|
|
track = session.get(Tracks, plr.track_id)
|
|
if not track:
|
|
print(f"Error retreiving track: {plr=}")
|
|
return False
|
|
if column == Col.TITLE.value:
|
|
track.title = str(value)
|
|
elif column == Col.ARTIST.value:
|
|
track.artist = str(value)
|
|
else:
|
|
print(f"Error updating track: {column=}, {value=}")
|
|
return False
|
|
elif column == Col.NOTE.value:
|
|
plr.note = str(value)
|
|
else:
|
|
# This is a header row
|
|
if column == HEADER_NOTES_COLUMN:
|
|
plr.note = str(value)
|
|
|
|
# Flush changes before refreshing data
|
|
session.flush()
|
|
self.refresh_row(session, row_number)
|
|
self.dataChanged.emit(index, index, [Qt.ItemDataRole.DisplayRole, role])
|
|
return True
|
|
|
|
return False
|
|
|
|
def sort_by_artist(self, row_numbers: List[int]) -> None:
|
|
"""
|
|
Sort selected rows by artist
|
|
"""
|
|
|
|
self.sort_by_attribute(row_numbers, "artist")
|
|
|
|
def sort_by_attribute(self, row_numbers: List[int], attr_name: str) -> None:
|
|
"""
|
|
Sort selected rows by passed attribute name where 'attribute' is a
|
|
key in PlaylistRowData
|
|
"""
|
|
|
|
# Create a subset of playlist_rows with the rows we are
|
|
# interested in
|
|
shortlist_rows = {k: self.playlist_rows[k] for k in row_numbers}
|
|
sorted_list = [
|
|
plr.plr_rownum
|
|
for plr in sorted(shortlist_rows.values(), key=attrgetter(attr_name))
|
|
]
|
|
self.move_rows(sorted_list, min(sorted_list))
|
|
|
|
def sort_by_duration(self, row_numbers: List[int]) -> None:
|
|
"""
|
|
Sort selected rows by duration
|
|
"""
|
|
|
|
self.sort_by_attribute(row_numbers, "duration")
|
|
|
|
def sort_by_lastplayed(self, row_numbers: List[int]) -> None:
|
|
"""
|
|
Sort selected rows by lastplayed
|
|
"""
|
|
|
|
self.sort_by_attribute(row_numbers, "lastplayed")
|
|
|
|
def sort_by_title(self, row_numbers: List[int]) -> None:
|
|
"""
|
|
Sort selected rows by title
|
|
"""
|
|
|
|
self.sort_by_attribute(row_numbers, "title")
|
|
|
|
def supportedDropActions(self) -> Qt.DropAction:
|
|
return Qt.DropAction.MoveAction | Qt.DropAction.CopyAction
|
|
|
|
def update_track_times(self) -> None:
|
|
"""
|
|
Update track start/end times in self.playlist_rows
|
|
"""
|
|
|
|
next_start_time: Optional[datetime] = None
|
|
update_rows: List[int] = []
|
|
|
|
for row_number in range(len(self.playlist_rows)):
|
|
stend = self.start_end_times[row_number] = StartEndTimes()
|
|
prd = self.playlist_rows[row_number]
|
|
|
|
# Reset start_time if this is the current row
|
|
if row_number == track_sequence.now.plr_rownum:
|
|
stend.start_time = track_sequence.now.start_time
|
|
stend.end_time = track_sequence.now.end_time
|
|
if not next_start_time:
|
|
next_start_time = stend.end_time
|
|
continue
|
|
|
|
# Set start time for next row if we have a current track
|
|
if (
|
|
row_number == track_sequence.next.plr_rownum
|
|
and track_sequence.now.end_time
|
|
):
|
|
stend.start_time = track_sequence.now.end_time
|
|
stend.end_time = stend.start_time + timedelta(milliseconds=prd.duration)
|
|
next_start_time = stend.end_time
|
|
update_rows.append(row_number)
|
|
continue
|
|
|
|
# Don't update times for tracks that have been played
|
|
if prd.played:
|
|
continue
|
|
|
|
# Don't schedule unplayable tracks
|
|
if file_is_unreadable(prd.path):
|
|
continue
|
|
|
|
# If we're between the current and next row, zero out
|
|
# times
|
|
if (
|
|
track_sequence.now.plr_rownum is not None
|
|
and track_sequence.next.plr_rownum is not None
|
|
and track_sequence.now.plr_rownum
|
|
< row_number
|
|
< track_sequence.next.plr_rownum
|
|
):
|
|
update_rows.append(row_number)
|
|
continue
|
|
|
|
# Reset start time if timing in header or at current track
|
|
if self.is_header_row(row_number):
|
|
header_time = get_embedded_time(prd.note)
|
|
if header_time:
|
|
next_start_time = header_time
|
|
else:
|
|
# This is an unplayed track; set start/end if we have a
|
|
# start time
|
|
if next_start_time is None:
|
|
continue
|
|
if stend.start_time != next_start_time:
|
|
stend.start_time = next_start_time
|
|
update_rows.append(row_number)
|
|
next_start_time += timedelta(
|
|
milliseconds=self.playlist_rows[row_number].duration
|
|
)
|
|
if stend.end_time != next_start_time:
|
|
stend.end_time = next_start_time
|
|
update_rows.append(row_number)
|
|
|
|
# Update start/stop times of rows that have changed
|
|
for updated_row in update_rows:
|
|
self.dataChanged.emit(
|
|
self.index(updated_row, Col.START_TIME.value),
|
|
self.index(updated_row, Col.END_TIME.value),
|
|
)
|
|
|
|
|
|
class ImportTrack(QObject):
|
|
import_finished = pyqtSignal()
|
|
|
|
def __init__(self, model: PlaylistModel, filenames: List[str], row_number) -> None:
|
|
super().__init__()
|
|
self.model = model
|
|
self.filenames = filenames
|
|
self.row_number = row_number
|
|
self.signals = MusicMusterSignals()
|
|
|
|
def run(self):
|
|
"""
|
|
Create track objects from passed files and add to visible playlist
|
|
"""
|
|
|
|
target_row = self.row_number
|
|
with Session() as session:
|
|
for fname in self.filenames:
|
|
self.signals.status_message_signal.emit(f"Importing {basename(fname)}", 5000)
|
|
metadata = get_file_metadata(fname)
|
|
try:
|
|
track = Tracks(session, **metadata)
|
|
except Exception as e:
|
|
self.signals.show_warning_signal.emit("Error importing track", e)
|
|
return
|
|
normalise_track(track.path)
|
|
self.model.insert_row(self.row_number, track.id)
|
|
# Insert next row under this one
|
|
target_row += 1
|
|
# We're importing potentially multiple tracks in a loop.
|
|
# If there's an error adding the track to the Tracks
|
|
# table, the session will rollback, thus losing any
|
|
# previous additions in this loop. So, commit now to
|
|
# lock in what we've just done.
|
|
session.commit()
|
|
self.signals.status_message_signal.emit(f"{len(self.filenames)} tracks imported", 10000)
|
|
self.import_finished.emit()
|