RemiAudio: common.cr at tip

Annotate Artifact Blame Check-ins Using Doc Download Hex
File src/remiaudio/common.cr from the latest check-in

#### 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