Skip to content

Commit

Permalink
up
Browse files Browse the repository at this point in the history
  • Loading branch information
twardoch committed Apr 29, 2023
1 parent ab3e4dc commit 33cbaca
Show file tree
Hide file tree
Showing 5 changed files with 402 additions and 0 deletions.
6 changes: 6 additions & 0 deletions .gitmodules
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
[submodule "vendors/stretch"]
path = vendors/stretch
url = https://github.com/dbry/audio-stretch
[submodule "vendors/resample"]
path = vendors/resample
url = https://github.com/dbry/audio-resampler
285 changes: 285 additions & 0 deletions src.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,285 @@
=== file: audiostretchy/__init__.py ===

```
import sys

if sys.version_info[:2] >= (3, 8):
# TODO: Import directly (no need for conditional) when `python_requires = >= 3.8`
from importlib.metadata import PackageNotFoundError, version # pragma: no cover
else:
from importlib_metadata import PackageNotFoundError, version # pragma: no cover

try:
# Change here if project is renamed and does not equal the package name
dist_name = __name__
__version__ = version(dist_name)
except PackageNotFoundError: # pragma: no cover
__version__ = "unknown"
finally:
del version, PackageNotFoundError

```

import sys

if sys.version_info[:2] >= (3, 8):
# TODO: Import directly (no need for conditional) when `python_requires = >= 3.8`
from importlib.metadata import PackageNotFoundError, version # pragma: no cover
else:
from importlib_metadata import PackageNotFoundError, version # pragma: no cover

try:
# Change here if project is renamed and does not equal the package name
dist_name = __name__
__version__ = version(dist_name)
except PackageNotFoundError: # pragma: no cover
__version__ = "unknown"
finally:
del version, PackageNotFoundError

```

===

=== file: audiostretchy/interface/stretch.py ===

```
import ctypes

stretch_lib = ctypes.cdll.LoadLibrary("_stretch.so")

class Stretch:
def __init__(self, shortest_period, longest_period, num_chans, flags):
self.stretch_init = stretch_lib.stretch_init
self.stretch_init.argtypes = [
ctypes.c_int,
ctypes.c_int,
ctypes.c_int,
ctypes.c_int,
]
self.stretch_init.restype = ctypes.c_void_p
self.handle = self.stretch_init(
shortest_period, longest_period, num_chans, flags
)
self.stretch_output_capacity = stretch_lib.stretch_output_capacity
self.stretch_output_capacity.argtypes = [
ctypes.c_void_p,
ctypes.c_int,
ctypes.c_float,
]
self.stretch_output_capacity.restype = ctypes.c_int
self.stretch_samples = stretch_lib.stretch_samples
self.stretch_samples.argtypes = [
ctypes.c_void_p,
np.ctypeslib.ndpointer(dtype=np.int16),
ctypes.c_int,
np.ctypeslib.ndpointer(dtype=np.int16),
ctypes.c_float,
]
self.stretch_samples.restype = ctypes.c_int
self.stretch_flush = stretch_lib.stretch_flush
self.stretch_flush.argtypes = [
ctypes.c_void_p,
np.ctypeslib.ndpointer(dtype=np.int16),
]
self.stretch_flush.restype = ctypes.c_int
self.stretch_reset = stretch_lib.stretch_reset
self.stretch_reset.argtypes = [ctypes.c_void_p]
self.stretch_reset.restype = None
self.stretch_deinit = stretch_lib.stretch_deinit
self.stretch_deinit.argtypes = [ctypes.c_void_p]
self.stretch_deinit.restype = None

def output_capacity(self, max_num_samples, max_ratio):
return self.stretch_output_capacity(self.handle, max_num_samples, max_ratio)

def samples(self, samples, num_samples, output, ratio):
return self.stretch_samples(self.handle, samples, num_samples, output, ratio)

def flush(self, output):
return self.stretch_flush(self.handle, output)

def reset(self):
self.stretch_reset(self.handle)

def deinit(self):
self.stretch_deinit(self.handle)
self.handle = None

```

import ctypes

stretch_lib = ctypes.cdll.LoadLibrary("_stretch.so")

class Stretch:
def __init__(self, shortest_period, longest_period, num_chans, flags):
self.stretch_init = stretch_lib.stretch_init
self.stretch_init.argtypes = [
ctypes.c_int,
ctypes.c_int,
ctypes.c_int,
ctypes.c_int,
]
self.stretch_init.restype = ctypes.c_void_p
self.handle = self.stretch_init(
shortest_period, longest_period, num_chans, flags
)
self.stretch_output_capacity = stretch_lib.stretch_output_capacity
self.stretch_output_capacity.argtypes = [
ctypes.c_void_p,
ctypes.c_int,
ctypes.c_float,
]
self.stretch_output_capacity.restype = ctypes.c_int
self.stretch_samples = stretch_lib.stretch_samples
self.stretch_samples.argtypes = [
ctypes.c_void_p,
np.ctypeslib.ndpointer(dtype=np.int16),
ctypes.c_int,
np.ctypeslib.ndpointer(dtype=np.int16),
ctypes.c_float,
]
self.stretch_samples.restype = ctypes.c_int
self.stretch_flush = stretch_lib.stretch_flush
self.stretch_flush.argtypes = [
ctypes.c_void_p,
np.ctypeslib.ndpointer(dtype=np.int16),
]
self.stretch_flush.restype = ctypes.c_int
self.stretch_reset = stretch_lib.stretch_reset
self.stretch_reset.argtypes = [ctypes.c_void_p]
self.stretch_reset.restype = None
self.stretch_deinit = stretch_lib.stretch_deinit
self.stretch_deinit.argtypes = [ctypes.c_void_p]
self.stretch_deinit.restype = None

def output_capacity(self, max_num_samples, max_ratio):
return self.stretch_output_capacity(self.handle, max_num_samples, max_ratio)

def samples(self, samples, num_samples, output, ratio):
return self.stretch_samples(self.handle, samples, num_samples, output, ratio)

def flush(self, output):
return self.stretch_flush(self.handle, output)

def reset(self):
self.stretch_reset(self.handle)

def deinit(self):
self.stretch_deinit(self.handle)
self.handle = None

```

===

=== file: audiostretchy/stretch.py ===

```
import wave
import numpy as np
import fire

from .interface.stretch import Stretch

def validate_ratio(ratio):
if not (0.25 <= ratio <= 4.0):
raise ValueError("Ratio must be from 0.25 to 4.0!")
return ratio


def read_wave_file(filename):
with wave.open(filename, "rb") as infile:
params = infile.getparams()
nchannels, sampwidth, framerate, nframes = params[:4]
audio_data = infile.readframes(nframes)
samples = np.frombuffer(audio_data, dtype=np.int16)

return params, samples


def write_wave_file(filename, params, output_samples, num_samples):
with wave.open(filename, "wb") as outfile:
outfile.setparams(params)
outfile.writeframes(output_samples[:num_samples].tobytes())


def process_audio(infilename, outfilename, ratio=1.0, silence_ratio=0.0):
ratio = validate_ratio(ratio)
silence_ratio = silence_ratio or ratio
silence_ratio = validate_ratio(silence_ratio)
params, samples = read_wave_file(infilename)
nchannels, sampwidth, framerate, nframes = params[:4]
stretcher = Stretch(framerate // 333, framerate // 55, nchannels, 0)
output_samples = np.zeros(stretcher.output_capacity(nframes, ratio), dtype=np.int16)
num_samples = stretcher.samples(samples, len(samples), output_samples, ratio)
num_samples += stretcher.flush(output_samples[num_samples:])
write_wave_file(outfilename, params, output_samples, num_samples)

stretcher.deinit()


def main():
fire.Fire(process_audio)


if __name__ == "__main__":
main()

```

import wave
import numpy as np
import fire

from .interface.stretch import Stretch

def validate_ratio(ratio):
if not (0.25 <= ratio <= 4.0):
raise ValueError("Ratio must be from 0.25 to 4.0!")
return ratio


def read_wave_file(filename):
with wave.open(filename, "rb") as infile:
params = infile.getparams()
nchannels, sampwidth, framerate, nframes = params[:4]
audio_data = infile.readframes(nframes)
samples = np.frombuffer(audio_data, dtype=np.int16)

return params, samples


def write_wave_file(filename, params, output_samples, num_samples):
with wave.open(filename, "wb") as outfile:
outfile.setparams(params)
outfile.writeframes(output_samples[:num_samples].tobytes())


def process_audio(infilename, outfilename, ratio=1.0, silence_ratio=0.0):
ratio = validate_ratio(ratio)
silence_ratio = silence_ratio or ratio
silence_ratio = validate_ratio(silence_ratio)
params, samples = read_wave_file(infilename)
nchannels, sampwidth, framerate, nframes = params[:4]
stretcher = Stretch(framerate // 333, framerate // 55, nchannels, 0)
output_samples = np.zeros(stretcher.output_capacity(nframes, ratio), dtype=np.int16)
num_samples = stretcher.samples(samples, len(samples), output_samples, ratio)
num_samples += stretcher.flush(output_samples[num_samples:])
write_wave_file(outfilename, params, output_samples, num_samples)

stretcher.deinit()


def main():
fire.Fire(process_audio)


if __name__ == "__main__":
main()

```

===

61 changes: 61 additions & 0 deletions src/audiostretchy/interface/stretch.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
import ctypes

stretch_lib = ctypes.cdll.LoadLibrary("_stretch.so")

class Stretch:
def __init__(self, shortest_period, longest_period, num_chans, flags):
self.stretch_init = stretch_lib.stretch_init
self.stretch_init.argtypes = [
ctypes.c_int,
ctypes.c_int,
ctypes.c_int,
ctypes.c_int,
]
self.stretch_init.restype = ctypes.c_void_p
self.handle = self.stretch_init(
shortest_period, longest_period, num_chans, flags
)
self.stretch_output_capacity = stretch_lib.stretch_output_capacity
self.stretch_output_capacity.argtypes = [
ctypes.c_void_p,
ctypes.c_int,
ctypes.c_float,
]
self.stretch_output_capacity.restype = ctypes.c_int
self.stretch_samples = stretch_lib.stretch_samples
self.stretch_samples.argtypes = [
ctypes.c_void_p,
np.ctypeslib.ndpointer(dtype=np.int16),
ctypes.c_int,
np.ctypeslib.ndpointer(dtype=np.int16),
ctypes.c_float,
]
self.stretch_samples.restype = ctypes.c_int
self.stretch_flush = stretch_lib.stretch_flush
self.stretch_flush.argtypes = [
ctypes.c_void_p,
np.ctypeslib.ndpointer(dtype=np.int16),
]
self.stretch_flush.restype = ctypes.c_int
self.stretch_reset = stretch_lib.stretch_reset
self.stretch_reset.argtypes = [ctypes.c_void_p]
self.stretch_reset.restype = None
self.stretch_deinit = stretch_lib.stretch_deinit
self.stretch_deinit.argtypes = [ctypes.c_void_p]
self.stretch_deinit.restype = None

def output_capacity(self, max_num_samples, max_ratio):
return self.stretch_output_capacity(self.handle, max_num_samples, max_ratio)

def samples(self, samples, num_samples, output, ratio):
return self.stretch_samples(self.handle, samples, num_samples, output, ratio)

def flush(self, output):
return self.stretch_flush(self.handle, output)

def reset(self):
self.stretch_reset(self.handle)

def deinit(self):
self.stretch_deinit(self.handle)
self.handle = None
Loading

0 comments on commit 33cbaca

Please sign in to comment.