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

Refactoring

Browse Source
This commit is contained in:
Keith Edmunds 2022-02-13 14:09:22 +00:00
parent fa2e1234e9
commit 04c3c2efbc
6 changed files with 1154 additions and 513 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

461
app/models.py
View File

@ -88,9 +88,9 @@ class NoteColours(Base):
for rec in (
session.query(NoteColours)
.filter(NoteColours.enabled.is_(True))
.order_by(NoteColours.order)
.all()
.filter(NoteColours.enabled.is_(True))
.order_by(NoteColours.order)
.all()
):
if rec.is_regex:
flags = re.UNICODE
@ -119,50 +119,42 @@ class Notes(Base):
row = Column(Integer, nullable=False)
note = Column(String(256), index=False)
def __init__(self, session, playlist_id, row, text):
"""Create note"""
DEBUG(f"Notes.__init__({playlist_id=}, {row=}, {text=})")
self.playlist_id = playlist_id
self.row = row
self.note = text
session.add(self)
session.commit()
def __repr__(self):
return (
f"<Note(id={self.id}, row={self.row}, note={self.note}>"
)
@classmethod
def add_note(cls, session, playlist_id, row, text):
"""Add note"""
DEBUG(f"add_note(playlist_id={playlist_id}, row={row}, text={text})")
note = Notes()
note.playlist_id = playlist_id
note.row = row
note.note = text
session.add(note)
session.commit()
return note
@staticmethod
def delete_note(session, note_id):
def delete_note(self, session):
"""Delete note"""
DEBUG(f"delete_note(id={note_id}")
DEBUG(f"delete_note({self.id=}")
session.query(Notes).filter(Notes.id == note_id).delete()
session.query(self).filter(id == self.id).delete()
session.commit()
@classmethod
def update_note(cls, session, note_id, row, text=None):
def update_note(self, session, row, text=None):
"""
Update note details. If text=None, don't change text.
"""
DEBUG(f"Notes.update_note(id={note_id}, row={row}, text={text})")
DEBUG(f"Notes.update_note({self.id=}, {row=}, {text=})")
note = session.query(Notes).filter(Notes.id == note_id).one()
note = session.query(self).filter(id == self.id).one()
note.row = row
if text:
note.note = text
session.commit()
return note
class Playdates(Base):
__tablename__ = 'playdates'
@ -181,7 +173,7 @@ class Playdates(Base):
pd.lastplayed = datetime.now()
pd.track_id = track.id
session.add(pd)
track.update_lastplayed(session, track.id)
track.update_lastplayed(session)
session.commit()
return pd
@ -258,7 +250,7 @@ class Playlists(Base):
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(32), nullable=False, unique=True)
last_used = Column(DateTime, default=None, nullable=True)
loaded = Column(Boolean, default=True)
loaded = Column(Boolean, default=True, nullable=False)
notes = relationship("Notes",
order_by="Notes.row",
back_populates="playlist")
@ -266,9 +258,36 @@ class Playlists(Base):
order_by="PlaylistTracks.row",
back_populates="playlists")
def __init__(self, session, name):
self.name = name
session.add(playlist)
session.commit()
def __repr__(self):
return f"<Playlists(id={self.id}, name={self.name}>"
def add_note(self, session, row, text):
"""Add note to playlist at passed row"""
return Notes(session, self.id, row, text)
def add_track(self, session, track, row=None):
"""
Add track to playlist at given row.
If row=None, add to end of playlist
"""
if not row:
last_row = session.query(
func.max(PlaylistTracks.row)
).filter_by(playlist_id=self.id).first()
if last_row:
row = last_row[0] + 1
else:
row = 0
PlaylistTracks(session, self.id, track.id, row)
def close(self, session):
"""Record playlist as no longer loaded"""
@ -277,52 +296,41 @@ class Playlists(Base):
session.commit()
@classmethod
def get_all_closed_playlists(cls, session):
"""Returns a list of all playlists not currently open"""
def get_all(cls, session):
"""Returns a list of all playlists ordered by last use"""
return (
session.query(cls)
.filter(
(cls.loaded == False) | # noqa E712
(cls.loaded.is_(None))
)
.order_by(cls.last_used.desc())
).all()
@classmethod
def get_all_playlists(cls, session):
"""Returns a list of all playlists"""
return session.query(cls).all()
def get_by_id(cls, session, playlist_id):
return (session.query(cls).filter(cls.id == playlist_id)).one()
@classmethod
def get_last_used(cls, session):
def get_closed(cls, session):
"""Returns a list of all closed playlists ordered by last use"""
return (
session.query(cls)
.filter(cls.loaded.is_(False))
.order_by(cls.last_used.desc())
).all()
@classmethod
def get_open(cls, session):
"""
Return a list of playlists marked "loaded", ordered by loaded date.
"""
return (
session.query(cls)
.filter(cls.loaded == True) # noqa E712
.filter(cls.loaded.is_(True))
.order_by(cls.last_used.desc())
).all()
@classmethod
def get_playlist_by_id(cls, session, playlist_id):
return (session.query(cls).filter(cls.id == playlist_id)).one()
@classmethod
def new(cls, session, name):
DEBUG(f"Playlists.new(name={name})")
playlist = cls()
playlist.name = name
session.add(playlist)
session.commit()
return playlist
def open(self, session):
def mark_open(self, session):
"""Mark playlist as loaded and used now"""
self.loaded = True
@ -330,6 +338,26 @@ class Playlists(Base):
session.add(self)
session.commit()
def remove_all_tracks(self, session):
"""
Remove all tracks from this playlist
"""
session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == self.id,
).delete()
session.commit()
def remove_track(self, session, row):
DEBUG(f"Playlist.remove_track({playlist_id=}, {row=})")
session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == self.id,
PlaylistTracks.row == row
).delete()
session.commit()
class PlaylistTracks(Base):
__tablename__ = 'playlisttracks'
@ -341,17 +369,13 @@ class PlaylistTracks(Base):
tracks = relationship("Tracks", back_populates="playlists")
playlists = relationship("Playlists", back_populates="tracks")
@staticmethod
def add_track(session, playlist_id, track_id, row):
DEBUG(
f"PlaylistTracks.add_track(playlist_id={playlist_id}, "
f"track_id={track_id}, row={row})"
)
plt = PlaylistTracks()
plt.playlist_id = playlist_id,
plt.track_id = track_id,
plt.row = row
session.add(plt)
def __init__(self, session, playlist_id, track_id, row):
DEBUG(f"PlaylistTracks.__init__({playlist_id=}, {track_id=}, {row=})")
self.playlist_id = playlist_id,
self.track_id = track_id,
self.row = row
session.add(self)
session.commit()
@staticmethod
@ -361,103 +385,37 @@ class PlaylistTracks(Base):
return session.query(PlaylistTracks).filter(
PlaylistTracks.track_id == track_id).all()
@staticmethod
def move_track(session, from_playlist_id, row, to_playlist_id):
DEBUG(
"PlaylistTracks.move_tracks(from_playlist_id="
f"{from_playlist_id}, row={row}, "
f"to_playlist_id={to_playlist_id})"
)
max_row = session.query(func.max(PlaylistTracks.row)).filter(
PlaylistTracks.playlist_id == to_playlist_id).scalar()
if max_row is None:
# Destination playlist is empty; use row 0
new_row = 0
else:
# Destination playlist has tracks; add to end
new_row = max_row + 1
try:
record = session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == from_playlist_id,
PlaylistTracks.row == row).one()
except NoResultFound:
# Issue #38?
ERROR(
f"No rows matched in query: "
f"PlaylistTracks.playlist_id == {from_playlist_id}, "
f"PlaylistTracks.row == {row}"
)
return
record.playlist_id = to_playlist_id
record.row = new_row
session.commit()
# @staticmethod
# def move_track(session, from_playlist_id, row, to_playlist_id):
@staticmethod
def new_row(session, playlist_id):
"""
Return row number > the largest existing row number
If there are no existing rows, return 0 (ie, first row number)
"""
last_row = session.query(func.max(PlaylistTracks.row)).one()[0]
if last_row:
return last_row + 1
else:
return 0
@staticmethod
def remove_all_tracks(session, playlist_id):
"""
Remove all tracks from passed playlist_id
"""
session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == playlist_id,
).delete()
session.commit()
@staticmethod
def remove_track(session, playlist_id, row):
DEBUG(
f"PlaylistTracks.remove_track(playlist_id={playlist_id}, "
f"row={row})"
)
session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == playlist_id,
PlaylistTracks.row == row
).delete()
session.commit()
@staticmethod
def update_row_track(session, playlist_id, row, track_id):
DEBUG(
f"PlaylistTracks.update_track_row(playlist_id={playlist_id}, "
f"row={row}, track_id={track_id})"
)
try:
plt = session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == playlist_id,
PlaylistTracks.row == row
).one()
except MultipleResultsFound:
ERROR(
f"Multiple rows matched in query: "
f"PlaylistTracks.playlist_id == {playlist_id}, "
f"PlaylistTracks.row == {row}"
)
return
except NoResultFound:
ERROR(
f"No rows matched in query: "
f"PlaylistTracks.playlist_id == {playlist_id}, "
f"PlaylistTracks.row == {row}"
)
return
plt.track_id = track_id
session.commit()
# DEBUG(
# "PlaylistTracks.move_tracks(from_playlist_id="
# f"{from_playlist_id}, row={row}, "
# f"to_playlist_id={to_playlist_id})"
# )
# max_row = session.query(func.max(PlaylistTracks.row)).filter(
# PlaylistTracks.playlist_id == to_playlist_id).scalar()
# if max_row is None:
# # Destination playlist is empty; use row 0
# new_row = 0
# else:
# # Destination playlist has tracks; add to end
# new_row = max_row + 1
# try:
# record = session.query(PlaylistTracks).filter(
# PlaylistTracks.playlist_id == from_playlist_id,
# PlaylistTracks.row == row).one()
# except NoResultFound:
# # Issue #38?
# ERROR(
# f"No rows matched in query: "
# f"PlaylistTracks.playlist_id == {from_playlist_id}, "
# f"PlaylistTracks.row == {row}"
# )
# return
# record.playlist_id = to_playlist_id
# record.row = new_row
# session.commit()
class Settings(Base):
@ -505,51 +463,60 @@ class Tracks(Base):
playlists = relationship("PlaylistTracks", back_populates="tracks")
playdates = relationship("Playdates", back_populates="tracks")
def __init__(self, session, path):
self.path = path
session.add(self)
session.commit()
def __repr__(self):
return (
f"<Track(id={self.id}, title={self.title}, "
f"artist={self.artist}, path={self.path}>"
)
# Not currently used 1 June 2021
# @staticmethod
# def get_note(session, id):
# return session.query(Notes).filter(Notes.id == id).one()
@staticmethod
def get_all_paths(session):
"""Return a list of paths of all tracks"""
return [a[0] for a in session.query(Tracks.path).all()]
@staticmethod
def get_all_tracks(session):
@classmethod
def get_all_tracks(cls, session):
"Return a list of all tracks"
=======
@classmethod
def get_all_tracks(cls, session):
"""Return a list of all tracks"""
return session.query(Tracks).all()
return session.query(cls).all()
@classmethod
def get_or_create(cls, session, path):
DEBUG(f"Tracks.get_or_create(path={path})")
"""
If a track with path exists, return it;
else created new track and return it
"""
DEBUG(f"Tracks.get_or_create(path=})")
try:
track = session.query(cls).filter(cls.path == path).one()
except NoResultFound:
track = Tracks()
track.path = path
session.add(track)
track = Tracks(session, path)
return track
@staticmethod
def get_duration(session, track_id):
try:
return session.query(
Tracks.duration).filter(Tracks.id == track_id).one()[0]
except NoResultFound:
ERROR(f"Can't find track id {track_id}")
return None
# @staticmethod
# def get_duration(session, id):
# try:
# return session.query(
# Tracks.duration).filter(Tracks.id == id).one()[0]
# except NoResultFound:
# ERROR(f"Can't find track id {id}")
# return None
@staticmethod
def get_track_from_filename(session, filename):
@classmethod
def get_from_filename(cls, session, filename):
"""
Return track if one and only one track in database has passed
filename (ie, basename of path). Return None if zero or more
@ -564,8 +531,13 @@ class Tracks(Base):
except (NoResultFound, MultipleResultsFound):
return None
@staticmethod
def get_track_from_path(session, path):
@classmethod
def get_from_id(cls, session, id):
return session.query(Tracks).filter(
Tracks.id == id).one()
@classmethod
def get_from_path(cls, session, path):
"""
Return track with passee path, or None.
"""
@ -574,19 +546,19 @@ class Tracks(Base):
return session.query(Tracks).filter(Tracks.path == path).first()
@staticmethod
def get_path(session, track_id):
"""Return path of passed track_id, or None"""
# @staticmethod
# def get_path(session, track_id):
# "Return path of passed track_id, or None"
#
# try:
# return session.query(Tracks.path).filter(
# Tracks.id == track_id).one()[0]
# except NoResultFound:
# ERROR(f"Can't find track id {track_id}")
# return None
try:
return session.query(Tracks.path).filter(
Tracks.id == track_id).one()[0]
except NoResultFound:
ERROR(f"Can't find track id {track_id}")
return None
@staticmethod
def get_track(session, track_id):
@classmethod
def get_by_id(cls, session, track_id):
"""Return track or None"""
try:
@ -598,8 +570,8 @@ class Tracks(Base):
return None
@staticmethod
def remove_path(session, path):
"""Remove track with passed path from database"""
def remove_by_path(session, path):
"Remove track with passed path from database"
DEBUG(f"Tracks.remove_path({path=})")
@ -609,66 +581,59 @@ class Tracks(Base):
except IntegrityError as exception:
ERROR(f"Can't remove track with {path=} ({exception=})")
@staticmethod
def search(session, title=None, artist=None, duration=None):
"""
Return any tracks matching passed criteria
"""
# @staticmethod
# def search(session, title=None, artist=None, duration=None):
# """
# Return any tracks matching passed criteria
# """
#
# DEBUG(
# f"Tracks.search({title=}, {artist=}), {duration=})"
# )
#
# if not title and not artist and not duration:
# return None
#
# q = session.query(Tracks).filter(False)
# if title:
# q = q.filter(Tracks.title == title)
# if artist:
# q = q.filter(Tracks.artist == artist)
# if duration:
# q = q.filter(Tracks.duration == duration)
#
# return q.all()
DEBUG(
f"Tracks.search({title=}, {artist=}), {duration=})"
)
@classmethod
def search_artists(cls, session, text):
if not title and not artist and not duration:
return None
q = session.query(Tracks).filter(False)
if title:
q = q.filter(Tracks.title == title)
if artist:
q = q.filter(Tracks.artist == artist)
if duration:
q = q.filter(Tracks.duration == duration)
return q.all()
@staticmethod
def search_artists(session, text):
return (
session.query(Tracks)
.filter(Tracks.artist.ilike(f"%{text}%"))
.order_by(Tracks.title)
.filter(Tracks.artist.ilike(f"%{text}%"))
.order_by(Tracks.title)
).all()
@staticmethod
def search_titles(session, text):
@classmethod
def search_titles(cls, session, text):
return (
session.query(Tracks)
.filter(Tracks.title.ilike(f"%{text}%"))
.order_by(Tracks.title)
.filter(Tracks.title.ilike(f"%{text}%"))
.order_by(Tracks.title)
).all()
@staticmethod
def track_from_id(session, track_id):
return session.query(Tracks).filter(
Tracks.id == track_id).one()
@staticmethod
def update_lastplayed(session, track_id):
track = session.query(Tracks).filter(Tracks.id == track_id).one()
track.lastplayed = datetime.now()
def update_lastplayed(self, session):
self.lastplayed = datetime.now()
session.add(self)
session.commit()
@staticmethod
def update_artist(session, track_id, artist):
track = session.query(Tracks).filter(Tracks.id == track_id).one()
track.artist = artist
def update_artist(self, session, artist):
self.artist = artist
session.add(self)
session.commit()
@staticmethod
def update_title(session, track_id, title):
track = session.query(Tracks).filter(Tracks.id == track_id).one()
track.title = title
def update_title(self, session, title):
self.title = title
session.add(self)
session.commit()
def update_path(self, newpath):

673
app/models.py.orig Normal file
View File

@ -0,0 +1,673 @@
#!/usr/bin/python3
import os.path
import re
import sqlalchemy
from datetime import datetime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import (
Boolean,
Column,
DateTime,
Float,
ForeignKey,
Integer,
String,
func
)
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import MultipleResultsFound, NoResultFound
from sqlalchemy.orm import relationship, sessionmaker, scoped_session
from config import Config
from log import DEBUG, ERROR
# Create session at the global level as per
# https://docs.sqlalchemy.org/en/13/orm/session_basics.html
Base = declarative_base()
Session = scoped_session(sessionmaker())
def db_init():
# Set up database connection
global Session
engine = sqlalchemy.create_engine(
f"{Config.MYSQL_CONNECT}?charset=utf8",
encoding='utf-8',
echo=Config.DISPLAY_SQL,
pool_pre_ping=True)
Session.configure(bind=engine)
Base.metadata.create_all(engine)
# Create a Session factory
Session = sessionmaker(bind=engine)
# Database classes
class NoteColours(Base):
__tablename__ = 'notecolours'
id = Column(Integer, primary_key=True, autoincrement=True)
substring = Column(String(256), index=False)
colour = Column(String(21), index=False)
enabled = Column(Boolean, default=True, index=True)
is_regex = Column(Boolean, default=False, index=False)
is_casesensitive = Column(Boolean, default=False, index=False)
order = Column(Integer, index=True)
def __repr__(self):
return (
f"<NoteColour(id={self.id}, substring={self.substring}, "
f"colour={self.colour}>"
)
@classmethod
def get_all(cls, session):
"""Return all records"""
return session.query(cls).all()
@classmethod
def get_by_id(cls, session, note_id):
"""Return record identified by id, or None if not found"""
return session.query(NoteColours).filter(
NoteColours.id == note_id).first()
@staticmethod
def get_colour(session, text):
"""
Parse text and return colour string if matched, else None
"""
for rec in (
session.query(NoteColours)
.filter(NoteColours.enabled.is_(True))
.order_by(NoteColours.order)
.all()
):
if rec.is_regex:
flags = re.UNICODE
if not rec.is_casesensitive:
flags |= re.IGNORECASE
p = re.compile(rec.substring, flags)
if p.match(text):
return rec.colour
else:
if rec.is_casesensitive:
if rec.substring in text:
return rec.colour
else:
if rec.substring.lower() in text.lower():
return rec.colour
return None
class Notes(Base):
__tablename__ = 'notes'
id = Column(Integer, primary_key=True, autoincrement=True)
playlist_id = Column(Integer, ForeignKey('playlists.id'))
playlist = relationship("Playlists", back_populates="notes")
row = Column(Integer, nullable=False)
note = Column(String(256), index=False)
def __repr__(self):
return (
f"<Note(id={self.id}, row={self.row}, note={self.note}>"
)
@classmethod
def add_note(cls, session, playlist_id, row, text):
"""Add note"""
DEBUG(f"add_note(playlist_id={playlist_id}, row={row}, text={text})")
note = Notes()
note.playlist_id = playlist_id
note.row = row
note.note = text
session.add(note)
session.commit()
return note
@staticmethod
def delete_note(session, note_id):
"""Delete note"""
DEBUG(f"delete_note(id={note_id}")
session.query(Notes).filter(Notes.id == note_id).delete()
session.commit()
@classmethod
def update_note(cls, session, note_id, row, text=None):
"""
Update note details. If text=None, don't change text.
"""
DEBUG(f"Notes.update_note(id={note_id}, row={row}, text={text})")
note = session.query(Notes).filter(Notes.id == note_id).one()
note.row = row
if text:
note.note = text
session.commit()
return note
class Playdates(Base):
__tablename__ = 'playdates'
id = Column(Integer, primary_key=True, autoincrement=True)
lastplayed = Column(DateTime, index=True, default=None)
track_id = Column(Integer, ForeignKey('tracks.id'))
tracks = relationship("Tracks", back_populates="playdates")
@classmethod
def add_playdate(cls, session, track):
"""Record that track was played"""
DEBUG(f"add_playdate(track={track})")
pd = Playdates()
pd.lastplayed = datetime.now()
pd.track_id = track.id
session.add(pd)
track.update_lastplayed(session)
session.commit()
return pd
@staticmethod
def last_played(session, track_id):
"""Return datetime track last played or None"""
last_played = session.query(Playdates.lastplayed).filter(
(Playdates.track_id == track_id)
).order_by(Playdates.lastplayed.desc()).first()
if last_played:
return last_played[0]
else:
return None
@staticmethod
def remove_track(session, track_id):
"""
Remove all records of track_id
"""
session.query(Playdates).filter(
Playdates.track_id == track_id,
).delete()
session.commit()
class Playlists(Base):
"""
Usage:
pl = session.query(Playlists).filter(Playlists.id == 1).one()
pl
<Playlist(id=1, name=Default>
pl.tracks
[<__main__.PlaylistTracks at 0x7fcd20181c18>,
<__main__.PlaylistTracks at 0x7fcd20181c88>,
<__main__.PlaylistTracks at 0x7fcd20181be0>,
<__main__.PlaylistTracks at 0x7fcd20181c50>]
[a.tracks for a in pl.tracks]
[<Track(id=3992, title=Yesterday Man, artist=Various, path=/h[...]
<Track(id=2238, title=These Boots Are Made For Walkin', arti[...]
<Track(id=3837, title=Babe, artist=Various, path=/home/kae/m[...]
<Track(id=2332, title=Such Great Heights - Remastered, artis[...]]
glue = PlaylistTracks(row=5)
tr = session.query(Tracks).filter(Tracks.id == 676).one()
tr
<Track(id=676, title=Seven Nation Army, artist=White Stripes,
path=/home/kae/music/White Stripes/Elephant/01. Seven Nation Army.flac>
glue.track_id = tr.id
pl.tracks.append(glue)
session.commit()
[a.tracks for a in pl.tracks]
[<Track(id=3992, title=Yesterday Man, artist=Various, path=/h[...]
<Track(id=2238, title=These Boots Are Made For Walkin', arti[...]
<Track(id=3837, title=Babe, artist=Various, path=/home/kae/m[...]
<Track(id=2332, title=Such Great Heights - Remastered, artis[...]
<Track(id=676, title=Seven Nation Army, artist=White Stripes[...]]
"""
__tablename__ = "playlists"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(32), nullable=False, unique=True)
last_used = Column(DateTime, default=None, nullable=True)
loaded = Column(Boolean, default=True)
notes = relationship("Notes",
order_by="Notes.row",
back_populates="playlist")
tracks = relationship("PlaylistTracks",
order_by="PlaylistTracks.row",
back_populates="playlists")
def __repr__(self):
return f"<Playlists(id={self.id}, name={self.name}>"
def close(self, session):
"""Record playlist as no longer loaded"""
self.loaded = False
session.add(self)
session.commit()
@classmethod
def get_all_closed_playlists(cls, session):
"""Returns a list of all playlists not currently open"""
return (
session.query(cls)
.filter(
(cls.loaded == False) | # noqa E712
(cls.loaded.is_(None))
)
.order_by(cls.last_used.desc())
).all()
@classmethod
def get_all_playlists(cls, session):
"""Returns a list of all playlists"""
return session.query(cls).all()
@classmethod
def get_last_used(cls, session):
"""
Return a list of playlists marked "loaded", ordered by loaded date.
"""
return (
session.query(cls)
.filter(cls.loaded == True) # noqa E712
.order_by(cls.last_used.desc())
).all()
@classmethod
def get_playlist_by_id(cls, session, playlist_id):
return (session.query(cls).filter(cls.id == playlist_id)).one()
@classmethod
def new(cls, session, name):
DEBUG(f"Playlists.new(name={name})")
playlist = cls()
playlist.name = name
session.add(playlist)
session.commit()
return playlist
def open(self, session):
"""Mark playlist as loaded and used now"""
self.loaded = True
self.last_used = datetime.now()
session.add(self)
session.commit()
class PlaylistTracks(Base):
__tablename__ = 'playlisttracks'
id = Column(Integer, primary_key=True, autoincrement=True)
playlist_id = Column(Integer, ForeignKey('playlists.id'), primary_key=True)
track_id = Column(Integer, ForeignKey('tracks.id'), primary_key=True)
row = Column(Integer, nullable=False)
tracks = relationship("Tracks", back_populates="playlists")
playlists = relationship("Playlists", back_populates="tracks")
@staticmethod
def add_track(session, playlist_id, track_id, row):
DEBUG(
f"PlaylistTracks.add_track(playlist_id={playlist_id}, "
f"track_id={track_id}, row={row})"
)
plt = PlaylistTracks()
plt.playlist_id = playlist_id,
plt.track_id = track_id,
plt.row = row
session.add(plt)
session.commit()
@staticmethod
def get_track_playlists(session, track_id):
"""Return all PlaylistTracks objects with this track_id"""
return session.query(PlaylistTracks).filter(
PlaylistTracks.track_id == track_id).all()
@staticmethod
def move_track(session, from_playlist_id, row, to_playlist_id):
DEBUG(
"PlaylistTracks.move_tracks(from_playlist_id="
f"{from_playlist_id}, row={row}, "
f"to_playlist_id={to_playlist_id})"
)
max_row = session.query(func.max(PlaylistTracks.row)).filter(
PlaylistTracks.playlist_id == to_playlist_id).scalar()
if max_row is None:
# Destination playlist is empty; use row 0
new_row = 0
else:
# Destination playlist has tracks; add to end
new_row = max_row + 1
try:
record = session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == from_playlist_id,
PlaylistTracks.row == row).one()
except NoResultFound:
# Issue #38?
ERROR(
f"No rows matched in query: "
f"PlaylistTracks.playlist_id == {from_playlist_id}, "
f"PlaylistTracks.row == {row}"
)
return
record.playlist_id = to_playlist_id
record.row = new_row
session.commit()
@staticmethod
def new_row(session, playlist_id):
"""
Return row number > the largest existing row number
If there are no existing rows, return 0 (ie, first row number)
"""
last_row = session.query(func.max(PlaylistTracks.row)).one()[0]
if last_row:
return last_row + 1
else:
return 0
@staticmethod
def remove_all_tracks(session, playlist_id):
"""
Remove all tracks from passed playlist_id
"""
session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == playlist_id,
).delete()
session.commit()
@staticmethod
def remove_track(session, playlist_id, row):
DEBUG(
f"PlaylistTracks.remove_track(playlist_id={playlist_id}, "
f"row={row})"
)
session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == playlist_id,
PlaylistTracks.row == row
).delete()
session.commit()
@staticmethod
def update_row_track(session, playlist_id, row, track_id):
DEBUG(
f"PlaylistTracks.update_track_row(playlist_id={playlist_id}, "
f"row={row}, track_id={track_id})"
)
try:
plt = session.query(PlaylistTracks).filter(
PlaylistTracks.playlist_id == playlist_id,
PlaylistTracks.row == row
).one()
except MultipleResultsFound:
ERROR(
f"Multiple rows matched in query: "
f"PlaylistTracks.playlist_id == {playlist_id}, "
f"PlaylistTracks.row == {row}"
)
return
except NoResultFound:
ERROR(
f"No rows matched in query: "
f"PlaylistTracks.playlist_id == {playlist_id}, "
f"PlaylistTracks.row == {row}"
)
return
plt.track_id = track_id
session.commit()
class Settings(Base):
__tablename__ = 'settings'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(32), nullable=False, unique=True)
f_datetime = Column(DateTime, default=None, nullable=True)
f_int = Column(Integer, default=None, nullable=True)
f_string = Column(String(128), default=None, nullable=True)
@classmethod
def get_int(cls, session, name):
try:
int_setting = session.query(cls).filter(
cls.name == name).one()
except NoResultFound:
int_setting = Settings()
int_setting.name = name
int_setting.f_int = None
session.add(int_setting)
session.commit()
return int_setting
def update(self, session, data):
for key, value in data.items():
assert hasattr(self, key)
setattr(self, key, value)
session.commit()
class Tracks(Base):
__tablename__ = 'tracks'
id = Column(Integer, primary_key=True, autoincrement=True)
title = Column(String(256), index=True)
artist = Column(String(256), index=True)
duration = Column(Integer, index=True)
start_gap = Column(Integer, index=False)
fade_at = Column(Integer, index=False)
silence_at = Column(Integer, index=False)
path = Column(String(2048), index=False, nullable=False)
mtime = Column(Float, index=True)
lastplayed = Column(DateTime, index=True, default=None)
playlists = relationship("PlaylistTracks", back_populates="tracks")
playdates = relationship("Playdates", back_populates="tracks")
def __repr__(self):
return (
f"<Track(id={self.id}, title={self.title}, "
f"artist={self.artist}, path={self.path}>"
)
# Not currently used 1 June 2021
# @staticmethod
# def get_note(session, id):
# return session.query(Notes).filter(Notes.id == id).one()
@staticmethod
def get_all_paths(session):
"""Return a list of paths of all tracks"""
return [a[0] for a in session.query(Tracks.path).all()]
@classmethod
def get_all_tracks(cls, session):
"Return a list of all tracks"
return session.query(Tracks).all()
@classmethod
def get_or_create(cls, session, path):
DEBUG(f"Tracks.get_or_create(path={path})")
try:
track = session.query(cls).filter(cls.path == path).one()
except NoResultFound:
track = Tracks()
track.path = path
session.add(track)
return track
# @staticmethod
# def get_duration(session, id):
# try:
# return session.query(
# Tracks.duration).filter(Tracks.id == id).one()[0]
# except NoResultFound:
# ERROR(f"Can't find track id {id}")
# return None
@classmethod
def get_from_filename(cls, session, filename):
"""
Return track if one and only one track in database has passed
filename (ie, basename of path). Return None if zero or more
than one track matches.
"""
DEBUG(f"Tracks.get_track_from_filename({filename=})")
try:
track = session.query(Tracks).filter(Tracks.path.ilike(
f'%{os.path.sep}{filename}')).one()
return track
except (NoResultFound, MultipleResultsFound):
return None
@classmethod
def get_from_id(cls, session, id):
return session.query(Tracks).filter(
Tracks.id == id).one()
@classmethod
def get_from_path(cls, session, path):
"""
Return track with passee path, or None.
"""
DEBUG(f"Tracks.get_track_from_path({path=})")
return session.query(Tracks).filter(Tracks.path == path).first()
# @staticmethod
# def get_path(session, track_id):
# "Return path of passed track_id, or None"
#
# try:
# return session.query(Tracks.path).filter(
# Tracks.id == track_id).one()[0]
# except NoResultFound:
# ERROR(f"Can't find track id {track_id}")
# return None
@classmethod
def get_by_id(cls, session, track_id):
"""Return track or None"""
try:
DEBUG(f"Tracks.get_track(track_id={track_id})")
track = session.query(Tracks).filter(Tracks.id == track_id).one()
return track
except NoResultFound:
ERROR(f"get_track({track_id}): not found")
return None
@staticmethod
def remove_by_path(session, path):
"Remove track with passed path from database"
DEBUG(f"Tracks.remove_path({path=})")
try:
session.query(Tracks).filter(Tracks.path == path).delete()
session.commit()
except IntegrityError as exception:
ERROR(f"Can't remove track with {path=} ({exception=})")
# @staticmethod
# def search(session, title=None, artist=None, duration=None):
# """
# Return any tracks matching passed criteria
# """
#
# DEBUG(
# f"Tracks.search({title=}, {artist=}), {duration=})"
# )
#
# if not title and not artist and not duration:
# return None
#
# q = session.query(Tracks).filter(False)
# if title:
# q = q.filter(Tracks.title == title)
# if artist:
# q = q.filter(Tracks.artist == artist)
# if duration:
# q = q.filter(Tracks.duration == duration)
#
# return q.all()
@classmethod
def search_artists(cls, session, text):
return (
session.query(Tracks)
.filter(Tracks.artist.ilike(f"%{text}%"))
.order_by(Tracks.title)
).all()
@classmethod
def search_titles(cls, session, text):
return (
session.query(Tracks)
.filter(Tracks.title.ilike(f"%{text}%"))
.order_by(Tracks.title)
).all()
def update_lastplayed(self, session):
self.lastplayed = datetime.now()
session.add(self)
session.commit()
def update_artist(self, session, artist):
self.artist = artist
session.add(self)
session.commit()
def update_title(self, session, title):
self.title = title
session.add(self)
session.commit()
def update_path(self, newpath):
self.path = newpath

29
app/musicmuster.py
View File

@ -211,7 +211,7 @@ class Window(QMainWindow, Ui_MainWindow):
ok = dlg.exec()
if ok:
with Session() as session:
playlist_db = Playlists.new(session, dlg.textValue())
playlist_db = Playlists(session, dlg.textValue())
self.load_playlist(session, playlist_db)
def change_volume(self, volume):
@ -260,6 +260,7 @@ class Window(QMainWindow, Ui_MainWindow):
note = Notes.add_note(
session, self.visible_playlist_tab().id, row, text)
# TODO: this needs to call playlist.add_note now
return note
@ -356,7 +357,7 @@ class Window(QMainWindow, Ui_MainWindow):
# Get playlist db object
with Session() as session:
playlist_db = Playlists.get_playlist_by_id(
playlist_db = Playlists.get_by_id(
session, self.visible_playlist_tab().id)
with open(path, "w") as f:
# Required directive on first line
@ -400,7 +401,7 @@ class Window(QMainWindow, Ui_MainWindow):
"Load the playlists that we loaded at end of last session"
with Session() as session:
for playlist_db in Playlists.get_last_used(session):
for playlist_db in Playlists.get_open(session):
self.load_playlist(session, playlist_db)
def load_playlist(self, session, playlist_db):
@ -410,7 +411,7 @@ class Window(QMainWindow, Ui_MainWindow):
"""
playlist_tab = PlaylistTab(self)
playlist_db.open(session)
playlist_db.mark_open(session)
playlist_tab.populate(session, playlist_db)
idx = self.tabPlaylist.addTab(playlist_tab, playlist_db.name)
self.tabPlaylist.setCurrentIndex(idx)
@ -418,8 +419,10 @@ class Window(QMainWindow, Ui_MainWindow):
def move_selected(self):
"Move selected rows to another playlist"
# TODO needs refactoring
with Session() as session:
playlist_dbs = [p for p in Playlists.get_all_playlists(session)
playlist_dbs = [p for p in Playlists.get_all(session)
if p.id != self.visible_playlist_tab().id]
dlg = SelectPlaylistDialog(self, playlist_dbs=playlist_dbs)
dlg.exec()
@ -444,19 +447,19 @@ class Window(QMainWindow, Ui_MainWindow):
self.visible_playlist_tab().get_selected_rows_and_tracks()
):
rows.append(row)
track = Tracks.track_from_id(session, track_id)
track = Tracks.get_from_id(session, track_id)
if destination_visible_playlist_tab:
# Insert with repaint=False to not update database
destination_visible_playlist_tab.insert_track(
session, track, repaint=False)
# Update database
# Update database for both source and destination playlists
PlaylistTracks.move_track(
session, self.visible_playlist_tab().id, row, dlg.plid)
# Update destination playlist if visible
if destination_visible_playlist_tab:
destination_visible_playlist_tab.repaint()
destination_visible_playlist_tab.update_display()
# Update source playlist
self.visible_playlist_tab().remove_rows(rows)
@ -514,7 +517,7 @@ class Window(QMainWindow, Ui_MainWindow):
next_track_id = self.current_track_playlist_tab.play_started()
if next_track_id is not None:
self.next_track = Tracks.get_track(session, next_track_id)
self.next_track = Tracks.get_by_id(session, next_track_id)
self.next_track_playlist_tab = self.current_track_playlist_tab
else:
self.next_track = self.next_track_playlist_tab = None
@ -548,11 +551,11 @@ class Window(QMainWindow, Ui_MainWindow):
def open_playlist(self):
with Session() as session:
playlist_dbs = Playlists.get_all_closed_playlists(session)
playlist_dbs = Playlists.get_closed(session)
dlg = SelectPlaylistDialog(self, playlist_dbs=playlist_dbs)
dlg.exec()
if dlg.plid:
playlist_db = Playlists.get_playlist_by_id(session, dlg.plid)
playlist_db = Playlists.get_by_id(session, dlg.plid)
self.load_playlist(session, playlist_db)
def select_next_row(self):
@ -601,7 +604,7 @@ class Window(QMainWindow, Ui_MainWindow):
self.set_tab_colour(self.next_track_playlist_tab,
QColor(Config.COLOUR_NEXT_TAB))
self.next_track = Tracks.get_track(session, next_track_id)
self.next_track = Tracks.get_by_id(session, next_track_id)
self.update_headers()
@ -880,7 +883,7 @@ class DbDialog(QDialog):
self.select_searchtext()
def add_track(self, track_id):
track = Tracks.track_from_id(self.session, track_id)
track = Tracks.get_from_id(self.session, track_id)
# Add to playlist on screen
# If we don't specify "repaint=False", playlist will
# also be saved to database

472
app/playlists.py
View File

@ -155,8 +155,8 @@ class PlaylistTab(QTableWidget):
)
with Session() as session:
self._save_playlist(session)
self._repaint()
self.save_playlist(session)
self.update_display()
def edit(self, index, trigger, event):
result = super(PlaylistTab, self).edit(index, trigger, event)
@ -270,8 +270,8 @@ class PlaylistTab(QTableWidget):
self.scrollToItem(titleitem, QAbstractItemView.PositionAtCenter)
if repaint:
self._save_playlist(session)
self._repaint(clear_selection=False)
self.save_playlist(session)
self.update_display(clear_selection=False)
return row
@ -324,8 +324,8 @@ class PlaylistTab(QTableWidget):
self._meta_set_unreadable(row)
if repaint:
self._save_playlist(session)
self._repaint(clear_selection=False)
self.save_playlist(session)
self.update_display(clear_selection=False)
return row
@ -333,13 +333,13 @@ class PlaylistTab(QTableWidget):
"Clear current track"
self._meta_clear_current()
self._repaint()
self.update_display()
def clear_next(self):
"""Clear next track"""
self._meta_clear_next()
self._repaint()
self.update_display()
def get_next_track_id(self):
"Return next track id"
@ -389,9 +389,9 @@ class PlaylistTab(QTableWidget):
self.removeRow(row)
with Session() as session:
self._save_playlist(session)
self.save_playlist(session)
self._repaint()
self.update_display()
def play_started(self):
"""
@ -418,13 +418,13 @@ class PlaylistTab(QTableWidget):
break
search_starting_row += 1
self._repaint()
self.update_display()
return next_track_id
def play_stopped(self):
self._meta_clear_current()
self.current_track_start_time = None
self._repaint()
self.update_display()
def populate(self, session, playlist_db):
"""
@ -464,12 +464,80 @@ class PlaylistTab(QTableWidget):
scroll_to = self.item(0, self.COL_INDEX)
self.scrollToItem(scroll_to, QAbstractItemView.PositionAtTop)
self._save_playlist(session)
self._repaint()
self.save_playlist(session)
self.update_display()
def repaint(self):
# Called when we change tabs
self._repaint()
def save_playlist(self, session):
"""
Save playlist to database.
For notes: check the database entry is correct and update it if
necessary. Playlists:Note is one:many, so there is only one notes
appearance in all playlists.
For tracks: erase the playlist tracks and recreate. This is much
simpler than trying to correct any Playlists:Tracks many:many
errors.
"""
# We need to add ourself to the session
playlist_db = session.query(Playlists).filter(
Playlists.id == self.id).one()
# Notes first
# Create dictionaries indexed by note_id
playlist_notes = {}
database_notes = {}
notes_rows = self._meta_get_notes()
# PlaylistTab
for row in notes_rows:
note_id = self._get_row_id(row)
if not note_id:
DEBUG(f"(_save_playlist(): no COL_INDEX data in row {row}")
continue
playlist_notes[note_id] = row
# Database
for note in playlist_db.notes:
database_notes[note.id] = note.row
# Notes to add to database
# This should never be needed as notes are added to a specific
# playlist upon creation
for note_id in set(playlist_notes.keys()) - set(database_notes.keys()):
ERROR(
f"_save_playlist(): Note.id={note_id} "
f"missing from playlist {playlist_db} in database"
)
# Notes to remove from database
for note_id in set(database_notes.keys()) - set(playlist_notes.keys()):
DEBUG(
f"_save_playlist(): Delete note note_id={note_id} "
f"from playlist {playlist_db} in database"
)
Notes.delete_note(session, note_id)
# Note rows to update in playlist database
for note_id in set(playlist_notes.keys()) & set(database_notes.keys()):
if playlist_notes[note_id] != database_notes[note_id]:
DEBUG(
f"_save_playlist(): Update database note.id {note_id} "
f"from row={database_notes[note_id]} to "
f"row={playlist_notes[note_id]}"
)
Notes.update_note(session, note_id, playlist_notes[note_id])
# Tracks
# Remove all tracks for us in datbase
playlist_db.remove_all_tracks(session)
# Iterate on-screen playlist and add tracks back in
for row in range(self.rowCount()):
if row in notes_rows:
continue
playlist_db.add_track(
session, self.id, self._get_row_id(row), row)
def select_next_row(self):
"""
@ -579,7 +647,140 @@ class PlaylistTab(QTableWidget):
return
self._set_next(self.currentRow())
self._repaint()
self.update_display()
def update_display(self, clear_selection=True):
"Set row colours, fonts, etc"
DEBUG(
f"playlist[{self.id}:{self.name}]."
f"_repaint(clear_selection={clear_selection}"
)
with Session() as session:
if clear_selection:
self.clearSelection()
current = self._meta_get_current()
next = self._meta_get_next()
notes = self._meta_get_notes()
unreadable = self._meta_get_unreadable()
# Set colours and start times
next_start_time = None
# Don't change start times for tracks that have been played.
# For unplayed tracks, if there's a 'current' or 'next'
# track marked, populate start times from then onwards. If
# neither, populate start times from first note with a start
# time.
if current and next:
start_times_row = min(current, next)
else:
start_times_row = current or next
if not start_times_row:
start_times_row = 0
# Cycle through all rows
for row in range(self.rowCount()):
# We can't calculate start times until next_start_time is
# set. That can be set by either a note with a time, or the
# current track.
if row in notes:
row_time = self._get_row_time(row)
if row_time:
next_start_time = row_time
# Set colour
note_text = self.item(row, self.COL_TITLE).text()
note_colour = NoteColours.get_colour(session, note_text)
if not note_colour:
note_colour = Config.COLOUR_NOTES_PLAYLIST
self._set_row_colour(
row, QColor(note_colour)
)
self._set_row_bold(row)
elif row in unreadable:
# Set colour
self._set_row_colour(
row, QColor(Config.COLOUR_UNREADABLE)
)
self._set_row_bold(row)
elif row == current:
# Set start time
self._set_row_start_time(
row, self.current_track_start_time)
last_played_str = get_relative_date(
self.current_track_start_time)
self.item(row, self.COL_LAST_PLAYED).setText(
last_played_str)
# Calculate next_start_time
next_start_time = self._calculate_next_start_time(
session, row, self.current_track_start_time)
# Set end time
self._set_row_end_time(row, next_start_time)
# Set colour
self._set_row_colour(row, QColor(
Config.COLOUR_CURRENT_PLAYLIST))
# Make bold
self._set_row_bold(row)
elif row == next:
# if there's a track playing, set start time from that
if self.current_track_start_time:
start_time = self._calculate_next_start_time(
session, current, self.current_track_start_time)
else:
# No current track to base from, but don't change
# time if it's already set
start_time = self._get_row_time(row)
if not start_time:
start_time = next_start_time
# Now set it
self._set_row_start_time(row, start_time)
next_start_time = self._calculate_next_start_time(
session, row, start_time)
# Set end time
self._set_row_end_time(row, next_start_time)
# Set colour
self._set_row_colour(
row, QColor(Config.COLOUR_NEXT_PLAYLIST))
# Make bold
self._set_row_bold(row)
else:
# Stripe remaining rows
if row % 2:
colour = QColor(Config.COLOUR_ODD_PLAYLIST)
else:
colour = QColor(Config.COLOUR_EVEN_PLAYLIST)
self._set_row_colour(row, colour)
track_id = self._get_row_id(row)
if track_id in self.played_tracks:
# Played today, so update last played column
last_playtime = Playdates.last_played(
session, track_id)
last_played_str = get_relative_date(last_playtime)
self.item(row, self.COL_LAST_PLAYED).setText(
last_played_str)
self._set_row_not_bold(row)
else:
# Set start/end times only if we haven't played it yet
if next_start_time and row >= start_times_row:
self._set_row_start_time(row, next_start_time)
next_start_time = self._calculate_next_start_time(
session, row, next_start_time)
# Set end time
self._set_row_end_time(row, next_start_time)
else:
# Clear start and end time
self._set_row_start_time(row, None)
self._set_row_end_time(row, None)
# Don't dim unplayed tracks
self._set_row_bold(row)
# ########## Internally called functions ##########
@ -594,7 +795,7 @@ class PlaylistTab(QTableWidget):
track_id = self._get_row_id(row)
if track_id:
with Session() as session:
track = Tracks.get_track(session, track_id)
track = Tracks.get_by_id(session, track_id)
open_in_audacity(track.path)
def _calculate_next_start_time(self, session, row, start):
@ -667,11 +868,11 @@ class PlaylistTab(QTableWidget):
# Reset row start time in case it used to have one
self._set_row_start_time(row, None)
DEBUG(
f"_cell_changed:Note {new} does not contain "
f"_ct ell_changed:Note {new} does not contain "
"start time"
)
else:
track = Tracks.get_track(session, row_id)
track = Tracks.get_by_id(session, row_id)
if column == self.COL_ARTIST:
update_meta(session, track, artist=new)
elif column == self.COL_TITLE:
@ -690,7 +891,7 @@ class PlaylistTab(QTableWidget):
# Call repaint to update start times, such as when a note has
# been edited
self._repaint()
self.update_display()
self.master_process.enable_play_next_controls()
@ -718,16 +919,16 @@ class PlaylistTab(QTableWidget):
# delete in reverse row order so row numbers don't
# change
for del_row in sorted(rows_to_delete, reverse=True):
id = self._get_row_id(del_row)
if del_row in notes:
for row in sorted(rows_to_delete, reverse=True):
id = self._get_row_id(row)
if row in notes:
Notes.delete_note(session, id)
else:
PlaylistTracks.remove_track(session, self.id, del_row)
self.removeRow(del_row)
self.remove_track(session, row)
self.removeRow(row)
self._save_playlist(session)
self._repaint()
self.save_playlist(session)
self.update_display()
def _drop_on(self, event):
index = self.indexAt(event.pos())
@ -774,7 +975,7 @@ class PlaylistTab(QTableWidget):
txt = f"Note: {note_text}"
else:
with Session() as session:
track = Tracks.get_track(session, id)
track = Tracks.get_by_id(session, id)
if not track:
txt = f"Track not found (track.id={id})"
else:
@ -989,143 +1190,10 @@ class PlaylistTab(QTableWidget):
else:
self._meta_set_unreadable(row)
track_id = None
self._repaint()
self.update_display()
return track_id
def _repaint(self, clear_selection=True):
"Set row colours, fonts, etc"
DEBUG(
f"playlist[{self.id}:{self.name}]."
f"_repaint(clear_selection={clear_selection}"
)
with Session() as session:
if clear_selection:
self.clearSelection()
current = self._meta_get_current()
next = self._meta_get_next()
notes = self._meta_get_notes()
unreadable = self._meta_get_unreadable()
# Set colours and start times
next_start_time = None
# Don't change start times for tracks that have been played.
# For unplayed tracks, if there's a 'current' or 'next'
# track marked, populate start times from then onwards. If
# neither, populate start times from first note with a start
# time.
if current and next:
start_times_row = min(current, next)
else:
start_times_row = current or next
if not start_times_row:
start_times_row = 0
# Cycle through all rows
for row in range(self.rowCount()):
# We can't calculate start times until next_start_time is
# set. That can be set by either a note with a time, or the
# current track.
if row in notes:
row_time = self._get_row_time(row)
if row_time:
next_start_time = row_time
# Set colour
note_text = self.item(row, self.COL_TITLE).text()
note_colour = NoteColours.get_colour(session, note_text)
if not note_colour:
note_colour = Config.COLOUR_NOTES_PLAYLIST
self._set_row_colour(
row, QColor(note_colour)
)
self._set_row_bold(row)
elif row in unreadable:
# Set colour
self._set_row_colour(
row, QColor(Config.COLOUR_UNREADABLE)
)
self._set_row_bold(row)
elif row == current:
# Set start time
self._set_row_start_time(
row, self.current_track_start_time)
last_played_str = get_relative_date(
self.current_track_start_time)
self.item(row, self.COL_LAST_PLAYED).setText(
last_played_str)
# Calculate next_start_time
next_start_time = self._calculate_next_start_time(
session, row, self.current_track_start_time)
# Set end time
self._set_row_end_time(row, next_start_time)
# Set colour
self._set_row_colour(row, QColor(
Config.COLOUR_CURRENT_PLAYLIST))
# Make bold
self._set_row_bold(row)
elif row == next:
# if there's a track playing, set start time from that
if self.current_track_start_time:
start_time = self._calculate_next_start_time(
session, current, self.current_track_start_time)
else:
# No current track to base from, but don't change
# time if it's already set
start_time = self._get_row_time(row)
if not start_time:
start_time = next_start_time
# Now set it
self._set_row_start_time(row, start_time)
next_start_time = self._calculate_next_start_time(
session, row, start_time)
# Set end time
self._set_row_end_time(row, next_start_time)
# Set colour
self._set_row_colour(
row, QColor(Config.COLOUR_NEXT_PLAYLIST))
# Make bold
self._set_row_bold(row)
else:
# Stripe remaining rows
if row % 2:
colour = QColor(Config.COLOUR_ODD_PLAYLIST)
else:
colour = QColor(Config.COLOUR_EVEN_PLAYLIST)
self._set_row_colour(row, colour)
track_id = self._get_row_id(row)
if track_id in self.played_tracks:
# Played today, so update last played column
last_playtime = Playdates.last_played(
session, track_id)
last_played_str = get_relative_date(last_playtime)
self.item(row, self.COL_LAST_PLAYED).setText(
last_played_str)
self._set_row_not_bold(row)
else:
# Set start/end times only if we haven't played it yet
if next_start_time and row >= start_times_row:
self._set_row_start_time(row, next_start_time)
next_start_time = self._calculate_next_start_time(
session, row, next_start_time)
# Set end time
self._set_row_end_time(row, next_start_time)
else:
# Clear start and end time
self._set_row_start_time(row, None)
self._set_row_end_time(row, None)
# Don't dim unplayed tracks
self._set_row_bold(row)
def _rescan(self, row):
"""
If passed row is track row, rescan it.
@ -1140,82 +1208,10 @@ class PlaylistTab(QTableWidget):
track_id = self._get_row_id(row)
if track_id:
with Session() as session:
track = Tracks.get_track(session, track_id)
track = Tracks.get_by_id(session, track_id)
create_track_from_file(session, track.path)
self._update_row(row, track)
def _save_playlist(self, session):
"""
Save playlist to database.
For notes: check the database entry is correct and update it if
necessary. Playlists:Note is one:many, so there is only one notes
appearance in all playlists.
For tracks: erase the playlist tracks and recreate. This is much
simpler than trying to correct any Playlists:Tracks many:many
errors.
"""
# We need to add ourself to the session
playlist_db = session.query(Playlists).filter(
Playlists.id == self.id).one()
# Notes first
# Create dictionaries indexed by note_id
playlist_notes = {}
database_notes = {}
notes_rows = self._meta_get_notes()
# PlaylistTab
for row in notes_rows:
note_id = self._get_row_id(row)
if not note_id:
DEBUG(f"(_save_playlist(): no COL_INDEX data in row {row}")
continue
playlist_notes[note_id] = row
# Database
for note in playlist_db.notes:
database_notes[note.id] = note.row
# Notes to add to database
# This should never be needed as notes are added to a specific
# playlist upon creation
for note_id in set(playlist_notes.keys()) - set(database_notes.keys()):
ERROR(
f"_save_playlist(): Note.id={note_id} "
f"missing from playlist {playlist_db} in database"
)
# Notes to remove from database
for note_id in set(database_notes.keys()) - set(playlist_notes.keys()):
DEBUG(
f"_save_playlist(): Delete note note_id={note_id} "
f"from playlist {playlist_db} in database"
)
Notes.delete_note(session, note_id)
# Note rows to update in playlist database
for note_id in set(playlist_notes.keys()) & set(database_notes.keys()):
if playlist_notes[note_id] != database_notes[note_id]:
DEBUG(
f"_save_playlist(): Update database note.id {note_id} "
f"from row={database_notes[note_id]} to "
f"row={playlist_notes[note_id]}"
)
Notes.update_note(session, note_id, playlist_notes[note_id])
# Tracks
# Remove all tracks for us in datbase
PlaylistTracks.remove_all_tracks(session, self.id)
# Iterate on-screen playlist and add tracks back in
for row in range(self.rowCount()):
if row in notes_rows:
continue
PlaylistTracks.add_track(
session, self.id, self._get_row_id(row), row)
def _select_event(self):
"""
Called when item selection changes.
@ -1322,4 +1318,4 @@ class PlaylistTab(QTableWidget):
item_duration = self.item(row, self.COL_DURATION)
item_duration.setText(helpers.ms_to_mmss(track.duration))
self._repaint()
self.update_display()

14
app/songdb.py
View File

@ -21,6 +21,9 @@ def main():
"Main loop"
DEBUG("Starting")
print("needs refactoring")
import sys
sys.exit(1)
# Parse command line
p = argparse.ArgumentParser()
@ -329,7 +332,7 @@ def update_db(session):
for path in list(os_paths - db_paths):
DEBUG(f"songdb.update_db: {path=} not in database")
# is filename in database?
track = Tracks.get_track_from_filename(session, os.path.basename(path))
track = Tracks.get_from_filename(session, os.path.basename(path))
if not track:
messages.append(f"Track missing from database: {path}")
else:
@ -345,7 +348,7 @@ def update_db(session):
# Remote any tracks from database whose paths don't exist
for path in list(db_paths - os_paths):
# Manage tracks listed in database but where path is invalid
track = Tracks.get_track_from_path(session, path)
track = Tracks.get_from_path(session, path)
messages.append(f"Remove from database: {path=} {track=}")
# Remove references from Playdates
@ -359,11 +362,12 @@ def update_db(session):
for pt in PlaylistTracks.get_track_playlists(session, track.id):
# Create note
Notes.add_note(session, pt.playlist_id, pt.row, note_txt)
# TODO: this needs to call playlist.add_note() now
# Remove playlist entry
PlaylistTracks.remove_track(session, pt.playlist_id, pt.row)
# Remove Track entry pointing to invalid path
Tracks.remove_path(session, path)
Tracks.remove_by_path(session, path)
# Output messages (so if running via cron, these will get sent to
# user)
@ -406,9 +410,9 @@ def update_meta(session, track, artist=None, title=None):
# Update database
with Session() as session:
if artist:
Tracks.update_artist(session, track.id, artist)
track.update_artist(session, artist)
if title:
Tracks.update_title(session, track.id, title)
track.update_title(session, title)
if __name__ == '__main__' and '__file__' in globals():

18
test_models.py
View File

@ -89,15 +89,15 @@ def test_playlist_close(session):
session.add(playlist2)
session.commit()
apl = Playlists.get_all_playlists(session)
apl = Playlists.get_all(session)
assert len(apl) == 2
cpl = Playlists.get_all_closed_playlists(session)
cpl = Playlists.get_closed(session)
assert len(cpl) == 0
playlist2.close(session)
cpl = Playlists.get_all_closed_playlists(session)
cpl = Playlists.get_closed(session)
assert len(cpl) == 1
@ -119,15 +119,15 @@ def test_playlist_get_last_user(session):
session.add(playlist3)
session.commit()
apl = Playlists.get_all_playlists(session)
apl = Playlists.get_all(session)
assert len(apl) == 3
playlist1.open(session)
playlist1.mark_open(session)
playlist2.close(session)
time.sleep(1)
playlist3.open(session)
playlist3.mark_open(session)
last_used = Playlists.get_last_used(session)
last_used = Playlists.get_open(session)
assert len(last_used) == 2
assert last_used[0].name == "Test playlist three"
@ -136,6 +136,6 @@ def test_playlist_new(session):
"""Test new function"""
plname = "This is a test name"
p = Playlists.new(session, plname)
n = Playlists.get_playlist_by_id(session, p.id)
p = Playlists(session, plname)
n = Playlists.get_by_id(session, p.id)
assert p.name == n.name