From 698fa4625aa1baf3c2f810b561f9c78354027717 Mon Sep 17 00:00:00 2001 From: Keith Edmunds Date: Tue, 7 Nov 2023 23:14:26 +0000 Subject: [PATCH] WIP V3: track start/stop times basics working Only updates from header rows or current track. Changing current track doesn't update correctly. --- app/config.py | 2 +- app/helpers.py | 89 ++++++++++++++++++++++++++----------------- app/playlistmodel.py | 84 ++++++++++++++++++++++++++++++---------- test_playlistmodel.py | 71 ++++++++++++++++++++++++++++++++++ 4 files changed, 190 insertions(+), 56 deletions(-) diff --git a/app/config.py b/app/config.py index ff1fb54..38d60a3 100644 --- a/app/config.py +++ b/app/config.py @@ -81,7 +81,7 @@ class Config(object): MAX_MISSING_FILES_TO_REPORT = 10 MILLISECOND_SIGFIGS = 0 MINIMUM_ROW_HEIGHT = 30 - NOTE_TIME_FORMAT = "%H:%M:%S" + NOTE_TIME_FORMAT = "%H:%M" OBS_HOST = "localhost" OBS_PASSWORD = "auster" OBS_PORT = 4455 diff --git a/app/helpers.py b/app/helpers.py index 35a1943..be197ff 100644 --- a/app/helpers.py +++ b/app/helpers.py @@ -4,6 +4,7 @@ from typing import Any, Dict, Optional import functools import os import psutil +import re import shutil import smtplib import ssl @@ -19,6 +20,8 @@ from tinytag import TinyTag # type: ignore from config import Config from log import log +start_time_re = re.compile(r"@\d\d:\d\d") + # Classes are defined after global functions so that classes can use # those functions. @@ -95,20 +98,47 @@ def get_audio_segment(path: str) -> Optional[AudioSegment]: return None -def get_tags(path: str) -> Dict[str, Any]: - """ - Return a dictionary of title, artist, duration-in-milliseconds and path. - """ +def get_embedded_time(text: str) -> Optional[datetime]: + """Return datetime specified as @hh:mm:ss in text""" - tag = TinyTag.get(path) + try: + match = start_time_re.search(text) + except TypeError: + return None + if not match: + return None - return dict( - title=tag.title, - artist=tag.artist, - bitrate=round(tag.bitrate), - duration=int(round(tag.duration, Config.MILLISECOND_SIGFIGS) * 1000), - path=path, - ) + try: + return datetime.strptime(match.group(0)[1:], Config.NOTE_TIME_FORMAT) + except ValueError: + return None + + +def get_file_metadata(filepath: str) -> dict: + """Return track metadata""" + + # Get title, artist, bitrate, duration, path + metadata: Dict[str, str | int | float] = get_tags(filepath) + + metadata["mtime"] = os.path.getmtime(filepath) + + # Set start_gap, fade_at and silence_at + audio = get_audio_segment(filepath) + if not audio: + audio_values = dict(start_gap=0, fade_at=0, silence_at=0) + else: + audio_values = dict( + start_gap=leading_silence(audio), + fade_at=int( + round(fade_point(audio) / 1000, Config.MILLISECOND_SIGFIGS) * 1000 + ), + silence_at=int( + round(trailing_silence(audio) / 1000, Config.MILLISECOND_SIGFIGS) * 1000 + ), + ) + metadata |= audio_values + + return metadata def get_relative_date( @@ -153,31 +183,20 @@ def get_relative_date( return f"{weeks} {weeks_str}, {days} {days_str} ago" -def get_file_metadata(filepath: str) -> dict: - """Return track metadata""" +def get_tags(path: str) -> Dict[str, Any]: + """ + Return a dictionary of title, artist, duration-in-milliseconds and path. + """ - # Get title, artist, bitrate, duration, path - metadata: Dict[str, str | int | float] = get_tags(filepath) + tag = TinyTag.get(path) - metadata["mtime"] = os.path.getmtime(filepath) - - # Set start_gap, fade_at and silence_at - audio = get_audio_segment(filepath) - if not audio: - audio_values = dict(start_gap=0, fade_at=0, silence_at=0) - else: - audio_values = dict( - start_gap=leading_silence(audio), - fade_at=int( - round(fade_point(audio) / 1000, Config.MILLISECOND_SIGFIGS) * 1000 - ), - silence_at=int( - round(trailing_silence(audio) / 1000, Config.MILLISECOND_SIGFIGS) * 1000 - ), - ) - metadata |= audio_values - - return metadata + return dict( + title=tag.title, + artist=tag.artist, + bitrate=round(tag.bitrate), + duration=int(round(tag.duration, Config.MILLISECOND_SIGFIGS) * 1000), + path=path, + ) def leading_silence( diff --git a/app/playlistmodel.py b/app/playlistmodel.py index 76e76e1..713df1f 100644 --- a/app/playlistmodel.py +++ b/app/playlistmodel.py @@ -1,4 +1,4 @@ -from datetime import datetime +from datetime import datetime, timedelta from enum import auto, Enum from typing import List, Optional @@ -17,7 +17,7 @@ from PyQt6.QtGui import ( from classes import track_sequence, MusicMusterSignals, PlaylistTrack from config import Config from dbconfig import scoped_session, Session -from helpers import file_is_unreadable +from helpers import file_is_unreadable, get_embedded_time from log import log from models import Playdates, PlaylistRows, Tracks @@ -43,14 +43,16 @@ class PlaylistRowData: Populate PlaylistRowData from database PlaylistRows record """ - self.start_gap: Optional[int] = None - self.title: str = "" self.artist: str = "" - self.duration: int = 0 - self.lastplayed: datetime = Config.EPOCH self.bitrate = 0 + self.duration: int = 0 + self.end_time: Optional[datetime] = None + self.lastplayed: datetime = Config.EPOCH self.path = "" self.played = False + self.start_gap: Optional[int] = None + self.start_time: Optional[datetime] = None + self.title: str = "" self.plrid: int = plr.id self.plr_rownum: int = plr.plr_rownum @@ -108,6 +110,7 @@ class PlaylistModel(QAbstractTableModel): with Session() as session: self.refresh_data(session) + self.update_track_times() def __repr__(self) -> str: return ( @@ -232,13 +235,15 @@ class PlaylistModel(QAbstractTableModel): - find next track """ + row_number = track_sequence.now.plr_rownum + # Sanity check if not track_sequence.now.track_id: log.error( "playlistmodel:current_track_started called with no current track" ) return - if track_sequence.now.plr_rownum is None: + if row_number is None: log.error( "playlistmodel:current_track_started called with no row number " f"({track_sequence.now=})" @@ -246,10 +251,14 @@ class PlaylistModel(QAbstractTableModel): return # Update display - self.invalidate_row(track_sequence.now.plr_rownum) + self.invalidate_row(row_number) # Update track times - # TODO + self.playlist_rows[row_number].start_time = datetime.now() + self.playlist_rows[row_number].end_time = datetime.now() + timedelta( + milliseconds=self.playlist_rows[row_number].duration + ) + self.update_track_times() # Update Playdates in database with Session() as session: @@ -271,11 +280,7 @@ class PlaylistModel(QAbstractTableModel): try: # Find next row after current track next_row = min( - [ - a - for a in unplayed_rows - if a > track_sequence.now.plr_rownum - ] + [a for a in unplayed_rows if a > track_sequence.now.plr_rownum] ) except ValueError: # Find first unplayed track @@ -349,9 +354,15 @@ class PlaylistModel(QAbstractTableModel): if column == Col.DURATION.value: return QVariant(prd.duration) if column == Col.START_TIME.value: - return QVariant("FIXME") + if prd.start_time: + return QVariant(prd.start_time.strftime(Config.TRACK_TIME_FORMAT)) + else: + return QVariant() if column == Col.END_TIME.value: - return QVariant("FIXME") + if prd.end_time: + return QVariant(prd.end_time.strftime(Config.TRACK_TIME_FORMAT)) + else: + return QVariant() if column == Col.LAST_PLAYED.value: return QVariant(prd.lastplayed) if column == Col.BITRATE.value: @@ -594,9 +605,7 @@ class PlaylistModel(QAbstractTableModel): # Sanity check if not track_sequence.previous.track_id: - log.error( - "playlistmodel:previous_track_ended called with no current track" - ) + log.error("playlistmodel:previous_track_ended called with no current track") return if track_sequence.previous.plr_rownum is None: log.error( @@ -701,5 +710,40 @@ class PlaylistModel(QAbstractTableModel): return False - def supportedDropActions(self): + def supportedDropActions(self) -> Qt.DropAction: return Qt.DropAction.MoveAction | Qt.DropAction.CopyAction + + def update_track_times(self) -> None: + """ + Update track start/end times in self.playlist_rows + """ + + next_start_time: Optional[datetime] = None + + for row_number in range(len(self.playlist_rows)): + plr = self.playlist_rows[row_number] + + # Reset start_time if this is the current row + if row_number == track_sequence.now.plr_rownum: + next_start_time = plr.end_time = track_sequence.now.end_time + continue + + # Don't update times for tracks that have been played + if plr.played: + continue + + # Reset start time if timing in hearer or at current track + if not plr.path: + # This is a header row + header_time = get_embedded_time(plr.note) + if header_time: + next_start_time = header_time + else: + # This is an unplayed track; set start/end if we have a + # start time + if next_start_time is None: + continue + plr.start_time = next_start_time + next_start_time = plr.end_time = next_start_time + timedelta( + milliseconds=self.playlist_rows[row_number].duration + ) diff --git a/test_playlistmodel.py b/test_playlistmodel.py index 38eb95b..688e30b 100644 --- a/test_playlistmodel.py +++ b/test_playlistmodel.py @@ -1,9 +1,37 @@ from app.models import ( Playlists, + Tracks, ) +from app.helpers import get_file_metadata from app import playlistmodel from dbconfig import scoped_session +test_tracks = [ + "testdata/isa.mp3", + "testdata/isa_with_gap.mp3", + "testdata/loser.mp3", + "testdata/lovecats-10seconds.mp3", + "testdata/lovecats.mp3", + "testdata/mom.mp3", + "testdata/sitting.mp3", +] + + +def create_model_with_tracks(session: scoped_session) -> "playlistmodel.PlaylistModel": + + playlist = Playlists(session, "test playlist") + model = playlistmodel.PlaylistModel(playlist.id) + + for row in range(len(test_tracks)): + track_path = test_tracks[row % len(test_tracks)] + metadata = get_file_metadata(track_path) + track = Tracks(session, **metadata) + model.insert_track_row(row, track.id, f"{row=}") + + session.commit() + return model + + def create_model_with_playlist_rows( session: scoped_session, rows: int ) -> "playlistmodel.PlaylistModel": @@ -197,3 +225,46 @@ def test_insert_header_row_middle(monkeypatch, session): model.edit_role(model.rowCount(), playlistmodel.Col.NOTE.value, prd) == note_text ) + + +def test_create_model_with_tracks(monkeypatch, session): + + monkeypatch.setattr(playlistmodel, "Session", session) + model = create_model_with_tracks(session) + assert len(model.playlist_rows) == len(test_tracks) + + +def test_timing_one_track(monkeypatch, session): + + START_ROW = 0 + END_ROW = 2 + + monkeypatch.setattr(playlistmodel, "Session", session) + model = create_model_with_tracks(session) + + model.insert_header_row(START_ROW, "start+") + model.insert_header_row(END_ROW, "-") + + prd = model.playlist_rows[END_ROW] + qv_value = model.display_role(END_ROW, playlistmodel.HEADER_NOTES_COLUMN, prd) + assert qv_value.value() == "4:23" + + +# def test_edit_header(monkeypatch, session): # edit header row in middle of playlist + +# monkeypatch.setattr(playlistmodel, "Session", session) +# note_text = "test text" +# initial_row_count = 11 +# insert_row = 6 + +# model = create_model_with_playlist_rows(session, initial_row_count) +# model.insert_header_row(insert_row, note_text) +# assert model.rowCount() == initial_row_count + 1 +# prd = model.playlist_rows[insert_row] +# # Test against edit_role because display_role for headers is +# # handled differently (sets up row span) +# assert ( +# model.edit_role(model.rowCount(), playlistmodel.Col.NOTE.value, prd) +# == note_text +# ) +