Showing the changes to python (3 of 3 files). Show all
Unified · Split
| @@ -0,0 +1,56 @@ | ||
| 1 | +from aubio import source, sink | |
| 2 | +import os | |
| 3 | + | |
| 4 | +def slice_source_at_stamps(source_file, timestamps, timestamps_end = None, | |
| 5 | + output_dir = None, | |
| 6 | + samplerate = 0, | |
| 7 | + hopsize = 256): | |
| 8 | + | |
| 9 | + source_base_name, source_ext = os.path.splitext(os.path.basename(source_file)) | |
| 10 | + if output_dir != None: | |
| 11 | + if not os.path.isdir(output_dir): | |
| 12 | + os.makedirs(output_dir) | |
| 13 | + source_base_name = os.path.join(output_dir, source_base_name) | |
| 14 | + | |
| 15 | + def new_sink_name(source_base_name, timestamp): | |
| 16 | + return source_base_name + '_%02.3f' % (timestamp) + '.wav' | |
| 17 | + | |
| 18 | + # reopen source file | |
| 19 | + s = source(source_file, samplerate, hopsize) | |
| 20 | + if samplerate == 0: samplerate = s.get_samplerate() | |
| 21 | + # create first sink at 0 | |
| 22 | + g = sink(new_sink_name(source_base_name, 0.), samplerate) | |
| 23 | + total_frames = 0 | |
| 24 | + # get next region | |
| 25 | + next_stamp = int(timestamps.pop(0)) | |
| 26 | + if not next_stamp: | |
| 27 | + next_stamp = int(timestamps.pop(0)) | |
| 28 | + | |
| 29 | + while True: | |
| 30 | + # get hopsize new samples from source | |
| 31 | + vec, read = s() | |
| 32 | + remaining = next_stamp - total_frames | |
| 33 | + # not enough frames remaining, time to split | |
| 34 | + if remaining <= read: | |
| 35 | + if remaining != 0: | |
| 36 | + # write remaining samples from current region | |
| 37 | + g(vec[0:remaining], remaining) | |
| 38 | + # close this file | |
| 39 | + del g | |
| 40 | + # create a new file for the new region | |
| 41 | + new_sink_path = new_sink_name(source_base_name, next_stamp / float(samplerate)) | |
| 42 | + #print "new slice", total_frames, "+", remaining, "=", next_stamp | |
| 43 | + g = sink(new_sink_path, samplerate) | |
| 44 | + if remaining != read: | |
| 45 | + # write the remaining samples in the new file | |
| 46 | + g(vec[remaining:read], read - remaining) | |
| 47 | + if len(timestamps): | |
| 48 | + next_stamp = int(timestamps.pop(0)) | |
| 49 | + else: | |
| 50 | + next_stamp = 1e120 | |
| 51 | + else: | |
| 52 | + g(vec[0:read], read) | |
| 53 | + total_frames += read | |
| 54 | + if read < hopsize: break | |
| 55 | + | |
| 56 | + del g | |
| @@ -0,0 +1,41 @@ | ||
| 1 | +#! /usr/bin/env python | |
| 2 | + | |
| 3 | +from numpy.testing import TestCase, run_module_suite | |
| 4 | +from numpy.testing import assert_equal, assert_almost_equal | |
| 5 | + | |
| 6 | +from aubio import slice_source_at_stamps | |
| 7 | +from utils import count_samples_in_file, count_samples_in_directory | |
| 8 | + | |
| 9 | +import tempfile | |
| 10 | +import shutil | |
| 11 | + | |
| 12 | +class aubio_slicing_test_case(TestCase): | |
| 13 | + | |
| 14 | + def setUp(self): | |
| 15 | + self.source_file = 'chocolate_1min.wav' | |
| 16 | + self.output_dir = tempfile.mkdtemp(suffix = 'aubio_slicing_test_case') | |
| 17 | + | |
| 18 | + def test_slice_start_only(self): | |
| 19 | + regions_start = [i*1000 for i in range(100)] | |
| 20 | + slice_source_at_stamps(self.source_file, regions_start, output_dir = self.output_dir) | |
| 21 | + | |
| 22 | + def test_slice_start_only_no_zero(self): | |
| 23 | + regions_start = [i*1000 for i in range(1, 100)] | |
| 24 | + slice_source_at_stamps(self.source_file, regions_start, output_dir = self.output_dir) | |
| 25 | + | |
| 26 | + def test_slice_start_beyond_end(self): | |
| 27 | + regions_start = [i*1000 for i in range(1, 100)] | |
| 28 | + regions_start += [count_samples_in_file(self.source_file)] | |
| 29 | + regions_start += [count_samples_in_file(self.source_file) + 1000] | |
| 30 | + slice_source_at_stamps(self.source_file, regions_start, output_dir = self.output_dir) | |
| 31 | + | |
| 32 | + def tearDown(self): | |
| 33 | + original_samples = count_samples_in_file(self.source_file) | |
| 34 | + written_samples = count_samples_in_directory(self.output_dir) | |
| 35 | + assert_equal(original_samples, written_samples, | |
| 36 | + "number samples written different from number of original samples") | |
| 37 | + shutil.rmtree(self.output_dir) | |
| 38 | + | |
| 39 | +if __name__ == '__main__': | |
| 40 | + from unittest import main | |
| 41 | + main() | |
| @@ -18,3 +18,25 @@ def array_from_yaml_file(filename): · 17 unmodified lines | ||
| 18 | 18 | yaml_data = yaml.safe_load(f) |
| 19 | 19 | f.close() |
| 20 | 20 | return yaml_data |
| 21 | + | |
| 22 | +def count_samples_in_file(file_path): | |
| 23 | + from aubio import source | |
| 24 | + hopsize = 256 | |
| 25 | + s = source(file_path, 0, hopsize) | |
| 26 | + total_frames = 0 | |
| 27 | + while True: | |
| 28 | + samples, read = s() | |
| 29 | + total_frames += read | |
| 30 | + if read < hopsize: break | |
| 31 | + return total_frames | |
| 32 | + | |
| 33 | +def count_samples_in_directory(samples_dir): | |
| 34 | + import os | |
| 35 | + total_frames = 0 | |
| 36 | + for f in os.walk(samples_dir): | |
| 37 | + if len(f[2]): | |
| 38 | + for each in f[2]: | |
| 39 | + file_path = os.path.join(f[0], each) | |
| 40 | + if file_path: | |
| 41 | + total_frames += count_samples_in_file(file_path) | |
| 42 | + return total_frames | |