Simple D-Bus wrapper for Mumble

libinput.rb

# frozen_string_literal: true

require 'open3'

class Libinput
  Event = Struct.new('Event', :device, :type, :time, :data)
  class DeviceEvent < Event
    def name
      data.first
    end

    def cap?(cap)
      data.last.include? cap
    end
  end
  class ButtonEvent < Event
    def button
      data.first.to_sym
    end
    
    def button_id
      data[1].delete('()').to_i
    end

    def pressed?
      data[2] == 'pressed'
    end
  end
  class KeyboardEvent < Event
    def button
      data.first.to_sym
    end
    
    def button_id
      data[1].delete('()').to_i
    end

    def pressed?
      data[2] == 'pressed'
    end
  end

  attr_accessor :keyboard, :only_event, :only_device

  def initialize(keyboard: false, only_event: nil, only_device: nil)
    @keyboard = keyboard

    @only_event = only_event
    @only_device = only_device
  end

  def wait_event
    line = readline

    data = line.split
    dev = data.first.delete_prefix('-').to_sym
    data.shift
    type = data.first.to_sym
    data.shift
    data.shift

    case type.to_s
    when /DEVICE_/
      dev_name = data.take_while { |d| !d.start_with?('seat') }.join(' ')
      cap = data.find { |d| d.start_with? 'cap:' }[4..].split('').map(&:to_sym)
      DeviceEvent.new(dev, type, Time.now, [dev_name, cap])
    when /_BUTTON/
      ButtonEvent.new(dev, type, Time.now, data.join(' ').split(',').first.split)
    when 'KEYBOARD_KEY'
      KeyboardEvent.new(dev, type, Time.now, data)
    else
      Event.new(dev, type, Time.now, data)
    end
  end

  def start
    return if @watcher

    puts "Polling libinput with pid #{watcher}"
  end

  def stop
    return unless @watcher

    pid = watcher
    Process.kill 'INT', pid
    Process.wait pid
  end

  def io
    @io ||= IO.pipe
  end

  def io_reader
    io.first
  end

  def io_writer
    io.last
  end

  def readline
    io_reader.readline(chomp: true)
  end

  def watcher
    @watcher ||= begin
      line = [[debug_events]]
      line << ["grep #{only_event.to_s.upcase} --line-buffered"] if only_event

      # puts "Launching pipeline; #{line}"
      p = Open3.pipeline_start(
        *line,
        out: io_writer, in: '/dev/null'
      )
      p.first.pid
    end
  end

  private 

  def debug_events_filter
    filters = []
    filters << "--device=#{only_device}" if only_device
    filters << "--show-keycodes" if keyboard

    filters.join ' '
  end

  def debug_events
    "stdbuf -oL -- libinput debug-events #{debug_events_filter}"
  end
end

mumble.rb

#!/bin/env ruby
# frozen_string_literal: true

require 'bundler/setup'

require 'dbus'
require_relative 'libinput'

class MumbleProxy
  attr_accessor :button
  attr_reader :bus, :input, :iface

  # Button 276 - Mouse 5
  def initialize(button: 276)
    @button = button
    @bus = DBus.session_bus

    @input = Libinput.new(only_event: :pointer_button)
    @iface = bus['net.sourceforge.mumble.mumble']['/']['net.sourceforge.mumble.Mumble']
  end

  def run!
    input.start

    loop do
      data = input.wait_event
      break if data.nil?

      case data
      when Libinput::ButtonEvent
        if data.button_id == button
          if data.pressed?
            iface.startTalking
          else
            iface.stopTalking
          end
        end
      else
        puts "Received #{data.inspect}"
      end
    end
  end
end

app = MumbleProxy.new
app.run!