Overengineered pulseaudio cli wrapper

pulseaudio.zsh

# Generic PulseAudio wrapper commands
if (( ! $+commands[pactl] ))
then
  return
fi

pulseaudio::list() {
  local invert
  local friendly_name
  local monitor

  zparseopts -D -K -E \
    v=invert -invert-match=invert \
    f=friendly_name -friendly-name=friendly_name \
    m=monitor -monitor=monitor

  local object_type="$1"
  local name="$2"

  if [[ ! "$object_type" =~ s$ ]]
  then
    object_type="${object_type}s" # append "s"
  fi
  local res=$(pactl list short "$object_type")

  # Skip monitors by default (unless -m|--monitor is set)
  if [[ -z "$monitor" ]]
  then
    res=$(grep -v monitor <<< "$res")
  fi

  # Check for direct matches (in case $2 is an actual device/sink/source name)
  # eg: alsa_input.usb-046d_HD_Pro_Webcam_C920_7E896ABF-02.analog-stereo
  if ! grep -qE "\s+${name}" <<< "$res"
  then
    # Replace non-alphanumeric chars with _
    name="${2//[^[:alnum:]]/_}"
  fi

  if [[ -n "$name" ]]
  then
    local match=$(awk -v 'IGNORECASE=1' \
      '/'${name}'/ { print $2; exit }' <<< "$res")

    if [[ -z "$match" ]]
    then
      echo_err "No match found for $name in $object_type"
      return 1
    fi

    local r
    # Only return names matching $2
    if [[ -n "$invert" ]]
    then
      res=("${(@f)$(awk -v 'IGNORECASE=1' '!/'${name}'/ { print $2 }' <<< "$res")}")
      for r in ${res}
      do
        if [[ -n "$friendly_name" ]]
        then
          pulseaudio::friendly-name "$object_type" "$r"
        else
          echo "$r"
        fi
      done
    else
      if [[ -n "$friendly_name" ]]
      then
        pulseaudio::friendly-name "$object_type" "$match"
      else
        echo "$match"
      fi
    fi
  else
    # Return all names
    res=("${(@f)$(awk '{ print $2 }' <<< "$res")}")
    for r in ${res}
    do
      if [[ -n "$friendly_name" ]]
      then
        pulseaudio::friendly-name "$object_type" "$r"
      else
        echo "$r"
      fi
    done
  fi
}

pulseaudio::get-device() {
  local object_type=$1
  local name=$2

  local data=$(pactl list "$object_type")
  local index
  local d

  # Remove trailing "s"
  object_type=${object_type[1,-2]}
  for index in $(awk -v IGNORECASE=1 "/^${object_type} #/ {print \$2}" <<< "$data")
  do
    d=$(sed -n "/^${object_type} ${index}/I,/^${object_type} #/Ip" <<< "$data")

    if grep -qE "^\s+Name: ${name}" <<< "$d"
    then
      # Remove last line (Next "Source/Sink #")
      awk 'NR>1{print buf}{buf = $0}' <<< "$d"
      return
    fi
  done
}

pulseaudio::get-prop() {
  local object_type=$1
  local name=$2
  local prop=$3

  pulseaudio::get-device "$object_type" "$name" | \
    sed -nr "s/.*${prop}\s*[:=]\s*\"?([^\"]+)\"?/\1/Ip"
}

pulseaudio::get-monitor() {
  local object_type=$1
  local name=$2

  if [[ -z "$name" ]]
  then
    echo_err "Missing device name"
    return 2
  fi

  # FIXME There's gotta be a way to do this in one command
  pactl list "$object_type" | \
    sed -nr "/Name:\s*/,\${p;/Monitor of.*:\s+${name}/q}" | \
    tac | awk '/Name:/ {print $NF; exit}'
}

pulseaudio::friendly-name() {
  local object_type=$1
  local name=$2

  local prop="alsa.card_name"
  if [[ "$name" =~ .*bluez.* ]]
  then
    prop="device.description"
  fi

  local res=$(pulseaudio::get-prop "$object_type" "$name" "$prop")

  if [[ -n "$res" ]]
  then
    echo $res
    return
  fi

  # Fall back to "Description"
  pulseaudio::get-prop "$object_type" "$name" "Description"
}

# Get device name from friendly name (or any other property)
pulseaudio::device-name() {
  local object_type=$1
  local val=$2

  # FIXME Use awk -v "val=${val}" instead of closing and re-opening the
  # singles quotes below
  local res=$(pactl list "$object_type" | \
    awk -F ':\\s+' \
      '{ IGNORECASE=1 }
       /Name:/ { line=NR; a=$2 }
       /'${val}'/ { found=1; exit }
     END { if(found == 1) print a; else exit 1; }')

  if [[ -z "$res" ]]
  then
    return 1
  fi
  echo "$res"
}

pulseaudio::info-extract() {
  local header="$*"
  pactl info | awk -v IGNORECASE=1 "/${header}: / {print \$NF}"
}

pulseaudio::get-default() {
  local friendly_name
  zparseopts -D -K -E \
    f=friendly_name -friendly-name=friendly_name

  local object_type=${1%s}  # remove trailing "s"
  local res=$(pulseaudio::info-extract "Default ${object_type}")

  if [[ -z "$res" ]]
  then
    echo_err "Failed to determine default $object_type"
    return 1
  fi

  if [[ -n "$friendly_name" ]]
  then
    pulseaudio::friendly-name "${object_type}s" "$res"
  else
    echo "$res"
  fi
}

pulseaudio::get-default-sink() {
  pulseaudio::get-default sink "$@"
}

pulseaudio::get-default-source() {
  pulseaudio::get-default source "$@"
}

pulseaudio::set-card-profile() {
  local name=$1
  local profile=$2
  local card=$(pulseaudio::list cards $name)

  if [[ -z "$card" ]]
  then
    echo_err "Failed to determine card name of $name"
    return 1
  fi

  echo_info "Setting profile of $name to $profile"
  pactl set-card-profile "$card" "$profile"
}

pulseaudio::set-default-sink() {
  local name=$1
  local sink=$(pulseaudio::list sinks $name)

  if [[ -z "$sink" ]]
  then
    echo_err "Failed to determine sink name of $name"
    return 1
  fi

  echo_info "Setting default sink to $sink"
  pactl set-default-sink "$sink"
}

pulseaudio::set-default-source() {
  local name=$1
  local source=$(pulseaudio::list sources $name)

  if [[ -z "$source" ]]
  then
    echo_err "Failed to determine source name of $name"
    return 1
  fi

  echo_info "Setting default source to $source"
  pacmd set-default-source "$source"
}

pulseaudio::set-source-volume() {
  local name=$1
  local volume_percent="${2:-100}"
  local source=$(pulseaudio::list sources $name)

  if [[ -z "$source" ]]
  then
    echo_err "Failed to determine source name of $name"
    return 1
  fi

  echo_info "Setting source volume of $name to ${volume_percent}%"
  pactl set-source-volume "$source" "${volume_percent}%"
}

pulseaudio::mute-unmute() {
  local device_type="$1"
  local device_name="$2"
  local action="${3:-mute}"

  local cmd
  case "$action" in
    mute|m)
      action=mute
      ;;
    unmute|u)
      action=unmute
      ;;
    *)
      echo_err "Unknown action: $action (Allowed values: mute|unmute)"
      return 2
      ;;
  esac

  case "$device_type" in
    sink*|output*)
      pulseaudio::${action}-sink "$device_name"
      ;;
    source*|input*)
      pulseaudio::${action}-source "$device_name"
      ;;
    *)
      echo_err "Unknown device type: $device_type"
      return 2
      ;;
  esac
}

pulseaudio::mute() {
  pulseaudio::mute-unmute "$1" "$2" mute
}

pulseaudio::unmute() {
  pulseaudio::mute-unmute "$1" "$2" unmute
}

pulseaudio::set-mute-source() {
  local name="$1"
  local action="${2:-toggle}"
  local source=$(pulseaudio::list -m sources $name)

  if [[ -z "$source" ]]
  then
    echo_err "Failed to determine source name of $name"
    return 1
  fi

  echo_info "Setting mute state for $name (-> $action)"
  pactl set-source-mute "$source" "$action"
}

pulseaudio::mute-source() {
  pulseaudio::set-mute-source "$1" true
}

pulseaudio::unmute-source() {
  pulseaudio::set-mute-source "$1" false
}

pulseaudio::set-mute-sink() {
  local name=$1
  local action="${2:-toggle}"
  local sink=$(pulseaudio::list -m sinks $name)

  if [[ -z "$sink" ]]
  then
    echo_err "Failed to determine sink name of $name"
    return 1
  fi

  echo_info "Setting mute state for $name (-> $action)"
  pactl set-sink-mute "$sink" "$action"
}

pulseaudio::mute-sink() {
  pulseaudio::set-mute-sink "$1" true
}

pulseaudio::unmute-sink() {
  pulseaudio::set-mute-sink "$1" false
}

pulseaudio::mute-all-sources() {
  local sources=($(pulseaudio::list sources))

  echo_info "Muting all sources"

  local s
  for s in ${sources[@]}
  do
    pulseaudio::set-mute-source "$s" true
  done
}

pulseaudio::mute-all-sinks() {
  local sinks=($(pulseaudio::list sinks))

  echo_info "Muting all sinks"

  local s
  for s in ${sinks[@]}
  do
    pulseaudio::set-mute-sink "$s" true
  done
}

pulseaudio::mute-default-source() {
  echo_info "Muting default source ($(pulseaudio::get-default-source -f))"
  pactl set-source-mute @DEFAULT_SOURCE@ true
}

pulseaudio::mute-default-sink() {
  echo_info "Muting default sink ($(pulseaudio::get-default-sink -f))"
  pactl set-sink-mute @DEFAULT_SINK@ true
}

pulseaudio::unmute-default-source() {
  echo_info "Unmuting default source ($(pulseaudio::get-default-source -f))"
  pactl set-source-mute @DEFAULT_SOURCE@ false

  # Check if noisetorch is running, if yes: unmute noisetorch's device
  local noisetorch_device=$(noisetorch::get-device-name)
  if [[ -n "$noisetorch_device" ]]
  then
    pulseaudio::unmute-source "$noisetorch_device"
    noisetorch::unmute
  fi
}

pulseaudio::unmute-default-sink() {
  echo_info "Unmuting default sink ($(pulseaudio::get-default-sink -f))"
  pactl set-sink-mute @DEFAULT_SINK@ false
}

pulseaudio::mute-all() {
  pulseaudio::mute-all-sources
  pulseaudio::mute-all-sinks
}

pulseaudio::unmute-all() {
  pulseaudio::unmute-all-sources
  pulseaudio::unmute-all-sinks
}

pulseaudio::unmute-default() {
  pulseaudio::unmute-default-source
  pulseaudio::unmute-default-sink
}

alias mute="pulseaudio::mute-all"
# alias unmute="pulseaudio::set-mute-sources-all; pulseaudio::set-mute-sinks-all"
alias unmute="pulseaudio::unmute-default"
alias umute="unmute"

pulseaudio::mute-all-sources-but() {
  local name=$1
  local source_unmute=$(pulseaudio::list sources $name)
  local sources=($(pulseaudio::list -v sources $name))

  echo_info "Muting all sources except $source_unmute"

  local s
  for s in ${sources[@]}
  do
    pulseaudio::set-mute-source "$s" true
  done

  pulseaudio::set-mute-source "$source_unmute" false
}

pulseaudio::mute-all-sources-but-default() {
  local default=$(pulseaudio::get-default-source)
  pulseaudio::mute-all-sources-but "$default"
}

# Check if a given source is muted, if no source name is provided, it defaults
# to checking the default source
pulseaudio::default-source-is-muted() {
  local src=${1:-$(pulseaudio::get-default-source)}
  pulseaudio::get-prop sources "$src" Mute | grep -iq "yes"
}

pulseaudio::toggle-mute-sources() {
  if pulseaudio::default-source-is-muted
  then
    pulseaudio::mute-all-sources-but-default && echo "unmuted"

    # Check if noisetorch is running, if yes: unmute noisetorch's device
    local noisetorch_device=$(noisetorch::get-device-name)
    if [[ -n "$noisetorch_device" ]]
    then
      pulseaudio::set-mute-source "$noisetorch_device" false
    fi
  else
    pulseaudio::mute-all-sources && echo "muted"
  fi
}

pulseaudio::setup-source() {
  local default
  local mute_others

  zparseopts -D -K -E \
    d=default -default=default \
    M=mute_others -mute-others=mute_others

  local name="$1"
  shift
  local volume
  local profile

  while [[ "$#" -gt 0 ]]
  do
    if [[ "$1" =~ [0-9]+ ]]
    then
      volume="$1"
    else
      profile="$1"
    fi
    shift

    if [[ -n "$volume" ]] && [[ -n "$profile" ]]
    then
      break
    fi
  done

  if [[ -n "$profile" ]]
  then
    pulseaudio::set-card-profile "$name" "$profile"
  fi

  if [[ -n "$default" ]]
  then
    pulseaudio::set-default-source "$name"
  fi

  if [[ -n "$volume" ]]
  then
    pulseaudio::set-source-volume "$name" "$volume"
  fi

  if [[ -n "$mute_others" ]]
  then
    pulseaudio::mute-all-sources-but "$name"
  fi

  # TODO Switch existing stream? like zoom for eg
}

noisetorch::get-device-name() {
  pulseaudio::list source | grep nui
}

noisetorch::list() {
  local device_type="$1"
  local noisetorch_device_prefix="nui_mic"

  pulseaudio::list --monitor "$device_type" | \
    awk "/^${noisetorch_device_prefix}/"
}

noisetorch::unmute() {
  local device_type
  for device_type in sinks sources
  do
    for device_name in $(noisetorch::list "$device_type")
    do
      pulseaudio::unmute "$device_type" "$device_name"
    done
  done
}

pulseaudio::get-module-index() {
  local name="$1"

  pactl list short modules | awk "/=${name} / {print \$1}"
}

pulseaudio::unload-module() {
  # Check if called with module index, if yes proceed by directly removing it
  if [[ "$1" =~ ^\d+$ ]]
  then
    pactl unload-module "$1"
    return "$?"
  fi

  local object_name="$1"
  local module_index="$(pulseaudio::get-module-index "$object_name")"
  if [[ -z "$module_index" ]]
  then
    echo_err "Failed to determine module index of $object_name"
    return 1
  fi

  echo_info "Module index of $object_name is $index"
  pactl unload-module "$index"
}

pulseaudio::fake-mic() {
  # Unload fake mic stuff first
  pulseaudio::fake-mic-unload

  # https://github.com/toadjaune/pulseaudio-config/blob/master/pulse_setup.sh
  local fake_source="fake-source"
  local fake_source_desc=$(sed 's/ /\\\ /g' <<< "Fake mic by ${0}")

  # Create the null sinks
  local fake_feedback_sink="fake-sink-without-feedback"
  local fake_feedback_sink_desc=$(sed 's/ /\\\ /g' <<< "Fake sink (with feedback) by ${0}")
  local fake_nofeedback_sink="fake-sink-with-feedback"
  local fake_nofeedback_sink_desc=$(sed 's/ /\\\ /g' <<< "Fake sink (w/o feedback) by ${0}")

  local real_source="$(pulseaudio::get-default-source)"
  local real_sink="$(pulseaudio::get-default-sink)"

  echo_info "Real source: $real_source"
  echo_info "Real sink: $real_sink"

  # fake_feedback_sink gets your audio sources (mplayer ...) that you want to hear and share
  # fake_nofeedback_sink gets all the audio you want to share (fake_nofeedback_sink + real mic)
  pactl load-module module-null-sink \
    sink_name=${fake_feedback_sink} \
    sink_properties=device.description=\"${fake_feedback_sink_desc}\"
  pactl load-module module-null-sink \
    sink_name=${fake_nofeedback_sink} \
    sink_properties=device.description=\"${fake_nofeedback_sink_desc}\"

  # Now create the loopback devices, all arguments are optional and can be configured with pavucontrol
  pactl load-module module-loopback \
    source="${fake_feedback_sink}.monitor" \
    sink="${real_sink}" \
    latency_msec=1

  pactl load-module module-loopback \
    source="${fake_feedback_sink}.monitor" \
    sink="${fake_nofeedback_sink}" \
    latency_msec=1

  # NoiseTorch does not follow the same pattern as most other mics
  # monitor_name=$name.monitor
  if [[ "$real_source" == "nui_mic_remap" ]]
  then
    pactl load-module module-loopback \
      source="nui_mic_denoised_out.monitor" \
      sink="${fake_nofeedback_sink}" \
      latency_msec=1
  else
    pactl load-module module-loopback \
      source="${real_source}.monitor" \
      sink="${fake_nofeedback_sink}" \
      latency_msec=1
  fi

  # This should not be necessary, however some programs (zoom) won't be able to see monitors
  # We could manually re-assign them with pavucontrol or similar, but creating a virtual source is more convenient
  pactl load-module module-virtual-source \
    source_name=${fake_source} \
    master=${fake_nofeedback_sink}.monitor \
    source_properties=device.description=\"${fake_source_desc}\"
}

pulseaudio::fake-mic-unload() {
  # FIXME
  # local name
  # for name in fake-source fake-sink-no-feedback fake-sink fake-sink-no-feedback.monitor fake-sink.monitor
  # do
  #   pulseaudio::unload-module "$name"
  # done

  local index
  for index in $(pactl list short modules | awk '/fake-sink/ { print $1 }' | tac)
  do
    pactl unload-module "$index"
  done
}

# Elgate Wave:3
elgato() {
  local name="Elgato Wave:3"
  local profile="input:mono-fallback"
  local volume="${1:-80}"

  pulseaudio::setup-source --default --mute-others \
    "$name" "$profile" "$volume"

  # TODO Noisetorch setup
  systemctl --user start "noisetorch@$(pulseaudio::list sources "$name")"

  # Set Noisetorch to be the default source
  # NOTE: This is already done by noisetorch@.service
  local noisetorch_device
  local tries=0

  while [[ "$tries" -lt 5 ]]
  do
    noisetorch_device=$(pulseaudio::device-name sources "NoiseTorch Microphone")
    if [[ -n "$noisetorch_device" ]]
    then
      pulseaudio::set-default-source "$noisetorch_device"
      return
    fi

    tries=$(( tries + 1))
    sleep 1
  done

  return 1
}