dtas.git  about / heads / tags
duct tape audio suite for *nix
blob 2ccdcf1d66f943bd13cd02c658d75ac78e884acf 4100 bytes (raw)
$ git show v0.16.1:lib/dtas/tfx.rb	# shows this blob on the CLI

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# Copyright (C) 2013-2016 all contributors <dtas-all@nongnu.org>
# License: GPL-3.0+ <https://www.gnu.org/licenses/gpl-3.0.txt>
# frozen_string_literal: true
require_relative '../dtas'
require_relative 'parse_time'
require_relative 'format'
require 'shellwords'

# this will represent a trim section inside -splitfx for applying
# effects to only a part of the output
class DTAS::TFX # :nodoc:
  include DTAS::ParseTime

  attr_reader :tbeg
  attr_reader :tlen
  attr_reader :cmd

  def initialize(args, format = DTAS::Format.new)
    @format = format
    args = args.dup
    case args.shift
    when :pad # [ :pad, start_time, end_time ]
      @tbeg = args.shift
      @tlen = args.shift - @tbeg
    when "trim"
      parse_trim!(args)
    when "all"
      @tbeg = 0
      @tlen = nil
    else
      raise ArgumentError, "#{args.inspect} not understood"
    end
    case tmp = args.shift
    when "sh" then @cmd = args
    when "sox" then tfx_sox(args)
    when "eca" then tfx_eca(args)
    when nil
      @cmd = []
    else
      raise ArgumentError, "unknown effect type: #{tmp}"
    end
  end

  def tfx_sox(args)
    @cmd = %w(sox $SOXIN $SOXOUT $TRIMFX)
    @cmd.concat(args)
  end

  def tfx_eca(args)
    @cmd = %w(sox $SOXIN $SOX2ECA $TRIMFX)
    @cmd.concat(%w(| ecasound $ECAFMT -i stdin -o stdout))
    @cmd.concat(args)
    @cmd.concat(%w(| sox $ECA2SOX - $SOXOUT))
  end

  def to_sox_arg
    if @tbeg && @tlen
      %W(trim #{@tbeg}s #{@tlen}s)
    elsif @tbeg
      return [] if @tbeg == 0
      %W(trim #{@tbeg}s)
    else
      []
    end
  end

  # tries to interpret "trim" time args the same way the sox trim effect does
  # This takes _time_ arguments only, not sample counts;
  # otherwise, deviations from sox are considered bugs in dtas
  def parse_trim!(args)
    tbeg = parse_time(args.shift)
    if args[0] =~ /\A=?[\d\.]+\z/
      tlen = args.shift.dup
      is_stop_time = tlen.sub!(/\A=/, "") ? true : false
      tlen = parse_time(tlen)
      tlen = tlen - tbeg if is_stop_time
      @tlen = (tlen * @format.rate).round
    else
      @tlen = nil
    end
    @tbeg = (tbeg * @format.rate).round
  end

  def <=>(other)
    tbeg <=> other.tbeg
  end

  # for stable sorting
  class TFXSort < Struct.new(:tfx, :idx) # :nodoc:
    def <=>(other)
      cmp = tfx <=> other.tfx
      0 == cmp ? idx <=> other.idx : cmp
    end
  end

  # sorts and converts an array of TFX objects into non-overlapping arrays
  # of epochs
  #
  # input:
  #   [ tfx1, tfx2, tfx3, ... ]
  #
  # output:
  #   [
  #     [ tfx1 ],         # first epoch
  #     [ tfx2, tfx3 ],   # second epoch
  #     ...
  #   ]
  # There are multiple epochs only if ranges overlap,
  # There is only one epoch if there are no overlaps
  def self.schedule(ary)
    sorted = []
    ary.each_with_index { |tfx, i| sorted << TFXSort[tfx, i] }
    sorted.sort!
    rv = []
    epoch = 0
    prev_end = 0
    defer = []

    begin
      while tfxsort = sorted.shift
        tfx = tfxsort.tfx
        if tfx.tbeg >= prev_end
          # great, no overlap, append to the current epoch
          prev_end = tfx.tbeg + tfx.tlen
          (rv[epoch] ||= []) << tfx
        else
          # overlapping region, we'll need a new epoch
          defer << tfxsort
        end
      end

      if defer[0] # do we need another epoch?
        epoch += 1
        sorted = defer
        defer = []
        prev_end = 0
      end
    end while sorted[0]

    rv
  end

  # like schedule, but fills in the gaps with pass-through (no-op) TFX objs
  # This does not change the number of epochs.
  def self.expand(ary, total_samples)
    rv = []
    schedule(ary).each_with_index do |sary, epoch|
      tip = 0
      dst = rv[epoch] = []
      while tfx = sary.shift
        if tfx.tbeg > tip
          # fill in the previous gap
          nfx = new([:pad, tip, tfx.tbeg])
          dst << nfx
          dst << tfx
          tip = tfx.tbeg + tfx.tlen
        end
      end
      if tip < total_samples # fill until the last chunk
        nfx = new([:pad, tip, total_samples])
        dst << nfx
      end
    end
    rv
  end
end

git clone git://80x24.org/dtas.git
git clone https://80x24.org/dtas.git