txtp_maker.py

#!/usr/bin/env python3

# ################################################### #
# TXTP MAKER by bnnm
# v20190122: initial release
# ################################################### #

from __future__ import division
import subprocess
import zlib
import os.path
import sys


if (len(sys.argv) <= 1):
    print("Usage: {} (filename) [options]".format(os.path.basename(sys.argv[0])) + "\n"
          "\n"
          "Creates a (filename)_(subsong).txtp for every subsong in (filename).\n"
          "Options:\n"
          " -c (name): set path to CLI (default: test.exe)\n"
          " -n (name): use (name)_(subsong).txtp format\n"
          " -z N: zero-fill subsong number (default: auto fill up to total subsongs)\n"
          " -d (dir): prepend dir to name (if the file will reside in a subdir)\n"
          " -m: create mini-txtp\n"
          " -o: overwrite existing .txtp (beware when using with internal names)\n"
          " -in: name TXTP using the subsong's internal name if found\n"
          " -ie: remove internal name's extension\n"
          " -ii: add subsong number when using internal name\n"
          " -l N: create multiple TXTP per subsong layers, every N channels\n"
          " -fd: filter duplicates (slower)\n"
          " -fcm N: filter min channels\n"
          " -fcM N: filter max channels\n"
          " -frm N: filter min sample rate\n"
          " -frM N: filter max sample rate\n"
          " -fsm N.N: filter min seconds\n"
          " -fsM N.N: filter max seconds\n"
          )
    exit()

# ####################################################

def make_cmd(cfg, fname_in, fname_out, target_subsong):
    if (cfg.test_dupes):
        cmd = "{} -s {} -i -o {} {}".format(cfg.cli, target_subsong, fname_out, fname_in)
    else:
        cmd = "{} -s {} -m -i -o {} {}".format(cfg.cli, target_subsong, fname_out, fname_in)
    return cmd


class ConfigParser(object):
    cli = "test.exe"

    base_name = ''
    zero_fill = -1
    subdir = ''
    mini_txtp = False
    overwrite = False
    layers = 0

    use_internal_name = False
    use_internal_ext = False
    use_internal_index = False

    test_dupes = False
    min_channels = 2
    max_channels = 0
    min_sample_rate = 0
    max_sample_rate = 0
    min_seconds = 0.0
    max_seconds = 0.0

    argv_len = 0
    index = 0


    def read_bool(self, command, default):
        if self.index > self.argv_len - 1:
            return default
        if self.argv[self.index] == command:
            val = True
            self.index += 1
            return val
        return default

    def read_value(self, command, default):
        if self.index > self.argv_len - 2:
            return default
        if self.argv[self.index] == command:
            val = self.argv[self.index+1]
            self.index += 2
            return val
        return default

    def read_string(self, command, default):
        return str(self.read_value(command, default))

    def read_int(self, command, default):
        return int(self.read_value(command, default))

    def read_float(self, command, default):
        return float(self.read_value(command, default))

    #todo improve this poop
    def __init__(self, argv):

        self.index = 2 #after file and
        self.argv = argv
        self.argv_len = len(argv)

        prev_index = self.index
        while self.index < len(self.argv):
            self.cli = self.read_string('-c', self.cli)
            self.base_name = self.read_string('-n', self.base_name)
            self.zero_fill = self.read_int('-z', self.zero_fill)
            self.subdir = self.read_string('-d', self.subdir)

            self.test_dupes = self.read_bool('-fd', self.test_dupes)
            self.min_channels = self.read_int('-fcm', self.min_channels)
            self.max_channels = self.read_int('-fcM', self.max_channels)
            self.min_sample_rate = self.read_int('-frm', self.min_sample_rate)
            self.max_sample_rate = self.read_int('-frM', self.max_sample_rate)
            self.min_seconds = self.read_float('-fsm', self.min_seconds)
            self.max_seconds = self.read_float('-fsM', self.max_seconds)

            self.mini_txtp = self.read_bool('-m', self.mini_txtp)
            self.overwrite = self.read_bool('-o', self.overwrite)
            self.layers = self.read_int('-l', self.layers)

            self.use_internal_name = self.read_bool('-in', self.use_internal_name)
            self.use_internal_ext = self.read_bool('-ie', self.use_internal_ext)
            self.use_internal_index = self.read_bool('-ii', self.use_internal_index)

            if prev_index == self.index:
                self.index += 1
            prev_index = self.index

        if (self.subdir != '') and not (self.subdir.endswith('/') or self.subdir.endswith('\\')):
            self.subdir += '/'

    def __str__(self):
        return str(self.__dict__)


class Cr32Helper(object):
    crc32_map = {}
    dupe = False


    def get_crc32(self, fname):
        buf_size = 0x8000
        with open(fname, 'rb') as file:
            buf = file.read(buf_size)
            crc32 = 0
            while len(buf) > 0:
                crc32 = zlib.crc32(buf, crc32)
                buf = file.read(buf_size)
        return crc32 & 0xFFFFFFFF

    def update(self, fname):
        self.dupe = False
        if cfg.test_dupes == 0:
            return
        if not os.path.exists(fname):
            return

        crc32_str = format(self.get_crc32(fname),'08x')
        if (crc32_str in self.crc32_map):
            self.dupe = True
            return
        self.crc32_map[crc32_str] = True

        os.remove(fname)
        return

    def is_dupe(self):
        return self.dupe

    def __init__(self, cfg):
        self.cfg = cfg


class TxtpMaker(object):
    channels = 0
    sample_rate = 0
    num_samples = 0
    stream_count = 0
    stream_index = 0
    stream_name = ''
    stream_seconds = 0

    def get_string(self, str):
        find_pos = self.output.find(str)
        if (find_pos == -1):
            return ''
        cut_pos = find_pos + len(str)
        str_cut = self.output[cut_pos:]
        return str_cut.split()[0]

    def get_value(self, str):
        res = self.get_string(str)
        if (res == ''):
           return 0;
        return int(res)

    def is_ignorable(self):
        if (self.channels < cfg.min_channels):
            return True;
        if (cfg.max_channels > 0 and self.channels > cfg.max_channels):
            return True;
        if (self.sample_rate < cfg.min_sample_rate):
            return True;
        if (cfg.max_sample_rate > 0 and self.sample_rate > cfg.max_sample_rate):
            return True;
        if (self.stream_seconds < cfg.min_seconds):
            return True;
        if (cfg.max_seconds > 0 and self.stream_seconds > cfg.max_seconds):
            return True;
        return False

    def get_stream_mask(self, layer):
        mask = '#c'

        loops = cfg.layers
        if layer + cfg.layers > self.channels:
            loops = self.channels - cfg.layers
        for ch in range(0,loops):
            mask += str(layer+ch) + ','

        mask = mask[:-1]
        return mask

    def get_stream_name(self):
        if not cfg.use_internal_name:
            return ''
        txt = self.stream_name

        # remove paths #todo maybe config/replace?
        pos = txt.rfind("\\")
        if (pos != -1):
            txt = txt[pos+1:]
        pos = txt.rfind("/")
        if (pos != -1):
            txt = txt[pos+1:]
        # remove bad chars
        txt = txt.replace("%", "_")
        txt = txt.replace("*", "_")
        txt = txt.replace("?", "_")
        txt = txt.replace(":", "_")
        txt = txt.replace("\"", "_")
        txt = txt.replace("|", "_")
        txt = txt.replace("<", "_")
        txt = txt.replace(">", "_")

        if not cfg.use_internal_ext:
            pos = txt.rfind(".")
            if (pos != -1):
                txt = txt[:pos]
        return txt

    def write(self, outname, line):
        outname += '.txtp'
        if not cfg.overwrite and os.path.exists(outname):
            raise ValueError('TXTP exists in path: ' + outname)
        ftxtp = open(outname,"w+")
        if line != '':
            ftxtp.write(line)
        ftxtp.close()
        print("created: " + outname)
        return

    def make(self, fname):
        if self.is_ignorable():
            return

        index = str(self.stream_index)
        if cfg.zero_fill < 0:
            index = index.zfill(len(str(self.stream_count)))
        else:
            index = index.zfill(cfg.zero_fill)

        if cfg.mini_txtp:
            outname = "{}#{}".format(fname, index)

            if cfg.layers > 0 and cfg.layers < self.channels:
                for layer in range(0, self.channels, cfg.layers):
                    mask = self.get_stream_mask(layer)
                    self.write(outname + mask, '')
            else:
                self.write(outname, '')

        else:
            stream_name = self.get_stream_name()
            if stream_name != '':
                outname = stream_name
                if cfg.use_internal_index:
                    outname += "_{}".format(index)
            else:
                if cfg.base_name != '':
                    txt = cfg.base_name
                else:
                    txt = fname
                    pos = txt.rfind(".")
                    if (pos != -1):
                        txt = txt[:pos]
                outname = "{}_{}".format(txt, index)

            line = ''
            if cfg.subdir != '':
                line += cfg.subdir
            line += "{}#{}".format(fname, self.stream_index)

            if cfg.layers > 0 and cfg.layers < self.channels:
                done = 0
                for layer in range(0, self.channels, cfg.layers):
                    sub = chr(ord('a') + done)
                    done += 1
                    mask = self.get_stream_mask(layer)
                    self.write(outname + sub, line + mask)
            else:
                self.write(outname, line)

    def has_more_subsongs(self, target_subsong):
        return target_subsong < self.stream_count

    def __init__(self, cfg, output_b):
        self.cfg = cfg

        self.output = str(output_b).replace("\\r","").replace("\\n","\n")
        self.channels = self.get_value("channels: ")
        self.sample_rate = self.get_value("sample rate: ")
        self.num_samples = self.get_value("stream total samples: ")
        self.stream_count = self.get_value("stream count: ")
        self.stream_index = self.get_value("stream index: ")
        self.stream_name = self.get_string("stream name: ")

        if self.channels == 0:
            raise ValueError('Incorrect command result')

        self.stream_seconds = self.num_samples / self.sample_rate

    def __str__(self):
        return str(self.__dict__)

# ####################################################

cfg = ConfigParser(sys.argv)

fname_in = sys.argv[1]
fname_out = fname_in + ".wav"
target_subsong = 1
crc32 = Cr32Helper(cfg)

while 1:
    cmd = make_cmd(cfg, fname_in, fname_out, target_subsong)
    output_b = subprocess.check_output(cmd, shell=True)

    maker = TxtpMaker(cfg, output_b)

    crc32.update(fname_out)
    if not crc32.is_dupe():
        maker.make(fname_in)

    if not maker.has_more_subsongs(target_subsong):
        break
    target_subsong += 1

print("done!")