unmix/unmix.py

#! /usr/bin/env python3
# encoding: utf-8

from dataclasses import dataclass
import json
import os
import over
import re
import sys
import version

# --------------------------------------------------
# Exceptions

class UnknownFileType(Exception):
	pass

class InvocationError(Exception):
	pass

class FileSyntaxError(Exception):
	pass

# --------------------------------------------------
# Functions

def detect_silence(path, min_length_s=2, level_dB=-32):
	"""
	Detects segments that are at least min_length_s long where
	the audio level does not exceed level_db.
	"""
	
	cmd = over.cmd.Command("ffmpeg", "-i", "INFILE", "-af", "silencedetect=noise=%.1fdB:d=%.2f" %(level_dB, min_length_s), "-f", "null", "-")
	cmd.INFILE = "file:" + path
	cmd.run(stderr=True)
	raw = cmd.get_all_output().decode("utf-8")
	ends = re.findall("silence_end: ([0-9]+\.[0-9]+) . silence_duration: ([0-9]+\.[0-9]+)", raw)
	
	## convert (end, duration) tuples to (start, duration)
	starts = []
	
	for end_s, dur_s in ends:
		dur = float(dur_s)
		start = float(end_s) - dur
		starts.append((start, dur))
	
	return starts

def get_duration(path):
	"""
	Returns the duration of the media file pointed to by path.
	"""
	
	cmd = over.cmd.Command("ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "INFILE")
	cmd.INFILE = "file:" + path
	cmd.run()
	raw = cmd.get_all_output().decode("utf-8")
	dic = json.loads(raw)
	
	return float(dic["format"]["duration"])

def floats(*args):
	for arg in args:
		yield float(arg)

def sanitize_timestamp(raw):
	parts = raw.split(":")
	parts.reverse()
	t = 0.0
	
	for order, part in enumerate(parts):
		t += float(part) * 60**order
	
	return t

# --------------------------------------------------
# Classes

class Script:
	def __init__(self, title):
		self.lines = ["## " + title]
	
	def define_die(self):
		self.lines.append("")
		self.lines.append("function die {")
		self.lines.append('\techo "failed while $@"')
		self.lines.append("\texit 5")
		self.lines.append("}")
	
	def chapter(self, comment):
		self.lines.append("")
		self.lines.append("# " + comment)
	
	def line(self, a, die=None):
		if type(a) == over.cmd.Command:
			line = " ".join(a.dump(pretty=True))
		else:
			line = a
		
		if die:
			line += " || die %s" %(die)
		
		self.lines.append(line)
	
	def dump(self):
		return "\n".join(self.lines)

@dataclass
class Track:
	start: float
	duration: float
	number: int
	title: str = None
	album: str = None
	artist: str = None
	genre: str = None
	date: str = None
	
	def __str__(self):
		return "Track(#%02d: %s +%s \"%s\"" %(self.number, self.start, self.duration, self.title or "-")
	
	def set_name(self, fmt):
		name = fmt
		name = name.replace("%n", str(self.number).zfill(2))
		name = name.replace("%a", self.artist or "")
		name = name.replace("%t", self.title or "")
		
		self.final_file = name
		self.wav_file = name + ".wav"

# --------------------------------------------------

if __name__ == "__main__":
	main = over.app.Main("unmix", version.str, "GPL-3")
	main.add_option("artist", "Force an album-wide artist name.", str, count=1, abbr="a")
	main.add_option("album", "Force an album-wide album name.", str, count=1, abbr="l")
	main.add_option("date", "Force an album-wide date.", str, count=1, abbr="d")
	main.add_option("genre", "Force an album-wide genre.", str, count=1, abbr="g")
	main.add_option("path", "Path to the media file.", str, count=1, abbr="f")
	main.add_option("playlist", "Name of the file containing the playlist. See [<W>Playlist<.>] for the what fors.", str, count=1, abbr="p")
	main.add_option("encoder", "Encoding command with placeholders for individial parameters.", str, ["opusenc --title TITLE --artist ARTIST --album ALBUM --tracknumber TRACKNUMBER --genre GENRE --date DATE INFILE OUTFILE"], count=1, abbr="e")
	main.add_option("detect-silence", "Autodetect silence instead of parsing a playlist. Takes 2 arguments: the sound level in dB, and the minimum duration. E.g. <W>--<G>detect-silence<.> <m>-40<.> <m>2.25<.>.", floats, count=2, abbr="s")
	main.add_option("pad", "Pad each track with silence. Takes 2 arguments: how much padding to add to the beginning and end of each track. E.g. <W>--<G>pad<.> <m>0.25<.> <m>0.5<.>.", floats, count=2, abbr="x")
	main.add_option("output-playback", "Output playback commands (to verify timestamps) instead of the splitting & encoding scripts.", bool, [False], abbr="P")
	main.add_option("output-playlist", "Output just the playlist. This really only makes sense with <W>--<G>detect-silence<.>.", bool, [False], abbr="L")
	main.add_option("track-name", "Track name format.", str, ["%n %a - %t.opus"], abbr="t")
	main.add_option("verbose", "Print playlist lines as they are being parsed to stderr.", bool, [False], abbr="v")
	main.add_doc("Description", ["Takes a playlist and turns it into a shell script that splits a large mix into individual audio files."])
	main.add_doc("Playlist", ["Empty lines and those starting with # are ignored. Each line represents a track. The format is: START [DURATION] ARTIST TITLE. Fields are tab-separated."])
	main.setup()
	
	## Sanity checks
	##
	if not (main.cfg.path and os.path.isfile(main.cfg.path)):
		main.print("<R>%s<.> is not a readable file" %(main.cfg.path), main.print.tl.fail)
		main.exit(1)
	
	if not (bool(main.cfg.playlist) ^ bool(main.cfg.detect_silence)):
		main.print("nothing to do, you must specify either <W>--<G>detect-silence<.> or <W>--<G>playlist<.>", main.print.tl.fail)
		main.exit(1)
	
	## autodetect silence or parse playlist
	##
	tracks = []
	
	if main.cfg.detect_silence:
		lvl, dur = main.cfg.detect_silence
		gaps = detect_silence(main.cfg.path, dur, lvl)
		pos = 0.0
		n = 1
		
		for gap_start, gap_length in gaps:
			tracks.append(Track(
				start = pos,
				duration = gap_start - pos,
				number = n
			))
			
			pos = gap_start + gap_length
			n += 1
	
	elif main.cfg.playlist:
		previous_track = None
		
		with open(main.cfg.playlist) as f:
			n = 1
			
			for line in f:
				line = line.strip()
				
				if not line or line[0] == "#":
					continue
				
				cols = line.split("\t")
				
				if len(cols) == 3:
					start_s, artist, title = cols
					duration = None
				else:
					start_s, duration_s, artist, title = cols
					duration = sanitize_timestamp(duration_s)
				
				start = sanitize_timestamp(start_s)
				
				track = Track(
					start = start,
					duration = duration,
					number = n,
					artist = artist.strip(),
					title = title.strip()
				)
				
				tracks.append(track)
				
				if previous_track and previous_track.duration is None:
					previous_track.duration = track.start - previous_track.start
				
				previous_track = track
				n += 1
			
			# if unset, set the last track's duration to the end of file
			if tracks[-1].duration is None:
				tracks[-1].duration = get_duration(main.cfg.path) - tracks[-1].start
	
	## apply global overrides
	##
	if main.cfg.artist:
		for track in tracks:
			track.artist = main.cfg.artist
	
	if main.cfg.album:
		for track in tracks:
			track.album = main.cfg.album
	
	if main.cfg.genre:
		for track in tracks:
			track.genre = main.cfg.genre
	
	if main.cfg.date:
		for track in tracks:
			track.date = main.cfg.date
	
	if main.cfg.pad:
		pad_start, pad_end = main.cfg.pad
		
		for track in tracks:
			track.start = max(0, track.start - pad_start)
			track.duration += pad_start + pad_end
	
	s = Script(main.invocation)
	
	## output just the playlist
	if main.cfg.output_playlist:
		s.chapter("playlist")
		
		for track in tracks:
			s.line("\t".join((
				"%7.02f" %(track.start),
				"%7.02f" %(track.duration),
				track.artist or "",
				track.title or ""
			)))
	
	## output the split command, if enabled
	##
	else:
		if main.cfg.output_playback:
			s.chapter("playback (to verify timestamps)")
			
			for track in tracks:
				cmd = over.cmd.Command("ffplay", "-ss", "SEEK", "-t", "DURATION", "-autoexit", "INFILE")
				cmd.SEEK = "%.2f" %(track.start)
				cmd.DURATION = "%.2f" %(track.duration)
				cmd.INFILE = "file:" + main.cfg.path
				s.line(cmd)
		else:
			s.define_die()
			s.chapter("split into individual tracks")
			
			for track in tracks:
				cmd = over.cmd.Command("ffmpeg", "-ss", "SEEK", "-i", "INFILE", "-t", "DURATION", "-vn", "-sn", "-codec:a", "pcm_s16le", "OUTFILE")
				cmd.SEEK = "%.2f" %(track.start)
				cmd.DURATION = "%.2f" %(track.duration)
				cmd.INFILE = "file:" + main.cfg.path
				track.set_name(main.cfg.track_name)
				cmd.OUTFILE = track.wav_file
				s.line(cmd, die="splitting track %d" %(track.number))
	
	## output encode commands
	##
	if not (main.cfg.output_playback or main.cfg.output_playlist):
		s.chapter("encode and set metadata")
		
		for track in tracks:
			# this must contain: TITLE ARTIST ALBUM TRACKNUMBER GENRE DATE INFILE OUTFILE
			cmd = over.cmd.Command(*main.cfg.encoder.split())
			cmd.TITLE = track.title or ""
			cmd.ARTIST = track.artist or ""
			cmd.ALBUM = track.album or ""
			cmd.TRACKNUMBER = str(track.number).zfill(2)
			cmd.GENRE = track.genre or ""
			cmd.DATE = track.date or ""
			cmd.INFILE = track.wav_file
			cmd.OUTFILE = track.final_file
			
			s.line(cmd, die="encoding track %d" %(track.number))
	
	## remove temporary files
	##
		s.chapter("remove temporary files")
		
		for track in tracks:
			cmd = over.cmd.Command("rm", "-f", "INFILE")
			cmd.INFILE = track.wav_file
			s.line(cmd, die="removing track %d wav file" %(track.number))
	
	## output the script itself
	print(s.dump())