media-sucker/src/cd.py

189 lines
5.6 KiB
Python
Raw Normal View History

2022-08-23 22:06:44 -06:00
#! /usr/bin/python3
import subprocess
import time
import logging
import re
import os
import socket
import io
import gnudb
SECOND = 1
MINUTE = 60 * SECOND
HOUR = 60 * MINUTE
2022-08-25 10:17:25 -06:00
def scan(state, device):
2022-11-06 10:34:43 -07:00
logging.info("Scanning CD to get disc ID")
2022-08-23 22:06:44 -06:00
p = subprocess.run(
[
"cd-discid",
device,
],
encoding="utf-8",
capture_output=True,
)
2022-08-25 10:17:25 -06:00
discid = p.stdout.strip()
2022-08-25 10:17:25 -06:00
state["discid"] = discid
2022-08-25 10:17:25 -06:00
cddb_id = discid.split()[0]
2022-08-23 22:06:44 -06:00
2022-11-06 10:34:43 -07:00
logging.info("Looking disc up on CDDB")
2022-08-23 22:06:44 -06:00
email = os.environ.get("EMAIL") # You should really set this variable, tho
if not email:
user = "user"
try:
user = os.getlogin()
except OSError:
pass
email = "%s@%s" % (user, socket.gethostname())
db_server = gnudb.Server(email)
disc = db_server.bestguess(discid)
if disc:
# There was a hit! Hooray!
# We're expected to be automatic here,
# so just use the first one.
for k in ("title", "artist", "genre", "year", "tracks"):
2022-08-25 10:17:25 -06:00
state[k] = disc[k]
2022-08-23 22:06:44 -06:00
else:
num_tracks = int(discid.split()[1])
2022-08-25 10:17:25 -06:00
state["title"] = "Unknown CD - %s" % cddb_id
state["tracks"] = ["Track %02d" % (i+1) for i in range(num_tracks)]
2022-08-23 22:06:44 -06:00
2022-11-06 10:34:43 -07:00
logging.info("CD scan complete")
2022-08-25 10:17:25 -06:00
def copy(state, device, directory):
2022-08-23 22:06:44 -06:00
# cdparanoia reports completion in samples
# use discid duration to figure out total number of samples
2022-08-25 10:17:25 -06:00
duration = int(state["discid"].split()[-1]) * SECOND # disc duration in seconds
2022-08-23 22:06:44 -06:00
total_samples = duration * (75 / SECOND) * 1176 # 75 sectors per second, 1176 samples per sector
2022-08-25 10:17:25 -06:00
state["total_samples"] = total_samples
2022-08-23 22:06:44 -06:00
track_num = 1
2022-08-25 10:17:25 -06:00
for track_name in state["tracks"]:
2022-11-06 10:34:43 -07:00
logging.info("Ripping track %d of %d", track_num, len(state["tracks"]))
2022-08-23 22:06:44 -06:00
p = subprocess.Popen(
[
"cdparanoia",
"--stderr-progress",
"--force-cdrom-device", device,
"--batch",
str(track_num),
],
cwd = directory,
stderr = subprocess.PIPE,
encoding = "utf-8",
)
for line in p.stderr:
line = line.strip()
if line.startswith("##: -2"):
samples = int(line.split()[-1])
2022-08-25 10:17:25 -06:00
yield samples / total_samples
2022-08-23 22:06:44 -06:00
track_num += 1
2022-08-25 10:17:25 -06:00
def encode(state, directory):
2022-08-23 22:06:44 -06:00
track_num = 1
2022-08-25 10:17:25 -06:00
total_tracks = len(state["tracks"])
2022-08-25 10:17:25 -06:00
durations = [int(d) for d in state["discid"].split()[2:-1]]
2022-08-25 10:17:25 -06:00
total_duration = sum(durations)
encoded_duration = 0
2022-08-25 10:17:25 -06:00
tag_script = io.StringIO()
tag_script.write("#! /bin/sh\n")
tag_script.write("\n")
tag_script.write("ALBUM=%s\n" % state["title"])
tag_script.write("ARTIST=%s\n" % state.get("artist", ""))
tag_script.write("GENRE=%s\n" % state.get("genre", ""))
tag_script.write("YEAR=%s\n" % state.get("year", ""))
tag_script.write("\n")
2022-08-25 10:17:25 -06:00
for track_name in state["tracks"]:
2022-11-06 10:34:43 -07:00
logging.info("Encoding track %d (%s)" % (track_num, track_name))
2022-08-25 10:17:25 -06:00
duration = durations[track_num-1]
2022-08-23 22:06:44 -06:00
argv = [
"lame",
2022-08-25 10:17:25 -06:00
"--brief",
"--nohist",
"--disptime", "1",
2022-08-23 22:06:44 -06:00
"--preset", "standard",
2022-08-25 10:17:25 -06:00
"--tl", state["title"],
2022-08-25 10:17:25 -06:00
"--tn", "%d/%d" % (track_num, total_tracks),
2022-08-23 22:06:44 -06:00
]
2022-08-25 10:17:25 -06:00
tag_script.write("id3v2")
tag_script.write(" --album \"$ALBUM\"")
tag_script.write(" --artist \"$ARTIST\"")
tag_script.write(" --genre \"$GENRE\"")
tag_script.write(" --year \"$YEAR\"")
2022-08-25 10:17:25 -06:00
if state.get("artist"):
argv.extend(["--ta", state["artist"]])
if state.get("genre"):
argv.extend(["--tg", state["genre"]])
if state.get("year"):
argv.extend(["--ty", state["year"]])
2022-08-23 22:06:44 -06:00
if track_name:
2022-08-25 10:17:25 -06:00
argv.extend(["--tt", track_name])
2022-08-25 10:17:25 -06:00
tag_script.write(" --song \"%s\"" % track_name)
2022-08-25 10:17:25 -06:00
outfn = "%02d - %s.mp3" % (track_num, track_name)
2022-08-23 22:06:44 -06:00
else:
2022-08-25 10:17:25 -06:00
outfn = "%02d.mp3" % track_num
2022-08-23 22:06:44 -06:00
argv.append("track%02d.cdda.wav" % track_num)
argv.append(outfn)
2022-08-25 10:17:25 -06:00
tag_script.write("\\\n ")
tag_script.write(" --track %d/%d" % (track_num, total_tracks))
tag_script.write(" \"%s\"\n" % outfn)
2022-08-23 22:06:44 -06:00
p = subprocess.Popen(
argv,
cwd = directory,
2022-08-25 10:17:25 -06:00
stderr = subprocess.PIPE,
2022-08-23 22:06:44 -06:00
encoding = "utf-8",
)
2022-08-25 10:17:25 -06:00
for line in p.stderr:
line = line.strip()
if "%)" in line:
p = line.split("(")[1]
p = p.split("%")[0]
pct = int(p) / 100
2022-08-25 10:17:25 -06:00
yield (encoded_duration + (duration * pct)) / total_duration
2022-08-25 10:17:25 -06:00
2022-08-25 10:17:25 -06:00
encoded_duration += duration
2022-08-23 22:06:44 -06:00
track_num += 1
2022-08-25 10:17:25 -06:00
with open(os.path.join(directory, "tag.sh"), "w") as f:
f.write(tag_script.getvalue())
2022-08-25 10:17:25 -06:00
2022-08-25 10:17:25 -06:00
def clean(state, directory):
for fn in os.listdir(directory):
if fn.endswith(".wav"):
os.remove(os.path.join(directory, fn))
2022-08-25 10:17:25 -06:00
2022-08-23 22:06:44 -06:00
if __name__ == "__main__":
import pprint
2022-08-25 10:17:25 -06:00
import sys
import json
2022-08-23 22:06:44 -06:00
2022-08-25 10:17:25 -06:00
logging.basicConfig(level=logging.DEBUG)
state = {}
scan(state, "/dev/sr0")
pprint.pprint(state)
directory = os.path.join(".", state["title"])
os.makedirs(directory, exist_ok=True)
with open(os.path.join(directory, "state.json"), "w") as f:
json.dump(f, state)
for pct in copy(state, "/dev/sr0", directory):
sys.stdout.write("Copying: %3d%%\r" % (pct*100))
pprint.pprint(state)
for pct in encode(state, directory):
sys.stdout.write("Encoding: %3d%%\r" % (pct*100))
pprint.pprint(state)
2022-08-23 22:06:44 -06:00
# vi: sw=4 ts=4 et ai