Source code for dscleaner.merger

from .fileinfo import FileInfo
from . import ifileinfo, filewriter, CsvFileInfo, utils
import numpy as np
import os
import soundfile as sf

[docs]class Merger(): """ Merger allows for creation of an empty soundfile to store multiple datasets easily. NOTE: W64 filetype is recommended, given it can store up to 18 exabytes of data. Args: channels: Number of channels the files parsed should have. path: the path where the new merger file should be created. samplerate: samplerate to write on the file. cutoff: how often should the file be written **NOT IMPLEMENTED** (eg. for each 1024MB of data reached a new file is created) mode: either 'a' or 'w' if the file should be appended or truncated, respectively. Default behavior: append """ def __init__(self, channels, path, samplerate, cutoff = None, mode='a'): self._channels = channels self._init_samples() if(isinstance(samplerate,int)): self._samplerate = samplerate else: raise TypeError("Samplerate not integer") self.path = path #if the file exists already, then there's no need to create it(FileInfo) #if theres no file, we need to create it (array) self.file_exists = (os.path.isfile(self.path)) self.mode = mode def _init_samples(self): self._samples = np.empty(shape=(0,self._channels)) def __enter__(self): return self def __exit__(self,type,value,traceback): pass
[docs] def add(self, *files): """ Adds new samples to the buffer array. When ``create_file`` method is executed the buffer gets emptied. Args: *files: An array, containing several pathes to files or IFileInfos specializations, although the latter is preferred. """ tmp = None for f in files: if(issubclass(type(f),ifileinfo.IFileInfo)): tmp = f.getSamples() else: with FileInfo(f) as this: tmp = this.getSamples() self._samples = np.append(self._samples,tmp,axis=0)
#if(sys.getsizeof(self._samples) > (2**29)): escreve no ficheiro; para tal é necessário ter o ficheiro pré definido #adicionar 2**29(512MB) como cutoff
[docs] def create_file(self, samplerate = None): """ Creates a new file with the filename, converts based on extension given in ``new_filename`` When executed the sample buffer will be emptied, so ``create_file`` should be executed frequently. Args: samplerate: Samplerate of the file. """ if(len(self._samples) == 0): raise TypeError("No samples are queued!") if(samplerate == None): samplerate = self._samplerate path = utils.path_splitter(self.path) if(path['file_name'] == None): raise TypeError("It doesn't contain an extension") #tries to create the directory try: import os os.mkdir(path['path']) except (OSError,FileNotFoundError) as e: #import errno pass #if(e.errno != errno.EEXIST): # raise #if there's an error thats not eexits(file/directory exists) #Appends or simply write over the file based on mode if(self.mode == 'a' and self.file_exists == True): with sf.SoundFile(path['full_path'], mode = 'r+') as wfile:,sf.SEEK_END) wfile.write(self._samples) self._init_samples() else: with CsvFileInfo(self._samples,self._samplerate) as infofile: self.file_exists = True self._init_samples() with filewriter.FileWriter(infofile) as fw: fw.create_file(path['full_path'], self._samplerate)