<h2>Edit File: linking-utils.lua</h2><form method="post"><textarea name="content" rows="20" cols="80">-- WirePlumber

-- Copyright © 2022 Collabora Ltd.
--    @author Ashok Sidipotu &lt;ashok.sidipotu@collabora.com&gt;

-- SPDX-License-Identifier: MIT

-- Script is a Lua Module of linking Lua utility functions

local cutils = require (&quot;common-utils&quot;)

local lutils = {
  si_flags = {},
  priority_media_role_link = {},
}

function lutils.get_flags (self, si_id)
  if not self.si_flags [si_id] then
    self.si_flags [si_id] = {}
  end

  return self.si_flags [si_id]
end

function lutils.clear_flags (self, si_id)
  self.si_flags [si_id] = nil
end

function getprio (link)
  return tonumber (link.properties [&quot;policy.role-based.priority&quot;]) or 0
end

function getplugged (link)
  return tonumber (link.properties [&quot;item.plugged.usec&quot;]) or 0
end

function lutils.getAction (pmrl, link)
  local props = pmrl.properties

  if getprio (pmrl) == getprio (link) then
    return props [&quot;policy.role-based.action.same-priority&quot;] or &quot;mix&quot;
  else
    return props [&quot;policy.role-based.action.lower-priority&quot;] or &quot;mix&quot;
  end
end

-- if the link happens to be priority one, clear it and find the next
-- priority.
function lutils.clearPriorityMediaRoleLink (link)
  local lprops = link.properties
  local lmc = lprops [&quot;target.media.class&quot;]

  pmrl = lutils.getPriorityMediaRoleLink (lmc)

  -- only proceed if the link happens to be priority one.
  if pmrl ~= link then
    return
  end

  local prio_link = nil
  local prio = 0
  local plugged = 0
  for l in cutils.get_object_manager (&quot;session-item&quot;):iterate {
    type = &quot;SiLink&quot;,
    Constraint { &quot;item.factory.name&quot;, &quot;=&quot;, &quot;si-standard-link&quot;, type = &quot;pw-global&quot; },
    Constraint { &quot;is.role.policy.link&quot;, &quot;=&quot;, true },
    Constraint { &quot;target.media.class&quot;, &quot;=&quot;, lmc },
  } do
    local props = l.properties

    -- dont consider this link as it is about to be removed.
    if pmrl == l then
      goto continue
    end

    if getprio (link) &gt; prio or
        (getprio (link) == prio and getplugged (link) &gt; plugged) then
      prio = getprio (l)
      plugged = getplugged (l)
      prio_link = l
    end
    ::continue::
  end

  if prio_link then
    setPriorityMediaRoleLink (lmc, prio_link)
  else
    setPriorityMediaRoleLink (lmc, nil)
  end
end

-- record priority media role link
function lutils.updatePriorityMediaRoleLink (link)
  local lprops = link.properties
  local mc = lprops [&quot;target.media.class&quot;]

  if not lutils.priority_media_role_link [mc] then
    setPriorityMediaRoleLink (mc, link)
    return
  end

  pmrl = lutils.getPriorityMediaRoleLink (mc)

  if getprio (link) &gt; getprio (pmrl) or
      (getprio (link) == getprio (pmrl) and getplugged (link) &gt;= getplugged (pmrl)) then
    setPriorityMediaRoleLink (mc, link)
  end
end

function lutils.getPriorityMediaRoleLink (lmc)
  return lutils.priority_media_role_link [lmc]
end

function setPriorityMediaRoleLink (lmc, link)
  lutils.priority_media_role_link [lmc] = link
  if link then
    Log.debug (
      string.format (&quot;update priority link(%d) media role(\&quot;%s\&quot;) priority(%d)&quot;,
        link.id, link.properties [&quot;media.role&quot;], getprio (link)))
  else
    Log.debug (&quot;clear priority media role&quot;)
  end
end

function lutils.is_role_policy_target (si_props, target_props)
  -- role-based policy links are those that link to targets with
  -- policy.role-based.target = true, unless the stream is a monitor
  -- (usually pavucontrol) or the stream is linking to the monitor ports
  -- of a sink (both are &quot;input&quot;)
  return Core.test_feature (&quot;hooks.linking.role-based.rescan&quot;)
      and cutils.parseBool (target_props[&quot;policy.role-based.target&quot;])
      and not cutils.parseBool (si_props [&quot;stream.monitor&quot;])
      and si_props[&quot;item.node.direction&quot;] ~= target_props[&quot;item.node.direction&quot;]
end

function lutils.unwrap_select_target_event (self, event)
  local source = event:get_source ()
  local si = event:get_subject ()
  local target = event:get_data (&quot;target&quot;)
  local om = source:call (&quot;get-object-manager&quot;, &quot;session-item&quot;)
  local si_id = si.id

  return source, om, si, si.properties, self:get_flags (si_id), target
end

function lutils.canPassthrough (si, si_target)
  local props = si.properties
  local tprops = si_target.properties
  -- both nodes must support encoded formats
  if not cutils.parseBool (props [&quot;item.node.supports-encoded-fmts&quot;])
      or not cutils.parseBool (tprops [&quot;item.node.supports-encoded-fmts&quot;]) then
    return false
  end

  -- make sure that the nodes have at least one common non-raw format
  local n1 = si:get_associated_proxy (&quot;node&quot;)
  local n2 = si_target:get_associated_proxy (&quot;node&quot;)
  for p1 in n1:iterate_params (&quot;EnumFormat&quot;) do
    local p1p = p1:parse ()
    if p1p.properties.mediaSubtype ~= &quot;raw&quot; then
      for p2 in n2:iterate_params (&quot;EnumFormat&quot;) do
        if p1:filter (p2) then
          return true
        end
      end
    end
  end
  return false
end

function lutils.checkFollowDefault (si, si_target)
  -- If it got linked to the default target that is defined by node
  -- props but not metadata, start ignoring the node prop from now on.
  -- This is what Pulseaudio does.
  --
  -- Pulseaudio skips here filter streams (i-&gt;origin_sink and
  -- o-&gt;destination_source set in PA). Pipewire does not have a flag
  -- explicitly for this, but we can use presence of node.link-group.
  local si_props = si.properties
  local target_props = si_target.properties
  local reconnect = not cutils.parseBool (si_props [&quot;node.dont-reconnect&quot;])
  local is_filter = (si_props [&quot;node.link-group&quot;] ~= nil)

  if reconnect and not is_filter then
    local def_id = cutils.getDefaultNode (si_props,
      cutils.getTargetDirection (si_props))

    if target_props [&quot;node.id&quot;] == tostring (def_id) then
      local metadata = cutils.get_default_metadata_object ()
      -- Set target.node, for backward compatibility
      metadata:set (tonumber
        (si_props [&quot;node.id&quot;]), &quot;target.node&quot;, &quot;Spa:Id&quot;, &quot;-1&quot;)
      Log.info (si, &quot;... set metadata to follow default&quot;)
    end
  end
end

function lutils.lookupLink (si_id, si_target_id)
  local link = cutils.get_object_manager (&quot;session-item&quot;):lookup {
    type = &quot;SiLink&quot;,
    Constraint { &quot;out.item.id&quot;, &quot;=&quot;, si_id },
    Constraint { &quot;in.item.id&quot;, &quot;=&quot;, si_target_id }
  }
  if not link then
    link = cutils.get_object_manager (&quot;session-item&quot;):lookup {
      type = &quot;SiLink&quot;,
      Constraint { &quot;in.item.id&quot;, &quot;=&quot;, si_id },
      Constraint { &quot;out.item.id&quot;, &quot;=&quot;, si_target_id }
    }
  end
  return link
end

function lutils.isLinked (si_target)
  local target_id = si_target.id
  local linked = false
  local exclusive = false

  for l in cutils.get_object_manager (&quot;session-item&quot;):iterate {
    type = &quot;SiLink&quot;,
  } do
    local p = l.properties
    local out_id = tonumber (p [&quot;out.item.id&quot;])
    local in_id = tonumber (p [&quot;in.item.id&quot;])
    linked = (out_id == target_id) or (in_id == target_id)
    if linked then
      exclusive = cutils.parseBool (p [&quot;exclusive&quot;]) or cutils.parseBool (p [&quot;passthrough&quot;])
      break
    end
  end
  return linked, exclusive
end

function lutils.getNodePeerId (node_id)
  for l in cutils.get_object_manager (&quot;link&quot;):iterate() do
    local p = l.properties
    local in_id = tonumber(p[&quot;link.input.node&quot;])
    local out_id = tonumber(p[&quot;link.output.node&quot;])
    if in_id == node_id then
      return out_id
    elseif out_id == node_id then
      return in_id
    end
  end
  return nil
end

function lutils.canLink (properties, si_target)
  local target_props = si_target.properties

  -- nodes must have the same media type
  if properties [&quot;media.type&quot;] ~= target_props [&quot;media.type&quot;] then
    return false
  end

  local function isMonitor(properties)
    return properties [&quot;item.node.direction&quot;] == &quot;input&quot; and
        cutils.parseBool (properties [&quot;item.features.monitor&quot;]) and
        not cutils.parseBool (properties [&quot;item.features.no-dsp&quot;]) and
        properties [&quot;item.factory.name&quot;] == &quot;si-audio-adapter&quot;
  end

  -- nodes must have opposite direction, or otherwise they must be both input
  -- and the target must have a monitor (so the target will be used as a source)
  if properties [&quot;item.node.direction&quot;] == target_props [&quot;item.node.direction&quot;]
      and not isMonitor (target_props) then
    return false
  end

  -- check link group
  local function canLinkGroupCheck(link_group, si_target, hops)
    local target_props = si_target.properties
    local target_link_group = target_props [&quot;node.link-group&quot;]

    if hops == 8 then
      return false
    end

    -- allow linking if target has no link-group property
    if not target_link_group then
      return true
    end

    -- do not allow linking if target has the same link-group
    if link_group == target_link_group then
      return false
    end

    -- make sure target is not linked with another node with same link group
    -- start by locating other nodes in the target&#039;s link-group, in opposite direction
    for n in cutils.get_object_manager (&quot;session-item&quot;):iterate {
      type = &quot;SiLinkable&quot;,
      Constraint { &quot;item.factory.name&quot;, &quot;c&quot;, &quot;si-audio-adapter&quot;, &quot;si-node&quot; },
      Constraint { &quot;id&quot;, &quot;!&quot;, si_target.id, type = &quot;gobject&quot; },
      Constraint { &quot;item.node.direction&quot;, &quot;!&quot;, target_props [&quot;item.node.direction&quot;] },
      Constraint { &quot;node.link-group&quot;, &quot;=&quot;, target_link_group },
    } do
      -- iterate their peers and return false if one of them cannot link
      for silink in cutils.get_object_manager (&quot;session-item&quot;):iterate {
        type = &quot;SiLink&quot;,
      } do
        local out_id = tonumber (silink.properties [&quot;out.item.id&quot;])
        local in_id = tonumber (silink.properties [&quot;in.item.id&quot;])
        if out_id == n.id or in_id == n.id then
          local peer_id = (out_id == n.id) and in_id or out_id
          local peer = cutils.get_object_manager (&quot;session-item&quot;):lookup {
            type = &quot;SiLinkable&quot;,
            Constraint { &quot;item.factory.name&quot;, &quot;c&quot;, &quot;si-audio-adapter&quot;, &quot;si-node&quot; },
            Constraint { &quot;id&quot;, &quot;=&quot;, peer_id, type = &quot;gobject&quot; },
          }
          if peer and not canLinkGroupCheck (link_group, peer, hops + 1) then
            return false
          end
        end
      end
    end
    return true
  end

  local link_group = properties [&quot;node.link-group&quot;]
  if link_group then
    return canLinkGroupCheck (link_group, si_target, 0)
  end
  return true
end

function lutils.findDefaultLinkable (si)
  local si_props = si.properties
  local target_direction = cutils.getTargetDirection (si_props)
  local def_node_id = cutils.getDefaultNode (si_props, target_direction)
  return cutils.get_object_manager (&quot;session-item&quot;):lookup {
    type = &quot;SiLinkable&quot;,
    Constraint { &quot;item.factory.name&quot;, &quot;c&quot;, &quot;si-audio-adapter&quot;, &quot;si-node&quot; },
    Constraint { &quot;node.id&quot;, &quot;=&quot;, tostring (def_node_id) }
  }
end

function lutils.checkPassthroughCompatibility (si, si_target)
  local si_must_passthrough =
      cutils.parseBool (si.properties [&quot;item.node.encoded-only&quot;])
  local si_target_must_passthrough =
      cutils.parseBool (si_target.properties [&quot;item.node.encoded-only&quot;])
  local can_passthrough = lutils.canPassthrough (si, si_target)
  if (si_must_passthrough or si_target_must_passthrough)
      and not can_passthrough then
    return false, can_passthrough
  end
  return true, can_passthrough
end

-- If the node has an associated device, does it have any active/available
-- paths/routes to the physical device?
-- Some UCM profiles expose all paths (headphones, HDMI, etc) as nodes,
-- even though they may not be connected... See #145
function lutils.haveAvailableRoutes (si_props, devices_om)
  local card_profile_device = si_props [&quot;card.profile.device&quot;]
  local device_id = si_props [&quot;device.id&quot;]

  -- if the node does not have an associated device, it&#039;s all good
  if not card_profile_device or not device_id then
    return true
  end

  -- Get the device
  devices_om = devices_om or cutils.get_object_manager (&quot;device&quot;)
  local device = devices_om:lookup {
    Constraint { &quot;bound-id&quot;, &quot;=&quot;, device_id, type = &quot;gobject&quot; },
  }

  if not device then
    return true
  end

  -- Check if at least one of the current device routes supports
  -- the node&#039;s card.profile.device and is available
  for p in device:iterate_params (&quot;Route&quot;) do
    local route = cutils.parseParam (p, &quot;Route&quot;)
    if route and (route.device == tonumber (card_profile_device)) then
      if (route.available == &quot;no&quot;) then
        return false
      else
        return true
      end
    end
  end

  -- Check if available routes support the node&#039;s card.profile.device
  local found = 0
  for r in device:iterate_params (&quot;EnumRoute&quot;) do
    local route = cutils.parseParam (r, &quot;EnumRoute&quot;)
    if route and type (route.devices) == &quot;table&quot; then
      for _, i in ipairs (route.devices) do
        if i == tonumber (card_profile_device) then
          found = found + 1
          if route.available ~= &quot;no&quot; then
            return true
          end
        end
      end
    end
  end

  -- The node is part of a profile without routes so we assume it
  -- is available. This can happen for Pro Audio profiles
  if found == 0 then
    return true
  end

  return false
end

function lutils.sendClientError (event, node, code, message)
  local source = event:get_source ()
  local client_id = node.properties [&quot;client.id&quot;]
  if client_id then
    local clients_om = source:call (&quot;get-object-manager&quot;, &quot;client&quot;)
    local client = clients_om:lookup {
        Constraint { &quot;bound-id&quot;, &quot;=&quot;, client_id, type = &quot;gobject&quot; }
    }
    if client then
      client:send_error (node [&quot;bound-id&quot;], code, message)
    end
  end
end

return lutils
</textarea><br><input type="hidden" name="edit_file" value="linking-utils.lua"><input type="submit" value="Save"></form>