Last active
March 14, 2020 23:05
-
-
Save sandhose/b6903fe3bca799063300cce28832dfdc to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# usage: python3 ableton-to-cues.py $ALS_FILENAME $REKORDBOX_XML_FILENAME | |
# converts ableton warp markers to rekordbox cue points. | |
# use '--reverse' to convert rekordbox cues to ableton warp markers instead | |
# writes output to output.xml if converting ableton to rekordbox | |
# in reverse mode, it writes to output.als | |
import argparse | |
import datetime | |
import gzip | |
import logging | |
import sys | |
from tkinter import Tk, Button | |
from tkinter.filedialog import askopenfile, asksaveasfile | |
from tkinter.scrolledtext import ScrolledText | |
from urllib.parse import unquote | |
import xml.etree.ElementTree as ET | |
logger = logging.getLogger(__name__) | |
def run(): | |
stderrHandler = logging.StreamHandler() | |
logger.addHandler(stderrHandler) | |
logger.setLevel(logging.INFO) | |
if len(sys.argv) > 1: | |
parser = argparse.ArgumentParser( | |
description='convert between cues and warp markers') | |
parser.add_argument( | |
'als_file', type=str, help='path to ALS file') | |
parser.add_argument( | |
'rekordbox_file', type=str, help='path to Rekordbox xml file') | |
parser.add_argument( | |
'--reverse', type=bool, default=False, | |
help='setto true to convert Rekordbox to Ableton') | |
args = parser.parse_args() | |
with open(args.als_file, mode='rb') as als_file, open(args.rekordbox_file, mode='rb') as rekordbox_file: | |
if not args.reverse: | |
with open('output.xml', 'wb') as outfile: | |
ableton_to_rekordbox(als_file, rekordbox_file, outfile) | |
else: | |
with open('output.als', 'wb') as outfile: | |
rekordbox_to_ableton(als_file, rekordbox_file, outfile) | |
logger.info('Finished writing ' + outfile) | |
else: | |
app = App(None) | |
app.title('Ableton to cues') | |
guiHandler = MyHandlerText(app.logbox) | |
logger.addHandler(guiHandler) | |
app.mainloop() | |
def normalize_time(time): | |
if time == 0: | |
return "0.001" | |
else: | |
return "{0:.3f}".format(time) | |
def get_memcue(time): | |
child = ET.Element('POSITION_MARK') | |
child.set('Name', '') | |
child.set('Type', '0') | |
child.set('Num', '-1') | |
child.set('Start', normalize_time(time)) | |
return child | |
def get_hotcue(time, num): | |
child = ET.Element('POSITION_MARK') | |
child.set('Name', '') | |
child.set('Type', '0') | |
child.set('Red', '40') | |
child.set('Green', '226') | |
child.set('Blue', '20') | |
child.set('Num', str(num)) | |
child.set('Start', normalize_time(time)) | |
return child | |
def get_warp_marker(time, num, bpm): | |
child = ET.Element('WarpMarker') | |
child.set('Id', str(num)) | |
child.set('SecTime', time) | |
beat_time = float(time) * bpm / 60 | |
child.set('BeatTime', str(beat_time)) | |
return child | |
def get_rekordbox_filename(rekordbox_track): | |
parts = rekordbox_track.get('Location').split('/') | |
return unquote(parts[-1]) | |
def get_ableton_filename(track): | |
return track.find('.//FileRef').find('./Name').get('Value') | |
def ableton_to_rekordbox(als_file, rekordbox_file, output): | |
logger.info('Converting Ableton warp markers to Rekordbox cues.') | |
tree = ET.parse(gzip.GzipFile(fileobj=als_file)) | |
tracks = tree.getroot().findall('.//AudioClip') | |
rekordbox_tree = ET.parse(rekordbox_file) | |
rekordbox_tracks = rekordbox_tree.getroot().findall('./COLLECTION/TRACK') | |
for track in tracks: | |
filename = get_ableton_filename(track) | |
warp_markers = track.findall('.//WarpMarkers/WarpMarker') | |
# Find the corresponding track in rekordbox | |
for rekordbox_track in rekordbox_tracks: | |
if (get_rekordbox_filename(rekordbox_track) == filename): | |
logger.info('processing ' + filename) | |
# clear all existing cues | |
for element in rekordbox_track.findall('./POSITION_MARK'): | |
rekordbox_track.remove(element) | |
# create a hotcue and mem cue for each warp marker | |
num = 0 | |
times = [float(marker.get('SecTime')) for marker in warp_markers] | |
times.sort() | |
# ignore last item cuz it gets duplicated for some reason | |
del times[-1] | |
for time in times: | |
hotcue = get_hotcue(time, num) | |
memcue = get_memcue(time) | |
num = num + 1 | |
rekordbox_track.append(hotcue) | |
rekordbox_track.append(memcue) | |
rekordbox_tree.write(output, encoding='UTF-8', xml_declaration=True) | |
def rekordbox_to_ableton(als_file, rekordbox_file, output): | |
logger.info('Converting Rekordbox cues to Ableton warp markers.') | |
tree = ET.parse(gzip.GzipFile(fileobj=als_file)) | |
tracks = tree.getroot().findall('.//AudioClip') | |
rekordbox_tree = ET.parse(rekordbox_file) | |
rekordbox_tracks = rekordbox_tree.getroot().findall('./COLLECTION/TRACK') | |
for rekordbox_track in rekordbox_tracks: | |
filename = get_rekordbox_filename(rekordbox_track) | |
# find the corresponding ableton track | |
for track in tracks: | |
if (get_ableton_filename(track) == filename): | |
logger.info('processing ' + filename) | |
# clear all existing warp markers | |
# create new <WarpMarkers> group | |
child = ET.Element('WarpMarkers') | |
# ableton warp marker IDs seem to start at 2?? | |
num = 2 | |
# get rekordbox bpm | |
bpm = float(rekordbox_track.find('./TEMPO').get('Bpm')) | |
time = None | |
for element in rekordbox_track.findall('./POSITION_MARK'): | |
time = element.get('Start') | |
marker = get_warp_marker(time, num, bpm) | |
num = num + 1 | |
child.append(marker) | |
# append the last marker a 2nd time; for some reason this seems | |
# needed otherwise it doesn't show up in ableton. | |
if time: | |
marker = get_warp_marker(str(float(time) + 0.1), num, bpm) | |
child.append(marker) | |
# attach the new WarpMarkers group | |
if len(child.getchildren()) > 0: | |
for markers in track.findall('./WarpMarkers'): | |
track.remove(markers) | |
track.append(child) | |
tree.write(output, encoding='UTF-8', xml_declaration=True) | |
class App(Tk): | |
def __init__(self, parent): | |
super().__init__(parent) | |
self.parent = parent | |
self.grid() | |
self.to_ableton = Button(self, text="Rekordbox to Ableton", command=self.to_ableton) | |
self.to_ableton.grid(column=0, row=0, ipadx=5, padx=5, pady=5) | |
self.to_rekordbox = Button(self, text="Ableton to Rekordbox", command=self.to_rekordbox) | |
self.to_rekordbox.grid(column=1, row=0, ipadx=5, padx=5, pady=5) | |
self.logbox = ScrolledText(self, state="disabled") | |
self.logbox.grid(column=0, row=1, columnspan=2) | |
self.flow = None | |
def to_ableton(self): | |
if self.flow is None: | |
self.flow = 'ableton' | |
self.after(0, self.ask_ableton) | |
else: | |
logger.warning('A flow is already running') | |
def to_rekordbox(self): | |
if self.flow is None: | |
self.flow = 'rekordbox' | |
self.after(0, self.ask_ableton) | |
else: | |
logger.warning('A flow is already running') | |
def abort(self): | |
self.flow = None | |
logger.warning('Aborting') | |
def ask_ableton(self): | |
logger.info("Asking the Ableton file") | |
self.als_file = askopenfile(mode='rb', title="Ableton input file", filetypes=[("Ableton files", ".als")], parent=self) | |
if self.als_file is None: | |
self.abort() | |
else: | |
self.after(0, self.ask_rekordbox) | |
def ask_rekordbox(self): | |
logger.info("Asking the Rekordbox file") | |
self.rekordbox_file = askopenfile(mode='rb', title="Rekordbox input file", filetypes=[("Rekordbox file", ".xml")], parent=self) | |
if self.rekordbox_file is None: | |
self.abort() | |
else: | |
self.after(0, self.ask_output) | |
def ask_output(self): | |
logger.info("Asking the output file") | |
if self.flow == 'ableton': | |
self.output_file = asksaveasfile(mode='wb', title="Ableton output file", defaultextension="als", parent=self) | |
else: | |
self.output_file = asksaveasfile(mode='wb', title="Rekordbox output file", defaultextension="xml", parent=self) | |
if self.output_file is None: | |
self.abort() | |
else: | |
self.after(0, self.run_flow) | |
def run_flow(self): | |
if self.flow == 'ableton': | |
rekordbox_to_ableton(self.als_file, self.rekordbox_file, self.output_file) | |
elif self.flow == 'rekordbox': | |
ableton_to_rekordbox(self.als_file, self.rekordbox_file, self.output_file) | |
else: | |
logger.error("No flow running") | |
logger.info("Processing done") | |
self.flow = None | |
class MyHandlerText(logging.StreamHandler): | |
def __init__(self, textctrl): | |
logging.StreamHandler.__init__(self) # initialize parent | |
self.setFormatter(logging.Formatter('%(asctime)s %(levelname)-8s\n %(message)s')) | |
self.textctrl = textctrl | |
def emit(self, record): | |
msg = self.format(record) | |
self.textctrl.config(state="normal") | |
self.textctrl.insert("end", msg + "\n") | |
self.flush() | |
self.textctrl.config(state="disabled") | |
if __name__ == '__main__': | |
run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment