#### RemiAudio
#### Copyright (C) 2022-2024 Remilia Scarlet <remilia@posteo.jp>
####
#### This program is free software: you can redistribute it and/or modify it
#### under the terms of the GNU Affero General Public License as published by
#### the Free Software Foundation, either version 3 of the License, or (at your
#### option) any later version.
####
#### This program is distributed in the hope that it will be useful, but WITHOUT
#### ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
#### FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public
#### License for more details.
####
#### You should have received a copy of the GNU Affero General Public License
#### along with this program. If not, see <https://www.gnu.org/licenses/>.
module RemiAudio
INT8_INV = 1 / Int8::MAX
INT8_INV_F32 = INT8_INV.to_f32!
INT16_INV = 1 / Int16::MAX
INT16_INV_F32 = INT16_INV.to_f32!
INT24_INV = 1 / ((2 ** 23) - 1)
INT24_INV_F32 = INT24_INV.to_f32!
INT32_INV = 1 / Int32::MAX
INT32_INV_F32 = INT32_INV.to_f32!
# Base class for exceptions that occur in RemiAudio.
class RemiAudioError < Exception
end
# Indicates that a buffer or buffers are of an unexpected size.
class BufferSizeMismatchError < RemiAudioError
end
# Indicates an unsupported format.
class UnsupportedFormatError < RemiAudioError
end
# Indicates an unsupported or invalid bit depth.
class InvalidBitDepthError < RemiAudioError
end
# Any of the supported sample formats.
alias Sample = UInt8|Int8|Int16|Int32|Int64|Float32|Float64
# An array or slice of unsigned LPCM samples.
alias UnsignedSampleData = Array(UInt8)|Slice(UInt8)
# An array or slice of signed LPCM samples.
alias LPCMSampleData = Array(Int8)|Slice(Int8)|
Array(Int16)|Slice(Int16)|
Array(Int32)|Slice(Int32)|
Array(Int64)|Slice(Int64)
# An array or slice of unsigned or signed LPCM samples.
alias LPCMSampleDataMixed = Array(UInt8|Int8|Int16|Int32|Int64)|
Slice(UInt8|Int8|Int16|Int32|Int64)
# An array or slice of float samples.
alias FloatSampleData = Array(Float32)|Slice(Float32)|
Array(Float64)|Slice(Float64)
# An array or slice of float samples.
alias FloatSampleDataMixed = Array(Float32|Float64)|Slice(Float32|Float64)
# An array or slice of any supported sample format.
alias SampleData = UnsignedSampleData|LPCMSampleData|FloatSampleData
# An array or slice of any supported sample format, possibly mixed together.
alias SampleDataMixed = Array(UInt8|Int8|Int16|Int32|Int64|Float32|Float64)|
Slice(UInt8|Int8|Int16|Int32|Int64|Float32|Float64)
# Used to describe the format of a sample when it may be ambiguous.
enum SampleFormat
F64
F32
I64
I32
I24
I16
#I12
I8
U8
# Returns the number of bits that this format represents.
@[AlwaysInline]
def getBits
case self
in .f64? then 64
in .f32? then 32
in .i64? then 64
in .i32? then 32
in .i24? then 24
in .i16? then 16
#in .i12? then 12
in .i8? then 8
in .u8? then 8
end
end
# Returns the class that is expected to store this format.
def getClass : Class
case self
in .f64? then Float64
in .f32? then Float32
in .i64? then Int64
in .i32? then Int32
in .i24? then Int32
in .i16? then Int16
#in .i12? then Int16
in .i8? then Int8
in .u8? then UInt8
end
end
def makeArray(size : Int) : SampleData
case self
in .f64? then Array(Float64).new(size, 0.0).as(SampleData)
in .f32? then Array(Float32).new(size, 0.0f32).as(SampleData)
in .i64? then Array(Int64).new(size, 0i64).as(SampleData)
in .i32?, .i24? then Array(Int32).new(size, 0i32).as(SampleData)
in .i16? then Array(Int16).new(size, 0i16).as(SampleData)
in .i8? then Array(Int8).new(size, 0i8).as(SampleData)
in .u8? then Array(UInt8).new(size, 0u8).as(SampleData)
end
end
end
# Indicates a stereo side.
enum Side
Left
Right
end
enum PlotType
GnuPlot
Octave
end
# Converts decibels to a linear float value.
@[AlwaysInline]
def self.decibelsToLinear(x : Float64) : Float64
10.0 ** (0.05 * x)
end
# :ditto:
@[AlwaysInline]
def self.decibelsToLinear(x : Float32) : Float32
10.0f32 ** (0.05f32 * x)
end
# Converts a linear float value to decibels. If `x` is 0, this always returns
# -144.0.
@[AlwaysInline]
def self.linearToDecibels(x : Float64) : Float64
return -144.0 if x == 0
20.0 * Math.log10(x)
end
# :ditto:
@[AlwaysInline]
def self.linearToDecibels(x : Float32) : Float32
return -144.0f32 if x == 0
20.0f32 * Math.log10(x)
end
end
###
### Extension methods for IO
###
abstract class IO
# Gets the next `::Char` in this IO, or `nil` if there is none. The internal
# position is not changed.
#
# The IO must support `::IO#pos` and `::IO#pos=`
def peekChar : Char?
oldPos = self.pos
c = self.read_char
self.pos = oldPos
c
end
# Attempts to read a "four CC" string from the IO, as used in RIFF formats.
# If four bytes cannot be read, this returns `nil`.
def readFourCC : String?
data = Bytes.new(4)
read_fully?(data) || return nil
data.map! do |val|
unless (32 <= val) && (val <= 126)
val = '?'.to_u8
else
val
end
end
String.new(data)
end
# Attempts to read a RIFF Chunk ID from the `IO`. The chunk ID must be equal
# to `chunkType`. If it is, this returns the chunk that was read, otherwise
# this raises an `IO::Error`.
#
# If `message` is specified, then the error uses that as the message,
# otherwise it uses a more generic error message that lists what it found and
# the expected chunk name.
def expectChunkType(chunkType : String, message : String? = nil) : String?
chunk = readFourCC
unless chunk == chunkType
raise IO::Error.new(message || "The chunk type must be '#{chunkType}', not '#{chunk}'")
end
chunk
end
# Attempts to read a RIFF Chunk ID from the `IO`. The chunk ID must be one of
# the chunk names listed in `chunkTypes`. If it is, this returns the chunk
# that was read, otherwise this raises an `IO::Error`.
#
# If `message` is specified, then the error uses that as the message,
# otherwise it uses a more generic error message that lists what it found and
# the expected chunk name.
def expectChunkType(chunkTypes : Array(String), message : String? = nil) : String?
chunk = readFourCC
unless chunkTypes.includes?(chunk)
raise IO::Error.new(message || "The chunk type must be one of '#{chunkTypes}', not '#{chunk}'")
end
chunk
end
end
###
### Extension methods for Array and Slice
###
class Array(T)
# Returns the sample at the given index for the given side. This treats
# this array as having interleaved left/right data, and so `index` must be
# less than half of the total `size`, or less than `sizePerSide`.
@[AlwaysInline]
def [](index : Int, side : RemiAudio::Side) : T
if side.left?
self[index + index]
else
self[index + index + 1]
end
end
# Sets the sample at the given index for the given side. This treats this
# array as having interleaved left/right data, and so `index` must be less
# than half of the total `size`, or less than `sizePerSide`.
@[AlwaysInline]
def []=(index : Int, side : RemiAudio::Side, value : T) : Nil
if side.left?
self[index + index] = value
else
self[index + index + 1] = value
end
end
# Returns the side for one side of the internal buffer. This treats this
# array as having interleaved left/right data.
@[AlwaysInline]
def sizePerSide : Int
self.size.tdiv(2)
end
# Interleaves this audio data with *other*. The size of *other* must be
# exactly the same as this instance's `#size`, or an `RemiAudioError` is
# raised.
#
# The size check can be disabled by passing `-Dremiaudio_wd40` to the compiler
# at build time.
@[AlwaysInline]
def interleave(other : Array(T)) : Array(T)
ret : Array(T) = Array(T).new(self.size * 2, T.zero)
interleave(ret)
ret
end
# Interleaves this audio data with *other*, storing the results in *dest*.
# The size of *other* must be exactly the same as this instance's `#size`, and
# the size of *dest* must be exactly twice to the size of this instance, or an
# `RemiAudioError` is raised.
#
# The size checks can be disabled by passing `-Dremiaudio_wd40` to the
# compiler at build time.
def interleave(other : Array(T), dest : Array(T)) : Array(T)
{% unless flag?(:remiaudio_wd40) %}
unless self.size == other.size
raise RemiAudioError.new("Cannot interleave arrays of two different sizes")
end
unless self.size * 2 <= dest.size
raise RemiAudioError.new("Cannot interleave array, destination not the correct size")
end
{% end %}
sidx : Int32 = 0
didx : Int32 = 0
while didx < dest.size
dest[didx] = self.[sidx]
dest[didx + 1] = other.[sidx]
sidx += 1
didx += 2
end
ret
end
# Treats this array as having interleaved left/right data, and splits the
# internal buffer into two new `Array` instances. This returns a `Tuple`
# where the 0th element is the left side, and the other is the right side.
@[AlwaysInline]
def deinterleave : Tuple(Array(T), Array(T))
left = Array(T).new(sizePerSide)
right = Array(T).new(sizePerSide)
deinterleave(left, right)
{left, right}
end
# Treats this array as having interleaved left/right data, and splits the
# internal buffer into *destLeft* and *destRight*. The size of *destLeft* and
# *destRight* must be exactly equal to the size of this instance.
# Additionally, this instance must have a size divisible by 2.
#
# If the size checks fail, this raises a `RemiAudioError`.
#
# The size checks can be disabled by passing `-Dremiaudio_wd40` to the
# compiler at build time.
def deinterleave(destLeft : Array(T), destRight : Array(T)) : Nil
{% unless flag?(:remiaudio_wd40) %}
unless self.size % 2 == 0
raise RemiAudioError.new("Size of source is not divisible by 2")
end
destSize = self.size.tdiv(2)
unless destLeft.size == selfSize && destRight.size == selfSize
raise RemiAudioError.new("Size of destinations are not equal to exactly half the source")
end
{% end %}
idx : Int32 = 0
while idx < self.size
left << self[idx]
right << self[idx + 1]
idx += 2
end
end
# Treats this array as having interleaved left/right data, yielding each
# element that would appear on the left side to the block. Returns `self`.
def eachLeftWithIndex(& : T ->)
pos = 0
while pos < self.size
yield self[pos]
pos += 2
end
self
end
# Treats this array as having interleaved left/right data, yielding each
# element that would appear on the right side to the block. Returns `self`.
def eachRight(& : T ->)
pos = 1
while pos < self.size
yield self[pos]
pos += 2
end
self
end
# Treats this array as having interleaved left/right data, yielding each
# element along with its side. Returns `self`.
def eachWithSide(& : Tuple(T, RemiAudio::Side) ->)
pos = 0
self.each do |samp|
yield samp, (pos % 2 == 0 ? RemiAudio::Side::Left : RemiAudio::Side::Right)
end
self
end
# Invokes the given block for each sample in `self`, along with the side that
# sample is on. This replaces the sample with the value returned by the
# block. Returns `self`.
def mapSides!(& : Tuple(T, RemiAudio::Side) ->)
pos = 0
while pos < self.size
self[pos] = yield ({samp, (pos % 2 == 0 ? RemiAudio::Side::Left : RemiAudio::Side::Right)})
end
self
end
end
struct Slice(T)
# Returns the sample at the given index for the given side. This treats
# this slice as having interleaved left/right data, and so `index` must be
# less than half of the total `size`, or less than `sizePerSide`.
@[AlwaysInline]
def [](index : Int, side : RemiAudio::Side) : T
if side.left?
self[index + index]
else
self[index + index + 1]
end
end
# Sets the sample at the given index for the given side. This treats this
# slice as having interleaved left/right data, and so `index` must be less
# than half of the total `size`, or less than `sizePerSide`.
@[AlwaysInline]
def []=(index : Int, side : RemiAudio::Side, value : T) : Nil
if side.left?
self[index + index] = value
else
self[index + index + 1] = value
end
end
# Returns the side for one side of the internal buffer. This treats this
# slice as having interleaved left/right data.
@[AlwaysInline]
def sizePerSide : Int
self.size.tdiv(2)
end
# Interleaves this audio data with *other*. The size of *other* must be
# exactly the same as this instance's `#size`, or an `RemiAudioError` is
# raised.
#
# The size check can be disabled by passing `-Dremiaudio_wd40` to the compiler
# at build time.
@[AlwaysInline]
def interleave(other : Slice(T)) : Slice(T)
ret : Slice(T) = Slice(T).new(self.size * 2, T.zero)
interleave(ret)
ret
end
# Interleaves this audio data with *other*, storing the results in *dest*.
# The size of *other* must be exactly the same as this instance's `#size`, and
# the size of *dest* must be exactly twice to the size of this instance, or an
# `RemiAudioError` is raised.
#
# The size checks can be disabled by passing `-Dremiaudio_wd40` to the
# compiler at build time.
def interleave(other : Slice(T), dest : Slice(T)) : Slice(T)
{% unless flag?(:remiaudio_wd40) %}
unless self.size == other.size
raise RemiAudioError.new("Cannot interleave arrays of two different sizes")
end
unless self.size * 2 <= dest.size
raise RemiAudioError.new("Cannot interleave array, destination not the correct size")
end
{% end %}
sidx : Int32 = 0
didx : Int32 = 0
while didx < dest.size
dest[didx] = self.[sidx]
dest[didx + 1] = other.[sidx]
sidx += 1
didx += 2
end
ret
end
# Treats this array as having interleaved left/right data, and splits the
# internal buffer into two new `Array` instances. This returns a `Tuple`
# where the 0th element is the left side, and the other is the right side.
@[AlwaysInline]
def deinterleave : Tuple(Slice(T), Slice(T))
left = Slice(T).new(sizePerSide)
right = Slice(T).new(sizePerSide)
deinterleave(left, right)
{left, right}
end
# Treats this array as having interleaved left/right data, and splits the
# internal buffer into *destLeft* and *destRight*. The size of *destLeft* and
# *destRight* must be exactly equal to the size of this instance.
# Additionally, this instance must have a size divisible by 2.
#
# If the size checks fail, this raises a `RemiAudioError`.
#
# The size checks can be disabled by passing `-Dremiaudio_wd40` to the
# compiler at build time.
def deinterleave(destLeft : Slice(T), destRight : Slice(T)) : Nil
{% unless flag?(:remiaudio_wd40) %}
unless self.size % 2 == 0
raise RemiAudioError.new("Size of source is not divisible by 2")
end
destSize = self.size.tdiv(2)
unless destLeft.size == selfSize && destRight.size == selfSize
raise RemiAudioError.new("Size of destinations are not equal to exactly half the source")
end
{% end %}
idx : Int32 = 0
while idx < self.size
left << self[idx]
right << self[idx + 1]
idx += 2
end
end
# Treats this slice as having interleaved left/right data, yielding each
# element that would appear on the left side to the block. Returns `self`.
def eachLeftWithIndex(& : T ->)
pos = 0
while pos < self.size
yield self[pos]
pos += 2
end
self
end
# Treats this slice as having interleaved left/right data, yielding each
# element that would appear on the right side to the block. Returns `self`.
def eachRight(& : T ->)
pos = 1
while pos < self.size
yield self[pos]
pos += 2
end
self
end
# Treats this slice as having interleaved left/right data, yielding each
# element along with its side. Returns `self`.
def eachWithSide(& : Tuple(T, RemiAudio::Side) ->)
pos = 0
self.each do |samp|
yield ({samp, (pos % 2 == 0 ? RemiAudio::Side::Left : RemiAudio::Side::Right)})
end
self
end
# Invokes the given block for each sample in `self`, along with the side that
# sample is on. This replaces the sample with the value returned by the
# block. Returns `self`.
def mapSides!(& : Tuple(T, RemiAudio::Side) ->)
pos = 0
while pos < self.size
self[pos] = yield ({samp, (pos % 2 == 0 ? RemiAudio::Side::Left : RemiAudio::Side::Right)})
end
self
end
end