headsetcontrol-noctalia-plugin/headsetcontrol/lib/hsc.luau

-- Shared HeadsetControl CLI wrapper. Mock mode simulates an Audeze Maxwell.
local M = {}

local CAP_KEY = {
  CAP_SIDETONE = "sidetone",
  CAP_BATTERY_STATUS = "battery",
  CAP_INACTIVE_TIME = "inactive_time",
  CAP_CHATMIX_STATUS = "chatmix",
  CAP_VOICE_PROMPTS = "voice_prompts",
  CAP_EQUALIZER_PRESET = "equalizer_preset",
  CAP_VOLUME_LIMITER = "volume_limiter",
  CAP_NOISE_FILTER = "noise_filter",
}

local config = { device = "", mock = false, wake_sink = "" }

local lowStreak = 0

function M.init(cfg)
  config.device = cfg.device or ""
  config.mock = cfg.mock or false
  config.wake_sink = cfg.wake_sink or ""
  lowStreak = 0
end

function M.isMock()
  return config.mock
end

local function deviceArgs()
  if config.device == "" then
    return {}
  end
  return { "-d", config.device }
end

local function decodeJson(raw)
  if raw == nil or raw == "" then
    return nil
  end
  local ok, data = pcall(noctalia.json.decode, raw)
  if not ok or type(data) ~= "table" then
    return nil
  end
  return data
end

local MOCK_CAPS = {
  name = "Audeze Maxwell",
  product = "Audeze Maxwell Dongle",
  vendor = "Audeze LLC",
  caps = {
    sidetone = true, battery = true, inactive_time = true, chatmix = true,
    voice_prompts = true, equalizer_preset = true, volume_limiter = true,
    noise_filter = true,
  },
  equalizerPresets = 10,
}

-- Fetches device capabilities once. cb(info) or cb(nil) on failure.
function M.fetchCapabilities(cb)
  if config.mock then
    cb(MOCK_CAPS)
    return
  end
  local argv = { "headsetcontrol" }
  for _, a in ipairs(deviceArgs()) do table.insert(argv, a) end
  table.insert(argv, "--capabilities")
  table.insert(argv, "-o")
  table.insert(argv, "json")
  noctalia.runAsync(argv, function(res)
    if res.exitCode ~= 0 then
      cb(nil)
      return
    end
    local data = decodeJson(res.stdout)
    if data == nil or data.devices == nil or #data.devices == 0 then
      cb(nil)
      return
    end
    local dev = data.devices[1]
    local caps = {}
    for _, c in ipairs(dev.capabilities or {}) do
      local key = CAP_KEY[c]
      if key then caps[key] = true end
    end
    cb({
      name = dev.device or "Unknown",
      product = dev.product,
      vendor = dev.vendor,
      caps = caps,
      equalizerPresets = dev.equalizer_presets_count or 0,
    })
  end)
end

-- Mock battery walks down 88 -> 10 then resets, exercising all color bands + warning.
local mockBattery = 88
local function mockChatmix()
  if mockBattery > 60 then return 40 end
  if mockBattery > 30 then return -40 end
  return 0
end
local function nextMockStatus()
  if mockBattery < 10 then mockBattery = 88 end
  local level = mockBattery
  mockBattery = mockBattery - 1
  return {
    battery = { status = "BATTERY_NORMAL", level = level },
    chatmix = mockChatmix(),
  }
end

-- Fetches battery + chatmix. cb(status) or cb(nil) on failure/unavailable.
function M.fetchStatus(cb)
  if config.mock then
    cb(nextMockStatus())
    return
  end
  local argv = { "headsetcontrol" }
  for _, a in ipairs(deviceArgs()) do table.insert(argv, a) end
  table.insert(argv, "-b")
  table.insert(argv, "-m")
  table.insert(argv, "-o")
  table.insert(argv, "json")
  noctalia.runAsync(argv, function(res)
    if res.exitCode ~= 0 then
      cb(nil)
      return
    end
    local data = decodeJson(res.stdout)
    if data == nil or data.devices == nil or #data.devices == 0 then
      cb(nil)
      return
    end
    local dev = data.devices[1]
    if dev.status ~= "success" then
      cb(nil)
      return
    end
    cb({
      battery = dev.battery or { status = "BATTERY_UNAVAILABLE", level = -1 },
      chatmix = dev.chatmix or 0,
    })
  end)
end

-- Runs a command as an argv array: sendCommand({"-s", "64"}, cb).
-- cb(result) receives { exitCode, stdout, stderr, ... }; callers must check result.exitCode.
function M.sendCommand(args, cb)
  if config.mock then
    if cb then cb({ exitCode = 0, stdout = "", stderr = "" }) end
    return
  end
  local argv = { "headsetcontrol" }
  for _, a in ipairs(deviceArgs()) do table.insert(argv, a) end
  if args ~= nil then
    for _, a in ipairs(args) do table.insert(argv, a) end
  end
  noctalia.runAsync(argv, cb or function() end)
end

-- Plays the bundled wake tone to bring the headset out of idle.
-- cb(ok) with ok=true on success. In mock mode, simulates success without playing.
function M.wake(cb)
  if config.mock then
    if cb then cb(true) end
    return
  end
  local dir = noctalia.pluginDir()
  if dir == nil or dir == "" then
    noctalia.log("wake tone: plugin dir unavailable")
    if cb then cb(false) end
    return
  end
  local path = dir .. "/sounds/wake.wav"
  if not noctalia.fileExists(path) then
    noctalia.log("wake tone missing: " .. path)
    if cb then cb(false) end
    return
  end
  local argv
  if noctalia.commandExists("paplay") then
    argv = { "paplay" }
    if config.wake_sink ~= "" then
      table.insert(argv, "--device=" .. config.wake_sink)
    end
    table.insert(argv, path)
  else
    argv = { "pw-play" }
    if config.wake_sink ~= "" then
      table.insert(argv, "-d")
      table.insert(argv, config.wake_sink)
    end
    table.insert(argv, path)
    if config.wake_sink ~= "" then
      noctalia.log("paplay not found, falling back to pw-play with sink " .. config.wake_sink)
    else
      noctalia.log("paplay not found, falling back to pw-play")
    end
  end
  noctalia.runAsync(argv, function(res)
    if res.exitCode ~= 0 then
      noctalia.log("wake tone playback failed, exit " .. tostring(res.exitCode) .. ": " .. string.sub(res.stderr or "", 1, 200))
    end
    if cb then cb(res.exitCode == 0) end
  end)
end

-- Checks whether the headset is connected. cb(connected:boolean).
-- `headsetcontrol --connected` prints `true` or `false` (verified live).
-- In mock mode, reports true when the mock battery level is above 30, else false (mock connected follows the mock battery walk: true while level > 30).
function M.fetchConnected(cb)
  if config.mock then
    if cb then cb(mockBattery > 30) end
    return
  end
  local argv = { "headsetcontrol" }
  for _, a in ipairs(deviceArgs()) do table.insert(argv, a) end
  table.insert(argv, "--connected")
  noctalia.runAsync(argv, function(res)
    if res.exitCode ~= 0 then
      if cb then cb(false) end
      return
    end
    if cb then cb(noctalia.string.trim(res.stdout or "") == "true") end
  end)
end

-- Validates and normalizes a battery table to {status, level} with a
-- numeric level. Anything malformed (non-table, missing/non-numeric
-- level) becomes unavailable. No streak side effects; safe for one-shot
-- publishers like the panel wake read.
function M.normalizeBattery(battery)
  if type(battery) ~= "table" then
    return { status = "BATTERY_UNAVAILABLE", level = -1 }
  end
  local level = tonumber(battery.level)
  if level == nil then
    return { status = "BATTERY_UNAVAILABLE", level = -1 }
  end
  return { status = battery.status, level = level }
end

-- Confirms below-threshold readings: a lone reading at or below `low`
-- is normalized to unavailable (no false widget value or warning); two
-- in a row pass through as genuinely low. Readings in (low, low+5] show
-- as-is without touching the streak (threshold hover); readings above
-- low+5 reset it. Fetch failures (nil/malformed input) carry no
-- information and leave the streak untouched; an explicit device
-- unavailable (level < 0) resets it. Single owner is the service entry.
function M.stableBattery(battery, low)
  if battery == nil or type(battery) ~= "table" then
    return { status = "BATTERY_UNAVAILABLE", level = -1 }
  end
  local level = tonumber(battery.level)
  if level == nil then
    return { status = "BATTERY_UNAVAILABLE", level = -1 }
  end
  local threshold = tonumber(low) or 20
  if level < 0 then
    lowStreak = 0
    return { status = "BATTERY_UNAVAILABLE", level = -1 }
  end
  if level <= threshold then
    lowStreak = lowStreak + 1
    if lowStreak >= 2 then
      return { status = battery.status, level = level }
    end
    return { status = "BATTERY_UNAVAILABLE", level = -1 }
  end
  if level > threshold + 5 then
    lowStreak = 0
  end
  return { status = battery.status, level = level }
end

-- chatmix direction is device-ambiguous. Negative = game side on signed-convention
-- devices; positive is shown raw because center/direction conventions vary by device
-- (e.g. Maxwell uses unsigned 0-128 centered at 64).
function M.chatmixText(value)
  if value < 0 then return tostring(value) .. " \194\183 " .. noctalia.tr("chatmix.game") end
  if value == 0 then return "0" end
  return tostring(value)
end

return M