#!/usr/bin/env python import argparse import os import shutil import tempfile from config import Config from log import DEBUG, INFO from model import Tracks, Playdates, PlaylistTracks, Session from mutagen.flac import FLAC from mutagen.mp3 import MP3 from pydub import AudioSegment, effects from tinytag import TinyTag def main(): "Main loop" INFO("Starting") # Parse command line p = argparse.ArgumentParser() p.add_argument('-u', '--update', action="store_true", dest="update", default=False, help="Update database") p.add_argument('-f', '--full-update', action="store_true", dest="full_update", default=False, help="Update database") args = p.parse_args() # Run as required if args.update: INFO("Updating database") with Session() as session: update_db(session) elif args.full_update: INFO("Full update of database") with Session() as session: full_update_db(session) else: INFO("No action specified") INFO("Finished") def create_track_from_file(session, path): """ Create track in database from passed path, or update database entry if path already in database. Return track. """ track = Tracks.get_or_create(session, path) t = get_music_info(path) track.title = t['title'] track.artist = t['artist'] track.duration = int(round( t['duration'], Config.MILLISECOND_SIGFIGS) * 1000) audio = get_audio_segment(path) track.start_gap = leading_silence(audio) track.fade_at = round(fade_point(audio) / 1000, Config.MILLISECOND_SIGFIGS) * 1000 track.silence_at = round(trailing_silence(audio) / 1000, Config.MILLISECOND_SIGFIGS) * 1000 track.mtime = os.path.getmtime(path) session.commit() if Config.NORMALISE_ON_IMPORT: # Check type ftype = os.path.splitext(path)[1][1:] if ftype not in ['mp3', 'flac']: INFO(f"File type {ftype} not implemented") return track # Get current file gid, uid and permissions stats = os.stat(path) try: # Copy original file fd, temp_path = tempfile.mkstemp() shutil.copyfile(path, temp_path) except Exception as err: DEBUG(f"songdb.create_track_from_file({path}): err1: {str(err)}") return # Overwrite original file with normalised output normalised = effects.normalize(audio) try: normalised.export(path, format=os.path.splitext(path)[1][1:]) # Fix up permssions and ownership os.chown(path, stats.st_uid, stats.st_gid) os.chmod(path, stats.st_mode) # Copy tags if ftype == 'flac': tag_handler = FLAC elif ftype == 'mp3': tag_handler = MP3 else: return track src = tag_handler(temp_path) dst = tag_handler(path) for tag in src: dst[tag] = src[tag] dst.save() except Exception as err: DEBUG(f"songdb.create_track_from_file({path}): err2: {str(err)}") # Restore original file shutil.copyfile(path, temp_path) finally: if os.path.exists(temp_path): os.remove(temp_path) return track def full_update_db(): "Rescan all entries in database" print("Full scan not yet implemented") def get_audio_segment(path): try: if path.endswith('.mp3'): return AudioSegment.from_mp3(path) elif path.endswith('.flac'): return AudioSegment.from_file(path, "flac") except AttributeError: return None def get_music_info(path): """ Return a dictionary of title, artist, duration-in-milliseconds and path. """ tag = TinyTag.get(path) return dict( title=tag.title, artist=tag.artist, duration=tag.duration, path=path ) def leading_silence(audio_segment, silence_threshold=Config.DBFS_SILENCE, chunk_size=Config.AUDIO_SEGMENT_CHUNK_SIZE): """ Returns the millisecond/index that the leading silence ends. audio_segment - the segment to find silence in silence_threshold - the upper bound for how quiet is silent in dFBS chunk_size - chunk size for interating over the segment in ms https://github.com/jiaaro/pydub/blob/master/pydub/silence.py """ trim_ms = 0 # ms assert chunk_size > 0 # to avoid infinite loop while ( audio_segment[trim_ms:trim_ms + chunk_size].dBFS < # noqa W504 silence_threshold and trim_ms < len(audio_segment)): trim_ms += chunk_size # if there is no end it should return the length of the segment return min(trim_ms, len(audio_segment)) def fade_point(audio_segment, fade_threshold=Config.DBFS_FADE, chunk_size=Config.AUDIO_SEGMENT_CHUNK_SIZE): """ Returns the millisecond/index of the point where the fade is down to fade_threshold and doesn't get louder again. audio_segment - the sdlg_search_database_uiegment to find silence in fade_threshold - the upper bound for how quiet is silent in dFBS chunk_size - chunk size for interating over the segment in ms """ assert chunk_size > 0 # to avoid infinite loop segment_length = audio_segment.duration_seconds * 1000 # ms trim_ms = segment_length - chunk_size while ( audio_segment[trim_ms:trim_ms + chunk_size].dBFS < fade_threshold and trim_ms > 0): # noqa W503 trim_ms -= chunk_size # if there is no trailing silence, return lenght of track (it's less # the chunk_size, but for chunk_size = 10ms, this may be ignored) return int(trim_ms) # Current unused (1 June 2021) # def rescan_database(session): # # tracks = Tracks.get_all_tracks(session) # total_tracks = len(tracks) # track_count = 0 # for track in tracks: # track_count += 1 # print(f"Track {track_count} of {total_tracks}") # audio = get_audio_segment(track.path) # track.start_gap = leading_silence(audio) # track.fade_at = fade_point(audio) # track.silence_at = trailing_silence(audio) # session.commit() def trailing_silence(audio_segment, silence_threshold=-50.0, chunk_size=Config.AUDIO_SEGMENT_CHUNK_SIZE): return fade_point(audio_segment, silence_threshold, chunk_size) def update_db(session): """ Repopulate database """ # Search for tracks in only one of directory and database db_paths = set(Tracks.get_all_paths(session)) os_paths_list = [] for root, dirs, files in os.walk(Config.ROOT): for f in files: path = os.path.join(root, f) ext = os.path.splitext(f)[1] if ext in [".flac", ".mp3"]: os_paths_list.append(path) os_paths = set(os_paths_list) # If a track is moved, only the path will have changed. # For any files we have found whose paths are not in the database, # check to see whether the filename (basename) is present in the # database: 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)) if not track: DEBUG(f"songdb.update_db: Adding to database: {path}") create_track_from_file(session, path) else: # Check track info matches found track t = get_music_info(path) if t['artist'] == track.artist and t['title'] == track.title: track.update_path(path) else: create_track_from_file(session, path) # Refresh database paths db_paths = set(Tracks.get_all_paths(session)) # Remote any tracks from database whose paths don't exist for path in list(db_paths - os_paths): track = Tracks.get_track_from_path(session, path) DEBUG(f"songdb.update_db(): remove from database: {path=} {track=}") played = Playdates.last_played(session, track) playlists = PlaylistTracks.get_playlists_containing_track_id( session, track.id) if played: INFO( f"songdb.update_db: Can't remove {track.id=} ({track.path=}) " f"as it's in playdates.id={played.id}" ) elif playlists: INFO( f"songdb.update_db: Can't remove {track.id=} ({track.path=} " f"as it's in playlists {[p.name for p in playlists]}" ) else: Tracks.remove_path(session, path) if __name__ == '__main__' and '__file__' in globals(): main()