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

Rename ds functions; fix add track to header

Browse Source
This commit is contained in:
Keith Edmunds 2025-04-13 09:12:33 +01:00
parent 0c717241ff
commit aec994bafd
14 changed files with 461 additions and 465 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
app/classes.py
View File

@ -207,7 +207,7 @@ class InsertTrack:
@dataclass @dataclass
class PlayTrack: class TrackAndPlaylist:
playlist_id: int playlist_id: int
track_id: int track_id: int
@ -229,7 +229,7 @@ class MusicMusterSignals(QObject):
search_songfacts_signal = pyqtSignal(str) search_songfacts_signal = pyqtSignal(str)
search_wikipedia_signal = pyqtSignal(str) search_wikipedia_signal = pyqtSignal(str)
show_warning_signal = pyqtSignal(str, str) show_warning_signal = pyqtSignal(str, str)
signal_add_track_to_header = pyqtSignal(InsertTrack) signal_add_track_to_header = pyqtSignal(int)
signal_begin_insert_rows = pyqtSignal(InsertRows) signal_begin_insert_rows = pyqtSignal(InsertRows)
signal_end_insert_rows = pyqtSignal(int) signal_end_insert_rows = pyqtSignal(int)
signal_insert_track = pyqtSignal(InsertTrack) signal_insert_track = pyqtSignal(InsertTrack)
@ -239,7 +239,7 @@ class MusicMusterSignals(QObject):
# specify that here as it requires us to import PlaylistRow from # specify that here as it requires us to import PlaylistRow from
# playlistrow.py, which itself imports MusicMusterSignals # playlistrow.py, which itself imports MusicMusterSignals
signal_set_next_track = pyqtSignal(object) signal_set_next_track = pyqtSignal(object)
signal_track_started = pyqtSignal(PlayTrack) signal_track_started = pyqtSignal(TrackAndPlaylist)
span_cells_signal = pyqtSignal(int, int, int, int, int) span_cells_signal = pyqtSignal(int, int, int, int, int)
status_message_signal = pyqtSignal(str, int) status_message_signal = pyqtSignal(str, int)
track_ended_signal = pyqtSignal() track_ended_signal = pyqtSignal()

18
app/dialogs.py
View File

@ -18,7 +18,12 @@ from PyQt6.QtWidgets import (
# Third party imports # Third party imports
# App imports # App imports
from classes import ApplicationError, InsertTrack, MusicMusterSignals from classes import (
ApplicationError,
InsertTrack,
MusicMusterSignals,
TrackAndPlaylist,
)
from helpers import ( from helpers import (
get_relative_date, get_relative_date,
ms_to_mmss, ms_to_mmss,
@ -94,8 +99,8 @@ class TrackInsertDialog(QDialog):
self.setLayout(layout) self.setLayout(layout)
self.resize(800, 600) self.resize(800, 600)
width = ds.get_setting("dbdialog_width") or 800 width = ds.setting_get("dbdialog_width") or 800
height = ds.get_setting("dbdialog_height") or 800 height = ds.setting_get("dbdialog_height") or 800
self.resize(width, height) self.resize(width, height)
self.signals = MusicMusterSignals() self.signals = MusicMusterSignals()
@ -141,7 +146,12 @@ class TrackInsertDialog(QDialog):
self.title_edit.setFocus() self.title_edit.setFocus()
if self.add_to_header: if self.add_to_header:
self.signals.signal_add_track_to_header.emit(insert_track_data) # The model will have the right-clicked row marked as a
# selected_row so we only need to pass the playlist_id and
# track_id.
self.signals.signal_add_track_to_header.emit(
TrackAndPlaylist(playlist_id=self.playlist_id, track_id=track_id)
)
self.accept() self.accept()
else: else:
self.signals.signal_insert_track.emit(insert_track_data) self.signals.signal_insert_track.emit(insert_track_data)

688
app/ds.py
View File

@ -78,7 +78,7 @@ def _remove_substring_case_insensitive(parent_string: str, substring: str) -> st
# Notecolour functions # Notecolour functions
def _all_notecolours(session: Session) -> list[NoteColoursDTO]: def _notecolours_all(session: Session) -> list[NoteColoursDTO]:
""" """
Return all notecolour records Return all notecolour records
""" """
@ -118,13 +118,13 @@ def _all_notecolours(session: Session) -> list[NoteColoursDTO]:
return results return results
def _get_colour_record(text: str) -> tuple[NoteColoursDTO | None, str]: def _notecolors_get_notecolours_dto(text: str) -> tuple[NoteColoursDTO | None, str]:
""" """
Parse text and return first matching colour record or None Parse text and return first matching colour record or None
""" """
with db.Session() as session: with db.Session() as session:
for rec in _all_notecolours(session): for rec in _notecolours_all(session):
if rec.is_regex: if rec.is_regex:
flags = re.UNICODE flags = re.UNICODE
if not rec.is_casesensitive: if not rec.is_casesensitive:
@ -151,13 +151,13 @@ def _get_colour_record(text: str) -> tuple[NoteColoursDTO | None, str]:
return (None, text) return (None, text)
def get_colour(text: str, foreground: bool = False) -> str: def notecolours_get_colour(text: str, foreground: bool = False) -> str:
""" """
Parse text and return background (foreground if foreground==True) Parse text and return background (foreground if foreground==True)
colour string if matched, else None colour string if matched, else None
""" """
(rec, _) = _get_colour_record(text) (rec, _) = _notecolors_get_notecolours_dto(text)
if rec is None: if rec is None:
return "" return ""
elif foreground: elif foreground:
@ -167,12 +167,12 @@ def get_colour(text: str, foreground: bool = False) -> str:
# @log_call # @log_call
def remove_colour_substring(text: str) -> str: def notecolours_remove_colour_substring(text: str) -> str:
""" """
Remove text that identifies the colour to be used if strip_substring is True Remove text that identifies the colour to be used if strip_substring is True
""" """
(rec, stripped_text) = _get_colour_record(text) (rec, stripped_text) = _notecolors_get_notecolours_dto(text)
return stripped_text return stripped_text
@ -239,17 +239,33 @@ def _tracks_where(query: BinaryExpression | ColumnElement[bool],) -> list[TrackD
return results return results
def track_by_path(path: str) -> TrackDTO | None: # @log_call
def track_add_to_header(playlistrow_id: int, track_id: int) -> None:
""" """
Return track with passed path or None Add a track to this (header) row
""" """
track_list = _tracks_where(Tracks.path.ilike(path)) with db.Session() as session:
if not track_list: session.execute(
return None update(PlaylistRows)
if len(track_list) > 1: .where(PlaylistRows.id == playlistrow_id)
raise ApplicationError(f"Duplicate {path=}") .values(track_id=track_id)
return track_list[0] )
session.commit()
def tracks_all() -> list[TrackDTO]:
"""Return a list of all tracks"""
return _tracks_where(Tracks.id > 0)
def tracks_by_artist(filter_str: str) -> list[TrackDTO]:
"""
Return tracks where artist is like filter
"""
return _tracks_where(Tracks.artist.ilike(f"%{filter_str}%"))
def track_by_id(track_id: int) -> TrackDTO | None: def track_by_id(track_id: int) -> TrackDTO | None:
@ -265,12 +281,17 @@ def track_by_id(track_id: int) -> TrackDTO | None:
return track_list[0] return track_list[0]
def tracks_by_artist(filter_str: str) -> list[TrackDTO]: def track_by_path(path: str) -> TrackDTO | None:
""" """
Return tracks where artist is like filter Return track with passed path or None
""" """
return _tracks_where(Tracks.artist.ilike(f"%{filter_str}%")) track_list = _tracks_where(Tracks.path.ilike(path))
if not track_list:
return None
if len(track_list) > 1:
raise ApplicationError(f"Duplicate {path=}")
return track_list[0]
def tracks_by_title(filter_str: str) -> list[TrackDTO]: def tracks_by_title(filter_str: str) -> list[TrackDTO]:
@ -281,13 +302,48 @@ def tracks_by_title(filter_str: str) -> list[TrackDTO]:
return _tracks_where(Tracks.title.ilike(f"%{filter_str}%")) return _tracks_where(Tracks.title.ilike(f"%{filter_str}%"))
def get_all_tracks() -> list[TrackDTO]: # @log_call
"""Return a list of all tracks""" def track_create(metadata: dict[str, str | int | float]) -> TrackDTO:
"""
Create a track db entry from a track path and return the DTO
"""
return _tracks_where(Tracks.id > 0) with db.Session() as session:
try:
track = Tracks(
session=session,
path=str(metadata["path"]),
title=str(metadata["title"]),
artist=str(metadata["artist"]),
duration=int(metadata["duration"]),
start_gap=int(metadata["start_gap"]),
fade_at=int(metadata["fade_at"]),
silence_at=int(metadata["silence_at"]),
bitrate=int(metadata["bitrate"]),
)
track_id = track.id
session.commit()
except Exception:
raise ApplicationError("Can't create Track")
new_track = track_by_id(track_id)
if not new_track:
raise ApplicationError("Unable to create new track")
return new_track
def get_filtered_tracks(filter: Filter) -> list[TrackDTO]: def track_delete(track_id: int) -> None:
"""Delete track"""
with db.Session() as session:
track = session.get(Tracks, track_id)
session.delete(track)
session.commit()
def tracks_filtered(filter: Filter) -> list[TrackDTO]:
""" """
Return tracks matching filter Return tracks matching filter
""" """
@ -378,55 +434,23 @@ def get_filtered_tracks(filter: Filter) -> list[TrackDTO]:
return results return results
# @log_call def track_set_intro(track_id: int, intro: int) -> None:
def add_track_to_header(playlistrow_id: int, track_id: int) -> None:
""" """
Add a track to this (header) row Set track intro time
""" """
with db.Session() as session: with db.Session() as session:
session.execute( session.execute(
update(PlaylistRows) update(Tracks)
.where(PlaylistRows.id == playlistrow_id) .where(Tracks.id == track_id)
.values(track_id=track_id) .values(intro=intro)
) )
session.commit()
session.commit()
# @log_call # @log_call
def create_track(path: str, metadata: dict[str, str | int | float]) -> TrackDTO: def track_update(
"""
Create a track db entry from a track path and return the DTO
"""
with db.Session() as session:
try:
track = Tracks(
session=session,
path=str(metadata["path"]),
title=str(metadata["title"]),
artist=str(metadata["artist"]),
duration=int(metadata["duration"]),
start_gap=int(metadata["start_gap"]),
fade_at=int(metadata["fade_at"]),
silence_at=int(metadata["silence_at"]),
bitrate=int(metadata["bitrate"]),
)
track_id = track.id
session.commit()
except Exception:
raise ApplicationError("Can't create Track")
new_track = track_by_id(track_id)
if not new_track:
raise ApplicationError("Unable to create new track")
return new_track
# @log_call
def update_track(
path: str, track_id: int, metadata: dict[str, str | int | float] path: str, track_id: int, metadata: dict[str, str | int | float]
) -> TrackDTO: ) -> TrackDTO:
""" """
@ -455,31 +479,58 @@ def update_track(
return updated_track return updated_track
def set_track_intro(track_id: int, intro: int) -> None:
"""
Set track intro time
"""
with db.Session() as session:
session.execute(
update(Tracks)
.where(Tracks.id == track_id)
.values(intro=intro)
)
session.commit()
def delete_track(track_id: int) -> None:
"""Delete track"""
with db.Session() as session:
track = session.get(Tracks, track_id)
session.delete(track)
session.commit()
# Playlist functions # Playlist functions
def _playlist_check_playlist(
session: Session, playlist_id: int, fix: bool = False
) -> None:
"""
Ensure the row numbers are contiguous. Fix and log if fix==True,
else raise ApplicationError.
"""
playlist_rows = (
session.execute(
select(PlaylistRows)
.where(PlaylistRows.playlist_id == playlist_id)
.order_by(PlaylistRows.row_number)
)
.scalars()
.all()
)
for idx, plr in enumerate(playlist_rows):
if plr.row_number == idx:
continue
msg = (
"_check_playlist_integrity: incorrect row number "
f"({plr.id=}, {plr.row_number=}, {idx=})"
)
if fix:
log.debug(msg)
plr.row_number = idx
else:
raise ApplicationError(msg)
# @log_call
def _playlist_shift_rows(
session: Session, playlist_id: int, starting_row: int, shift_by: int
) -> None:
"""
Shift rows from starting_row by shift_by. If shift_by is +ve, shift rows
down; if -ve, shift them up.
"""
session.execute(
update(PlaylistRows)
.where(
(PlaylistRows.playlist_id == playlist_id),
(PlaylistRows.row_number >= starting_row),
)
.values(row_number=PlaylistRows.row_number + shift_by)
)
# @log_call # @log_call
def _playlists_where( def _playlists_where(
query: BinaryExpression | ColumnElement[bool], query: BinaryExpression | ColumnElement[bool],
@ -513,6 +564,12 @@ def _playlists_where(
return results return results
def playlists_all():
"""Return all playlists"""
return _playlists_where(Playlists.id > 0)
# @log_call # @log_call
def playlist_by_id(playlist_id: int) -> PlaylistDTO | None: def playlist_by_id(playlist_id: int) -> PlaylistDTO | None:
""" """
@ -527,6 +584,26 @@ def playlist_by_id(playlist_id: int) -> PlaylistDTO | None:
return playlist_list[0] return playlist_list[0]
def playlist_copy(src_id: int, dst_id: int) -> None:
"""Copy playlist entries"""
with db.Session() as session:
src_rows = session.scalars(
select(PlaylistRows).where(PlaylistRows.playlist_id == src_id)
).all()
for plr in src_rows:
PlaylistRows(
session=session,
playlist_id=dst_id,
row_number=plr.row_number,
note=plr.note,
track_id=plr.track_id,
)
session.commit()
def playlists_closed() -> list[PlaylistDTO]: def playlists_closed() -> list[PlaylistDTO]:
""" """
Return a list of closed playlists Return a list of closed playlists
@ -535,44 +612,32 @@ def playlists_closed() -> list[PlaylistDTO]:
return _playlists_where(Playlists.open.is_(False)) return _playlists_where(Playlists.open.is_(False))
def playlists_open() -> list[PlaylistDTO]: # @log_call
def playlist_create(name: str, template_id: int, as_template: bool = False) -> PlaylistDTO:
""" """
Return a list of open playlists Create playlist and return DTO.
""" """
return _playlists_where(Playlists.open.is_(True)) with db.Session() as session:
try:
playlist = Playlists(session, name, template_id)
playlist.is_template = as_template
playlist_id = playlist.id
session.commit()
except Exception:
raise ApplicationError("Can't create Playlist")
if template_id != 0:
playlist_copy(template_id, playlist_id)
new_playlist = playlist_by_id(playlist_id)
if not new_playlist:
raise ApplicationError("Can't retrieve new Playlist")
return new_playlist
def playlists_template_by_id(playlist_id: int) -> PlaylistDTO | None: def playlist_delete(playlist_id: int) -> None:
"""
Return a list of closed playlists
"""
playlist_list = _playlists_where(
Playlists.playlist_id == playlist_id, Playlists.is_template.is_(True)
)
if not playlist_list:
return None
if len(playlist_list) > 1:
raise ApplicationError(f"Duplicate {playlist_id=}")
return playlist_list[0]
def playlists_templates() -> list[PlaylistDTO]:
"""
Return a list of playlist templates
"""
return _playlists_where(Playlists.is_template.is_(True))
def get_all_playlists():
"""Return all playlists"""
return _playlists_where(Playlists.id > 0)
def delete_playlist(playlist_id: int) -> None:
"""Delete playlist""" """Delete playlist"""
with db.Session() as session: with db.Session() as session:
@ -581,61 +646,44 @@ def delete_playlist(playlist_id: int) -> None:
session.commit() session.commit()
def save_as_template(playlist_id: int, template_name: str) -> None: # @log_call
def playlist_insert_row(
playlist_id: int, row_number: int, track_id: int | None, note: str
) -> PlaylistRowDTO:
""" """
Save playlist as templated Insert a new row into playlist and return new row DTO
"""
new_template = create_playlist(template_name, 0, as_template=True)
copy_playlist(playlist_id, new_template.id)
def playlist_rename(playlist_id: int, new_name: str) -> None:
"""
Rename playlist
""" """
with db.Session() as session: with db.Session() as session:
session.execute( # Sanity check
update(Playlists) _playlist_check_playlist(session, playlist_id, fix=False)
.where(Playlists.id == playlist_id)
.values(name=new_name) # Make space for new row
_playlist_shift_rows(
session=session,
playlist_id=playlist_id,
starting_row=row_number,
shift_by=1,
) )
session.commit() playlist_row = PlaylistRows(
session=session,
playlist_id=playlist_id,
def _check_playlist_integrity( row_number=row_number,
session: Session, playlist_id: int, fix: bool = False note=note,
) -> None: track_id=track_id,
"""
Ensure the row numbers are contiguous. Fix and log if fix==True,
else raise ApplicationError.
"""
playlist_rows = (
session.execute(
select(PlaylistRows)
.where(PlaylistRows.playlist_id == playlist_id)
.order_by(PlaylistRows.row_number)
) )
.scalars() session.commit()
.all() playlist_row_id = playlist_row.id
)
for idx, plr in enumerate(playlist_rows):
if plr.row_number == idx:
continue
msg = ( # Sanity check
"_check_playlist_integrity: incorrect row number " _playlist_check_playlist(session, playlist_id, fix=False)
f"({plr.id=}, {plr.row_number=}, {idx=})"
) new_playlist_row = playlistrow_by_id(playlistrow_id=playlist_row_id)
if fix: if not new_playlist_row:
log.debug(msg) raise ApplicationError("Can't retrieve new playlist row")
plr.row_number = idx
else: return new_playlist_row
raise ApplicationError(msg)
# @log_call # @log_call
@ -653,26 +701,7 @@ def playlist_mark_status(playlist_id: int, open: bool) -> None:
# @log_call # @log_call
def _shift_rows( def playlist_move_rows(
session: Session, playlist_id: int, starting_row: int, shift_by: int
) -> None:
"""
Shift rows from starting_row by shift_by. If shift_by is +ve, shift rows
down; if -ve, shift them up.
"""
session.execute(
update(PlaylistRows)
.where(
(PlaylistRows.playlist_id == playlist_id),
(PlaylistRows.row_number >= starting_row),
)
.values(row_number=PlaylistRows.row_number + shift_by)
)
# @log_call
def move_rows(
from_rows: list[int], from_rows: list[int],
from_playlist_id: int, from_playlist_id: int,
to_row: int, to_row: int,
@ -698,18 +727,18 @@ def move_rows(
with db.Session() as session: with db.Session() as session:
# Sanity check row numbers # Sanity check row numbers
_check_playlist_integrity(session, from_playlist_id, fix=False) _playlist_check_playlist(session, from_playlist_id, fix=False)
if from_playlist_id != to_playlist_id: if from_playlist_id != to_playlist_id:
_check_playlist_integrity(session, to_playlist_id, fix=False) _playlist_check_playlist(session, to_playlist_id, fix=False)
# Check there are no playlist rows with playlist_id == PENDING_MOVE # Check there are no playlist rows with playlist_id == PENDING_MOVE
pending_move_rows = get_playlist_rows(Config.PLAYLIST_PENDING_MOVE) pending_move_rows = playlistrows_by_playlist(Config.PLAYLIST_PENDING_MOVE)
if pending_move_rows: if pending_move_rows:
raise ApplicationError(f"move_rows_to_playlist: {pending_move_rows=}") raise ApplicationError(f"move_rows_to_playlist: {pending_move_rows=}")
# We need playlist length if we're moving within a playlist. Get # We need playlist length if we're moving within a playlist. Get
# that now before we remove rows. # that now before we remove rows.
from_playlist_length = len(get_playlist_rows(from_playlist_id)) from_playlist_length = len(playlistrows_by_playlist(from_playlist_id))
# Put rows to be moved into PENDING_MOVE playlist # Put rows to be moved into PENDING_MOVE playlist
session.execute( session.execute(
update(PlaylistRows) update(PlaylistRows)
@ -721,7 +750,7 @@ def move_rows(
) )
# Resequence remaining row numbers # Resequence remaining row numbers
_check_playlist_integrity(session, from_playlist_id, fix=True) _playlist_check_playlist(session, from_playlist_id, fix=True)
session.commit() session.commit()
# Make space for moved rows. If moving within one playlist, # Make space for moved rows. If moving within one playlist,
@ -738,13 +767,13 @@ def move_rows(
to_row - overflow - len([a for a in from_rows if a > to_row]) to_row - overflow - len([a for a in from_rows if a > to_row])
) )
_shift_rows(session, to_playlist_id, space_row, len(from_rows)) _playlist_shift_rows(session, to_playlist_id, space_row, len(from_rows))
# Move the PENDING_MOVE rows back and fixup row numbers # Move the PENDING_MOVE rows back and fixup row numbers
update_list: list[dict[str, int]] = [] update_list: list[dict[str, int]] = []
next_row = space_row next_row = space_row
# PLAYLIST_PENDING_MOVE may have gaps so don't check it # PLAYLIST_PENDING_MOVE may have gaps so don't check it
for row_to_move in get_playlist_rows( for row_to_move in playlistrows_by_playlist(
Config.PLAYLIST_PENDING_MOVE, check_playlist_itegrity=False Config.PLAYLIST_PENDING_MOVE, check_playlist_itegrity=False
): ):
update_list.append( update_list.append(
@ -758,71 +787,32 @@ def move_rows(
session.commit() session.commit()
# Sanity check row numbers # Sanity check row numbers
_check_playlist_integrity(session, from_playlist_id, fix=False) _playlist_check_playlist(session, from_playlist_id, fix=False)
if from_playlist_id != to_playlist_id: if from_playlist_id != to_playlist_id:
_check_playlist_integrity(session, to_playlist_id, fix=False) _playlist_check_playlist(session, to_playlist_id, fix=False)
# @log_call def playlists_open() -> list[PlaylistDTO]:
def update_row_numbers(
playlist_id: int, id_to_row_number: list[dict[int, int]]
) -> None:
""" """
Update playlistrows rownumbers for passed playlistrow_ids Return a list of open playlists
playlist_id is only needed for sanity checking """
return _playlists_where(Playlists.open.is_(True))
def playlist_rename(playlist_id: int, new_name: str) -> None:
"""
Rename playlist
""" """
with db.Session() as session: with db.Session() as session:
session.execute(update(PlaylistRows), id_to_row_number) session.execute(
session.commit() update(Playlists)
.where(Playlists.id == playlist_id)
.values(name=new_name)
)
# Sanity check session.commit()
_check_playlist_integrity(session, playlist_id, fix=False)
# @log_call
def create_playlist(name: str, template_id: int, as_template: bool = False) -> PlaylistDTO:
"""
Create playlist and return DTO.
"""
with db.Session() as session:
try:
playlist = Playlists(session, name, template_id)
playlist.is_template = as_template
playlist_id = playlist.id
session.commit()
except Exception:
raise ApplicationError("Can't create Playlist")
if template_id != 0:
copy_playlist(template_id, playlist_id)
new_playlist = playlist_by_id(playlist_id)
if not new_playlist:
raise ApplicationError("Can't retrieve new Playlist")
return new_playlist
def copy_playlist(src_id: int, dst_id: int) -> None:
"""Copy playlist entries"""
with db.Session() as session:
src_rows = session.scalars(
select(PlaylistRows).where(PlaylistRows.playlist_id == src_id)
).all()
for plr in src_rows:
PlaylistRows(
session=session,
playlist_id=dst_id,
row_number=plr.row_number,
note=plr.note,
track_id=plr.track_id,
)
session.commit()
def playlist_row_count(playlist_id: int) -> int: def playlist_row_count(playlist_id: int) -> int:
@ -840,48 +830,58 @@ def playlist_row_count(playlist_id: int) -> int:
return count return count
# @log_call def playlist_save_as_template(playlist_id: int, template_name: str) -> None:
def insert_row(
playlist_id: int, row_number: int, track_id: int | None, note: str
) -> PlaylistRowDTO:
""" """
Insert a new row into playlist and return new row DTO Save playlist as templated
"""
new_template = playlist_create(template_name, 0, as_template=True)
playlist_copy(playlist_id, new_template.id)
def playlists_templates_all() -> list[PlaylistDTO]:
"""
Return a list of playlist templates
"""
return _playlists_where(Playlists.is_template.is_(True))
def playlists_template_by_id(playlist_id: int) -> PlaylistDTO | None:
"""
Return a list of closed playlists
"""
playlist_list = _playlists_where(
Playlists.playlist_id == playlist_id, Playlists.is_template.is_(True)
)
if not playlist_list:
return None
if len(playlist_list) > 1:
raise ApplicationError(f"Duplicate {playlist_id=}")
return playlist_list[0]
# @log_call
def playlist_update_row_numbers(
playlist_id: int, id_to_row_number: list[dict[int, int]]
) -> None:
"""
Update playlistrows rownumbers for passed playlistrow_ids
playlist_id is only needed for sanity checking
""" """
with db.Session() as session: with db.Session() as session:
# Sanity check session.execute(update(PlaylistRows), id_to_row_number)
_check_playlist_integrity(session, playlist_id, fix=False)
# Make space for new row
_shift_rows(
session=session,
playlist_id=playlist_id,
starting_row=row_number,
shift_by=1,
)
playlist_row = PlaylistRows(
session=session,
playlist_id=playlist_id,
row_number=row_number,
note=note,
track_id=track_id,
)
session.commit() session.commit()
playlist_row_id = playlist_row.id
# Sanity check # Sanity check
_check_playlist_integrity(session, playlist_id, fix=False) _playlist_check_playlist(session, playlist_id, fix=False)
new_playlist_row = get_playlist_row(playlistrow_id=playlist_row_id)
if not new_playlist_row:
raise ApplicationError("Can't retrieve new playlist row")
return new_playlist_row
# @log_call # @log_call
def remove_comments(playlist_id: int, row_numbers: list[int]) -> None: def playlist_remove_comments(playlist_id: int, row_numbers: list[int]) -> None:
""" """
Remove comments from rows in playlist Remove comments from rows in playlist
""" """
@ -899,7 +899,7 @@ def remove_comments(playlist_id: int, row_numbers: list[int]) -> None:
# @log_call # @log_call
def remove_rows(playlist_id: int, row_numbers: list[int]) -> None: def playlist_remove_rows(playlist_id: int, row_numbers: list[int]) -> None:
""" """
Remove rows from playlist Remove rows from playlist
@ -915,24 +915,11 @@ def remove_rows(playlist_id: int, row_numbers: list[int]) -> None:
) )
) )
# Fixup row number to remove gaps # Fixup row number to remove gaps
_check_playlist_integrity(session, playlist_id, fix=True) _playlist_check_playlist(session, playlist_id, fix=True)
session.commit() session.commit()
# @log_call
def update_template_favourite(template_id: int, favourite: bool) -> None:
"""Update template favourite"""
with db.Session() as session:
session.execute(
update(Playlists)
.where(Playlists.id == template_id)
.values(favourite=favourite)
)
session.commit()
# @log_call # @log_call
def playlist_save_tabs(playlist_id_to_tab: dict[int, int]) -> None: def playlist_save_tabs(playlist_id_to_tab: dict[int, int]) -> None:
""" """
@ -955,10 +942,23 @@ def playlist_save_tabs(playlist_id_to_tab: dict[int, int]) -> None:
session.commit() session.commit()
# @log_call
def playlist_update_template_favourite(template_id: int, favourite: bool) -> None:
"""Update template favourite"""
with db.Session() as session:
session.execute(
update(Playlists)
.where(Playlists.id == template_id)
.values(favourite=favourite)
)
session.commit()
# Playlist Rows # Playlist Rows
# @log_call # @log_call
def get_playlist_row(playlistrow_id: int) -> PlaylistRowDTO | None: def playlistrow_by_id(playlistrow_id: int) -> PlaylistRowDTO | None:
""" """
Return specific row DTO Return specific row DTO
""" """
@ -988,7 +988,7 @@ def get_playlist_row(playlistrow_id: int) -> PlaylistRowDTO | None:
return dto return dto
def get_playlist_rows( def playlistrows_by_playlist(
playlist_id: int, check_playlist_itegrity: bool = True playlist_id: int, check_playlist_itegrity: bool = True
) -> list[PlaylistRowDTO]: ) -> list[PlaylistRowDTO]:
@ -996,7 +996,7 @@ def get_playlist_rows(
# TODO: would be good to be confident at removing this # TODO: would be good to be confident at removing this
if check_playlist_itegrity: if check_playlist_itegrity:
_check_playlist_integrity( _playlist_check_playlist(
session=session, playlist_id=playlist_id, fix=False session=session, playlist_id=playlist_id, fix=False
) )
@ -1029,7 +1029,7 @@ def get_playlist_rows(
# Playdates # Playdates
# @log_call # @log_call
def get_last_played_dates(track_id: int, limit: int = 5) -> str: def playdates_get_last(track_id: int, limit: int = 5) -> str:
""" """
Return the most recent 'limit' dates that this track has been played Return the most recent 'limit' dates that this track has been played
as a text list as a text list
@ -1051,7 +1051,7 @@ def get_last_played_dates(track_id: int, limit: int = 5) -> str:
) )
def update_playdates(track_id: int) -> None: def playdates_update(track_id: int) -> None:
""" """
Update playdates for passed track Update playdates for passed track
""" """
@ -1132,7 +1132,7 @@ def _queries_where(
return results return results
def get_all_queries(favourites_only: bool = False) -> list[QueryDTO]: def queries_all(favourites_only: bool = False) -> list[QueryDTO]:
"""Return a list of all queries""" """Return a list of all queries"""
query = Queries.id > 0 query = Queries.id > 0
@ -1150,44 +1150,7 @@ def query_by_id(query_id: int) -> QueryDTO | None:
return query_list[0] return query_list[0]
def update_query_filter(query_id: int, filter: Filter) -> None: def query_create(name: str, filter: Filter) -> QueryDTO:
"""Update query filter"""
with db.Session() as session:
session.execute(
update(Queries).where(Queries.id == query_id).values(filter=filter)
)
session.commit()
def delete_query(query_id: int) -> None:
"""Delete query"""
with db.Session() as session:
query = session.get(Queries, query_id)
session.delete(query)
session.commit()
def update_query_name(query_id: int, name: str) -> None:
"""Update query name"""
with db.Session() as session:
session.execute(update(Queries).where(Queries.id == query_id).values(name=name))
session.commit()
def update_query_favourite(query_id: int, favourite: bool) -> None:
"""Update query favourite"""
with db.Session() as session:
session.execute(
update(Queries).where(Queries.id == query_id).values(favourite=favourite)
)
session.commit()
def create_query(name: str, filter: Filter) -> QueryDTO:
""" """
Create a query and return the DTO Create a query and return the DTO
""" """
@ -1207,8 +1170,45 @@ def create_query(name: str, filter: Filter) -> QueryDTO:
return new_query return new_query
def query_delete(query_id: int) -> None:
"""Delete query"""
with db.Session() as session:
query = session.get(Queries, query_id)
session.delete(query)
session.commit()
def query_update_favourite(query_id: int, favourite: bool) -> None:
"""Update query favourite"""
with db.Session() as session:
session.execute(
update(Queries).where(Queries.id == query_id).values(favourite=favourite)
)
session.commit()
def query_update_filter(query_id: int, filter: Filter) -> None:
"""Update query filter"""
with db.Session() as session:
session.execute(
update(Queries).where(Queries.id == query_id).values(filter=filter)
)
session.commit()
def query_update_name(query_id: int, name: str) -> None:
"""Update query name"""
with db.Session() as session:
session.execute(update(Queries).where(Queries.id == query_id).values(name=name))
session.commit()
# Misc # Misc
def get_setting(name: str) -> int | None: def setting_get(name: str) -> int | None:
""" """
Get int setting Get int setting
""" """
@ -1225,7 +1225,7 @@ def get_setting(name: str) -> int | None:
return record.f_int return record.f_int
def set_setting(name: str, value: int) -> None: def setting_set(name: str, value: int) -> None:
""" """
Add int setting Add int setting
""" """
@ -1244,7 +1244,7 @@ def set_setting(name: str, value: int) -> None:
session.commit() session.commit()
def get_db_name() -> str: def db_name_get() -> str:
"""Return database name""" """Return database name"""
with db.Session() as session: with db.Session() as session:

2
app/file_importer.py
View File

@ -640,7 +640,7 @@ class DoTrackImport(QThread):
if self.track_id == 0: if self.track_id == 0:
track_dto = ds.create_track(self.destination_track_path, metadata) track_dto = ds.create_track(self.destination_track_path, metadata)
else: else:
track_dto = ds.update_track( track_dto = ds.track_update(
self.destination_track_path, self.track_id, metadata self.destination_track_path, self.track_id, metadata
) )

6
app/musicmuster.py
View File

@ -67,7 +67,7 @@ from classes import (
Filter, Filter,
MusicMusterSignals, MusicMusterSignals,
PlaylistDTO, PlaylistDTO,
PlayTrack, TrackAndPlaylist,
QueryDTO, QueryDTO,
TrackInfo, TrackInfo,
) )
@ -1595,7 +1595,7 @@ class Window(QMainWindow):
except subprocess.CalledProcessError as exc_info: except subprocess.CalledProcessError as exc_info:
git_tag = str(exc_info.output) git_tag = str(exc_info.output)
dbname = ds.get_db_name() dbname = ds.db_name_get()
QMessageBox.information( QMessageBox.information(
self, self,
@ -2142,7 +2142,7 @@ class Window(QMainWindow):
# Notify others # Notify others
self.signals.signal_track_started.emit( self.signals.signal_track_started.emit(
PlayTrack(self.track_sequence.current.playlist_id, TrackAndPlaylist(self.track_sequence.current.playlist_id,
self.track_sequence.current.track_id) self.track_sequence.current.track_id)
) )

22
app/playlistmodel.py
View File

@ -36,7 +36,7 @@ from classes import (
InsertRows, InsertRows,
InsertTrack, InsertTrack,
MusicMusterSignals, MusicMusterSignals,
PlayTrack, TrackAndPlaylist,
) )
from config import Config from config import Config
from helpers import ( from helpers import (
@ -102,7 +102,7 @@ class PlaylistModel(QAbstractTableModel):
self.signals.signal_track_started.connect(self.track_started) self.signals.signal_track_started.connect(self.track_started)
# Populate self.playlist_rows # Populate self.playlist_rows
for dto in ds.get_playlist_rows(self.playlist_id): for dto in ds.playlistrows_by_playlist(self.playlist_id):
self.playlist_rows[dto.row_number] = PlaylistRow(dto) self.playlist_rows[dto.row_number] = PlaylistRow(dto)
self.update_track_times() self.update_track_times()
@ -148,16 +148,16 @@ class PlaylistModel(QAbstractTableModel):
return header_row return header_row
# @log_call # @log_call
def add_track_to_header(self, track_details: InsertTrack) -> None: def add_track_to_header(self, track_and_playlist: TrackAndPlaylist) -> None:
""" """
Handle signal_add_track_to_header Handle signal_add_track_to_header
""" """
if track_details.playlist_id != self.playlist_id: if track_and_playlist.playlist_id != self.playlist_id:
return return
if not self.selected_rows: if not self.selected_rows:
return raise ApplicationError("Add track to header but no row selected")
if len(self.selected_rows) > 1: if len(self.selected_rows) > 1:
self.signals.show_warning_signal.emit( self.signals.show_warning_signal.emit(
@ -172,9 +172,7 @@ class PlaylistModel(QAbstractTableModel):
) )
return return
if selected_row.note: selected_row.track_id = track_and_playlist.track_id
selected_row.note += " " + track_details.note
selected_row.track_id = track_details.track_id
# Update local copy # Update local copy
self.refresh_row(selected_row.row_number) self.refresh_row(selected_row.row_number)
@ -257,7 +255,7 @@ class PlaylistModel(QAbstractTableModel):
return len(Col) return len(Col)
# @log_call # @log_call
def track_started(self, play_track: PlayTrack) -> None: def track_started(self, play_track: TrackAndPlaylist) -> None:
""" """
Notification from musicmuster that the current track has just Notification from musicmuster that the current track has just
started playing started playing
@ -296,7 +294,7 @@ class PlaylistModel(QAbstractTableModel):
self.obs_scene_change(row_number) self.obs_scene_change(row_number)
# Update Playdates in database # Update Playdates in database
ds.update_playdates(track_id) ds.playdates_update(track_id)
# Mark track as played in playlist # Mark track as played in playlist
playlist_dto.played = True playlist_dto.played = True
@ -713,7 +711,7 @@ class PlaylistModel(QAbstractTableModel):
super().beginInsertRows(QModelIndex(), new_row_number, new_row_number) super().beginInsertRows(QModelIndex(), new_row_number, new_row_number)
_ = ds.insert_row( _ = ds.playlist_insert_row(
playlist_id=self.playlist_id, playlist_id=self.playlist_id,
row_number=new_row_number, row_number=new_row_number,
track_id=track_id, track_id=track_id,
@ -1022,7 +1020,7 @@ class PlaylistModel(QAbstractTableModel):
# build a new playlist_rows # build a new playlist_rows
new_playlist_rows: dict[int, PlaylistRow] = {} new_playlist_rows: dict[int, PlaylistRow] = {}
for dto in ds.get_playlist_rows(self.playlist_id): for dto in ds.playlistrows_by_playlist(self.playlist_id):
if dto.playlistrow_id not in plrid_to_row: if dto.playlistrow_id not in plrid_to_row:
new_playlist_rows[dto.row_number] = PlaylistRow(dto) new_playlist_rows[dto.row_number] = PlaylistRow(dto)
else: else:

10
app/playlistrow.py
View File

@ -161,7 +161,7 @@ class PlaylistRow:
if self.track_id > 0: if self.track_id > 0:
raise ApplicationError("Attempting to add track to row with existing track ({self=}") raise ApplicationError("Attempting to add track to row with existing track ({self=}")
ds.add_track_to_header(track_id) ds.add_track_to_header(playlistrow_id=self.playlistrow_id, track_id=track_id)
# Need to update with track information # Need to update with track information
track = ds.track_by_id(track_id) track = ds.track_by_id(track_id)
@ -169,12 +169,6 @@ class PlaylistRow:
for attr, value in track.__dataclass_fields__.items(): for attr, value in track.__dataclass_fields__.items():
setattr(self, attr, value) setattr(self, attr, value)
# TODO: set up write access to track_id. Should only update if
# track_id == 0. Need to update all other track fields at the
# same time.
print(f"set track_id attribute for {self=}, {value=}")
pass
# Expose PlaylistRowDTO fields as properties # Expose PlaylistRowDTO fields as properties
@property @property
def note(self): def note(self):
@ -583,7 +577,7 @@ class TrackSequence:
for ts in [self.next, self.current, self.previous]: for ts in [self.next, self.current, self.previous]:
if not ts: if not ts:
continue continue
playlist_row_dto = ds.get_playlist_row(ts.playlistrow_id) playlist_row_dto = ds.playlistrow_by_id(ts.playlistrow_id)
if not playlist_row_dto: if not playlist_row_dto:
raise ApplicationError(f"(Can't retrieve PlaylistRows entry, {self=}") raise ApplicationError(f"(Can't retrieve PlaylistRows entry, {self=}")
ts = PlaylistRow(playlist_row_dto) ts = PlaylistRow(playlist_row_dto)

10
app/playlists.py
View File

@ -40,7 +40,7 @@ from classes import (
Col, Col,
MusicMusterSignals, MusicMusterSignals,
PlaylistStyle, PlaylistStyle,
PlayTrack, TrackAndPlaylist,
TrackInfo TrackInfo
) )
from config import Config from config import Config
@ -519,7 +519,9 @@ class PlaylistTab(QTableView):
return menu_item return menu_item
def _add_track(self) -> None: def _add_track(self) -> None:
"""Add a track to a section header making it a normal track row""" """
Add a track to a section header making it a normal track row.
"""
dlg = TrackInsertDialog( dlg = TrackInsertDialog(
parent=self.musicmuster, parent=self.musicmuster,
@ -688,7 +690,7 @@ class PlaylistTab(QTableView):
self.resizeRowsToContents() self.resizeRowsToContents()
# Save settings # Save settings
ds.set_setting( ds.setting_set(
f"playlist_col_{column_number}_width", self.columnWidth(column_number) f"playlist_col_{column_number}_width", self.columnWidth(column_number)
) )
@ -724,7 +726,7 @@ class PlaylistTab(QTableView):
cb.setText(track_path, mode=cb.Mode.Clipboard) cb.setText(track_path, mode=cb.Mode.Clipboard)
# @log_call # @log_call
def track_started(self, play_track: PlayTrack) -> None: def track_started(self, play_track: TrackAndPlaylist) -> None:
""" """
Called when track starts playing Called when track starts playing
""" """

2
app/querylistmodel.py
View File

@ -271,4 +271,4 @@ class QuerylistModel(QAbstractTableModel):
track_id = self.querylist_rows[row].track_id track_id = self.querylist_rows[row].track_id
if not track_id: if not track_id:
return QVariant() return QVariant()
return ds.get_last_played_dates(track_id) return ds.playdates_get_last(track_id)

4
app/utilities.py
View File

@ -26,7 +26,7 @@ def check_db() -> None:
Check all paths in database exist Check all paths in database exist
""" """
db_paths = set([a.path for a in ds.get_all_tracks()]) db_paths = set([a.path for a in ds.tracks_all()])
os_paths_list = [] os_paths_list = []
for root, _dirs, files in os.walk(Config.ROOT): for root, _dirs, files in os.walk(Config.ROOT):
@ -88,7 +88,7 @@ def update_bitrates() -> None:
Update bitrates on all tracks in database Update bitrates on all tracks in database
""" """
for track in ds.get_all_tracks(): for track in ds.tracks_all():
try: try:
t = get_tags(track.path) t = get_tags(track.path)
# TODO this won't persist as we're updating DTO # TODO this won't persist as we're updating DTO

2
tests/X_test_file_importer.py
View File

@ -55,7 +55,7 @@ class MyTestCase(unittest.TestCase):
# Create a playlist for all tests # Create a playlist for all tests
playlist_name = "file importer playlist" playlist_name = "file importer playlist"
playlist = ds.create_playlist(name=playlist_name, template_id=0) playlist = ds.playlist_create(name=playlist_name, template_id=0)
cls.widget._open_playlist(playlist) cls.widget._open_playlist(playlist)
# Create our musicstore # Create our musicstore

20
tests/test_ds.py
View File

@ -163,7 +163,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 1, 2, 4, 5, 3, 6, 7, 8, 9] assert new_order == [0, 1, 2, 4, 5, 3, 6, 7, 8, 9]
@ -177,7 +177,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 1, 2, 4, 3, 5, 6, 7, 8, 9] assert new_order == [0, 1, 2, 4, 3, 5, 6, 7, 8, 9]
@ -191,7 +191,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 1, 4, 2, 3, 5, 6, 7, 8, 9] assert new_order == [0, 1, 4, 2, 3, 5, 6, 7, 8, 9]
@ -205,7 +205,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 2, 3, 6, 7, 8, 1, 4, 5, 10, 9] assert new_order == [0, 2, 3, 6, 7, 8, 1, 4, 5, 10, 9]
@ -219,7 +219,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 1, 2, 4, 5, 3, 6, 7, 8, 9, 10] assert new_order == [0, 1, 2, 4, 5, 3, 6, 7, 8, 9, 10]
@ -233,7 +233,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 1, 2, 4, 7, 8, 9, 10, 3, 5, 6] assert new_order == [0, 1, 2, 4, 7, 8, 9, 10, 3, 5, 6]
@ -247,7 +247,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 1, 2, 3, 4, 7, 8, 10, 5, 6, 9] assert new_order == [0, 1, 2, 3, 4, 7, 8, 10, 5, 6, 9]
@ -262,7 +262,7 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order = [] new_order = []
for row in ds.get_playlist_rows(playlist.playlist_id): for row in ds.playlistrows_by_playlist(playlist.playlist_id):
new_order.append(int(row.note)) new_order.append(int(row.note))
assert new_order == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10] assert new_order == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
@ -280,11 +280,11 @@ class MyTestCase(unittest.TestCase):
# Check we have all rows and plr_rownums are correct # Check we have all rows and plr_rownums are correct
new_order_src = [] new_order_src = []
for row in ds.get_playlist_rows(playlist_src.playlist_id): for row in ds.playlistrows_by_playlist(playlist_src.playlist_id):
new_order_src.append(int(row.note)) new_order_src.append(int(row.note))
assert new_order_src == [0, 1, 3, 5, 7, 8, 9, 10] assert new_order_src == [0, 1, 3, 5, 7, 8, 9, 10]
new_order_dst = [] new_order_dst = []
for row in ds.get_playlist_rows(playlist_dst.playlist_id): for row in ds.playlistrows_by_playlist(playlist_dst.playlist_id):
new_order_dst.append(int(row.note)) new_order_dst.append(int(row.note))
assert new_order_dst == [0, 1, 2, 3, 4, 2, 4, 6, 5, 6, 7, 8, 9, 10] assert new_order_dst == [0, 1, 2, 3, 4, 2, 4, 6, 5, 6, 7, 8, 9, 10]

2
tests/test_file_importer.py
View File

@ -52,7 +52,7 @@ class MyTestCase(unittest.TestCase):
# Create a playlist for all tests # Create a playlist for all tests
playlist_name = "file importer playlist" playlist_name = "file importer playlist"
playlist = ds.create_playlist(name=playlist_name, template_id=0) playlist = ds.playlist_create(name=playlist_name, template_id=0)
cls.widget._open_playlist(playlist) cls.widget._open_playlist(playlist)
# Create our musicstore # Create our musicstore

134
tests/test_playlistmodel.py
View File

@ -8,14 +8,14 @@ from PyQt6.QtCore import Qt, QModelIndex
# App imports # App imports
from app.helpers import get_all_track_metadata from app.helpers import get_all_track_metadata
from app import playlistmodel from app import ds, playlistmodel
from app.models import ( from app.models import (
db, db,
Playlists, Playlists,
Tracks, Tracks,
) )
from classes import ( from classes import (
InsertTrack, TrackAndPlaylist,
) )
@ -36,18 +36,13 @@ class TestMMMiscTracks(unittest.TestCase):
db.create_all() db.create_all()
# Create a playlist and model # Create a playlist and model
with db.Session() as session: self.playlist = ds.playlist_create(PLAYLIST_NAME, template_id=0)
self.playlist = Playlists(session, PLAYLIST_NAME, template_id=0) self.model = playlistmodel.PlaylistModel(self.playlist.playlist_id, is_template=False)
self.model = playlistmodel.PlaylistModel(
self.playlist.id, is_template=False
)
for row in range(len(self.test_tracks)): for row in range(len(self.test_tracks)):
track_path = self.test_tracks[row % len(self.test_tracks)] track_path = self.test_tracks[row % len(self.test_tracks)]
track = Tracks(session, **get_all_track_metadata(track_path)) track = ds.track_create(**get_all_track_metadata(track_path))
self.model.insert_row(track_id=track.id, note=f"{row=}") self.model.insert_row(track_id=track.id, note=f"{row=}")
session.commit()
def tearDown(self): def tearDown(self):
db.drop_all() db.drop_all()
@ -98,28 +93,25 @@ class TestMMMiscNoPlaylist(unittest.TestCase):
def test_insert_track_new_playlist(self): def test_insert_track_new_playlist(self):
# insert a track into a new playlist # insert a track into a new playlist
with db.Session() as session: playlist = ds.playlist_create(self.PLAYLIST_NAME, template_id=0)
playlist = Playlists(session, self.PLAYLIST_NAME, template_id=0) # Create a model
# Create a model model = playlistmodel.PlaylistModel(playlist.id, is_template=False)
model = playlistmodel.PlaylistModel(playlist.id, is_template=False) # test repr
# test repr _ = str(model)
_ = str(model)
track_path = self.test_tracks[0] track_path = self.test_tracks[0]
metadata = get_all_track_metadata(track_path) metadata = get_all_track_metadata(track_path)
track = Tracks(session, **metadata) track = ds.track_create(metadata)
model.insert_row(track_id=track.id) model.insert_row(track_id=track.id)
prd = model.playlist_rows[model.rowCount() - 1] prd = model.playlist_rows[model.rowCount() - 1]
# test repr # test repr
_ = str(prd) _ = str(prd)
assert ( assert (
model._edit_role( model._edit_role(model.rowCount() - 1, playlistmodel.Col.TITLE.value, prd)
model.rowCount() - 1, playlistmodel.Col.TITLE.value, prd == metadata["title"]
) )
== metadata["title"]
)
class TestMMMiscRowMove(unittest.TestCase): class TestMMMiscRowMove(unittest.TestCase):
@ -129,15 +121,10 @@ class TestMMMiscRowMove(unittest.TestCase):
def setUp(self): def setUp(self):
db.create_all() db.create_all()
with db.Session() as session: self.playlist = ds.playlist_create(self.PLAYLIST_NAME, template_id=0)
self.playlist = Playlists(session, self.PLAYLIST_NAME, template_id=0) self.model = playlistmodel.PlaylistModel(self.playlist.id, is_template=False)
self.model = playlistmodel.PlaylistModel( for row in range(self.ROWS_TO_CREATE):
self.playlist.id, is_template=False self.model.insert_row(note=str(row))
)
for row in range(self.ROWS_TO_CREATE):
self.model.insert_row(note=str(row))
session.commit()
def tearDown(self): def tearDown(self):
db.drop_all() db.drop_all()
@ -166,6 +153,9 @@ class TestMMMiscRowMove(unittest.TestCase):
note_text = "test text" note_text = "test text"
insert_row = 6 insert_row = 6
# Fake selected row in model
self.model.selected_rows = [self.model.playlist_rows[insert_row]]
self.model.insert_row(note=note_text) self.model.insert_row(note=note_text)
assert self.model.rowCount() == self.ROWS_TO_CREATE + 1 assert self.model.rowCount() == self.ROWS_TO_CREATE + 1
prd = self.model.playlist_rows[insert_row] prd = self.model.playlist_rows[insert_row]
@ -189,13 +179,8 @@ class TestMMMiscRowMove(unittest.TestCase):
self.model.selected_rows = [self.model.playlist_rows[insert_row]] self.model.selected_rows = [self.model.playlist_rows[insert_row]]
prd = self.model.playlist_rows[1] prd = self.model.playlist_rows[1]
import pdb; pdb.set_trace()
self.model.add_track_to_header( self.model.add_track_to_header(
InsertTrack( TrackAndPlaylist(playlist_id=self.model.playlist_id, track_id=prd.track_id)
playlist_id=self.model.playlist_id,
track_id=prd.track_id,
note=note_text,
)
) )
def test_reverse_row_groups_one_row(self): def test_reverse_row_groups_one_row(self):
@ -216,17 +201,19 @@ class TestMMMiscRowMove(unittest.TestCase):
def test_move_one_row_between_playlists_to_end(self): def test_move_one_row_between_playlists_to_end(self):
from_rows = [3] from_rows = [3]
to_row = self.ROWS_TO_CREATE to_row = self.ROWS_TO_CREATE
destination_playlist = "destination" destination_playlist_name = "destination"
model_src = self.model model_src = self.model
with db.Session() as session: playlist_dst = ds.playlist_create(destination_playlist_name, template_id=0)
playlist_dst = Playlists(session, destination_playlist, template_id=0) model_dst = playlistmodel.PlaylistModel(
model_dst = playlistmodel.PlaylistModel(playlist_dst.id, is_template=False) playlist_dst.playlist_id, is_template=False
for row in range(self.ROWS_TO_CREATE): )
model_dst.insert_row(note=str(row)) for row in range(self.ROWS_TO_CREATE):
model_dst.insert_row(note=str(row))
model_src.move_rows_between_playlists(from_rows, to_row, model_dst.playlist_id) ds.playlist_move_rows(
model_dst.refresh_data(session) from_rows, self.playlist.playlist_id, to_row, playlist_dst.playlist_id
)
assert len(model_src.playlist_rows) == self.ROWS_TO_CREATE - len(from_rows) assert len(model_src.playlist_rows) == self.ROWS_TO_CREATE - len(from_rows)
assert len(model_dst.playlist_rows) == self.ROWS_TO_CREATE + len(from_rows) assert len(model_dst.playlist_rows) == self.ROWS_TO_CREATE + len(from_rows)
@ -237,17 +224,19 @@ class TestMMMiscRowMove(unittest.TestCase):
def test_move_one_row_between_playlists_to_middle(self): def test_move_one_row_between_playlists_to_middle(self):
from_rows = [3] from_rows = [3]
to_row = 2 to_row = 2
destination_playlist = "destination" destination_playlist_name = "destination"
model_src = self.model model_src = self.model
with db.Session() as session: playlist_dst = ds.playlist_create(destination_playlist_name, template_id=0)
playlist_dst = Playlists(session, destination_playlist, template_id=0) model_dst = playlistmodel.PlaylistModel(
model_dst = playlistmodel.PlaylistModel(playlist_dst.id, is_template=False) playlist_dst.playlist_id, is_template=False
for row in range(self.ROWS_TO_CREATE): )
model_dst.insert_row(note=str(row)) for row in range(self.ROWS_TO_CREATE):
model_dst.insert_row(note=str(row))
model_src.move_rows_between_playlists(from_rows, to_row, model_dst.playlist_id) ds.playlist_move_rows(
model_dst.refresh_data(session) from_rows, self.playlist.playlist_id, to_row, playlist_dst.playlist_id
)
# Check the rows of the destination model # Check the rows of the destination model
row_notes = [] row_notes = []
@ -264,17 +253,20 @@ class TestMMMiscRowMove(unittest.TestCase):
def test_move_multiple_rows_between_playlists_to_end(self): def test_move_multiple_rows_between_playlists_to_end(self):
from_rows = [1, 3, 4] from_rows = [1, 3, 4]
to_row = 2 to_row = 2
destination_playlist = "destination" destination_playlist_name = "destination"
model_src = self.model model_src = self.model
with db.Session() as session:
playlist_dst = Playlists(session, destination_playlist, template_id=0)
model_dst = playlistmodel.PlaylistModel(playlist_dst.id, is_template=False)
for row in range(self.ROWS_TO_CREATE):
model_dst.insert_row(note=str(row))
model_src.move_rows_between_playlists(from_rows, to_row, model_dst.playlist_id) playlist_dst = ds.playlist_create(destination_playlist_name, template_id=0)
model_dst.refresh_data(session) model_dst = playlistmodel.PlaylistModel(
playlist_dst.playlist_id, is_template=False
)
for row in range(self.ROWS_TO_CREATE):
model_dst.insert_row(note=str(row))
ds.playlist_move_rows(
from_rows, self.playlist.id, playlist_dst.playlist_id, to_row
)
# Check the rows of the destination model # Check the rows of the destination model
row_notes = [] row_notes = []