morphagene_batch.py

# -*- coding: utf-8 -*-
"""
Created on Sun Nov 25 21:15:52 2018

@author: ferri

updated version of mor.py with wavio functions

Reads 8, 16 bit wav files, mono or stereo.
    
Allows for the compilation of multiple wav samples in one folder,
    either mono or stereo of any bit depth (rate ideally 48000Hz to avoid
    change in pitch), to be written as one 32-bit float 48000Hz wav, directly
    compatible with the Make Noise Morphagene.
     
See the Morphagene manual for naming conventions of output files:    
    http://www.makenoisemusic.com/content/manuals/morphagene-manual.pdf
"""
# see http://stackoverflow.com/questions/15576798/create-32bit-float-wav-file-in-python
# see... http://blog.theroyweb.com/extracting-wav-file-header-information-using-a-python-script
# marker code from Joseph Basquin [https://gist.github.com/josephernest/3f22c5ed5dabf1815f16efa8fa53d476]
# modified scipy wav read code to handle uint8 data

from __future__ import division, print_function, absolute_import
import os
import struct
import warnings
import collections
from scipy import interpolate
import numpy as np

class WavFileWarning(UserWarning):
    pass
    
_ieee = False
    
# assumes file pointer is immediately
#  after the 'fmt ' id
def _read_fmt_chunk(fid):
    res = struct.unpack('<ihHIIHH',fid.read(20))
    size, comp, noc, rate, sbytes, ba, bits = res
    if (comp != 1 or size > 16):
        if (comp == 3):
          global _ieee
          _ieee = True
          #warnings.warn("IEEE format not supported", WavFileWarning)        
        else: 
          warnings.warn("Unfamiliar format bytes", WavFileWarning)
        if (size>16):
            fid.read(size-16)
    return size, comp, noc, rate, sbytes, ba, bits

# assumes file pointer is immediately
#   after the 'data' id
def _read_data_chunk(fid, noc, bits, normalized=False):
    size = struct.unpack('<i',fid.read(4))[0]

    if bits == 8 or bits == 24:
        dtype = 'u1'
        bytes = 1
    else:
        bytes = bits//8
        dtype = '<i%d' % bytes
        
    if bits == 32 and _ieee:
       dtype = 'float32'
    
    data = np.fromfile(fid, dtype=dtype, count=size//bytes)
    
    if bits == 24:
        # handle 24 bit file by using samplewidth=3, no native 24-bit type
        a = np.empty((len(data) // 3, 4), dtype='u1')
        a[:, :3] = data.reshape((-1, 3))
        a[:, 3:] = (a[:, 3 - 1:3] >> 7) * 255
        data = a.view('<i4').reshape(a.shape[:-1])
    
    if noc > 1: 
        # handle stereo
        data = data.reshape(-1,noc)
        
    if bool(size & 1):     
      # if odd number of bytes, move 1 byte further (data chunk is word-aligned)
      fid.seek(1,1)    

    if normalized:
        if bits == 16 or bits == 24 or bits == 32: 
            normfactor = 2 ** (bits-1)
            data = np.float32(data) * 1.0 / normfactor
        elif bits == 8:
            if isinstance(data[0], (int, np.uint8)):
                # handle uint8 data by shifting to center at 0
                normfactor = 2 ** (bits-1)
                data = (np.float32(data) * 1.0 / normfactor) -\
                                ((normfactor)/(normfactor-1))
    return data

def _skip_unknown_chunk(fid):
    data = fid.read(4)
    size = struct.unpack('<i', data)[0]
    if bool(size & 1):     
      # if odd number of bytes, move 1 byte further (data chunk is word-aligned)
      size += 1 
    fid.seek(size, 1)

def _read_riff_chunk(fid):
    str1 = fid.read(4)
    if str1 != b'RIFF':
        raise ValueError("Not a WAV file.")
    fsize = struct.unpack('<I', fid.read(4))[0] + 8
    str2 = fid.read(4)
    if (str2 != b'WAVE'):
        raise ValueError("Not a WAV file.")
    return fsize

def read(file, readmarkers=False, readmarkerlabels=False, 
         readmarkerslist=False, readloops=False, readpitch=False, 
         normalized=False, forcestereo=False):
    """
    Return the sample rate (in samples/sec) and data from a WAV file
    Parameters
    ----------
    file : file
        Input wav file.
    Returns
    -------
    rate : int
        Sample rate of wav file
    data : np array
        Data read from wav file
    Notes
    -----
    * The file can be an open file or a filename.
    * The returned sample rate is a Python integer
    * The data is returned as a np array with a
      data-type determined from the file.
    """
    if hasattr(file,'read'):
        fid = file
    else:
        fid = open(file, 'rb')

    fsize = _read_riff_chunk(fid)
    noc = 1
    bits = 8
    #_cue = []
    #_cuelabels = []
    _markersdict = collections.defaultdict(lambda: {'position': -1, 'label': ''})
    loops = []
    pitch = 0.0
    while (fid.tell() < fsize):
        # read the next chunk
        chunk_id = fid.read(4)
        if chunk_id == b'fmt ':
            size, comp, noc, rate, sbytes, ba, bits = _read_fmt_chunk(fid)
        elif chunk_id == b'data':
            data = _read_data_chunk(fid, noc, bits, normalized)
        elif chunk_id == b'cue ':
            str1 = fid.read(8)
            size, numcue = struct.unpack('<ii',str1)
            for c in range(numcue):
                str1 = fid.read(24)
                id, position, datachunkid, chunkstart, blockstart, sampleoffset = struct.unpack('<iiiiii', str1)
                #_cue.append(position)
                # needed to match labels and markers
                _markersdict[id]['position'] = position                    

        elif chunk_id == b'LIST':
            str1 = fid.read(8)
            size, type = struct.unpack('<ii', str1)
        elif chunk_id in [b'ICRD', b'IENG', b'ISFT', b'ISTJ']:   
             # see http://www.pjb.com.au/midi/sfspec21.html#i5
            _skip_unknown_chunk(fid)
        elif chunk_id == b'labl':
            str1 = fid.read(8)
            size, id = struct.unpack('<ii',str1)
            # the size should be even, see WAV specfication, e.g. 16=>16, 23=>24
            size = size + (size % 2)      
            # remove the trailing null characters                        
            label = fid.read(size-4).rstrip('\x00')               
            #_cuelabels.append(label)
            # needed to match labels and markers
            _markersdict[id]['label'] = label                          

        elif chunk_id == b'smpl':
            str1 = fid.read(40)
            size, manuf, prod, sampleperiod, midiunitynote,\
            midipitchfraction, smptefmt, smpteoffs, numsampleloops, \
                samplerdata = struct.unpack('<iiiiiIiiii', str1)
            cents = midipitchfraction * 1./(2**32-1)
            pitch = 440. * 2 ** ((midiunitynote + cents - 69.)/12)
            for i in range(numsampleloops):
                str1 = fid.read(24)
                cuepointid, type, start, end, \
                fraction, playcount = struct.unpack('<iiiiii', str1) 
                loops.append([start, end])
        else:
            warnings.warn("Chunk " + chunk_id + " skipped", WavFileWarning)
            _skip_unknown_chunk(fid)
    fid.close()

    if data.ndim == 1 and forcestereo:
        data = np.column_stack((data, data))

    _markerslist = sorted([_markersdict[l] for l in _markersdict], key=lambda k: k['position'])  # sort by position
    _cue = [m['position'] for m in _markerslist]
    _cuelabels = [m['label'] for m in _markerslist]
    
    return (rate, data, bits, ) \
        + ((_cue,) if readmarkers else ()) \
        + ((_cuelabels,) if readmarkerlabels else ()) \
        + ((_markerslist,) if readmarkerslist else ()) \
        + ((loops,) if readloops else ()) \
        + ((pitch,) if readpitch else ())
    
def get_list_of_files(dir_name,file_type):
    '''
    See:
    https://thispointer.com/python-how-to-get-list-of-files-in-directory-and-sub-directories/
    Create a list of files within a directory and it's subdirectories
    Modified to filter so that full filname has to match fileType
    '''
    list_of_files = os.listdir(dir_name)
    all_files = list()
    # Iterate over all the entries
    for entry in list_of_files:
        # Create full path
        full_path = os.path.join(dir_name, entry)
        # If entry is a directory then get the list of files in this directory 
        if os.path.isdir(full_path):
            all_files = all_files + get_list_of_files(full_path,file_type)
        else:
            if file_type in full_path:
                all_files.append(full_path)
    return all_files

def change_samplerate_interp(old_audio,old_rate,new_rate):
    '''
    Change sample rate to new sample rate by simple interpolation.
    If old_rate > new_rate, there may be aliasing / data loss,
    Input should be in column format, as the interpolation will be completed
        on each channel this way.
    Modified from:
    https://stackoverflow.com/questions/33682490/
        how-to-read-a-wav-file-using-scipy-at-a-different-sampling-rate
    '''    
    if old_rate != new_rate:
        # duration of audio
        duration = old_audio.shape[0] / old_rate
        
        # length of old and new audio
        time_old  = np.linspace(0, duration, old_audio.shape[0])
        time_new  = np.linspace(0, duration, int(old_audio.shape[0] * new_rate / old_rate))
        
        # fit old_audio into new_audio length by interpolation
        interpolator = interpolate.interp1d(time_old, old_audio.T)
        new_audio = interpolator(time_new).T
        return new_audio
    else:
        return old_audio # conversion not needed

def float32_wav_file(file_name, sample_array, sample_rate, 
                     markers=None, verbose=False):
    '''
    Write 32-bit WAV
    '''
    (M,N)=sample_array.shape
    #print "len sample_array=(%d,%d)" % (M,N)
    byte_count = M * N * 4 # (len(sample_array)) * 4  # 32-bit floats
    wav_file = ""
    # write the header
    wav_file += struct.pack('<ccccIccccccccIHHIIHH',
        'R', 'I', 'F', 'F',
        byte_count + 0x2c - 8,  # header size
        'W', 'A', 'V', 'E', 'f', 'm', 't', ' ',
        0x10,  # size of 'fmt ' header
        3,  # format 3 = floating-point PCM
        M,  # channels
        sample_rate,  # samples / second
        sample_rate * 4,  # bytes / second
        4,  # block alignment
        32)  # bits / sample
    wav_file += struct.pack('<ccccI',
        'd', 'a', 't', 'a', byte_count)
    if verbose:
        print("packing...")
    for j in range(0,N):
        for k in range(0,M):
            wav_file += struct.pack("<f", sample_array[k,j])
    if verbose:
        print("saving audio...")
    fid=open(file_name,'wb')
    for value in wav_file:
        fid.write(value)
    if markers:    # != None and != []
        if verbose:
            print("saving cue markers...")
        if isinstance(markers[0], dict):       
            # then we have [{'position': 100, 'label': 'marker1'}, ...]
            labels = [m['label'] for m in markers]
            markers = [m['position'] for m in markers]
        else:
            labels = ['' for m in markers]
        fid.write(b'cue ')
        size = 4 + len(markers) * 24
        fid.write(struct.pack('<ii', size, len(markers)))
        for i, c in enumerate(markers):
            # 1635017060 is struct.unpack('<i',b'data')
            s = struct.pack('<iiiiii', i + 1, c, 1635017060, 0, 0, c)           
            fid.write(s)
        lbls = ''
        for i, lbl in enumerate(labels):
            lbls += b'labl'
            label = lbl + ('\x00' if len(lbl) % 2 == 1 else '\x00\x00')
            size = len(lbl) + 1 + 4          # because \x00
            lbls += struct.pack('<ii', size, i + 1)
            lbls += label
        fid.write(b'LIST')
        size = len(lbls) + 4
        fid.write(struct.pack('<i', size))   
        # https://web.archive.org/web/20141226210234/http://www.sonicspot.com/guide/wavefiles.html#list                      
        fid.write(b'adtl') 
        fid.write(lbls) 
    fid.close()
    return wav_file

if __name__ == "__main__":
    # folder containing samples of type [extension]
    folder = 'sh3'
    extension = '.wav'
    # morphagene settings
    morph_samplerate = 48000 # required sample rate for Morphagene
    morph_bitrate = 32 # 32-bit float for Morphagene
    # speed and attenuation
    db = -1. #decibels to adjust
    volscale = 10**(np.array(db)/10)
    speed_change = 2. # 1 = maintain original speed, >1 faster, <1 slower
    # read/write options
    write_label_file = True # write labels to a file compatible with Audacity
    simple_read = False # handles unsigned integer 8bit files

    ###########################################################################
    '''
    Write drum samples to one stereo Morphagene 32bit WAV file with scaled 
        volume sampled at 48000Hz
    '''
    ###########################################################################
    # load each sample name
    files = get_list_of_files(folder,extension)
    # volume of each repeated sample, leave as [1] to have one repeat of each
    #   sample at full volume
    scale = [volscale]
    # preallocate
    arrays, labs_comp, lab_tot = [],[], 0.
    # loop over files
    for fi in files:
        # repeat each sample at each volume
        for s in scale:
            # read rate, data, and bitdepth, normalized
            rate, array, _ = read(fi, normalized=True)
            # scale volume
            array = array*s
            # convert to Morphagene rate if not already that rate
            # do this to avoid pitch change upon rate mismatch
            if rate != morph_samplerate: 
                new_rate = float(morph_samplerate) * (1 / speed_change)
                array = change_samplerate_interp(array,float(rate),
                                 new_rate)   
            # warn (not error) if old rate is greater than Morphagene rate
            if rate > morph_samplerate: 
                print('Morphagene sample rate greater than original, aliasing possible')
            if array.ndim == 1: # mono
                # make mono sample stereo
                stereo = np.tile(array,(2,1))
                # append samples
                arrays.append(stereo)
                # calculate new splice location [in frames] and append to list
                lab_tot += stereo.shape[1]
                labs_comp.append(lab_tot)
            elif array.ndim == 2: #stereo
                # append samples
                arrays.append(array.T) # transpose to orientation of other arrays
                # calculate new splice location [in frames] and append to list
                lab_tot += array.T.shape[1]
                labs_comp.append(lab_tot)
            else:
                raise ValueError('File must be either mono or stereo')
    # concatenate samples into long array
    arrays = np.hstack(arrays)
    # convert splice location to int for correct format and make dict
    # ignore last splice point as there is no audio after it
    frame_labs = np.array(labs_comp)[:-1].astype('int')
    frame_dict = [{'position': l, 'label': 'marker%i'%(i+1)} for i,l in enumerate(frame_labs)]
    # test if resultant file exceeds the splice and
    #   file length limitations of the Morphagene
    if len(frame_dict) > 300 or (arrays.shape[1]/morph_samplerate)/60. > 2.9:
        raise ValueError('Number of splices (%i) and/or audio length (%2.1f minutes) exceed Morphagene limits'%(len(frame_dict),(arrays.shape[1]/morph_samplerate)/60.))
    else:
        # write concatenated sample arrays and labels to single morphagene wav file
        # also optionally write text file of Audacity-compatible labels
        wavefile=float32_wav_file("mgsh3_r.wav",arrays,morph_samplerate,markers=frame_dict)
        if write_label_file:
            labs = np.tile(frame_labs,(2,1)).T/float(morph_samplerate)
            np.savetxt('mgsh3_r.txt',labs,fmt='%03.6f',delimiter='\t')